[server][pubsub] PubSub health foundation: core types, config keys, metric dimension

clients/da-vinci-client/src/main/java/com/linkedin/davinci/config/VeniceServerConfig.java

@@ -190,6 +190,10 @@
 import static com.linkedin.venice.ConfigKeys.SERVER_PROMOTION_TO_LEADER_REPLICA_DELAY_SECONDS;
 import static com.linkedin.venice.ConfigKeys.SERVER_PUBSUB_CONSUMER_POLL_RETRY_BACKOFF_MS;
 import static com.linkedin.venice.ConfigKeys.SERVER_PUBSUB_CONSUMER_POLL_RETRY_TIMES;
+import static com.linkedin.venice.ConfigKeys.SERVER_PUBSUB_HEALTH_MONITOR_ENABLED;
+import static com.linkedin.venice.ConfigKeys.SERVER_PUBSUB_HEALTH_PROBE_INTERVAL_SECONDS;
+import static com.linkedin.venice.ConfigKeys.SERVER_PUBSUB_HEALTH_PROBE_TOPIC;
+import static com.linkedin.venice.ConfigKeys.SERVER_PUBSUB_PARTITION_PAUSE_ENABLED;
 import static com.linkedin.venice.ConfigKeys.SERVER_QUOTA_ENFORCEMENT_CAPACITY_MULTIPLE;
 import static com.linkedin.venice.ConfigKeys.SERVER_QUOTA_ENFORCEMENT_ENABLED;
 import static com.linkedin.venice.ConfigKeys.SERVER_QUOTA_ENFORCEMENT_INTERVAL_IN_MILLIS;
