Skip to content

Add lag-based wait for future-version OFFLINE->STANDBY transition #2976

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We鈥檒l occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jingy-li wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Add lag-based wait for future-version OFFLINE->STANDBY transition #2976

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 33 additions & 0 deletions clients/da-vinci-client/src/main/java/com/linkedin/davinci/config/VeniceServerConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -128,6 +128,10 @@
import static com.linkedin.venice.ConfigKeys.SERVER_ENABLE_LIVE_CONFIG_BASED_KAFKA_THROTTLING;
import static com.linkedin.venice.ConfigKeys.SERVER_ENABLE_PARALLEL_BATCH_GET;
import static com.linkedin.venice.ConfigKeys.SERVER_FORKED_PROCESS_JVM_ARGUMENT_LIST;
import static com.linkedin.venice.ConfigKeys.SERVER_FUTURE_VERSION_STANDBY_LAG_CHECK_ENABLED;
import static com.linkedin.venice.ConfigKeys.SERVER_FUTURE_VERSION_STANDBY_LAG_CHECK_POLL_INTERVAL_MINUTES;
import static com.linkedin.venice.ConfigKeys.SERVER_FUTURE_VERSION_STANDBY_LAG_CHECK_TIMEOUT_MINUTES;
import static com.linkedin.venice.ConfigKeys.SERVER_FUTURE_VERSION_STANDBY_LAG_THRESHOLD;
import static com.linkedin.venice.ConfigKeys.SERVER_GLOBAL_RT_DIV_ENABLED;
import static com.linkedin.venice.ConfigKeys.SERVER_HEARTBEAT_REPORTER_INTERVAL_SECONDS;
import static com.linkedin.venice.ConfigKeys.SERVER_HELIX_JOIN_AS_UNKNOWN;
Expand Down Expand Up @@ -768,6 +772,11 @@ public class VeniceServerConfig extends VeniceClusterConfig {
private final int lagBasedReplicaAutoResubscribeThresholdInSeconds;
private final int lagBasedReplicaAutoResubscribeMaxReplicaCount;

private final boolean futureVersionStandbyLagCheckEnabled;
private final long futureVersionStandbyLagThreshold;
private final int futureVersionStandbyLagCheckTimeoutMinutes;
private final int futureVersionStandbyLagCheckPollIntervalMinutes;

private final int serverIngestionInfoLogLineLimit;

private final boolean parallelResourceShutdownEnabled;
Expand Down Expand Up @@ -1354,6 +1363,14 @@ public VeniceServerConfig(VeniceProperties serverProperties, Map<String, Map<Str
serverProperties.getInt(SERVER_LAG_BASED_REPLICA_AUTO_RESUBSCRIBE_THRESHOLD_IN_SECONDS, 600);
this.lagBasedReplicaAutoResubscribeMaxReplicaCount =
serverProperties.getInt(SERVER_LAG_BASED_REPLICA_AUTO_RESUBSCRIBE_MAX_REPLICA_COUNT, 3);
this.futureVersionStandbyLagCheckEnabled =
serverProperties.getBoolean(SERVER_FUTURE_VERSION_STANDBY_LAG_CHECK_ENABLED, false);
this.futureVersionStandbyLagThreshold =
serverProperties.getLong(SERVER_FUTURE_VERSION_STANDBY_LAG_THRESHOLD, 1000L);
this.futureVersionStandbyLagCheckTimeoutMinutes =
serverProperties.getInt(SERVER_FUTURE_VERSION_STANDBY_LAG_CHECK_TIMEOUT_MINUTES, 2 * 60);
this.futureVersionStandbyLagCheckPollIntervalMinutes =
serverProperties.getInt(SERVER_FUTURE_VERSION_STANDBY_LAG_CHECK_POLL_INTERVAL_MINUTES, 15);
this.useMetricsBasedPositionInLagComputation =
serverProperties.getBoolean(SERVER_USE_METRICS_BASED_POSITION_IN_LAG_COMPUTATION, false);
this.useUpstreamPubSubPositionWithFallback =
Expand Down Expand Up @@ -2435,6 +2452,22 @@ public int getLagBasedReplicaAutoResubscribeMaxReplicaCount() {
return lagBasedReplicaAutoResubscribeMaxReplicaCount;
}

public boolean isFutureVersionStandbyLagCheckEnabled() {
return futureVersionStandbyLagCheckEnabled;
}

public long getFutureVersionStandbyLagThreshold() {
return futureVersionStandbyLagThreshold;
}

public int getFutureVersionStandbyLagCheckTimeoutMinutes() {
return futureVersionStandbyLagCheckTimeoutMinutes;
}

public int getFutureVersionStandbyLagCheckPollIntervalMinutes() {
return futureVersionStandbyLagCheckPollIntervalMinutes;
}

public boolean isUseMetricsBasedPositionInLagComputationEnabled() {
return this.useMetricsBasedPositionInLagComputation;
}
Expand Down
60 changes: 60 additions & 0 deletions ...da-vinci-client/src/main/java/com/linkedin/davinci/helix/AbstractPartitionStateModel.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,11 @@
package com.linkedin.davinci.helix;

import com.linkedin.davinci.config.VeniceServerConfig;
import com.linkedin.davinci.config.VeniceStoreVersionConfig;
import com.linkedin.davinci.ingestion.IngestionBackend;
import com.linkedin.davinci.kafka.consumer.PartitionReplicaIngestionContext;
import com.linkedin.davinci.kafka.consumer.StoreIngestionService;
import com.linkedin.davinci.kafka.consumer.StoreIngestionTask;
import com.linkedin.davinci.stats.ParticipantStateTransitionStats;
import com.linkedin.venice.exceptions.VeniceException;
import com.linkedin.venice.helix.HelixPartitionStatusAccessor;
Expand Down Expand Up @@ -408,6 +410,64 @@ protected void waitConsumptionCompleted(String resourceName, StateModelIngestion
}
}

/**
* Best-effort wait, applicable to a future-version replica whose push is still in progress (i.e. neither the
* current version, nor a future version which has already finished ingesting), for the replica's local version
* topic lag to drop to or below an acceptable threshold before completing the OFFLINE -> STANDBY transition.
*
* Callers are expected to only invoke this method when
* {@link VeniceServerConfig#isFutureVersionStandbyLagCheckEnabled()} is true.
*
* Unlike {@link #waitConsumptionCompleted}, this does not wait for ingestion to fully complete: it only waits
* until the measured lag is within {@link VeniceServerConfig#getFutureVersionStandbyLagThreshold()}, or until
* {@link VeniceServerConfig#getFutureVersionStandbyLagCheckTimeoutMinutes()} elapses, whichever happens first.
* If lag cannot be measured, or the ingestion task is not found, this method returns immediately, preserving
* the pre-existing (no-wait) behavior for this case.
*/
protected void waitUntilFutureVersionLagAcceptable(String resourceName) {
VeniceServerConfig serverConfig = storeAndServerConfigs;
String replicaId = Utils.getReplicaId(resourceName, partition);
StoreIngestionTask ingestionTask = getStoreIngestionService().getStoreIngestionTask(resourceName);
if (ingestionTask == null) {
logger.warn(
"No ingestion task found for replica {} when checking future version standby lag, proceeding to STANDBY without waiting.",
replicaId);
return;
}
long lagThreshold = serverConfig.getFutureVersionStandbyLagThreshold();
long timeoutMs = TimeUnit.MINUTES.toMillis(serverConfig.getFutureVersionStandbyLagCheckTimeoutMinutes());
long pollIntervalMs = TimeUnit.MINUTES.toMillis(serverConfig.getFutureVersionStandbyLagCheckPollIntervalMinutes());
long deadlineMs = System.currentTimeMillis() + timeoutMs;
Comment on lines +438 to +440
while (true) {
long lag = ingestionTask.getLocalVersionTopicLag(partition);
if (lag == Long.MAX_VALUE) {
logger.warn(
"Could not measure local version topic lag for replica {}, proceeding to STANDBY without waiting.",
replicaId);
return;
}
if (lag <= lagThreshold) {
logger.info(
"Future version replica {} local version topic lag {} is within threshold {}, proceeding to STANDBY.",
replicaId,
lag,
lagThreshold);
return;
}
long remainingMs = deadlineMs - System.currentTimeMillis();
if (remainingMs <= 0) {
logger.warn(
"Future version replica {} local version topic lag {} still above threshold {} after {}min timeout, proceeding to STANDBY.",
replicaId,
lag,
lagThreshold,
serverConfig.getFutureVersionStandbyLagCheckTimeoutMinutes());
return;
}
Utils.sleep(Math.min(pollIntervalMs, remainingMs));

@sixpluszero sixpluszero Aug 14, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This loop synchronously sleeps on the Helix state-transition executor. With the defaults, each lagging partition can hold one worker for up to 120 minutes. VeniceServerConfig still defaults LEADER_FOLLOWER_STATE_TRANSITION_THREAD_POOL_STRATEGY to SINGLE_POOL_STRATEGY (20 workers), so enough concurrent future-version transitions could delay current or backup-version transitions as well.

What is the current rollout/adoption of DUAL_POOL_STRATEGY in production? Is enabling this lag check intended to be coupled with the dual-pool rollout? If some clusters still use the single pool, should we validate or reject that configuration combination, or otherwise avoid holding a transition worker while polling? Even with dual pool, it would be useful to understand expected concurrency versus the default future-version pool size.

}
}

private void waitPartitionPushStatusAccessor() throws Exception {
if (partitionPushStatusAccessor == null) {
partitionPushStatusAccessor =
Expand Down
5 changes: 5 additions & 0 deletions ...ci-client/src/main/java/com/linkedin/davinci/helix/LeaderFollowerPartitionStateModel.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,6 +105,7 @@ public void onBecomeStandbyFromOffline(Message message, NotificationContext cont
// A future version is ready to serve if it's status is either PUSHED or ONLINE
// PUSHED is set for future versions of a target region push with deferred swap
// ONLINE is set for future versions of a push with deferred swap
boolean isFutureVersion = Utils.isFutureVersion(resourceName, getStoreRepo());
boolean isFutureVersionReady = Utils.isFutureVersionReady(resourceName, getStoreRepo());
/**
* For current version and already completed future versions, firstly create a latch, then start ingestion and wait
Expand Down Expand Up @@ -136,6 +137,10 @@ public void onBecomeStandbyFromOffline(Message message, NotificationContext cont
Utils.getReplicaId(message.getResourceName(), getPartition()));
if (isCurrentVersion || isFutureVersionReady) {
waitConsumptionCompleted(resourceName, notifier);
} else if (isFutureVersion && getStoreAndServerConfigs().isFutureVersionStandbyLagCheckEnabled()) {

@sixpluszero sixpluszero Aug 14, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could we restrict this wait to an explicitly active future-version push, e.g. version.getStatus() == VersionStatus.STARTED, rather than using "future and not PUSHED/ONLINE" as the proxy?

isFutureVersionReady() is false for terminal states such as KILLED, ERROR, and ROLLED_BACK. Those states can overlap with an OFFLINE -> STANDBY callback because killing and Helix-resource cleanup are asynchronous: the transition may already be running or queued when the version is marked KILLED. In the normal kill path the ingestion task will eventually clear its PCS, so this wait should fail open on a later poll, but it can still occupy the transition worker until that poll; delayed kill or cleanup can extend this to the timeout.

It would also be useful to re-check the status while polling, since a push can be STARTED when the wait begins and become KILLED or ERROR afterward.

// Future version whose push is still in progress: best-effort wait for lag to become acceptable before
// this replica becomes eligible for leader election (no-op unless explicitly enabled).
waitUntilFutureVersionLagAcceptable(resourceName);
}
});
}
Expand Down
18 changes: 18 additions & 0 deletions ...da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3869,6 +3869,24 @@ protected static long measureLagWithCallToPubSub(
return diff - 1;
}

/**
* Best-effort measurement of how far behind {@code partition}'s local version topic consumption is, relative to
* the end of the local version topic partition. Unlike {@link #isReadyToServe(PartitionConsumptionState)}, this
* does not require END_OF_PUSH to have been received, so it can be used to gate in-progress (future version)
* pushes.
*
* @return the lag in number of records, or {@link Long#MAX_VALUE} if it could not be measured (e.g. no
* {@link PartitionConsumptionState} yet for the partition, or a PubSub error occurred).
*/
public long getLocalVersionTopicLag(int partition) {
PartitionConsumptionState pcs = getPartitionConsumptionStateMap().get(partition);
if (pcs == null) {
return Long.MAX_VALUE;
}
PubSubTopicPartition topicPartition = new PubSubTopicPartitionImpl(versionTopic, partition);
return measureLagWithCallToPubSub(localKafkaServer, topicPartition, pcs.getLatestProcessedVtPosition());
}

public abstract int getWriteComputeErrorCode();

public abstract void updateLeaderTopicOnFollower(PartitionConsumptionState partitionConsumptionState);
Expand Down
30 changes: 30 additions & 0 deletions ...nts/da-vinci-client/src/test/java/com/linkedin/davinci/config/VeniceServerConfigTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,10 @@
import static com.linkedin.venice.ConfigKeys.SERVER_CROSS_TP_PARALLEL_PROCESSING_THREAD_POOL_SIZE;
import static com.linkedin.venice.ConfigKeys.SERVER_DEAD_LEADER_READY_TO_SERVE_FALLBACK_THRESHOLD_MS;
import static com.linkedin.venice.ConfigKeys.SERVER_FORKED_PROCESS_JVM_ARGUMENT_LIST;
import static com.linkedin.venice.ConfigKeys.SERVER_FUTURE_VERSION_STANDBY_LAG_CHECK_ENABLED;
import static com.linkedin.venice.ConfigKeys.SERVER_FUTURE_VERSION_STANDBY_LAG_CHECK_POLL_INTERVAL_MINUTES;
import static com.linkedin.venice.ConfigKeys.SERVER_FUTURE_VERSION_STANDBY_LAG_CHECK_TIMEOUT_MINUTES;
import static com.linkedin.venice.ConfigKeys.SERVER_FUTURE_VERSION_STANDBY_LAG_THRESHOLD;
import static com.linkedin.venice.ConfigKeys.SERVER_INGESTION_OTEL_STATS_ENABLED;
import static com.linkedin.venice.ConfigKeys.SERVER_LEADER_COMPLETE_STATE_CHECK_IN_FOLLOWER_VALID_INTERVAL_MS;
import static com.linkedin.venice.ConfigKeys.SERVER_LEADER_HANDOVER_USE_DOL_MECHANISM_FOR_SYSTEM_STORES;
Expand Down Expand Up @@ -68,6 +72,32 @@ public void testForkedJVMParams() {
assertEquals(jvmArgs.get(1), "-Xmx256G");
}

@Test
public void testFutureVersionStandbyLagCheckDefaults() {
Properties props = populatedBasicProperties();
VeniceServerConfig config = new VeniceServerConfig(new VeniceProperties(props));

assertFalse(config.isFutureVersionStandbyLagCheckEnabled());
assertEquals(config.getFutureVersionStandbyLagThreshold(), 1000L);
assertEquals(config.getFutureVersionStandbyLagCheckTimeoutMinutes(), 2 * 60);
assertEquals(config.getFutureVersionStandbyLagCheckPollIntervalMinutes(), 15);
}

@Test
public void testFutureVersionStandbyLagCheckOverrides() {
Properties props = populatedBasicProperties();
props.setProperty(SERVER_FUTURE_VERSION_STANDBY_LAG_CHECK_ENABLED, "true");
props.setProperty(SERVER_FUTURE_VERSION_STANDBY_LAG_THRESHOLD, "2000");
props.setProperty(SERVER_FUTURE_VERSION_STANDBY_LAG_CHECK_TIMEOUT_MINUTES, "60");
props.setProperty(SERVER_FUTURE_VERSION_STANDBY_LAG_CHECK_POLL_INTERVAL_MINUTES, "5");
VeniceServerConfig config = new VeniceServerConfig(new VeniceProperties(props));

assertTrue(config.isFutureVersionStandbyLagCheckEnabled());
assertEquals(config.getFutureVersionStandbyLagThreshold(), 2000L);
assertEquals(config.getFutureVersionStandbyLagCheckTimeoutMinutes(), 60);
assertEquals(config.getFutureVersionStandbyLagCheckPollIntervalMinutes(), 5);
}

@Test
public void testAaDcrBugInjectionEnabledForStore() {
// Default: empty map, nothing is enabled.
Expand Down
Loading
Loading