diff --git a/clients/da-vinci-client/src/main/java/com/linkedin/davinci/config/VeniceServerConfig.java b/clients/da-vinci-client/src/main/java/com/linkedin/davinci/config/VeniceServerConfig.java index 970b2778e55..bdf148c019e 100644 --- a/clients/da-vinci-client/src/main/java/com/linkedin/davinci/config/VeniceServerConfig.java +++ b/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 expectedValues = CollectionUtils.mapBuilder() + .put(PubSubHealthCategory.BROKER, "broker") + .put(PubSubHealthCategory.METADATA_SERVICE, "metadata_service") + .build(); + new VeniceDimensionTestFixture<>( + PubSubHealthCategory.class, + VeniceMetricsDimensions.VENICE_PUBSUB_HEALTH_CATEGORY, + expectedValues).assertAll(); + } +} diff --git a/internal/venice-client-common/src/test/java/com/linkedin/venice/stats/dimensions/VeniceMetricsDimensionsTest.java b/internal/venice-client-common/src/test/java/com/linkedin/venice/stats/dimensions/VeniceMetricsDimensionsTest.java index b60bdbe704f..fd630483d94 100644 --- a/internal/venice-client-common/src/test/java/com/linkedin/venice/stats/dimensions/VeniceMetricsDimensionsTest.java +++ b/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); } diff --git a/internal/venice-common/src/main/java/com/linkedin/venice/ConfigKeys.java b/internal/venice-common/src/main/java/com/linkedin/venice/ConfigKeys.java index 321708bca6b..8c8b88e2805 100644 --- a/internal/venice-common/src/main/java/com/linkedin/venice/ConfigKeys.java +++ b/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. diff --git a/internal/venice-common/src/main/java/com/linkedin/venice/pubsub/PubSubHealthCategory.java b/internal/venice-common/src/main/java/com/linkedin/venice/pubsub/PubSubHealthCategory.java new file mode 100644 index 00000000000..74d6e24b166 --- /dev/null +++ b/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; + } +} diff --git a/internal/venice-common/src/main/java/com/linkedin/venice/pubsub/PubSubHealthChangeListener.java b/internal/venice-common/src/main/java/com/linkedin/venice/pubsub/PubSubHealthChangeListener.java new file mode 100644 index 00000000000..6f64cf5f08c --- /dev/null +++ b/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); +} diff --git a/internal/venice-common/src/main/java/com/linkedin/venice/pubsub/PubSubHealthSignalProvider.java b/internal/venice-common/src/main/java/com/linkedin/venice/pubsub/PubSubHealthSignalProvider.java new file mode 100644 index 00000000000..9bb217151c5 --- /dev/null +++ b/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. + * + *

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) { + } +} diff --git a/internal/venice-common/src/main/java/com/linkedin/venice/pubsub/PubSubHealthStatus.java b/internal/venice-common/src/main/java/com/linkedin/venice/pubsub/PubSubHealthStatus.java new file mode 100644 index 00000000000..57b16da7427 --- /dev/null +++ b/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 +}