From c5433745cb58d745c8ee472937a810d451c2218c Mon Sep 17 00:00:00 2001 From: Michelle Kwong Date: Mon, 20 Jul 2026 13:00:16 -0700 Subject: [PATCH 1/2] [server][pubsub] PubSub health foundation: core types, config keys, metric dimension MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit First of a stacked series splitting the PubSub Health Monitoring feature (#2534) into reviewable chunks. Introduces only the foundational, behavior-free building blocks — no wiring, all default-off. - New pubsub health types (venice-common): - PubSubHealthStatus (HEALTHY/UNHEALTHY) - PubSubHealthCategory (BROKER/METADATA_SERVICE) as a metric dimension - PubSubHealthSignalProvider (pluggable detection interface) - PubSubHealthChangeListener (state-transition callback interface) - New config keys (all default-off) + VeniceServerConfig parsing: - server.pubsub.health.monitor.enabled - server.pubsub.partition.pause.enabled - server.pubsub.health.probe.interval.seconds - server.pubsub.health.probe.topic - New OTel dimension VENICE_PUBSUB_HEALTH_CATEGORY (required by PubSubHealthCategory) + dimension unit tests. No call sites yet; compiles and tests independently. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../davinci/config/VeniceServerConfig.java | 30 ++++++++++++++++ .../dimensions/VeniceMetricsDimensions.java | 5 ++- .../dimensions/PubSubHealthCategoryTest.java | 21 ++++++++++++ .../VeniceMetricsDimensionsTest.java | 9 +++++ .../java/com/linkedin/venice/ConfigKeys.java | 27 +++++++++++++++ .../venice/pubsub/PubSubHealthCategory.java | 22 ++++++++++++ .../pubsub/PubSubHealthChangeListener.java | 16 +++++++++ .../pubsub/PubSubHealthSignalProvider.java | 34 +++++++++++++++++++ .../venice/pubsub/PubSubHealthStatus.java | 8 +++++ 9 files changed, 171 insertions(+), 1 deletion(-) create mode 100644 internal/venice-client-common/src/test/java/com/linkedin/venice/stats/dimensions/PubSubHealthCategoryTest.java create mode 100644 internal/venice-common/src/main/java/com/linkedin/venice/pubsub/PubSubHealthCategory.java create mode 100644 internal/venice-common/src/main/java/com/linkedin/venice/pubsub/PubSubHealthChangeListener.java create mode 100644 internal/venice-common/src/main/java/com/linkedin/venice/pubsub/PubSubHealthSignalProvider.java create mode 100644 internal/venice-common/src/main/java/com/linkedin/venice/pubsub/PubSubHealthStatus.java 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 +} From 0bf6263174236fb6598db89a48b3e870302bdb84 Mon Sep 17 00:00:00 2001 From: Michelle Kwong Date: Tue, 21 Jul 2026 16:51:24 -0700 Subject: [PATCH 2/2] [server][pubsub] Add VeniceServerConfig test coverage for pubsub health configs Covers the diffCoverage gate for da-vinci-client: exercises the new pubsub-health config getters and both sides of the `pubSubHealthMonitorEnabled && ...` short-circuit in partition-pause resolution (monitor-on honors pause, monitor-off gates it off), plus the default values. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../config/VeniceServerConfigTest.java | 33 +++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/config/VeniceServerConfigTest.java b/clients/da-vinci-client/src/test/java/com/linkedin/davinci/config/VeniceServerConfigTest.java index 9135195b45c..cc0d91060f3 100644 --- a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/config/VeniceServerConfigTest.java +++ b/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()); + } }