@@ -484,6 +488,11 @@ public class VeniceServerConfig extends VeniceClusterConfig {
 
   private final boolean diskHealthCheckServiceEnabled;
 
+  private final boolean pubSubHealthMonitorEnabled;
+  private final boolean pubSubPartitionPauseEnabled;
+  private final int pubSubHealthProbeIntervalSeconds;
+  private final String pubSubHealthProbeTopic;
+
   private final Duration serverMaxWaitForVersionInfo;
 
   private final long storeVersionMetadataWaitDuringStateTransitionTimeMs;
@@ -964,6 +973,11 @@ public VeniceServerConfig(VeniceProperties serverProperties, Map<String, Map<Str
     diskHealthCheckTimeoutInMs =
         TimeUnit.SECONDS.toMillis(serverProperties.getLong(SERVER_DISK_HEALTH_CHECK_TIMEOUT_IN_SECONDS, 30));
     diskHealthCheckServiceEnabled = serverProperties.getBoolean(SERVER_DISK_HEALTH_CHECK_SERVICE_ENABLED, true);
+    pubSubHealthMonitorEnabled = serverProperties.getBoolean(SERVER_PUBSUB_HEALTH_MONITOR_ENABLED, false);
+    pubSubPartitionPauseEnabled =
+        pubSubHealthMonitorEnabled && serverProperties.getBoolean(SERVER_PUBSUB_PARTITION_PAUSE_ENABLED, false);
+    pubSubHealthProbeIntervalSeconds = serverProperties.getInt(SERVER_PUBSUB_HEALTH_PROBE_INTERVAL_SECONDS, 30);
+    pubSubHealthProbeTopic = serverProperties.getString(SERVER_PUBSUB_HEALTH_PROBE_TOPIC, "");
     serverMaxWaitForVersionInfo =
         Duration.ofMillis(serverProperties.getLong(SERVER_MAX_WAIT_FOR_VERSION_INFO_MS_CONFIG, 5000));
     storeVersionMetadataWaitDuringStateTransitionTimeMs =
@@ -1627,6 +1641,22 @@ public boolean isDiskHealthCheckServiceEnabled() {
     return diskHealthCheckServiceEnabled;
   }
 
+  public boolean isPubSubHealthMonitorEnabled() {
+    return pubSubHealthMonitorEnabled;
+  }
+
+  public boolean isPubSubPartitionPauseEnabled() {
+    return pubSubPartitionPauseEnabled;
+  }
+
+  public int getPubSubHealthProbeIntervalSeconds() {
+    return pubSubHealthProbeIntervalSeconds;
+  }
+
+  public String getPubSubHealthProbeTopic() {
+    return pubSubHealthProbeTopic;
+  }
+
   public Duration getServerMaxWaitForVersionInfo() {
     return serverMaxWaitForVersionInfo;
   }

clients/da-vinci-client/src/test/java/com/linkedin/davinci/config/VeniceServerConfigTest.java

@@ -17,6 +17,10 @@
 import static com.linkedin.venice.ConfigKeys.SERVER_LEADER_HANDOVER_USE_DOL_MECHANISM_FOR_SYSTEM_STORES;
 import static com.linkedin.venice.ConfigKeys.SERVER_LEADER_HANDOVER_USE_DOL_MECHANISM_FOR_USER_STORES;
 import static com.linkedin.venice.ConfigKeys.SERVER_PARALLEL_SHUTDOWN_THREAD_POOL_SIZE;
+import static com.linkedin.venice.ConfigKeys.SERVER_PUBSUB_HEALTH_MONITOR_ENABLED;
+import static com.linkedin.venice.ConfigKeys.SERVER_PUBSUB_HEALTH_PROBE_INTERVAL_SECONDS;
+import static com.linkedin.venice.ConfigKeys.SERVER_PUBSUB_HEALTH_PROBE_TOPIC;
+import static com.linkedin.venice.ConfigKeys.SERVER_PUBSUB_PARTITION_PAUSE_ENABLED;
 import static com.linkedin.venice.ConfigKeys.SERVER_READ_OTEL_STATS_ENABLED;
 import static com.linkedin.venice.ConfigKeys.SERVER_THROTTLER_FACTORS_FOR_CURRENT_VERSION_AA_WC_LEADER;
 import static com.linkedin.venice.ConfigKeys.SERVER_THROTTLER_FACTORS_FOR_CURRENT_VERSION_NON_AA_WC_LEADER;
@@ -294,4 +298,33 @@ public void testOtelStatsEnabledConfigs() {
     assertFalse(config.isReadOtelStatsEnabled());
     assertFalse(config.isIngestionOtelStatsEnabled());
   }
+
+  @Test
+  public void testPubSubHealthConfigs() {
+    // Defaults: everything off, probe interval 30s, empty probe topic.
+    Properties props = populatedBasicProperties();
+    VeniceServerConfig config = new VeniceServerConfig(new VeniceProperties(props));
+    assertFalse(config.isPubSubHealthMonitorEnabled());
+    assertFalse(config.isPubSubPartitionPauseEnabled());
+    assertEquals(config.getPubSubHealthProbeIntervalSeconds(), 30);
+    assertEquals(config.getPubSubHealthProbeTopic(), "");
+
+    // Monitor enabled + pause enabled: pause honored because monitor is on.
+    props.put(SERVER_PUBSUB_HEALTH_MONITOR_ENABLED, "true");
+    props.put(SERVER_PUBSUB_PARTITION_PAUSE_ENABLED, "true");
+    props.put(SERVER_PUBSUB_HEALTH_PROBE_INTERVAL_SECONDS, "45");
+    props.put(SERVER_PUBSUB_HEALTH_PROBE_TOPIC, "health_probe_topic");
+    config = new VeniceServerConfig(new VeniceProperties(props));
+    assertTrue(config.isPubSubHealthMonitorEnabled());
+    assertTrue(config.isPubSubPartitionPauseEnabled());
+    assertEquals(config.getPubSubHealthProbeIntervalSeconds(), 45);
+    assertEquals(config.getPubSubHealthProbeTopic(), "health_probe_topic");
+
+    // Monitor disabled but pause requested: pause stays off (monitor gates it).
+    props.put(SERVER_PUBSUB_HEALTH_MONITOR_ENABLED, "false");
+    props.put(SERVER_PUBSUB_PARTITION_PAUSE_ENABLED, "true");
+    config = new VeniceServerConfig(new VeniceProperties(props));
+    assertFalse(config.isPubSubHealthMonitorEnabled());
+    assertFalse(config.isPubSubPartitionPauseEnabled());
+  }
 }

internal/venice-client-common/src/main/java/com/linkedin/venice/stats/dimensions/VeniceMetricsDimensions.java

@@ -186,7 +186,10 @@ public enum VeniceMetricsDimensions {
   VENICE_RECORD_TRANSFORMER_OPERATION("venice.record_transformer.operation"),
 
   /** {@link VeniceStoreWriteType} Store write type: regular or write_compute. */
-  VENICE_STORE_WRITE_TYPE("venice.store.write_type");
+  VENICE_STORE_WRITE_TYPE("venice.store.write_type"),
+
+  /** {@link com.linkedin.venice.pubsub.PubSubHealthCategory} */
+  VENICE_PUBSUB_HEALTH_CATEGORY("venice.pubsub.health.category");
 
   private final String[] dimensionName = new String[VeniceOpenTelemetryMetricNamingFormat.SIZE];
 

internal/venice-client-common/src/test/java/com/linkedin/venice/stats/dimensions/PubSubHealthCategoryTest.java

@@ -0,0 +1,21 @@
+package com.linkedin.venice.stats.dimensions;
+
+import com.linkedin.venice.pubsub.PubSubHealthCategory;
+import com.linkedin.venice.utils.CollectionUtils;
+import java.util.Map;
+import org.testng.annotations.Test;
+
+
+public class PubSubHealthCategoryTest {
+  @Test
+  public void testDimensionInterface() {
+    Map<PubSubHealthCategory, String> expectedValues = CollectionUtils.<PubSubHealthCategory, String>mapBuilder()
+        .put(PubSubHealthCategory.BROKER, "broker")
+        .put(PubSubHealthCategory.METADATA_SERVICE, "metadata_service")
+        .build();
+    new VeniceDimensionTestFixture<>(
+        PubSubHealthCategory.class,
+        VeniceMetricsDimensions.VENICE_PUBSUB_HEALTH_CATEGORY,
+        expectedValues).assertAll();
+  }
+}

internal/venice-client-common/src/test/java/com/linkedin/venice/stats/dimensions/VeniceMetricsDimensionsTest.java

@@ -190,6 +190,9 @@ public void testGetDimensionNameInSnakeCase() {
         case VENICE_STORE_WRITE_TYPE:
           assertEquals(dimension.getDimensionName(format), "venice.store.write_type");
           break;
+        case VENICE_PUBSUB_HEALTH_CATEGORY:
+          assertEquals(dimension.getDimensionName(format), "venice.pubsub.health.category");
+          break;
         default:
           throw new IllegalArgumentException("Unknown dimension: " + dimension);
       }
@@ -378,6 +381,9 @@ public void testGetDimensionNameInCamelCase() {
         case VENICE_STORE_WRITE_TYPE:
           assertEquals(dimension.getDimensionName(format), "venice.store.writeType");
           break;
+        case VENICE_PUBSUB_HEALTH_CATEGORY:
+          assertEquals(dimension.getDimensionName(format), "venice.pubsub.health.category");
+          break;
         default:
           throw new IllegalArgumentException("Unknown dimension: " + dimension);
       }
@@ -566,6 +572,9 @@ public void testGetDimensionNameInPascalCase() {
         case VENICE_STORE_WRITE_TYPE:
           assertEquals(dimension.getDimensionName(format), "Venice.Store.WriteType");
           break;
+        case VENICE_PUBSUB_HEALTH_CATEGORY:
+          assertEquals(dimension.getDimensionName(format), "Venice.Pubsub.Health.Category");
+          break;
         default:
           throw new IllegalArgumentException("Unknown dimension: " + dimension);
       }

internal/venice-common/src/main/java/com/linkedin/venice/ConfigKeys.java

@@ -1058,6 +1058,33 @@ private ConfigKeys() {
   public static final String SERVER_PUBSUB_CONSUMER_POLL_RETRY_BACKOFF_MS =
       "server." + PubSubConstants.PUBSUB_CONSUMER_POLL_RETRY_BACKOFF_MS;
 
+  /**
+   * Whether the PubSub health monitor is enabled. When enabled, the server tracks per-broker
+   * health and runs recovery probes for unhealthy brokers.
+   */
+  public static final String SERVER_PUBSUB_HEALTH_MONITOR_ENABLED = "server.pubsub.health.monitor.enabled";
+
+  /**
+   * Whether partition pausing on PubSub outage is enabled. Requires
+   * {@link #SERVER_PUBSUB_HEALTH_MONITOR_ENABLED} to also be enabled — if the health monitor
+   * is disabled, this config is ignored.
+   */
+  public static final String SERVER_PUBSUB_PARTITION_PAUSE_ENABLED = "server.pubsub.partition.pause.enabled";
+
+  /**
+   * The interval in seconds between recovery probe attempts for unhealthy brokers.
+   */
+  public static final String SERVER_PUBSUB_HEALTH_PROBE_INTERVAL_SECONDS =
+      "server.pubsub.health.probe.interval.seconds";
+
+  /**
+   * The PubSub topic name used for recovery probes. The probe sends a metadata request for this
+   * topic to verify broker reachability. Must be a topic that is guaranteed to exist on the broker
+   * (e.g., a store's version topic). If empty or not set, recovery probes are skipped and paused
+   * partitions will not be automatically resumed.
+   */
+  public static final String SERVER_PUBSUB_HEALTH_PROBE_TOPIC = "server.pubsub.health.probe.topic";
+
   /**
    * Maximum duration (in milliseconds) to wait for the version information to become available in the store metadata
    * repository before skipping Heartbeat (HB) lag monitor setup activity during state transition.

internal/venice-common/src/main/java/com/linkedin/venice/pubsub/PubSubHealthCategory.java

@@ -0,0 +1,22 @@
+package com.linkedin.venice.pubsub;
+
+import com.linkedin.venice.stats.dimensions.VeniceDimensionInterface;
+import com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions;
+
+
+/**
+ * Categories of PubSub health tracked independently by the health monitor.
+ * A broker can be healthy while the metadata service is unhealthy, or vice versa.
+ */
+public enum PubSubHealthCategory implements VeniceDimensionInterface {
+  /** PubSub broker health — affects produce and consume operations */
+  BROKER,
+
+  /** PubSub metadata service health — affects topic creation, deletion, and metadata queries */
+  METADATA_SERVICE;
+
+  @Override
+  public VeniceMetricsDimensions getDimensionName() {
+    return VeniceMetricsDimensions.VENICE_PUBSUB_HEALTH_CATEGORY;
+  }
+}

internal/venice-common/src/main/java/com/linkedin/venice/pubsub/PubSubHealthChangeListener.java

@@ -0,0 +1,16 @@
+package com.linkedin.venice.pubsub;
+
+/**
+ * Listener for PubSub health status changes. Implementations are notified asynchronously
+ * when a PubSub target transitions between health states.
+ */
+public interface PubSubHealthChangeListener {
+  /**
+   * Called when a PubSub target's health status changes. Implementations must be non-blocking.
+   *
+   * @param pubSubAddress the broker or metadata service address
+   * @param category      whether this is a BROKER or METADATA_SERVICE status change
+   * @param newStatus     the new health status
+   */
+  void onHealthStatusChanged(String pubSubAddress, PubSubHealthCategory category, PubSubHealthStatus newStatus);
+}

internal/venice-common/src/main/java/com/linkedin/venice/pubsub/PubSubHealthSignalProvider.java

@@ -0,0 +1,34 @@
+package com.linkedin.venice.pubsub;
+
+/**
+ * A pluggable provider of health signals for PubSub targets. The PubSub health monitor
+ * aggregates signals from all registered providers to determine overall broker health.
+ *
+ * <p>The first implementation is exception-based (any PubSub exception marks the target unhealthy).
+ * Future implementations can detect non-exception failures such as growing heartbeat lag or
+ * elevated latency.
+ */
+public interface PubSubHealthSignalProvider {
+  /**
+   * @return a descriptive name for this provider, used in logging and metrics (e.g., "exception", "heartbeat-lag")
+   */
+  String getName();
+
+  /**
+   * @return true if this provider currently considers the given target unhealthy
+   */
+  boolean isUnhealthy(String pubSubAddress, PubSubHealthCategory category);
+
+  /**
+   * Called by the health monitor when a recovery probe succeeds for the given target.
+   * Implementations should clear any failure state for this target.
+   */
+  default void onProbeSuccess(String pubSubAddress, PubSubHealthCategory category) {
+  }
+
+  /**
+   * Called by the health monitor when a recovery probe fails for the given target.
+   */
+  default void onProbeFailure(String pubSubAddress, PubSubHealthCategory category) {
+  }
+}

internal/venice-common/src/main/java/com/linkedin/venice/pubsub/PubSubHealthStatus.java

@@ -0,0 +1,8 @@
+package com.linkedin.venice.pubsub;
+
+/**
+ * Health status of a PubSub target (broker or metadata service).
+ */
+public enum PubSubHealthStatus {
+  HEALTHY, UNHEALTHY
+}