natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Pull RetryStrategy for CMS commit operations into DatabaseDescriptor

This commit is contained in:
Sam Tunnicliffe 2026-06-23 12:21:14 +01:00
parent 31090ed3b0
commit 5a0dd82cd2
3 changed files with 43 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -124,8 +124,10 @@ import org.apache.cassandra.security.JREProvider;
import org.apache.cassandra.security.SSLFactory;
import org.apache.cassandra.service.CacheService.CacheType;
import org.apache.cassandra.service.FileSystemOwnershipCheck;
import org.apache.cassandra.service.RetryStrategy;
import org.apache.cassandra.service.StartupChecks;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.service.TimeoutStrategy;
import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.service.accord.api.AccordWaitStrategies;
import org.apache.cassandra.service.consensus.TransactionalMode;
@ -270,6 +272,15 @@ public class DatabaseDescriptor
public static volatile boolean allowUnlimitedConcurrentValidations = ALLOW_UNLIMITED_CONCURRENT_VALIDATIONS.getBoolean();
/**
* RetryStrategy which provides exponential backoff with full jitter, for use by both CMS and non-CMS members
* when submitting a Commit request. The range and increments of the backoff times are defined by
* conf.cms_commit_retry_initial_delay and conf.cms_commit_retry_max_delay. Both are hot properties and so
* changing either one causes this retry strategy to be reconstructed.
*/
private static volatile RetryStrategy cms_commit_retry_strategy;
/**
* The configuration for guardrails.
*/
@ -580,6 +591,8 @@ public class DatabaseDescriptor
applyAccord();
applyCMS();
applyStartupChecks();
}
@ -1388,6 +1401,25 @@ public class DatabaseDescriptor
AccordService.applyProtocolModifiers(getAccord());
}
private static void applyCMS()
{
try
{
long initialDelayMs = conf.cms_commit_retry_initial_delay.to(TimeUnit.MILLISECONDS);
long maxDelayMs = conf.cms_commit_retry_max_delay.to(TimeUnit.MILLISECONDS);
// range of backoff wait time starts at 0ms backing off exponentially at initialDelayMs * 2^attempts
String spec = String.format("0ms ... %dms * 2^attempts <= %dms", initialDelayMs, maxDelayMs);
logger.debug("Initializing cms_commit_retry_strategy from spec: " + spec);
cms_commit_retry_strategy = RetryStrategy.parse(spec,
TimeoutStrategy.LatencySourceFactory.none(),
RetryStrategy.randomizers.uniform());
}
catch (Exception e)
{
throw new ConfigurationException("Invalid configuration for cms_commit_retry_strategy. " + e.getMessage(), e);
}
}
public static StartupChecksConfiguration getStartupChecksConfiguration()
{
return startupChecksConfiguration;
@ -6200,6 +6232,8 @@ public class DatabaseDescriptor
}
}
public static DurationSpec getCmsCommitRetryInitialDelay()
{
return conf.cms_commit_retry_initial_delay;
@ -6211,6 +6245,7 @@ public class DatabaseDescriptor
{
logger.info("Setting cms_commit_retry_initial_delay to {}ms", delayInMillis);
conf.cms_commit_retry_initial_delay = new DurationSpec.LongMillisecondsBound(delayInMillis);
applyCMS();
}
}
@ -6225,9 +6260,15 @@ public class DatabaseDescriptor
{
logger.info("Setting cms_commit_retry_max_delay to {}ms", delayInMillis);
conf.cms_commit_retry_max_delay = new DurationSpec.LongMillisecondsBound(delayInMillis);
applyCMS();
}
}
public static RetryStrategy getCmsCommitRetryStrategy()
{
return cms_commit_retry_strategy;
}
public static int getEpochAwareDebounceInFlightTrackerMaxSize()
{
return conf.epoch_aware_debounce_inflight_tracker_max_size;

9
src/java/org/apache/cassandra/tcm/ClusterMetadataService.java
View File

@ -58,7 +58,6 @@ import org.apache.cassandra.schema.DistributedSchema;
import org.apache.cassandra.schema.Keyspaces;
import org.apache.cassandra.schema.ReplicationParams;
import org.apache.cassandra.service.RetryStrategy;
import org.apache.cassandra.service.TimeoutStrategy;
import org.apache.cassandra.service.accord.topology.AccordFastPath;
import org.apache.cassandra.service.accord.topology.AccordStaleReplicas;
import org.apache.cassandra.service.consensus.migration.ConsensusMigrationState;
@ -733,13 +732,7 @@ public class ClusterMetadataService
// with jitter works to desynchronize retry waves after a CMS await timeout. For CMS members committing
// locally, it helps to space Paxos CAS retries in the local commit loop.
long deadlineNanos = nanoTime() + DatabaseDescriptor.getCmsCommitTimeout().to(TimeUnit.NANOSECONDS);
long initialDelayMs = DatabaseDescriptor.getCmsCommitRetryInitialDelay().to(TimeUnit.MILLISECONDS);
long maxDelayMs = DatabaseDescriptor.getCmsCommitRetryMaxDelay().to(TimeUnit.MILLISECONDS);
// range of backoff wait time starts at 0ms backing off exponentially at initialDelayMs * 2^attempts
String spec = String.format("0ms ... %dms * 2^attempts <= %dms", initialDelayMs, maxDelayMs);
RetryStrategy backoffWithJitter = RetryStrategy.parse(spec,
TimeoutStrategy.LatencySourceFactory.none(),
RetryStrategy.randomizers.uniform());
RetryStrategy backoffWithJitter = DatabaseDescriptor.getCmsCommitRetryStrategy();
retryPolicy = Retry.until(deadlineNanos, TCMMetrics.instance.commitRetries, backoffWithJitter);
}
return retryPolicy;

9
src/java/org/apache/cassandra/tcm/Commit.java
View File

@ -43,7 +43,6 @@ import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.service.RetryStrategy;
import org.apache.cassandra.service.TimeoutStrategy;
import org.apache.cassandra.tcm.log.Entry;
import org.apache.cassandra.tcm.log.LogState;
import org.apache.cassandra.tcm.membership.Directory;
@ -396,13 +395,7 @@ public class Commit
long now = MonotonicClock.Global.preciseTime.now();
long localDeadlineNanos = Math.max(now + casWriteRpcTimeoutNanos,
message.expiresAtNanos() - casWriteRpcTimeoutNanos);
long initialDelayMs = DatabaseDescriptor.getCmsCommitRetryInitialDelay().to(TimeUnit.MILLISECONDS);
long maxDelayMs = DatabaseDescriptor.getCmsCommitRetryMaxDelay().to(TimeUnit.MILLISECONDS);
String spec = String.format("... %dms^attempts <= %dms", initialDelayMs, maxDelayMs);
RetryStrategy backoffWithJitter = RetryStrategy.parse(spec,
TimeoutStrategy.LatencySourceFactory.none(),
RetryStrategy.randomizers.uniform());
RetryStrategy backoffWithJitter = DatabaseDescriptor.getCmsCommitRetryStrategy();
Retry retryPolicy = Retry.until(localDeadlineNanos, TCMMetrics.instance.commitRetries, backoffWithJitter);
Result result = processor.commit(message.payload.entryId, message.payload.transform, message.payload.lastKnown, retryPolicy);
if (result.isSuccess())