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

Add policy for selecting CMS host when submitting commit request 

Patch by Jon Meredith; reviewed by Sam Tunnicliffe for CASSANDRA-21456
This commit is contained in:
Jon Meredith 2026-06-12 14:26:26 +01:00 committed by Sam Tunnicliffe
parent 44ee9d6167
commit 9a896cbaeb
7 changed files with 308 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
src/java/org/apache/cassandra/config/Config.java
View File

@ -191,6 +191,7 @@ public class Config
public volatile DurationSpec.IntMillisecondsBound cms_default_max_retry_backoff = null;
public String cms_retry_delay = "50ms*attempts <= 500ms ... 100ms*attempts <= 1s,retries=10";
public volatile CMSCommitMemberPreferencePolicy cms_commit_member_preference_policy = CMSCommitMemberPreferencePolicy.random;
public volatile int epoch_aware_debounce_inflight_tracker_max_size = 100;
/**
@ -1254,6 +1255,31 @@ public class Config
group
}
/**
* Strategy for selecting which CMS member to contact for TCM commits.
*/
public enum CMSCommitMemberPreferencePolicy
{
/** Shuffle candidates randomly (original behavior) */
random,
/** Shuffle local DC candidates randomly, followed by shuffled non-local */
local_random,
/** Sort by address - all nodes converge on same member with the goal
* of reducing Paxos contention, but will increase hot-spotting on the
* determined member
*/
deterministic,
/** Sort local members first by name, then non-local. Nodes in each DC
* will converge on same member with the goal of reducing Paxos contention d
* below the random but with lower latency, though higher contention than
* globally deterministic.
*/
local_deterministic
}
public enum FlushCompression
{
none,
@ -1560,4 +1586,4 @@ public class Config
* 6.0 and later.
*/
public volatile boolean gossip_quarantine_disabled = false;
}
}

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

