From 44ee9d6167aa291a81c74b9f0df891ead83b7565 Mon Sep 17 00:00:00 2001 From: Jon Meredith Date: Fri, 12 Jun 2026 14:54:15 +0100 Subject: [PATCH 1/2] Unable to catch up TCM Log from peer with gaps in log sequence Patch by Jon Meredith and Marcus Eriksson; reviewed by Sam Tunnicliffe for CASSANDRA-21455 Co-authored-by: Marcus Eriksson --- CHANGES.txt | 1 + .../apache/cassandra/tcm/log/LogReader.java | 11 ++- .../test/log/CatchupViaSnapshotTest.java | 87 +++++++++++++++++++ .../tcm/log/LocalStorageLogStateTest.java | 48 ++++++++++ 4 files changed, 146 insertions(+), 1 deletion(-) create mode 100644 test/distributed/org/apache/cassandra/distributed/test/log/CatchupViaSnapshotTest.java diff --git a/CHANGES.txt b/CHANGES.txt index 735693f672..c2d495450a 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 6.0-alpha2 + * Fix TCM log catchup from peer with snapshots and gaps in the log sequence (CASSANDRA-21455) * Speed up nodetool doc generation by producing all command help in a single jvm (CASSANDRA-21444) * Always send TCM commit failures as Messaging failures (CASSANDRA-21457) * Fix ReadCommand serializedSize() using incorrect epoch (CASSANDRA-21438) diff --git a/src/java/org/apache/cassandra/tcm/log/LogReader.java b/src/java/org/apache/cassandra/tcm/log/LogReader.java index e7fcd2b308..4be476475f 100644 --- a/src/java/org/apache/cassandra/tcm/log/LogReader.java +++ b/src/java/org/apache/cassandra/tcm/log/LogReader.java @@ -79,7 +79,11 @@ public interface LogReader if (snapshotEpochs.size() <= 1 || !allowSnapshots) { entries = getEntries(startEpoch); - if (entries.isContinuous()) + // Only return entries directly if they are continuous AND they reach at least as far as any + // known snapshot. If a node caught up via a ForceSnapshot (which is not written to local_metadata_log), + // its log will have a gap between the old entries and the snapshot epoch. In that case entries may + // appear continuous up to their last epoch, but there is a snapshot beyond them that is needed. + if (entries.isContinuous() && (snapshotEpochs.isEmpty() || !snapshotEpochs.get(0).isAfter(entries.latestEpoch()))) return new LogState(null, entries.immutable()); else if (!allowSnapshots) throw new IllegalStateException("Can't construct a continuous log since " + startEpoch + " and inclusion of snapshots is disallowed"); @@ -136,6 +140,11 @@ public interface LogReader entries.add(entry); } + private Epoch latestEpoch() + { + return entries.isEmpty() ? since : entries.last().epoch; + } + private boolean isContinuous() { Epoch prev = since; diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/CatchupViaSnapshotTest.java b/test/distributed/org/apache/cassandra/distributed/test/log/CatchupViaSnapshotTest.java new file mode 100644 index 0000000000..4393c927a8 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/log/CatchupViaSnapshotTest.java @@ -0,0 +1,87 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.distributed.test.log; + +import org.junit.Test; + +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.distributed.api.IInvokableInstance; +import org.apache.cassandra.distributed.test.TestBaseImpl; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.tcm.ClusterMetadata; +import org.apache.cassandra.tcm.ClusterMetadataService; +import org.apache.cassandra.tcm.Epoch; + +import static org.junit.Assert.assertEquals; + +public class CatchupViaSnapshotTest extends TestBaseImpl +{ + @Test + public void catchupViaSnapshotTest() throws Exception + { + try (Cluster cluster = init(builder().withNodes(3) + .withConfig(c -> c.set("metadata_snapshot_frequency", "10")) + .start())) + { + cluster.schemaChange(withKeyspace("alter keyspace %s with replication = {'class':'SimpleStrategy', 'replication_factor':3}")); + cluster.schemaChange(withKeyspace("create table %s.tbl (id int primary key)")); + + // isolate node2 and node3 + cluster.filters().inbound().from(1).to(2,3).drop(); + + for (int i = 0; i < 30; i++) + cluster.coordinator(1).execute(withKeyspace("alter table %s.tbl with comment='abc" + i + "'"), ConsistencyLevel.ONE); + + // Snapshot needs to be the last transformation in the log + cluster.get(1).nodetoolResult("cms", "snapshot").asserts().success(); + + // allow node2 to catch up: + cluster.filters().reset(); + cluster.filters().inbound().from(1).to(3).drop(); + String node1Address = cluster.get(1).config().broadcastAddress().getHostString(); + + fetchLogFromPeerAsync(cluster.get(2), node1Address); + // allow node3 to catch up + cluster.filters().reset(); + String node2Address = cluster.get(2).config().broadcastAddress().getHostString(); + // by now node2 has an incomplete log, it only caught up from the snapshot above + // this means the log is continuous, but its current epoch is beyond the last entry in the log + fetchLogFromPeerAsync(cluster.get(3), node2Address); + + long expectedEpoch = cluster.get(1).callOnInstance(() -> ClusterMetadata.current().epoch.getEpoch()); + for (IInvokableInstance i : cluster) + assertEquals(expectedEpoch, (long)i.callOnInstance(() -> ClusterMetadata.current().epoch.getEpoch())); + } + } + + private static void fetchLogFromPeerAsync(IInvokableInstance i, String address) + { + i.runOnInstance(() -> { + try + { + ClusterMetadataService.instance().fetchLogFromPeerAsync(InetAddressAndPort.getByNameUnchecked(address), Epoch.create(30)).get(); + } + catch (Exception e) + { + throw new RuntimeException(e); + } + }); + } +} diff --git a/test/unit/org/apache/cassandra/tcm/log/LocalStorageLogStateTest.java b/test/unit/org/apache/cassandra/tcm/log/LocalStorageLogStateTest.java index 5bc6ec0fa8..3460468abb 100644 --- a/test/unit/org/apache/cassandra/tcm/log/LocalStorageLogStateTest.java +++ b/test/unit/org/apache/cassandra/tcm/log/LocalStorageLogStateTest.java @@ -21,13 +21,16 @@ package org.apache.cassandra.tcm.log; import java.io.IOException; import org.junit.BeforeClass; +import org.junit.Test; import org.apache.cassandra.ServerTestUtils; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.cql3.UntypedResultSet; import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.SystemKeyspace; import org.apache.cassandra.db.commitlog.CommitLog; import org.apache.cassandra.dht.Murmur3Partitioner; +import org.apache.cassandra.distributed.test.log.ClusterMetadataTestHelper; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.tcm.Epoch; import org.apache.cassandra.tcm.MetadataSnapshots; @@ -37,6 +40,8 @@ import org.apache.cassandra.tcm.transformations.TriggerSnapshot; import static org.apache.cassandra.cql3.QueryProcessor.executeInternal; import static org.apache.cassandra.db.SystemKeyspace.METADATA_LOG; import static org.apache.cassandra.schema.SchemaConstants.SYSTEM_KEYSPACE_NAME; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; public class LocalStorageLogStateTest extends LogStateTestBase { @@ -109,4 +114,47 @@ public class LocalStorageLogStateTest extends LogStateTestBase }; } + @Test + public void catchUpViaForceSnapshotLeavesGappedLog() throws Exception + { + // Simulates a non-CMS node that caught up via a ForceSnapshot (real or synthetic). + // LocalLog.append(LogState) converts the received baseState into a synthetic ForceSnapshot, + // which is processed but not written to local_metadata_log. + // The snapshot is stored in metadata_snapshots by MetadataSnapshotListener. + // The resulting on-disk state is: entries 1..X in local_metadata_log, snapshot at S sometime after X in + // metadata_snapshots, with no entries between X and S. + // Any peer requesting log since epoch <= X must receive the snapshot — not just the continuous + // run of entries up to X, which would leave it unable to advance past the gap. + MetadataSnapshots realSnapshots = new MetadataSnapshots.SystemKeyspaceMetadataSnapshots(); + LogStateSUT sut = getSystemUnderTest(realSnapshots); + sut.cleanup(); + ColumnFamilyStore.getIfExists(SYSTEM_KEYSPACE_NAME, SystemKeyspace.SNAPSHOT_TABLE_NAME) + .truncateBlockingWithoutSnapshot(); + + // insertRegularEntry inserts 2 entries on the first call (epoch 1 and 2) due to the + // Epoch.FIRST pre-init entry, then 1 per subsequent call. After 3 calls: epochs 1..4. + sut.insertRegularEntry(); + sut.insertRegularEntry(); + sut.insertRegularEntry(); + + // Simulate ForceSnapshot at epoch 50: snapshot stored, no intermediate log entries written + Epoch gapSnapshotEpoch = Epoch.create(50); + realSnapshots.storeSnapshot(ClusterMetadataTestHelper.minimalForTesting(Murmur3Partitioner.instance) + .forceEpoch(gapSnapshotEpoch)); + + // A peer at epoch 3 sees entry [4] which is continuous, but does not bridge to epoch 50. + // Must return the snapshot rather than just entry 4, which would leave the peer stuck. + LogState state = sut.getLogState(Epoch.create(3)); + assertEquals(gapSnapshotEpoch, state.baseState.epoch); + assertTrue(state.entries.isEmpty()); + + // A peer already at epoch 4 (the last log entry) previously got an empty response and stalled. + state = sut.getLogState(Epoch.create(4)); + assertEquals(gapSnapshotEpoch, state.baseState.epoch); + assertTrue(state.entries.isEmpty()); + + ColumnFamilyStore.getIfExists(SYSTEM_KEYSPACE_NAME, SystemKeyspace.SNAPSHOT_TABLE_NAME) + .truncateBlockingWithoutSnapshot(); + } + } From 9a896cbaeb233d5b36c5cc27c4c23057f39cd86a Mon Sep 17 00:00:00 2001 From: Jon Meredith Date: Fri, 12 Jun 2026 14:26:26 +0100 Subject: [PATCH 2/2] Add policy for selecting CMS host when submitting commit request Patch by Jon Meredith; reviewed by Sam Tunnicliffe for CASSANDRA-21456 --- .../org/apache/cassandra/config/Config.java | 28 ++- .../cassandra/config/DatabaseDescriptor.java | 16 ++ .../cassandra/service/StorageService.java | 13 ++ .../service/StorageServiceMBean.java | 12 ++ .../apache/cassandra/tcm/RemoteProcessor.java | 74 +++++++- .../config/DatabaseDescriptorRefTest.java | 1 + .../cassandra/tcm/RemoteProcessorTest.java | 166 ++++++++++++++++++ 7 files changed, 308 insertions(+), 2 deletions(-) diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index 3a485fddc5..aca2bbcdde 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -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; -} +} \ No newline at end of file diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 9ee09db707..e03f13723f 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -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; diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index d32d19460d..0b46109334 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -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> getConcurrency(List stageNames) { Stream stageStream = stageNames.isEmpty() ? stream(Stage.values()) : stageNames.stream().map(Stage::fromPoolName); diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java index b9f47b3b2d..34db8b4180 100644 --- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java +++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java @@ -1170,6 +1170,18 @@ public interface StorageServiceMBean extends NotificationEmitter List getAccordManagedKeyspaces(); List 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 { diff --git a/src/java/org/apache/cassandra/tcm/RemoteProcessor.java b/src/java/org/apache/cassandra/tcm/RemoteProcessor.java index b14e73feb8..69830b64da 100644 --- a/src/java/org/apache/cassandra/tcm/RemoteProcessor.java +++ b/src/java/org/apache/cassandra/tcm/RemoteProcessor.java @@ -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 candidates) + { + Config.CMSCommitMemberPreferencePolicy policy = DatabaseDescriptor.getCmsCommitMemberPreferencePolicy(); + sortCandidates(candidates, policy, DatabaseDescriptor.getLocator()); + } + + @VisibleForTesting + static void sortCandidates(List 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 candidates, Locator locator) + { + Location local = locator.local(); + + List localDc = new ArrayList<>(); + List 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 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) { diff --git a/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java b/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java index 73a58b75a4..33b5c6f513 100644 --- a/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java +++ b/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java @@ -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", diff --git a/test/unit/org/apache/cassandra/tcm/RemoteProcessorTest.java b/test/unit/org/apache/cassandra/tcm/RemoteProcessorTest.java index 61b1b37e51..85dcfac8a1 100644 --- a/test/unit/org/apache/cassandra/tcm/RemoteProcessorTest.java +++ b/test/unit/org/apache/cassandra/tcm/RemoteProcessorTest.java @@ -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 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 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 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 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 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 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 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 localDcEndpoints = new HashSet<>(Arrays.asList(dc1_1, dc1_2)); + Set 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 candidates1 = new ArrayList<>(Arrays.asList( + InetAddressAndPort.getByNameUnchecked("127.0.0.5"), + InetAddressAndPort.getByNameUnchecked("127.0.0.1"), + InetAddressAndPort.getByNameUnchecked("127.0.0.3") + )); + List 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 locationMap; + + public TestLocator(Location localLocation, Map 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); + } + } }