@ -175,6 +175,7 @@ import static org.apache.cassandra.io.util.FileUtils.ONE_GIB;
import static org.apache.cassandra.io.util.FileUtils.ONE_MIB;
import static org.apache.cassandra.journal.Params.FlushMode.PERIODIC;
import static org.apache.cassandra.utils.Clock.Global.logInitializationOutcome;
import static org.apache.cassandra.utils.LocalizeString.toLowerCaseLocalized;
public class DatabaseDescriptor
{
@ -6092,6 +6093,21 @@ public class DatabaseDescriptor
return conf.cms_await_timeout;
}
public static Config.CMSCommitMemberPreferencePolicy getCmsCommitMemberPreferencePolicy()
{
return conf.cms_commit_member_preference_policy;
}
public static void setCmsCommitMemberPreferencePolicy(Config.CMSCommitMemberPreferencePolicy policy)
{
conf.cms_commit_member_preference_policy = policy;
}
public static void setCmsCommitMemberPreferencePolicy(String policy)
{
setCmsCommitMemberPreferencePolicy(Config.CMSCommitMemberPreferencePolicy.valueOf(toLowerCaseLocalized(policy)));
}
public static int getEpochAwareDebounceInFlightTrackerMaxSize()
{
return conf.epoch_aware_debounce_inflight_tracker_max_size;

13
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -1761,6 +1761,19 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
return pojoMapToString(snapshotAsMap, format);
}
@Override
public String getCmsCommitMemberPreferencePolicy()
{
return DatabaseDescriptor.getCmsCommitMemberPreferencePolicy().name();
}
@Override
public void setCmsCommitMemberPreferencePolicy(String policy)
{
DatabaseDescriptor.setCmsCommitMemberPreferencePolicy(policy);
logger.info("Set cms_commit_member_preference_policy to {}", policy);
}
public Map<String,List<Integer>> getConcurrency(List<String> stageNames)
{
Stream<Stage> stageStream = stageNames.isEmpty() ? stream(Stage.values()) : stageNames.stream().map(Stage::fromPoolName);

12
src/java/org/apache/cassandra/service/StorageServiceMBean.java
View File

@ -1170,6 +1170,18 @@ public interface StorageServiceMBean extends NotificationEmitter
List<String> getAccordManagedKeyspaces();
List<String> getAccordManagedTables();
/** Get the CMS commit member preference policy
*
* @return how to choose the cms member preference order for commits
*/
public String getCmsCommitMemberPreferencePolicy();
/** Update the CMS commit member preference policy
*
* @param policy see Config.CMSCommitMemberPreferencePolicy
*/
public void setCmsCommitMemberPreferencePolicy(String policy);
/** Gets the concurrency settings for processing stages*/
static class StageConcurrency implements Serializable
{

74
src/java/org/apache/cassandra/tcm/RemoteProcessor.java
View File

@ -31,16 +31,19 @@ import java.util.function.Consumer;
import java.util.function.Supplier;
import com.codahale.metrics.Timer;
import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.exceptions.RequestFailure;
import org.apache.cassandra.exceptions.RequestFailureReason;
import org.apache.cassandra.gms.FailureDetector;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.Locator;
import org.apache.cassandra.metrics.TCMMetrics;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessageDelivery;
@ -52,6 +55,7 @@ import org.apache.cassandra.tcm.Discovery.DiscoveredNodes;
import org.apache.cassandra.tcm.log.Entry;
import org.apache.cassandra.tcm.log.LocalLog;
import org.apache.cassandra.tcm.log.LogState;
import org.apache.cassandra.tcm.membership.Location;
import org.apache.cassandra.utils.AbstractIterator;
import org.apache.cassandra.utils.Clock;
import org.apache.cassandra.utils.FBUtilities;
@ -128,11 +132,79 @@ public final class RemoteProcessor implements Processor
}
}
Collections.shuffle(candidates);
sortCandidates(candidates);
return candidates;
}
private static void sortCandidates(List<InetAddressAndPort> candidates)
{
Config.CMSCommitMemberPreferencePolicy policy = DatabaseDescriptor.getCmsCommitMemberPreferencePolicy();
sortCandidates(candidates, policy, DatabaseDescriptor.getLocator());
}
@VisibleForTesting
static void sortCandidates(List<InetAddressAndPort> candidates,
Config.CMSCommitMemberPreferencePolicy policy,
Locator locator)
{
switch (policy)
{
case random:
Collections.shuffle(candidates);
break;
case local_random:
shuffleLocalDcFirstThenShuffleRest(candidates, locator);
break;
case deterministic:
Collections.sort(candidates);
break;
case local_deterministic:
sortLocalDcFirstThenByAddress(candidates, locator);
break;
default:
throw new IllegalStateException(policy.toString());
}
}
@VisibleForTesting
static void shuffleLocalDcFirstThenShuffleRest(List<InetAddressAndPort> candidates, Locator locator)
{
Location local = locator.local();
List<InetAddressAndPort> localDc = new ArrayList<>();
List<InetAddressAndPort> remoteDc = new ArrayList<>();
for (InetAddressAndPort ep : candidates)
{
if (local.sameDatacenter(locator.location(ep)))
localDc.add(ep);
else
remoteDc.add(ep);
}
Collections.shuffle(localDc);
Collections.shuffle(remoteDc);
candidates.clear();
candidates.addAll(localDc);
candidates.addAll(remoteDc);
}
@VisibleForTesting
static void sortLocalDcFirstThenByAddress(List<InetAddressAndPort> candidates, Locator locator)
{
Location local = locator.local();
candidates.sort((a, b) -> {
boolean aLocal = local.sameDatacenter(locator.location(a));
boolean bLocal = local.sameDatacenter(locator.location(b));
if (aLocal != bLocal)
return aLocal ? -1 : 1;
return a.compareTo(b);
});
}
@Override
public ClusterMetadata fetchLogAndWait(Epoch waitFor, Retry retryPolicy)
{

1
test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java
View File

@ -96,6 +96,7 @@ public class DatabaseDescriptorRefTest
"org.apache.cassandra.config.CassandraRelevantProperties$PropertyConverter",
"org.apache.cassandra.config.Config",
"org.apache.cassandra.config.Config$1",
"org.apache.cassandra.config.Config$CMSCommitMemberPreferencePolicy",
"org.apache.cassandra.config.Config$CommitFailurePolicy",
"org.apache.cassandra.config.Config$CQLStartTime",
"org.apache.cassandra.config.Config$CommitLogSync",

166
test/unit/org/apache/cassandra/tcm/RemoteProcessorTest.java
View File

@ -19,13 +19,19 @@
package org.apache.cassandra.tcm;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.junit.Test;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.Locator;
import org.apache.cassandra.tcm.membership.Location;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@ -107,4 +113,164 @@ public class RemoteProcessorTest
}
return allEndpoints;
}
// ========== CMS Member Selection Tests ==========
@Test
public void testCandidatesDeterministic()
{
// Create endpoints in non-sorted order
List<InetAddressAndPort> candidates = new ArrayList<>(Arrays.asList(
InetAddressAndPort.getByNameUnchecked("127.0.0.5"),
InetAddressAndPort.getByNameUnchecked("127.0.0.1"),
InetAddressAndPort.getByNameUnchecked("127.0.0.9"),
InetAddressAndPort.getByNameUnchecked("127.0.0.3")
));
// Deterministic policy should sort by address
RemoteProcessor.sortCandidates(candidates, Config.CMSCommitMemberPreferencePolicy.deterministic, null);
assertEquals(InetAddressAndPort.getByNameUnchecked("127.0.0.1"), candidates.get(0));
assertEquals(InetAddressAndPort.getByNameUnchecked("127.0.0.3"), candidates.get(1));
assertEquals(InetAddressAndPort.getByNameUnchecked("127.0.0.5"), candidates.get(2));
assertEquals(InetAddressAndPort.getByNameUnchecked("127.0.0.9"), candidates.get(3));
}
@Test
public void testCandidatesRandom()
{
// Create endpoints
List<InetAddressAndPort> original = Arrays.asList(
InetAddressAndPort.getByNameUnchecked("127.0.0.1"),
InetAddressAndPort.getByNameUnchecked("127.0.0.2"),
InetAddressAndPort.getByNameUnchecked("127.0.0.3"),
InetAddressAndPort.getByNameUnchecked("127.0.0.4")
);
List<InetAddressAndPort> candidates = new ArrayList<>(original);
// Random policy should shuffle (but contain same elements)
RemoteProcessor.sortCandidates(candidates, Config.CMSCommitMemberPreferencePolicy.random, null);
// Same elements, possibly different order
assertEquals(new HashSet<>(original), new HashSet<>(candidates));
assertEquals(original.size(), candidates.size());
}
@Test
public void testCandidatesLocalDeterministic()
{
// DC1 endpoints (local)
InetAddressAndPort dc1_1 = InetAddressAndPort.getByNameUnchecked("127.0.0.5");
InetAddressAndPort dc1_2 = InetAddressAndPort.getByNameUnchecked("127.0.0.1");
// DC2 endpoints (remote)
InetAddressAndPort dc2_1 = InetAddressAndPort.getByNameUnchecked("127.0.0.3");
InetAddressAndPort dc2_2 = InetAddressAndPort.getByNameUnchecked("127.0.0.2");
List<InetAddressAndPort> candidates = new ArrayList<>(Arrays.asList(dc2_1, dc1_1, dc2_2, dc1_2));
// Create a test locator
Location dc1 = new Location("DC1", "rack1");
Location dc2 = new Location("DC2", "rack1");
Map<InetAddressAndPort, Location> locationMap = new HashMap<>();
locationMap.put(dc1_1, dc1);
locationMap.put(dc1_2, dc1);
locationMap.put(dc2_1, dc2);
locationMap.put(dc2_2, dc2);
TestLocator locator = new TestLocator(dc1, locationMap);
// local_deterministic: local DC sorted first, then remote DC sorted
RemoteProcessor.sortCandidates(candidates, Config.CMSCommitMemberPreferencePolicy.local_deterministic, locator);
// Local DC first (sorted), then remote DC (sorted)
assertEquals(dc1_2, candidates.get(0)); // 127.0.0.1 (DC1)
assertEquals(dc1_1, candidates.get(1)); // 127.0.0.5 (DC1)
assertEquals(dc2_2, candidates.get(2)); // 127.0.0.2 (DC2)
assertEquals(dc2_1, candidates.get(3)); // 127.0.0.3 (DC2)
}
@Test
public void testCandidatesLocalRandom()
{
// DC1 endpoints (local)
InetAddressAndPort dc1_1 = InetAddressAndPort.getByNameUnchecked("127.0.0.1");
InetAddressAndPort dc1_2 = InetAddressAndPort.getByNameUnchecked("127.0.0.2");
// DC2 endpoints (remote)
InetAddressAndPort dc2_1 = InetAddressAndPort.getByNameUnchecked("127.0.0.3");
InetAddressAndPort dc2_2 = InetAddressAndPort.getByNameUnchecked("127.0.0.4");
List<InetAddressAndPort> candidates = new ArrayList<>(Arrays.asList(dc2_1, dc1_1, dc2_2, dc1_2));
// Create a test locator
Location dc1 = new Location("DC1", "rack1");
Location dc2 = new Location("DC2", "rack1");
Map<InetAddressAndPort, Location> locationMap = new HashMap<>();
locationMap.put(dc1_1, dc1);
locationMap.put(dc1_2, dc1);
locationMap.put(dc2_1, dc2);
locationMap.put(dc2_2, dc2);
TestLocator locator = new TestLocator(dc1, locationMap);
// local_random: local DC shuffled first, then remote DC shuffled
RemoteProcessor.sortCandidates(candidates, Config.CMSCommitMemberPreferencePolicy.local_random, locator);
// Local DC should be in first 2 positions, remote DC in last 2
Set<InetAddressAndPort> localDcEndpoints = new HashSet<>(Arrays.asList(dc1_1, dc1_2));
Set<InetAddressAndPort> remoteDcEndpoints = new HashSet<>(Arrays.asList(dc2_1, dc2_2));
assertTrue(localDcEndpoints.contains(candidates.get(0)));
assertTrue(localDcEndpoints.contains(candidates.get(1)));
assertTrue(remoteDcEndpoints.contains(candidates.get(2)));
assertTrue(remoteDcEndpoints.contains(candidates.get(3)));
}
@Test
public void testDeterministicPolicyProducesSameOrderAcrossCalls()
{
List<InetAddressAndPort> candidates1 = new ArrayList<>(Arrays.asList(
InetAddressAndPort.getByNameUnchecked("127.0.0.5"),
InetAddressAndPort.getByNameUnchecked("127.0.0.1"),
InetAddressAndPort.getByNameUnchecked("127.0.0.3")
));
List<InetAddressAndPort> candidates2 = new ArrayList<>(Arrays.asList(
InetAddressAndPort.getByNameUnchecked("127.0.0.3"),
InetAddressAndPort.getByNameUnchecked("127.0.0.5"),
InetAddressAndPort.getByNameUnchecked("127.0.0.1")
));
RemoteProcessor.sortCandidates(candidates1, Config.CMSCommitMemberPreferencePolicy.deterministic, null);
RemoteProcessor.sortCandidates(candidates2, Config.CMSCommitMemberPreferencePolicy.deterministic, null);
// Both should produce identical ordering regardless of initial order
assertEquals(candidates1, candidates2);
}
/**
* Test Locator implementation for unit testing
*/
private static class TestLocator extends Locator
{
private final Location localLocation;
private final Map<InetAddressAndPort, Location> locationMap;
public TestLocator(Location localLocation, Map<InetAddressAndPort, Location> locationMap)
{
super(null, null, () -> localLocation, null);
this.localLocation = localLocation;
this.locationMap = locationMap;
}
@Override
public Location local()
{
return localLocation;
}
@Override
public Location location(InetAddressAndPort endpoint)
{
return locationMap.getOrDefault(endpoint, Location.UNKNOWN);
}
}
}