You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by bd...@apache.org on 2026/07/31 19:32:23 UTC
(cassandra) 03/03: CEP-58: Satellite replication quorums and failover states
This is an automated email from the ASF dual-hosted git repository.
bdeggleston pushed a commit to branch cep-58-satellite-datacenters
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 119b71981b727f6dbdb508f95bb6105bb97d5b6d
Author: Blake Eggleston <bl...@ultrablake.com>
AuthorDate: Wed Apr 1 16:14:38 2026 -0700
CEP-58: Satellite replication quorums and failover states
Implements SatelliteReplicationStrategy read/write/paxos plans with
quorum-of-quorums semantics, built on a new CompositeTracker (replacing
PerDcResponseTracker) that requires N of M child quorums to succeed.
Adds SatelliteFailoverState, tracking per-range NORMAL / TRANSITION_ACK /
TRANSITION state so replica layout and consistency requirements can vary
by failover phase. Paxos runs within the primary DC but includes
satellite/secondary summary nodes on prepare and commit via
SatellitePaxosParticipants; PaxosCommitRemoteMutationVerbHandler rejects
remote paxos commit mutations during TRANSITION_ACK so stale commits
can't leak into the other DCs during failover.
Patch by Blake Eggleston; reviewed by Francisco Guerrero and Ariel Weisberg for CASSANDRA-21106
---
.../config/CassandraRelevantProperties.java | 6 +
.../db/PaxosCommitRemoteMutationVerbHandler.java | 68 +
.../apache/cassandra/index/IndexStatusManager.java | 44 +-
.../locator/AbstractReplicationStrategy.java | 50 +-
.../apache/cassandra/locator/CompositeTracker.java | 168 +++
.../cassandra/locator/PerDcResponseTracker.java | 186 ---
.../org/apache/cassandra/locator/ReplicaPlan.java | 67 +-
.../org/apache/cassandra/locator/ReplicaPlans.java | 38 +-
.../cassandra/locator/SatelliteFailoverState.java | 272 ++++
.../locator/SatelliteReplicationStrategy.java | 1398 ++++++++++++++++++++
.../org/apache/cassandra/metrics/PaxosMetrics.java | 1 +
src/java/org/apache/cassandra/net/Verb.java | 3 +-
.../service/AbstractWriteResponseHandler.java | 2 +-
.../DatacenterSyncWriteResponseHandler.java | 2 +-
.../org/apache/cassandra/service/paxos/Paxos.java | 34 +-
.../cassandra/service/paxos/PaxosCommit.java | 197 ++-
.../cassandra/service/paxos/PaxosPrepare.java | 17 +-
.../service/paxos/SatellitePaxosParticipants.java | 98 ++
.../service/reads/range/RangeCommandIterator.java | 5 +-
.../service/reads/tracked/ReadReconciliations.java | 29 +-
.../unit/org/apache/cassandra/ServerTestUtils.java | 7 +-
.../db/virtual/MutationJournalTableTest.java | 6 +-
.../cassandra/index/IndexStatusManagerTest.java | 2 +-
.../cassandra/locator/CompositeTrackerTest.java | 427 ++++++
.../locator/PerDcResponseTrackerTest.java | 401 ------
.../cassandra/locator/SatelliteCommitPlanTest.java | 199 +++
.../locator/SatellitePaxosFailoverTest.java | 394 ++++++
.../cassandra/locator/SatelliteReadPlanTest.java | 291 ++++
.../locator/SatelliteReplicationStrategyTest.java | 125 +-
.../SatelliteReplicationStrategyTestBase.java | 166 +++
.../cassandra/locator/SatelliteWritePlanTest.java | 153 +++
.../service/paxos/AugmentedCommitTest.java | 217 +++
.../service/paxos/PaxosCommitPropertyTest.java | 2 +-
.../service/paxos/SatellitePaxosCommitTest.java | 286 ++++
34 files changed, 4612 insertions(+), 749 deletions(-)
diff --git a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
index b39a2f1cd1..116fce2c81 100644
--- a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
+++ b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
@@ -525,7 +525,13 @@ public enum CassandraRelevantProperties
/** Controls the maximum top-k limit for vector search */
SAI_VECTOR_SEARCH_MAX_TOP_K("cassandra.sai.vector_search.max_top_k", "1000"),
+ /**
+ * enables additional correctness checks in the satellite datacenter replication strategy
+ */
+ SATELLITE_REPLICATION_ADDITIONAL_CHECKS("cassandra.satellite_replication_addl_checks", "true"),
+
SCHEMA_PULL_INTERVAL_MS("cassandra.schema_pull_interval_ms", "60000"),
+
SCHEMA_UPDATE_HANDLER_FACTORY_CLASS("cassandra.schema.update_handler_factory.class"),
SEARCH_CONCURRENCY_FACTOR("cassandra.search_concurrency_factor", "1"),
diff --git a/src/java/org/apache/cassandra/db/PaxosCommitRemoteMutationVerbHandler.java b/src/java/org/apache/cassandra/db/PaxosCommitRemoteMutationVerbHandler.java
new file mode 100644
index 0000000000..adc29cf375
--- /dev/null
+++ b/src/java/org/apache/cassandra/db/PaxosCommitRemoteMutationVerbHandler.java
@@ -0,0 +1,68 @@
+/*
+ * 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.db;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.exceptions.RequestFailureReason;
+import org.apache.cassandra.locator.AbstractReplicationStrategy;
+import org.apache.cassandra.locator.SatelliteFailoverState;
+import org.apache.cassandra.locator.SatelliteReplicationStrategy;
+import org.apache.cassandra.net.IVerbHandler;
+import org.apache.cassandra.net.Message;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.tcm.ClusterMetadata;
+
+/**
+ * Verb handler for PAXOS2_COMMIT_REMOTE_REQ that wraps {@link MutationVerbHandler} with a failover state check
+ * for SatelliteReplicationStrategy keyspaces.
+ *
+ * PAXOS2_COMMIT_REMOTE_REQ sends a normal {@link Mutation} that doesn't indicate it came from a paxos commit
+ * so the standard MutationVerbHandler has no awareness that it originated from a paxos commit. This wrapper
+ * rejects mutations during {@link SatelliteFailoverState.State#TRANSITION_ACK} to prevent stale paxos commits
+ * from being applied to satellite/secondary DCs during failover.
+ */
+public class PaxosCommitRemoteMutationVerbHandler implements IVerbHandler<Mutation>
+{
+ public static final PaxosCommitRemoteMutationVerbHandler instance = new PaxosCommitRemoteMutationVerbHandler();
+ private static final Logger logger = LoggerFactory.getLogger(PaxosCommitRemoteMutationVerbHandler.class);
+
+ @Override
+ public void doVerb(Message<Mutation> message)
+ {
+ Mutation mutation = message.payload;
+ Keyspace keyspace = Keyspace.open(mutation.getKeyspaceName());
+ AbstractReplicationStrategy strategy = keyspace.getReplicationStrategy();
+
+ if (strategy instanceof SatelliteReplicationStrategy)
+ {
+ SatelliteReplicationStrategy srs = (SatelliteReplicationStrategy) strategy;
+ ClusterMetadata metadata = ClusterMetadata.current();
+ SatelliteFailoverState.FailoverInfo failoverInfo = srs.getFailoverInfo(mutation.key().getToken(), metadata);
+ if (failoverInfo.getState() == SatelliteFailoverState.State.TRANSITION_ACK)
+ {
+ logger.debug("Rejecting PAXOS2_COMMIT_REMOTE_REQ for {} during TRANSITION_ACK", mutation.getKeyspaceName());
+ MessagingService.instance().respondWithFailure(RequestFailureReason.UNKNOWN, message);
+ return;
+ }
+ }
+
+ MutationVerbHandler.instance.doVerb(message);
+ }
+}
diff --git a/src/java/org/apache/cassandra/index/IndexStatusManager.java b/src/java/org/apache/cassandra/index/IndexStatusManager.java
index 169980e9e0..fcda9f2db3 100644
--- a/src/java/org/apache/cassandra/index/IndexStatusManager.java
+++ b/src/java/org/apache/cassandra/index/IndexStatusManager.java
@@ -85,23 +85,21 @@ public class IndexStatusManager
/**
* Remove endpoints whose indexes are not queryable for the specified {@link Index.QueryPlan}.
*
- * @param liveEndpoints current live endpoints where non-queryable endpoints will be removed
- * @param keyspace to be queried
+ * @param liveEndpoints current live endpoints where non-queryable endpoints will be removed
+ * @param keyspace to be queried
* @param indexQueryPlan index query plan used in the read command
- * @param level consistency level of read command
*/
- public <E extends Endpoints<E>> E filterForQuery(E liveEndpoints, Keyspace keyspace, Index.QueryPlan indexQueryPlan, ConsistencyLevel level)
+ public <E extends Endpoints<E>> E filterForQuery(E liveEndpoints, String keyspace, Index.QueryPlan indexQueryPlan, Map<InetAddressAndPort, Index.Status> indexStatusMap)
{
// UNKNOWN states are transient/rare; only a few replicas should have this state at any time. See CASSANDRA-19400
Set<Replica> queryableNonSucceeded = new HashSet<>(4);
- Map<InetAddressAndPort, Index.Status> indexStatusMap = new HashMap<>();
E queryableEndpoints = liveEndpoints.filter(replica -> {
boolean allBuilt = true;
for (Index index : indexQueryPlan.getIndexes())
{
- Index.Status status = getIndexStatus(replica.endpoint(), keyspace.getName(), index.getIndexMetadata().name);
+ Index.Status status = getIndexStatus(replica.endpoint(), keyspace, index.getIndexMetadata().name);
if (!index.isQueryable(status))
{
indexStatusMap.put(replica.endpoint(), status);
@@ -122,6 +120,28 @@ public class IndexStatusManager
if (!queryableNonSucceeded.isEmpty() && queryableNonSucceeded.size() != queryableEndpoints.size())
queryableEndpoints = queryableEndpoints.sorted(Comparator.comparingInt(e -> queryableNonSucceeded.contains(e) ? 1 : -1));
+ return queryableEndpoints;
+ }
+
+
+ public void readFailureException(int filtered, Iterable<Replica> removed, Map<InetAddressAndPort, Index.Status> indexStatusMap, ConsistencyLevel level, int required)
+ {
+ Map<InetAddressAndPort, RequestFailureReason> failureReasons = new HashMap<>();
+ removed.forEach(replica -> {
+ Index.Status status = indexStatusMap.get(replica.endpoint());
+ if (status == Index.Status.FULL_REBUILD_STARTED)
+ failureReasons.put(replica.endpoint(), RequestFailureReason.INDEX_BUILD_IN_PROGRESS);
+ else
+ failureReasons.put(replica.endpoint(), RequestFailureReason.INDEX_NOT_AVAILABLE);
+ });
+
+ throw new ReadFailureException(level, filtered, required, false, failureReasons);
+ }
+
+ public <E extends Endpoints<E>> E filterForQueryOrThrow(E liveEndpoints, Keyspace keyspace, Index.QueryPlan indexQueryPlan, ConsistencyLevel level)
+ {
+ Map<InetAddressAndPort, Index.Status> indexStatusMap = new HashMap<>();
+ E queryableEndpoints = filterForQuery(liveEndpoints, keyspace.getName(), indexQueryPlan, indexStatusMap);
int initial = liveEndpoints.size();
int filtered = queryableEndpoints.size();
@@ -132,17 +152,7 @@ public class IndexStatusManager
int required = level.blockFor(keyspace.getReplicationStrategy());
if (required <= initial && required > filtered)
{
- Map<InetAddressAndPort, RequestFailureReason> failureReasons = new HashMap<>();
- liveEndpoints.without(queryableEndpoints.endpoints())
- .forEach(replica -> {
- Index.Status status = indexStatusMap.get(replica.endpoint());
- if (status == Index.Status.FULL_REBUILD_STARTED)
- failureReasons.put(replica.endpoint(), RequestFailureReason.INDEX_BUILD_IN_PROGRESS);
- else
- failureReasons.put(replica.endpoint(), RequestFailureReason.INDEX_NOT_AVAILABLE);
- });
-
- throw new ReadFailureException(level, filtered, required, false, failureReasons);
+ readFailureException(filtered, liveEndpoints.without(queryableEndpoints.endpoints()), indexStatusMap, level, level.blockFor(keyspace.getReplicationStrategy()));
}
}
diff --git a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java
index ffa75d7208..f43eff4d88 100644
--- a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java
+++ b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java
@@ -40,8 +40,8 @@ import com.google.common.base.Preconditions;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.db.Keyspace;
-import org.apache.cassandra.db.PartitionPosition;
import org.apache.cassandra.db.Mutation;
+import org.apache.cassandra.db.PartitionPosition;
import org.apache.cassandra.db.WriteType;
import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.dht.Range;
@@ -59,6 +59,7 @@ import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.DatacenterSyncWriteResponseHandler;
import org.apache.cassandra.service.DatacenterWriteResponseHandler;
import org.apache.cassandra.service.WriteResponseHandler;
+import org.apache.cassandra.service.paxos.Commit.Agreed;
import org.apache.cassandra.service.paxos.Paxos;
import org.apache.cassandra.service.reads.ReadCoordinator;
import org.apache.cassandra.service.reads.SpeculativeRetryPolicy;
@@ -69,6 +70,7 @@ import org.apache.cassandra.tcm.ownership.DataPlacement;
import org.apache.cassandra.transport.Dispatcher;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
+import org.apache.cassandra.utils.concurrent.Future;
import static org.apache.cassandra.locator.ReplicaLayout.forTokenWriteLiveAndDown;
@@ -471,23 +473,13 @@ public abstract class AbstractReplicationStrategy
return new CoordinationPlan.ForWrite(plan, tracker);
}
- protected ReplicaPlan.ForWrite createReplicaPlanForWrite(ClusterMetadata metadata,
- Keyspace keyspace,
- ConsistencyLevel consistencyLevel,
- Function<ClusterMetadata, ReplicaLayout.ForTokenWrite> liveAndDown,
- ReplicaPlans.Selector selector)
- {
- return ReplicaPlans.forWrite(metadata, keyspace, consistencyLevel, liveAndDown, selector);
- }
-
public CoordinationPlan.ForWriteWithIdeal planForWrite(ClusterMetadata metadata,
Keyspace keyspace,
ConsistencyLevel consistencyLevel,
Function<ClusterMetadata, ReplicaLayout.ForTokenWrite> liveAndDown,
ReplicaPlans.Selector selector)
{
- ReplicaPlan.ForWrite plan = createReplicaPlanForWrite(metadata, keyspace, consistencyLevel, liveAndDown, selector);
- ResponseTracker tracker = createTrackerForWrite(consistencyLevel, plan, plan.pending, metadata);
+ CoordinationPlan.ForWrite actual = planForWriteInternal(metadata, keyspace, consistencyLevel, liveAndDown, selector);
CoordinationPlan.ForWrite ideal = null;
ConsistencyLevel idealCL = DatabaseDescriptor.getIdealConsistencyLevel();
@@ -495,16 +487,15 @@ public abstract class AbstractReplicationStrategy
{
if (idealCL == consistencyLevel)
{
- ideal = new CoordinationPlan.ForWrite(plan, tracker);
+ ideal = actual;
}
else
{
- ideal = new CoordinationPlan.ForWrite(createReplicaPlanForWrite(metadata, keyspace, idealCL, liveAndDown, selector),
- createTrackerForWrite(idealCL, plan, plan.pending, metadata));
+ ideal = planForWriteInternal(metadata, keyspace, idealCL, liveAndDown, selector);
}
}
- return new CoordinationPlan.ForWriteWithIdeal(plan, tracker, ideal);
+ return new CoordinationPlan.ForWriteWithIdeal(actual.replicas(), actual.responses(), ideal);
}
public CoordinationPlan.ForWriteWithIdeal planForWrite(ClusterMetadata metadata,
@@ -693,6 +684,27 @@ public abstract class AbstractReplicationStrategy
(cm) -> Paxos.Participants.get(cm, table, actualToken, consistencyForConsensus));
}
+ /**
+ * Hook for replication strategies to send additional mutations alongside a paxos commit.
+ * Called from PaxosCommit.start() after local synchronous execution for tracked keyspaces.
+ *
+ * If the method doesn't return null, the returned future is composed with the paxos consensus:
+ * onDone fires only after both the paxos quorum decision AND this future complete, or after
+ * one of them fails.
+ */
+ public Future<Void> sendPaxosCommitMutations(Agreed commit, boolean isUrgent)
+ {
+ return null;
+ }
+
+ /**
+ * Check whether paxos operations should be rejected for the given token.
+ */
+ public boolean shouldRejectPaxos(Token token)
+ {
+ return false;
+ }
+
/**
* Create ResponseTracker for write operation based on consistency level.
*/
@@ -749,7 +761,7 @@ public abstract class AbstractReplicationStrategy
}
case EACH_QUORUM:
- return createPerDcTracker(plan, pending, metadata);
+ return createCompositeTracker(plan, pending, metadata);
default:
throw new UnsupportedOperationException("Unsupported consistency level for writes: " + cl);
@@ -759,7 +771,7 @@ public abstract class AbstractReplicationStrategy
/**
* Create per-datacenter tracker for EACH_QUORUM.
*/
- private ResponseTracker createPerDcTracker(ReplicaPlan.ForWrite plan, Endpoints<?> pending, ClusterMetadata metadata)
+ private ResponseTracker createCompositeTracker(ReplicaPlan.ForWrite plan, Endpoints<?> pending, ClusterMetadata metadata)
{
Map<String, ResponseTracker> trackerPerDc = new HashMap<>();
Locator locator = metadata.locator;
@@ -807,6 +819,6 @@ public abstract class AbstractReplicationStrategy
}
}
- return new PerDcResponseTracker(trackerPerDc, locator);
+ return new CompositeTracker(trackerPerDc.size(), trackerPerDc.values());
}
}
diff --git a/src/java/org/apache/cassandra/locator/CompositeTracker.java b/src/java/org/apache/cassandra/locator/CompositeTracker.java
new file mode 100644
index 0000000000..0eb236d6e0
--- /dev/null
+++ b/src/java/org/apache/cassandra/locator/CompositeTracker.java
@@ -0,0 +1,168 @@
+/*
+ * 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.locator;
+
+import java.util.Collection;
+import java.util.function.ToIntFunction;
+
+/**
+ * Composite response tracker: broadcasts responses to all child trackers and succeeds when at least
+ * `blockFor` children are individually successful, or fails when `failures > count - blockFor`
+ *
+ * <ul>
+ * <li>{@code blockFor == children.length} gives AND semantics (all must succeed)</li>
+ * <li>{@code blockFor == quorum(children.length)} gives majority-quorum semantics</li>
+ * </ul>
+ */
+public class CompositeTracker implements ResponseTracker
+{
+ private final ResponseTracker[] children;
+ private final int blockFor;
+
+ public CompositeTracker(int blockFor, ResponseTracker... children)
+ {
+ if (children == null || children.length == 0)
+ throw new IllegalArgumentException("children cannot be null or empty");
+ if (blockFor < 1 || blockFor > children.length)
+ throw new IllegalArgumentException("blockFor must be between 1 and " + children.length);
+
+ this.children = children;
+ this.blockFor = blockFor;
+ }
+
+ public CompositeTracker(int blockFor, Collection<ResponseTracker> children)
+ {
+ this(blockFor, children.toArray(ResponseTracker[]::new));
+ }
+
+ public static int quorum(int count)
+ {
+ return (count / 2) + 1;
+ }
+
+ @Override
+ public boolean isSuccessful()
+ {
+ int successful = 0;
+ for (ResponseTracker child : children)
+ if (child.isSuccessful())
+ successful++;
+ return successful >= blockFor;
+ }
+
+ @Override
+ public void onResponse(InetAddressAndPort from)
+ {
+ for (ResponseTracker child : children)
+ child.onResponse(from);
+ }
+
+ @Override
+ public void onFailure(InetAddressAndPort from)
+ {
+ for (ResponseTracker child : children)
+ child.onFailure(from);
+ }
+
+ @Override
+ public boolean isComplete()
+ {
+ if (isSuccessful())
+ return true;
+
+ // Count children that have definitively failed
+ int failed = 0;
+ for (ResponseTracker child : children)
+ if (child.isComplete() && !child.isSuccessful())
+ failed++;
+ int maxPossible = children.length - failed;
+ return maxPossible < blockFor;
+ }
+
+ @Override
+ public int required()
+ {
+ return count(ResponseTracker::required);
+ }
+
+ @Override
+ public int received()
+ {
+ return count(ResponseTracker::received);
+ }
+
+ @Override
+ public int failures()
+ {
+ return count(ResponseTracker::failures);
+ }
+
+ @Override
+ public boolean countsTowardQuorum(InetAddressAndPort from)
+ {
+ for (ResponseTracker child : children)
+ if (child.countsTowardQuorum(from))
+ return true;
+ return false;
+ }
+
+ @Override
+ public boolean isPending(InetAddressAndPort from)
+ {
+ for (ResponseTracker child : children)
+ if (child.isPending(from))
+ return true;
+ return false;
+ }
+
+ @Override
+ public int totalContacts()
+ {
+ return count(ResponseTracker::totalContacts);
+ }
+
+ @Override
+ public int pendingContacts()
+ {
+ return count(ResponseTracker::pendingContacts);
+ }
+
+ @Override
+ public String toString()
+ {
+ return String.format("CompositeTracker[children=%d, blockFor=%d]", children.length, blockFor);
+ }
+
+ private int count(ToIntFunction<ResponseTracker> function)
+ {
+ int total = 0;
+ for (ResponseTracker child : children)
+ total += function.applyAsInt(child);
+ return total;
+ }
+
+ @Override
+ public ResponseTracker resetCopy()
+ {
+ ResponseTracker[] resetTrackers = new ResponseTracker[children.length];
+ for (int i=0; i<children.length; i++)
+ resetTrackers[i] = children[i].resetCopy();
+ return new CompositeTracker(blockFor, resetTrackers);
+ }
+}
diff --git a/src/java/org/apache/cassandra/locator/PerDcResponseTracker.java b/src/java/org/apache/cassandra/locator/PerDcResponseTracker.java
deleted file mode 100644
index 7b0466e2ae..0000000000
--- a/src/java/org/apache/cassandra/locator/PerDcResponseTracker.java
+++ /dev/null
@@ -1,186 +0,0 @@
-/*
- * 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.locator;
-
-import java.util.Map;
-import java.util.function.ToIntFunction;
-
-import org.apache.cassandra.exceptions.RequestFailureReason;
-
-/**
- * Per-datacenter response tracker that requires ALL datacenters to independently reach quorum.
- * <p>
- * Composes multiple ResponseTrackers (one per datacenter) and delegates operations
- * to the appropriate tracker based on endpoint datacenter. Used for EACH_QUORUM consistency.
- * <p>
- * The composed trackers can be any ResponseTracker implementation:
- * <ul>
- * <li>{@link SimpleResponseTracker} for basic quorum tracking</li>
- * <li>{@link WriteResponseTracker} for writes with pending replicas (double-count model)</li>
- * </ul>
- * <p>
- * Thread-safe through delegation to thread-safe ResponseTracker implementations.
- */
-public class PerDcResponseTracker implements ResponseTracker
-{
- private final Map<String, ResponseTracker> trackerPerDc;
- private final Locator locator;
-
- /**
- * Create per-DC tracker with pre-built trackers for each datacenter.
- *
- * @param trackerPerDc map of datacenter name to tracker (must be non-empty)
- * @param locator for looking up datacenter from endpoint
- */
- public PerDcResponseTracker(Map<String, ResponseTracker> trackerPerDc, Locator locator)
- {
- if (trackerPerDc == null || trackerPerDc.isEmpty())
- throw new IllegalArgumentException("trackerPerDc cannot be null or empty");
- if (locator == null)
- throw new IllegalArgumentException("locator cannot be null");
-
- this.locator = locator;
- this.trackerPerDc = trackerPerDc;
- }
-
- private int count(ToIntFunction<ResponseTracker> getter)
- {
- int total = 0;
- for (ResponseTracker tracker : trackerPerDc.values())
- total += getter.applyAsInt(tracker);
- return total;
- }
-
- @Override
- public void onResponse(InetAddressAndPort from)
- {
- ResponseTracker tracker = getTrackerForEndpoint(from);
- if (tracker != null)
- tracker.onResponse(from);
- }
-
- @Override
- public void onFailure(InetAddressAndPort from, RequestFailureReason reason)
- {
- ResponseTracker tracker = getTrackerForEndpoint(from);
- if (tracker != null)
- tracker.onFailure(from, reason);
- }
-
- @Override
- public boolean isComplete()
- {
- // Complete when ALL DCs are complete (either success or definite failure)
- for (ResponseTracker tracker : trackerPerDc.values())
- {
- if (!tracker.isComplete())
- return false;
- }
- return true;
- }
-
- @Override
- public boolean isSuccessful()
- {
- // Successful only if ALL DCs are successful
- for (ResponseTracker tracker : trackerPerDc.values())
- {
- if (!tracker.isSuccessful())
- return false;
- }
- return true;
- }
-
- @Override
- public int required()
- {
- return count(ResponseTracker::required);
- }
-
- @Override
- public int received()
- {
- return count(ResponseTracker::received);
- }
-
- @Override
- public int failures()
- {
- return count(ResponseTracker::failures);
- }
-
- @Override
- public boolean countsTowardQuorum(InetAddressAndPort from)
- {
- String dc = locator.location(from).datacenter;
- if (dc == null)
- return false;
- ResponseTracker tracker = trackerPerDc.get(dc);
- return tracker != null && tracker.countsTowardQuorum(from);
- }
-
- private ResponseTracker getTrackerForEndpoint(InetAddressAndPort from)
- {
- String dc = locator.location(from).datacenter;
- return trackerPerDc.get(dc);
- }
-
- /**
- * @return the tracker for the specified datacenter, or null if not tracked
- */
- public ResponseTracker getTrackerForDc(String datacenter)
- {
- return trackerPerDc.get(datacenter);
- }
-
- @Override
- public String toString()
- {
- return String.format("PerDcResponseTracker[datacenters=%s, trackerPerDc=%s]",
- trackerPerDc.keySet(), trackerPerDc);
- }
-
- @Override
- public boolean isPending(InetAddressAndPort from)
- {
- String dc = locator.location(from).datacenter;
- if (dc == null)
- return false;
- ResponseTracker tracker = trackerPerDc.get(dc);
- return tracker != null && tracker.isPending(from);
- }
-
- @Override
- public int totalRequired()
- {
- return count(ResponseTracker::totalRequired);
- }
-
- @Override
- public int totalContacts()
- {
- return count(ResponseTracker::totalContacts);
- }
-
- @Override
- public int pendingContacts()
- {
- return count(ResponseTracker::pendingContacts);
- }
-}
diff --git a/src/java/org/apache/cassandra/locator/ReplicaPlan.java b/src/java/org/apache/cassandra/locator/ReplicaPlan.java
index ee8198aac3..e3ceabfa65 100644
--- a/src/java/org/apache/cassandra/locator/ReplicaPlan.java
+++ b/src/java/org/apache/cassandra/locator/ReplicaPlan.java
@@ -134,11 +134,12 @@ public interface ReplicaPlan<E extends Endpoints<E>, P extends ReplicaPlan<E, P>
E contacts,
E liveAndDown,
Function<ClusterMetadata, P> recompute,
- Epoch epoch)
+ Epoch epoch,
+ int readQuorum)
{
super(keyspace, replicationStrategy, consistencyLevel, contacts, liveAndDown, recompute, epoch);
this.candidates = candidates;
- this.readQuorum = consistencyLevel.blockFor(replicationStrategy);
+ this.readQuorum = readQuorum;
}
public int readQuorum() { return readQuorum; }
@@ -208,15 +209,30 @@ public interface ReplicaPlan<E extends Endpoints<E>, P extends ReplicaPlan<E, P>
EndpointsForToken liveAndDown,
Function<ClusterMetadata, ReplicaPlan.ForTokenRead> recompute,
Function<ReplicaPlan<?, ?>, ReplicaPlan.ForWrite> repairPlan,
- Epoch epoch)
+ Epoch epoch,
+ int readQuorum)
{
- super(keyspace, replicationStrategy, consistencyLevel, candidates, contacts, liveAndDown, recompute, epoch);
+ super(keyspace, replicationStrategy, consistencyLevel, candidates, contacts, liveAndDown, recompute, epoch, readQuorum);
this.repairPlan = repairPlan;
}
+
+ public ForTokenRead(Keyspace keyspace,
+ AbstractReplicationStrategy replicationStrategy,
+ ConsistencyLevel consistencyLevel,
+ EndpointsForToken candidates,
+ EndpointsForToken contacts,
+ EndpointsForToken liveAndDown,
+ Function<ClusterMetadata, ReplicaPlan.ForTokenRead> recompute,
+ Function<ReplicaPlan<?, ?>, ReplicaPlan.ForWrite> repairPlan,
+ Epoch epoch)
+ {
+ this(keyspace, replicationStrategy, consistencyLevel, candidates, contacts, liveAndDown, recompute, repairPlan, epoch, consistencyLevel.blockFor(replicationStrategy));
+ }
+
public ForTokenRead withContacts(EndpointsForToken newContacts)
{
- ForTokenRead res = new ForTokenRead(keyspace, replicationStrategy, consistencyLevel, candidates, newContacts, liveAndDown, recompute, repairPlan, epoch);
+ ForTokenRead res = new ForTokenRead(keyspace, replicationStrategy, consistencyLevel, candidates, newContacts, liveAndDown, recompute, repairPlan, epoch, readQuorum);
res.contacted.addAll(contacted);
return res;
}
@@ -246,14 +262,31 @@ public interface ReplicaPlan<E extends Endpoints<E>, P extends ReplicaPlan<E, P>
int vnodeCount,
Function<ClusterMetadata, ReplicaPlan.ForRangeRead> recompute,
BiFunction<ReplicaPlan<?, ?>, Token, ReplicaPlan.ForWrite> repairPlan,
- Epoch epoch)
+ Epoch epoch,
+ int readQuorum)
{
- super(keyspace, replicationStrategy, consistencyLevel, candidates, contact, liveAndDown, recompute, epoch);
+ super(keyspace, replicationStrategy, consistencyLevel, candidates, contact, liveAndDown, recompute, epoch, readQuorum);
this.range = range;
this.vnodeCount = vnodeCount;
this.repairPlan = repairPlan;
}
+
+ public ForRangeRead(Keyspace keyspace,
+ AbstractReplicationStrategy replicationStrategy,
+ ConsistencyLevel consistencyLevel,
+ AbstractBounds<PartitionPosition> range,
+ EndpointsForRange candidates,
+ EndpointsForRange contact,
+ EndpointsForRange liveAndDown,
+ int vnodeCount,
+ Function<ClusterMetadata, ReplicaPlan.ForRangeRead> recompute,
+ BiFunction<ReplicaPlan<?, ?>, Token, ReplicaPlan.ForWrite> repairPlan,
+ Epoch epoch)
+ {
+ this(keyspace, replicationStrategy, consistencyLevel, range, candidates, contact, liveAndDown, vnodeCount, recompute, repairPlan, epoch, consistencyLevel.blockFor(replicationStrategy));
+ }
+
public AbstractBounds<PartitionPosition> range() { return range; }
/**
@@ -263,7 +296,7 @@ public interface ReplicaPlan<E extends Endpoints<E>, P extends ReplicaPlan<E, P>
public ForRangeRead withContacts(EndpointsForRange newContact)
{
- ForRangeRead res = new ForRangeRead(keyspace, replicationStrategy, consistencyLevel, range, readCandidates(), newContact, liveAndDown, vnodeCount, recompute, repairPlan, epoch);
+ ForRangeRead res = new ForRangeRead(keyspace, replicationStrategy, consistencyLevel, range, readCandidates(), newContact, liveAndDown, vnodeCount, recompute, repairPlan, epoch, readQuorum);
res.contacted.addAll(contacted);
return res;
}
@@ -289,13 +322,27 @@ public interface ReplicaPlan<E extends Endpoints<E>, P extends ReplicaPlan<E, P>
EndpointsForRange contact,
EndpointsForRange liveAndDown,
int vnodeCount,
- Epoch epoch)
+ Epoch epoch,
+ int readQuorum)
{
// A FullRangeRead plan, as part of a top K query, is not recomputed to check that it still applies should
// the epoch change during the course of query execution so no recomputation function is supplied. Likewise,
// no read repair is expected to be performed during this type of query so a null is also used in place of a
// function for calculating the repair plan.
- super(keyspace, replicationStrategy, consistencyLevel, range, candidates, contact, liveAndDown, vnodeCount, null, null, epoch);
+ super(keyspace, replicationStrategy, consistencyLevel, range, candidates, contact, liveAndDown, vnodeCount, null, null, epoch, readQuorum);
+ }
+
+ public ForFullRangeRead(Keyspace keyspace,
+ AbstractReplicationStrategy replicationStrategy,
+ ConsistencyLevel consistencyLevel,
+ AbstractBounds<PartitionPosition> range,
+ EndpointsForRange candidates,
+ EndpointsForRange contact,
+ EndpointsForRange liveAndDown,
+ int vnodeCount,
+ Epoch epoch)
+ {
+ this(keyspace, replicationStrategy, consistencyLevel, range, candidates, contact, liveAndDown, vnodeCount, epoch, consistencyLevel.blockFor(replicationStrategy));
}
@Override
diff --git a/src/java/org/apache/cassandra/locator/ReplicaPlans.java b/src/java/org/apache/cassandra/locator/ReplicaPlans.java
index c5c78623c1..c2f7623554 100644
--- a/src/java/org/apache/cassandra/locator/ReplicaPlans.java
+++ b/src/java/org/apache/cassandra/locator/ReplicaPlans.java
@@ -821,7 +821,7 @@ public class ReplicaPlans
{
E replicas = consistencyLevel.isDatacenterLocal() ? liveNaturalReplicas.filter(InOurDc.replicas()) : liveNaturalReplicas;
- return indexQueryPlan != null ? IndexStatusManager.instance.filterForQuery(replicas, keyspace, indexQueryPlan, consistencyLevel) : replicas;
+ return indexQueryPlan != null ? IndexStatusManager.instance.filterForQueryOrThrow(replicas, keyspace, indexQueryPlan, consistencyLevel) : replicas;
}
private static <E extends Endpoints<E>> E contactForEachQuorumRead(Locator locator, NetworkTopologyStrategy replicationStrategy, E candidates)
@@ -982,6 +982,31 @@ public class ReplicaPlans
return forRead(metadata, keyspace, tableId, token, indexQueryPlan, consistencyLevel, retry, coordinator, true);
}
+ public static ReplicaPlan.ForTokenRead forRead(ClusterMetadata metadata,
+ Keyspace keyspace,
+ TableId tableId,
+ Token token,
+ AbstractReplicationStrategy replicationStrategy,
+ ReplicaLayout.ForTokenRead forTokenReadLiveAndDown,
+ ReplicaLayout.ForTokenRead forTokenReadLive,
+ @Nullable Index.QueryPlan indexQueryPlan,
+ ConsistencyLevel consistencyLevel,
+ SpeculativeRetryPolicy retry,
+ ReadCoordinator coordinator,
+ boolean throwOnInsufficientLiveReplicas)
+ {
+ EndpointsForToken candidates = candidatesForRead(keyspace, indexQueryPlan, consistencyLevel, forTokenReadLive.all());
+ EndpointsForToken contacts = contactForRead(metadata.locator, replicationStrategy, consistencyLevel, retry.equals(AlwaysSpeculativeRetryPolicy.INSTANCE), candidates);
+
+ if (throwOnInsufficientLiveReplicas)
+ assureSufficientLiveReplicasForRead(metadata.locator, replicationStrategy, consistencyLevel, contacts);
+
+ return new ReplicaPlan.ForTokenRead(keyspace, replicationStrategy, consistencyLevel, candidates, contacts, forTokenReadLiveAndDown.all(),
+ (newClusterMetadata) -> forRead(newClusterMetadata, keyspace, tableId, token, indexQueryPlan, consistencyLevel, retry, coordinator, false),
+ (self) -> forReadRepair(self, metadata, keyspace, tableId, consistencyLevel, token, FailureDetector.isReplicaAlive, coordinator),
+ metadata.epoch);
+ }
+
private static ReplicaPlan.ForTokenRead forRead(ClusterMetadata metadata,
Keyspace keyspace,
TableId tableId,
@@ -995,16 +1020,7 @@ public class ReplicaPlans
AbstractReplicationStrategy replicationStrategy = keyspace.getReplicationStrategy();
ReplicaLayout.ForTokenRead forTokenReadLiveAndDown = ReplicaLayout.forTokenReadSorted(metadata, keyspace, replicationStrategy, tableId, token, coordinator);
ReplicaLayout.ForTokenRead forTokenReadLive = forTokenReadLiveAndDown.filter(FailureDetector.isReplicaAlive);
- EndpointsForToken candidates = candidatesForRead(keyspace, indexQueryPlan, consistencyLevel, forTokenReadLive.all());
- EndpointsForToken contacts = contactForRead(metadata.locator, replicationStrategy, consistencyLevel, retry.equals(AlwaysSpeculativeRetryPolicy.INSTANCE), candidates);
-
- if (throwOnInsufficientLiveReplicas)
- assureSufficientLiveReplicasForRead(metadata.locator, replicationStrategy, consistencyLevel, contacts);
-
- return new ReplicaPlan.ForTokenRead(keyspace, replicationStrategy, consistencyLevel, candidates, contacts, forTokenReadLiveAndDown.all(),
- (newClusterMetadata) -> forRead(newClusterMetadata, keyspace, tableId, token, indexQueryPlan, consistencyLevel, retry, coordinator, false),
- (self) -> forReadRepair(self, metadata, keyspace, tableId, consistencyLevel, token, FailureDetector.isReplicaAlive, coordinator),
- metadata.epoch);
+ return forRead(metadata, keyspace, tableId, token, replicationStrategy, forTokenReadLiveAndDown, forTokenReadLive, indexQueryPlan, consistencyLevel, retry, coordinator, throwOnInsufficientLiveReplicas);
}
/**
diff --git a/src/java/org/apache/cassandra/locator/SatelliteFailoverState.java b/src/java/org/apache/cassandra/locator/SatelliteFailoverState.java
new file mode 100644
index 0000000000..7661af87a6
--- /dev/null
+++ b/src/java/org/apache/cassandra/locator/SatelliteFailoverState.java
@@ -0,0 +1,272 @@
+/*
+ * 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.locator;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import com.google.common.base.Preconditions;
+
+import org.apache.cassandra.dht.Range;
+import org.apache.cassandra.dht.Token;
+
+/**
+ * Failover state management for satellite datacenter replication
+ *
+ * Tracks per-range failover state to support different replica layouts / consistency requirements
+ * during different failover states to maintain correctness
+ *
+ * NOTE: Initial implementation stores state in memory. Future work will integrate
+ * with TCM for cluster-wide coordination and persistence.
+ */
+public class SatelliteFailoverState
+{
+ /**
+ * Failover state for a range or keyspace.
+ *
+ * Each state corresponds to a specific phase in the failover process as defined
+ * in CEP-58, with different consistency level requirements.
+ */
+ public enum State
+ {
+ /**
+ * Normal operation: Primary DC + satellite active.
+ * Read/Write CL: QUORUM_OF_QUORUMS (primary + satellite OR secondary)
+ */
+ NORMAL,
+
+ /**
+ * First stage of failover. Waits for a QoQ to acknowledge new state before moving to
+ * TRANSITION state. During TRANSITION_ACK, coordinators will not start paxos operations, and the
+ * primary transition state machine will not transition to the transition state until it confirms
+ * that a QoQ nodes for a range are also on TRANSITION_ACK. This temporary gap in paxos availability
+ * prevents the different full dcs from performing conflicting paxos operations concurrently.
+ */
+ TRANSITION_ACK,
+
+ /**
+ *
+ * Currently syncing to new primary, once sync is complete, failover state will transition to NORMAL
+ * for this range
+ */
+ TRANSITION,
+ }
+
+ /**
+ * Failover state with DC context.
+ *
+ * Tracks which DC we're failing over from. The target DC is always the current
+ * primary DC configured in the replication strategy, enabling failover between
+ * any configured DCs via schema change (ALTER KEYSPACE ... WITH replication = {'primary': 'DC2'}).
+ */
+ public static class FailoverInfo
+ {
+ private final State state;
+ private final String fromDC; // DC we're failing over from (old primary), null for NORMAL
+
+ /**
+ * Create failover info.
+ *
+ * @param state The failover state
+ * @param fromDC The old primary DC we're failing from (null for NORMAL state)
+ */
+ private FailoverInfo(State state, String fromDC)
+ {
+ this.state = state;
+ this.fromDC = fromDC;
+
+ switch (state)
+ {
+ case NORMAL:
+ Preconditions.checkArgument(fromDC == null);
+ break;
+ case TRANSITION_ACK:
+ case TRANSITION:
+ Preconditions.checkArgument(fromDC != null);
+ break;
+ default:
+ throw new IllegalArgumentException("Unknown state: " + state);
+ }
+
+ }
+
+ /**
+ * Get the failover state.
+ */
+ public State getState()
+ {
+ return state;
+ }
+
+ /**
+ * Get the DC we're failing over FROM (old primary).
+ * Returns null for NORMAL state.
+ */
+ public String getFromDC()
+ {
+ return fromDC;
+ }
+
+ /**
+ * Create NORMAL state (no failover in progress).
+ */
+ public static FailoverInfo normal()
+ {
+ return new FailoverInfo(State.NORMAL, null);
+ }
+
+ /**
+ * Create TRANSITION_ACK state.
+ *
+ * @param fromDC The old primary DC we're failing over from
+ */
+ public static FailoverInfo transitionAck(String fromDC)
+ {
+ return new FailoverInfo(State.TRANSITION_ACK, fromDC);
+ }
+
+ /**
+ * Create TRANSITION state
+ *
+ * @param fromDC The old primary DC we're transferring from
+ */
+ public static FailoverInfo transition(String fromDC)
+ {
+ return new FailoverInfo(State.TRANSITION, fromDC);
+ }
+
+ @Override
+ public String toString()
+ {
+ if (fromDC == null)
+ return state.toString();
+ return String.format("%s(from=%s)", state, fromDC);
+ }
+
+ public boolean isTransitioning()
+ {
+ switch (state)
+ {
+ case TRANSITION:
+ case TRANSITION_ACK:
+ return true;
+ default:
+ return false;
+ }
+ }
+ }
+
+ /**
+ * Container for per-range failover state.
+ *
+ * Stores failover state indexed by token range, enabling different ranges
+ * to be in different failover states (though initial implementation uses
+ * uniform state across all ranges).
+ *
+ * NOTE: This is in-memory for initial implementation. Future work will
+ * store this in TCM for persistence and cluster-wide coordination.
+ */
+ public static class FailoverStateMap
+ {
+ private final Map<Range<Token>, FailoverInfo> perRangeState;
+ private final FailoverInfo defaultState; // Used for ranges not explicitly set
+
+ private FailoverStateMap(Map<Range<Token>, FailoverInfo> perRangeState,
+ FailoverInfo defaultState)
+ {
+ this.perRangeState = Collections.unmodifiableMap(new HashMap<>(perRangeState));
+ this.defaultState = defaultState;
+ }
+
+ private FailoverStateMap(FailoverInfo globalState)
+ {
+ this.perRangeState = Collections.emptyMap();
+ this.defaultState = globalState;
+ }
+
+ public FailoverInfo getFailoverInfo(Range<Token> range)
+ {
+ // FIXME: This is just meant for testing at the moment and assumes that we never query ranges that don't
+ // align with the state map. Fix as part of the failover process implementation
+ return getFailoverInfo(range.right);
+ }
+
+ public FailoverInfo getFailoverInfo(Token token)
+ {
+ for (Map.Entry<Range<Token>, FailoverInfo> entry : perRangeState.entrySet())
+ {
+ if (entry.getKey().contains(token))
+ return entry.getValue();
+ }
+ return defaultState;
+ }
+
+ public static class Builder
+ {
+ private final Map<Range<Token>, FailoverInfo> state = new HashMap<>();
+ private FailoverInfo defaultState = FailoverInfo.normal();
+
+ /**
+ * Set the default state for ranges not explicitly set.
+ */
+ public Builder setDefault(FailoverInfo info)
+ {
+ this.defaultState = info;
+ return this;
+ }
+
+ /**
+ * Set failover state for a specific range.
+ */
+ public Builder setState(Range<Token> range, FailoverInfo info)
+ {
+ state.put(range, info);
+ return this;
+ }
+
+ /**
+ * Set failover state for multiple ranges.
+ */
+ public Builder setState(Collection<Range<Token>> ranges, FailoverInfo info)
+ {
+ ranges.forEach(r -> state.put(r, info));
+ return this;
+ }
+
+ public FailoverStateMap build()
+ {
+ return new FailoverStateMap(state, defaultState);
+ }
+ }
+
+ public static Builder builder()
+ {
+ return new Builder();
+ }
+
+ /**
+ * Convenience factory: Create state map with single state for all ranges.
+ */
+ public static FailoverStateMap allRanges(FailoverInfo info)
+ {
+ return new FailoverStateMap(info);
+ }
+ }
+}
diff --git a/src/java/org/apache/cassandra/locator/SatelliteReplicationStrategy.java b/src/java/org/apache/cassandra/locator/SatelliteReplicationStrategy.java
index ce351a1236..7b9149843a 100644
--- a/src/java/org/apache/cassandra/locator/SatelliteReplicationStrategy.java
+++ b/src/java/org/apache/cassandra/locator/SatelliteReplicationStrategy.java
@@ -17,6 +17,8 @@
*/
package org.apache.cassandra.locator;
+import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
@@ -24,21 +26,61 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.function.BiFunction;
+import java.util.function.Function;
+import java.util.function.Predicate;
+import javax.annotation.Nullable;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Multimap;
+import com.google.common.collect.Sets;
+
+import org.agrona.collections.IntHashSet;
+import org.agrona.collections.Object2IntHashMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.cassandra.config.CassandraRelevantProperties;
+import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.Mutation;
+import org.apache.cassandra.db.PartitionPosition;
import org.apache.cassandra.db.guardrails.Guardrails;
+import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.exceptions.RequestFailure;
+import org.apache.cassandra.exceptions.UnavailableException;
+import org.apache.cassandra.gms.FailureDetector;
+import org.apache.cassandra.index.Index;
+import org.apache.cassandra.index.IndexStatusManager;
+import org.apache.cassandra.net.Message;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.net.NoPayload;
+import org.apache.cassandra.net.RequestCallback;
+import org.apache.cassandra.net.Verb;
+import org.apache.cassandra.replication.MutationId;
+import org.apache.cassandra.replication.MutationTrackingService;
+import org.apache.cassandra.schema.KeyspaceMetadata;
import org.apache.cassandra.schema.ReplicationType;
import org.apache.cassandra.schema.SchemaConstants;
+import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.ClientWarn;
+import org.apache.cassandra.service.paxos.Commit.Agreed;
+import org.apache.cassandra.service.paxos.Paxos;
+import org.apache.cassandra.service.paxos.SatellitePaxosParticipants;
+import org.apache.cassandra.service.reads.AlwaysSpeculativeRetryPolicy;
+import org.apache.cassandra.service.reads.ReadCoordinator;
+import org.apache.cassandra.service.reads.SpeculativeRetryPolicy;
import org.apache.cassandra.service.replication.migration.KeyspaceMigrationInfo;
import org.apache.cassandra.service.replication.migration.MutationTrackingMigrationState;
import org.apache.cassandra.tcm.ClusterMetadata;
@@ -48,6 +90,9 @@ import org.apache.cassandra.tcm.membership.Directory;
import org.apache.cassandra.tcm.ownership.DataPlacement;
import org.apache.cassandra.tcm.ownership.ReplicaGroups;
import org.apache.cassandra.tcm.ownership.VersionedEndpoints;
+import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.concurrent.AsyncPromise;
+import org.apache.cassandra.utils.concurrent.Future;
import static java.lang.String.format;
@@ -79,6 +124,8 @@ public class SatelliteReplicationStrategy extends AbstractReplicationStrategy
{
private static final Logger logger = LoggerFactory.getLogger(SatelliteReplicationStrategy.class);
+ private static final boolean ADDL_CHECKS_ENABLED = CassandraRelevantProperties.SATELLITE_REPLICATION_ADDITIONAL_CHECKS.getBoolean();
+
private static final String PRIMARY_DC_KEY = "primary";
private static final String SATELLITE_KEY_PATTERN = ".satellite.";
private static final String DISABLED_KEY_SUFFIX = ".disabled";
@@ -95,6 +142,17 @@ public class SatelliteReplicationStrategy extends AbstractReplicationStrategy
private final ReplicationFactor aggregateRf;
+ /**
+ * Per-range failover state.
+ * Volatile for visibility across threads.
+ *
+ * NOTE: In-memory for initial implementation. Future work will pull this
+ * from TCM (Transactional Cluster Metadata) for persistence and coordination.
+ *
+ * Initialized to NORMAL state for all ranges.
+ */
+ private volatile SatelliteFailoverState.FailoverStateMap failoverState;
+
private static class SatelliteInfo
{
public final String parentDC;
@@ -257,6 +315,9 @@ public class SatelliteReplicationStrategy extends AbstractReplicationStrategy
this.aggregateRf = ReplicationFactor.withTransient(totalReplicas, totalTransient);
+ this.failoverState = SatelliteFailoverState.FailoverStateMap.allRanges(
+ SatelliteFailoverState.FailoverInfo.normal());
+
if (disabledDCs.isEmpty())
logger.info("Configured satellite datacenter replication for keyspace {} with full datacenters {} (primary: {}), satellites {}",
keyspaceName, fullDCs, primaryDC, satellites);
@@ -339,6 +400,14 @@ public class SatelliteReplicationStrategy extends AbstractReplicationStrategy
"Use 'AND replication_type = tracked' when creating/updating the keyspace");
}
+ // must use paxos v2
+ if (DatabaseDescriptor.getPaxosVariant() != Config.PaxosVariant.v2)
+ {
+ throw new ConfigurationException(
+ getClass().getSimpleName() + " requires paxos_variant=v2. " +
+ "Set 'paxos_variant: v2' in cassandra.yaml");
+ }
+
// must have witness replicas enabled
if (!DatabaseDescriptor.isTransientReplicationEnabled())
{
@@ -497,6 +566,52 @@ public class SatelliteReplicationStrategy extends AbstractReplicationStrategy
return disabledDCs.contains(dc);
}
+ @Override
+ public Paxos.Participants paxosParticipants(ClusterMetadata metadata,
+ TableMetadata table,
+ Token token,
+ ConsistencyLevel consistencyForConsensus,
+ Predicate<Replica> isReplicaAlive)
+ {
+ // Reject paxos operations during TRANSITION_ACK to prevent conflicting paxos operations
+ // across different full DCs. The temporary gap in paxos availability ensures that the old
+ // primary has no in-flight proposals before the new primary begins serving paxos operations.
+ SatelliteFailoverState.FailoverInfo failoverInfo = getFailoverInfo(token, metadata);
+ if (failoverInfo.getState() == SatelliteFailoverState.State.TRANSITION_ACK)
+ throw UnavailableException.create(consistencyForConsensus, 1, 0);
+
+ KeyspaceMetadata keyspaceMetadata = metadata.schema.getKeyspaceMetadata(table.keyspace);
+ ReplicaLayout.ForTokenWrite liveAndDownLayout = ReplicaLayout.forTokenWriteLiveAndDown(metadata, keyspaceMetadata, token);
+
+ // Paxos consensus operates entirely within the primary DC
+ Predicate<Replica> inPrimaryDC = rp -> metadata.locator.location(rp.endpoint()).datacenter.equals(primaryDC);
+ ReplicaLayout.ForTokenWrite primaryAll = liveAndDownLayout.filter(inPrimaryDC);
+
+ // Satellite/secondary DC endpoints for reads during prepare and writes during commit.
+ // Only include the primary DC's satellite and other full DCs — not satellites of other DCs.
+ String primarySatellite = getSatelliteForDC(primaryDC);
+ Predicate<Replica> isParticipatingNonPrimary = rp -> {
+ String dc = metadata.locator.location(rp.endpoint()).datacenter;
+ if (dc.equals(primaryDC))
+ return false;
+ if (dc.equals(primarySatellite))
+ return true;
+ return fullDCs.containsKey(dc);
+ };
+ EndpointsForToken satelliteEndpoints = liveAndDownLayout.all().filter(isParticipatingNonPrimary);
+
+ EndpointsForToken live = liveAndDownLayout.all().filter(isReplicaAlive);
+
+ return new SatellitePaxosParticipants(metadata.epoch,
+ Keyspace.open(table.keyspace),
+ consistencyForConsensus,
+ primaryAll,
+ primaryAll, // DC filtering happens earlier
+ live,
+ (cm) -> Paxos.Participants.get(cm, table, token, consistencyForConsensus),
+ satelliteEndpoints);
+ }
+
@Override
public boolean hasSameSettings(AbstractReplicationStrategy other)
{
@@ -513,4 +628,1287 @@ public class SatelliteReplicationStrategy extends AbstractReplicationStrategy
primaryDC.equals(otherStrategy.primaryDC) &&
disabledDCs.equals(otherStrategy.disabledDCs);
}
+
+ private AbstractReplicationStrategy strategy(ClusterMetadata metadata)
+ {
+ return metadata.schema.getKeyspaceMetadata(keyspaceName).replicationStrategy;
+ }
+
+ static abstract class CoordinationPlanner<E extends Endpoints<E>, L extends ReplicaLayout<E>, P extends ReplicaPlan<E, P>>
+ {
+ final ClusterMetadata metadata;
+ final Keyspace keyspace;
+ final ConsistencyLevel cl;
+ final SatelliteReplicationStrategy strategy;
+ final String primary;
+ final String satellite;
+ final List<String> dcs;
+ final L liveAndDownLayout;
+ final L liveLayout;
+ final Map<String, L> liveAndDownLayouts;
+ final Map<String, L> liveLayouts;
+ final Map<String, E> liveAndDownEndpoints;
+ final Map<String, E> liveEndpoints;
+
+ public CoordinationPlanner(ClusterMetadata metadata, Keyspace keyspace, ConsistencyLevel cl, SatelliteReplicationStrategy strategy, String primary, L liveAndDownLayout)
+ {
+ this.metadata = metadata;
+ this.keyspace = keyspace;
+ this.cl = cl;
+ this.strategy = strategy;
+ this.primary = primary;
+ this.satellite = strategy.getSatelliteForDC(primary);
+
+ this.dcs = createDcList();
+
+ this.liveAndDownLayout = filterLayout(liveAndDownLayout, rp -> this.dcs.contains(metadata.locator.location(rp.endpoint()).datacenter));
+ this.liveLayout = filterLayout(this.liveAndDownLayout, FailureDetector.isReplicaAlive);
+
+ this.liveAndDownLayouts = Maps.newHashMapWithExpectedSize(dcs.size());
+ this.liveLayouts = Maps.newHashMapWithExpectedSize(dcs.size());
+
+ this.liveAndDownEndpoints = Maps.newHashMapWithExpectedSize(dcs.size());
+ this.liveEndpoints = Maps.newHashMapWithExpectedSize(dcs.size());
+
+ populateDcLayouts();
+ }
+
+ AbstractReplicationStrategy strategy(ClusterMetadata metadata)
+ {
+ return metadata.schema.getKeyspaceMetadata(keyspace.getName()).replicationStrategy;
+ }
+
+ List<String> createDcList()
+ {
+ List<String> results = new ArrayList<>();
+ if (!strategy.disabledDCs.contains(primary))
+ results.add(primary);
+
+ if (satellite != null && !strategy.disabledDCs.contains(satellite))
+ results.add(satellite);
+
+ for (String dc : strategy.fullDCs.keySet())
+ {
+ if (dc.equals(primary) || strategy.disabledDCs.contains(dc))
+ continue;
+ results.add(dc);
+ }
+
+ Preconditions.checkState(!results.isEmpty(), "No DCs available for request (primary=%s, all disabled?)", primary);
+ return results;
+ }
+
+ void populateDcLayouts()
+ {
+ for (String dc : dcs)
+ {
+ Predicate<Replica> dcFilter = rp -> metadata.locator.location(rp.endpoint()).datacenter.equals(dc);
+ L liveAndDown = filterLayout(liveAndDownLayout, dcFilter);
+ L live = filterLayout(liveLayout, dcFilter);
+
+ liveAndDownLayouts.put(dc, liveAndDown);
+ liveLayouts.put(dc, live);
+
+ liveAndDownEndpoints.put(dc, liveAndDown.natural());
+ liveEndpoints.put(dc, live.natural());
+ }
+
+ if (ADDL_CHECKS_ENABLED)
+ Preconditions.checkState(liveAndDownLayouts.size() == dcs.size(), "populateDcLayouts: liveAndDownLayouts.size()=%s != dcs.size()=%s", liveAndDownLayouts.size(), dcs.size());
+ }
+
+ ResponseTrackerBuilder<E, L, P> createResponseTrackerBuilder()
+ {
+ switch (cl)
+ {
+ case ONE:
+ case LOCAL_ONE:
+ case NODE_LOCAL:
+ case ANY:
+ case TWO:
+ case THREE:
+ case ALL:
+ return new ResponseTrackerBuilder.Simple<>(this);
+
+ case QUORUM:
+ case LOCAL_QUORUM:
+ // LOCAL_QUORUM an QUORUM mean the same thing in SRS
+ return new ResponseTrackerBuilder.Quorum<>(this);
+
+ case EACH_QUORUM:
+ return new ResponseTrackerBuilder.EachQuorum<>(this);
+
+ default:
+ throw new IllegalStateException("Unsupported consistency level: " + cl);
+ }
+ }
+
+ ResponseTracker createResponseTracker()
+ {
+ ResponseTrackerBuilder<E, L, P> builder = createResponseTrackerBuilder();
+ return builder.build();
+ }
+
+ abstract L filterLayout(L layout, Predicate<Replica> predicate);
+ abstract ResponseTracker createDcResponseTracker(String dc);
+
+ abstract P recomputePlan(ClusterMetadata metadata);
+ abstract P createReplicaPlan();
+
+ interface IndexReadExceptionCheck<E extends Endpoints<E>>
+ {
+ void maybeThrowIndexReadException(int initial, int filtered, Map<String, E> candidates, Map<InetAddressAndPort, Index.Status> indexStatusMap);
+ }
+
+ static class ForWrite extends CoordinationPlanner<EndpointsForToken, ReplicaLayout.ForTokenWrite, ReplicaPlan.ForWrite>
+ {
+ final ReplicaPlans.Selector selector;
+
+ public ForWrite(ClusterMetadata metadata, Keyspace keyspace, ConsistencyLevel cl, SatelliteReplicationStrategy strategy, String primary, ReplicaLayout.ForTokenWrite liveAndDownLayout, ReplicaPlans.Selector selector)
+ {
+ super(metadata, keyspace, cl, strategy, primary, liveAndDownLayout);
+ this.selector = selector;
+ if (ADDL_CHECKS_ENABLED)
+ Preconditions.checkState(!strategy.disabledDCs.contains(primary));
+ }
+
+ @Override
+ ResponseTracker createDcResponseTracker(String dc)
+ {
+ // this basically hardcodes ReplicaPlans.writeAll
+ Predicate<InetAddressAndPort> dcFilter = ep -> metadata.locator.location(ep).datacenter.equals(dc);
+
+ ReplicaLayout.ForTokenWrite dcLayout = liveAndDownLayouts.get(dc);
+ Preconditions.checkNotNull(dcLayout);
+
+ int blockFor = strategy.calculateQuorum(dc);
+ int totalContacts = dcLayout.all().size();
+
+ // Only count pending replicas that are actually in the selected contacts
+ int pendingInDc = dcLayout.pending().size();
+
+ if (pendingInDc == 0)
+ {
+ return new SimpleResponseTracker(blockFor, totalContacts, dcFilter);
+ }
+ else
+ {
+ int naturalContacts = totalContacts - pendingInDc;
+ int totalBlockFor = blockFor + pendingInDc;
+ Predicate<InetAddressAndPort> isPending = ep -> dcLayout.pending.endpoints().contains(ep);
+ return new WriteResponseTracker(blockFor, totalBlockFor,
+ naturalContacts, pendingInDc,
+ isPending, dcFilter);
+ }
+ }
+
+ @Override
+ ReplicaPlan.ForWrite recomputePlan(ClusterMetadata metadata)
+ {
+ Token token = liveAndDownLayout.token();
+ return strategy(metadata).planForWrite(metadata, keyspace, cl,
+ cm -> ReplicaLayout.forTokenWriteLiveAndDown(cm, keyspace, token),
+ selector).replicas();
+ }
+
+ @Override
+ ReplicaLayout.ForTokenWrite filterLayout(ReplicaLayout.ForTokenWrite layout, Predicate<Replica> predicate)
+ {
+ return layout.filter(predicate);
+ }
+
+ private ReplicaLayout.ForTokenWrite mergeLayouts(Collection<ReplicaLayout.ForTokenWrite> layouts)
+ {
+ Preconditions.checkArgument(layouts != null && !layouts.isEmpty());
+ ReplicaCollection.Builder<EndpointsForToken> naturalBuilder = null;
+ ReplicaCollection.Builder<EndpointsForToken> pendingBuilder = null;
+
+ for (ReplicaLayout.ForTokenWrite layout : layouts)
+ {
+ EndpointsForToken natural = layout.natural();
+ EndpointsForToken pending = layout.pending();
+ if (naturalBuilder == null)
+ {
+ naturalBuilder = natural.newBuilder(natural.size());
+ pendingBuilder = pending.newBuilder(pending.size());
+ }
+
+ naturalBuilder.addAll(natural);
+ pendingBuilder.addAll(pending);
+ }
+
+ return new ReplicaLayout.ForTokenWrite(strategy, naturalBuilder.build(), pendingBuilder.build());
+ }
+
+ @Override
+ ReplicaPlan.ForWrite createReplicaPlan()
+ {
+
+ ReplicaLayout.ForTokenWrite liveAndDownLayout = mergeLayouts(liveAndDownLayouts.values());
+ ReplicaLayout.ForTokenWrite liveLayout = mergeLayouts(liveLayouts.values());
+ EndpointsForToken contacts = selector.select(cl, liveAndDownLayout, liveLayout);
+
+ return new ReplicaPlan.ForWrite(keyspace,
+ strategy,
+ cl,
+ liveAndDownLayout.pending(),
+ liveAndDownLayout.all(),
+ liveLayout.all(),
+ contacts,
+ this::recomputePlan,
+ metadata.epoch);
+ }
+ }
+
+ static abstract class ForRead<E extends Endpoints<E>, L extends ReplicaLayout<E>, P extends ReplicaPlan.ForRead<E, P>> extends CoordinationPlanner<E, L, P>
+ {
+ final @Nullable Index.QueryPlan indexQueryPlan;
+ final boolean alwaysSpeculate;
+
+ public ForRead(ClusterMetadata metadata, Keyspace keyspace, ConsistencyLevel cl, SatelliteReplicationStrategy strategy, String primary, L liveAndDownLayout, @Nullable Index.QueryPlan indexQueryPlan, boolean alwaysSpeculate)
+ {
+ super(metadata, keyspace, cl, strategy, primary, liveAndDownLayout);
+ this.indexQueryPlan = indexQueryPlan;
+ this.alwaysSpeculate = alwaysSpeculate;
+ }
+
+ @Override
+ ResponseTracker createDcResponseTracker(String dc)
+ {
+ Predicate<InetAddressAndPort> dcFilter = ep -> metadata.locator.location(ep).datacenter.equals(dc);
+
+ L dcLayout = liveAndDownLayouts.get(dc);
+ int blockFor = strategy.calculateQuorum(dc);
+ int totalContacts = dcLayout.all().size();
+
+ return new SimpleResponseTracker(blockFor, totalContacts, dcFilter);
+ }
+
+ Map<String, E> candidates(IndexReadExceptionCheck<E> check)
+ {
+ if (indexQueryPlan == null)
+ return liveEndpoints;
+
+ Map<String, E> candidates = new HashMap<>();
+ Map<InetAddressAndPort, Index.Status> indexStatusMap = new HashMap<>();
+ int initial = 0;
+ int filtered = 0;
+ for (Map.Entry<String, E> entry : liveEndpoints.entrySet())
+ {
+ E liveEndpoints = entry.getValue();
+ E queryableEndpoints = IndexStatusManager.instance.filterForQuery(liveEndpoints, keyspace.getName(), indexQueryPlan, indexStatusMap);
+ initial += liveEndpoints.size();
+ filtered += queryableEndpoints.size();
+ candidates.put(entry.getKey(), queryableEndpoints);
+ }
+
+ if (initial != filtered)
+ check.maybeThrowIndexReadException(initial, filtered, candidates, indexStatusMap);
+
+ return candidates;
+ }
+
+ @Override
+ ResponseTracker createResponseTracker()
+ {
+ ResponseTracker tracker = super.createResponseTracker();
+
+ // read trackers don't wait on pending nodes
+ Preconditions.checkState(tracker.pendingContacts() == 0);
+ return tracker;
+ }
+
+
+ abstract P replicaPlanBuilder(E candidates, E contacts, E liveAndDown, int quorumSize);
+
+ ReadReplicaPlanBuilder<E, L, P> replicaPlanBuilder()
+ {
+ switch (cl)
+ {
+ case ONE:
+ case LOCAL_ONE:
+ case NODE_LOCAL:
+ case ANY:
+ case TWO:
+ case THREE:
+ case ALL:
+ return new ReadReplicaPlanBuilder.Simple<>(this);
+
+ case QUORUM:
+ case LOCAL_QUORUM:
+ case EACH_QUORUM:
+ return new ReadReplicaPlanBuilder.DcQuorum<>(this);
+
+ default:
+ throw new IllegalStateException("Unsupported consistency level: " + cl);
+ }
+ }
+
+ @Override
+ P createReplicaPlan()
+ {
+ ReadReplicaPlanBuilder<E, L, P> builder = replicaPlanBuilder();
+ return builder.build();
+ }
+ }
+
+ static class ForTokenRead extends ForRead<EndpointsForToken, ReplicaLayout.ForTokenRead, ReplicaPlan.ForTokenRead>
+ {
+ static final Function<ReplicaPlan<?, ?>, ReplicaPlan.ForWrite> READ_REPAIR_PLAN = (ReplicaPlan<?, ?> self) -> {throw new IllegalStateException("Cannot create read repair plans for SatelliteDatacenterReplicationStrategy");};
+ final Token token;
+ final TableId tableId;
+ final SpeculativeRetryPolicy retry;
+ final ReadCoordinator coordinator;
+
+ public ForTokenRead(ClusterMetadata metadata, Keyspace keyspace, Token token, ConsistencyLevel cl, SatelliteReplicationStrategy strategy, String primary, ReplicaLayout.ForTokenRead liveAndDownLayout, @Nullable Index.QueryPlan indexQueryPlan, SpeculativeRetryPolicy retry, TableId tableId, ReadCoordinator coordinator)
+ {
+ super(metadata, keyspace, cl, strategy, primary, liveAndDownLayout, indexQueryPlan, retry == AlwaysSpeculativeRetryPolicy.INSTANCE);
+ this.token = token;
+ this.tableId = tableId;
+ this.retry = retry;
+ this.coordinator = coordinator;
+ }
+
+ @Override
+ ReplicaLayout.ForTokenRead filterLayout(ReplicaLayout.ForTokenRead layout, Predicate<Replica> predicate)
+ {
+ return layout.filter(predicate);
+ }
+
+ @Override
+ ReplicaPlan.ForTokenRead recomputePlan(ClusterMetadata metadata)
+ {
+ return strategy(metadata).planForTokenRead(metadata, keyspace, tableId, token,
+ indexQueryPlan, cl, retry, coordinator).replicas();
+ }
+
+ @Override
+ ReplicaPlan.ForTokenRead replicaPlanBuilder(EndpointsForToken candidates, EndpointsForToken contacts, EndpointsForToken liveAndDown, int quorumSize)
+ {
+ return new ReplicaPlan.ForTokenRead(keyspace, strategy, cl,
+ candidates, contacts, liveAndDown,
+ this::recomputePlan,
+ READ_REPAIR_PLAN,
+ metadata.epoch,
+ quorumSize);
+ }
+ }
+
+ static class ForRangeRead extends ForRead<EndpointsForRange, ReplicaLayout.ForRangeRead, ReplicaPlan.ForRangeRead>
+ {
+ static final BiFunction<ReplicaPlan<?, ?>, Token, ReplicaPlan.ForWrite> READ_REPAIR_PLAN = (ReplicaPlan<?, ?> self, Token token) -> {throw new IllegalStateException("Cannot create read repair plans for SatelliteDatacenterReplicationStrategy");};
+ final AbstractBounds<PartitionPosition> range;
+ final int vnodeCount;
+ final TableId tableId;
+
+ public ForRangeRead(ClusterMetadata metadata, Keyspace keyspace, AbstractBounds<PartitionPosition> range, int vnodeCount, ConsistencyLevel cl, SatelliteReplicationStrategy strategy, String primary, ReplicaLayout.ForRangeRead liveAndDownLayout, @Nullable Index.QueryPlan indexQueryPlan, TableId tableId)
+ {
+ super(metadata, keyspace, cl, strategy, primary, liveAndDownLayout, indexQueryPlan, false);
+ this.range = range;
+ this.vnodeCount = vnodeCount;
+ this.tableId = tableId;
+ }
+
+ @Override
+ ReplicaLayout.ForRangeRead filterLayout(ReplicaLayout.ForRangeRead layout, Predicate<Replica> predicate)
+ {
+ return layout.filter(predicate);
+ }
+
+ @Override
+ ReplicaPlan.ForRangeRead recomputePlan(ClusterMetadata metadata)
+ {
+ return strategy(metadata).planForRangeRead(metadata, keyspace, tableId,
+ indexQueryPlan, cl, range, vnodeCount).replicas();
+ }
+
+ @Override
+ ReplicaPlan.ForRangeRead replicaPlanBuilder(EndpointsForRange candidates, EndpointsForRange contacts, EndpointsForRange liveAndDown, int quorumSize)
+ {
+ return new ReplicaPlan.ForRangeRead(keyspace, strategy, cl, range,
+ candidates, contacts, liveAndDown, vnodeCount,
+ this::recomputePlan,
+ READ_REPAIR_PLAN,
+ metadata.epoch,
+ quorumSize);
+ }
+ }
+ }
+
+ static abstract class ResponseTrackerBuilder<E extends Endpoints<E>, L extends ReplicaLayout<E>, P extends ReplicaPlan<E, P>>
+ {
+ final CoordinationPlanner<E, L, P> planner;
+
+ public ResponseTrackerBuilder(CoordinationPlanner<E, L, P> planner)
+ {
+ this.planner = planner;
+ }
+
+ Map<String, ResponseTracker> responseTrackersForPrimary()
+ {
+ Map<String, ResponseTracker> result = Maps.newHashMapWithExpectedSize(planner.dcs.size());
+
+ for (String dc : planner.dcs)
+ result.put(dc, planner.createDcResponseTracker(dc));
+
+ return result;
+ }
+
+ abstract ResponseTracker aggregateTrackers(Map<String, ResponseTracker> dcTrackers);
+
+ ResponseTracker build()
+ {
+ Map<String, ResponseTracker> dcTrackers = responseTrackersForPrimary();
+ return aggregateTrackers(dcTrackers);
+ }
+
+ static class Simple<E extends Endpoints<E>, L extends ReplicaLayout<E>, P extends ReplicaPlan<E, P>> extends ResponseTrackerBuilder<E, L, P>
+ {
+ public Simple(CoordinationPlanner<E, L, P> planner)
+ {
+ super(planner);
+ }
+
+ @Override
+ ResponseTracker aggregateTrackers(Map<String, ResponseTracker> dcTrackers)
+ {
+ int totalContacts = 0;
+ int totalPending = 0;
+
+
+ List<ResponseTracker> trackers = Lists.newArrayList();
+ Set<String> blockingDatacenters = Sets.newHashSet();
+
+ for (Map.Entry<String, ResponseTracker> entry : dcTrackers.entrySet())
+ {
+ String dc = entry.getKey();
+ ResponseTracker tracker = entry.getValue();
+ if (planner.cl == ConsistencyLevel.ALL || dc.equals(planner.primary) || dc.equals(planner.satellite))
+ {
+ totalContacts += tracker.totalContacts();
+ totalPending += tracker.pendingContacts();
+ trackers.add(tracker);
+ blockingDatacenters.add(dc);
+ }
+ }
+
+ Predicate<InetAddressAndPort> filter = i -> {
+ String dc = planner.metadata.locator.location(i).datacenter;
+ return dc != null && blockingDatacenters.contains(dc);
+ };
+
+ Preconditions.checkState(!trackers.isEmpty(), "No blocking DCs for CL %s (primary=%s, satellite=%s)", planner.cl, planner.primary, planner.satellite);
+
+
+ if (planner.cl == ConsistencyLevel.ALL)
+ return new SimpleResponseTracker(totalContacts, totalContacts, filter);
+
+ int blockFor = planner.cl.blockFor(planner.strategy);
+ int totalBlockFor = blockFor + totalPending;
+
+ if (blockFor > totalContacts)
+ throw UnavailableException.create(planner.cl, blockFor, totalBlockFor, totalContacts - totalPending, totalContacts);
+
+ if (totalPending == 0)
+ return new SimpleResponseTracker(blockFor, totalContacts, filter);
+
+ Predicate<InetAddressAndPort> pending = i -> {
+ for (ResponseTracker tracker : trackers)
+ {
+ if (tracker.isPending(i))
+ return true;
+ }
+ return false;
+ };
+ return new WriteResponseTracker(blockFor, totalBlockFor, totalContacts - totalPending, totalPending, pending, filter);
+ }
+ }
+
+ static class Quorum<E extends Endpoints<E>, L extends ReplicaLayout<E>, P extends ReplicaPlan<E, P>> extends ResponseTrackerBuilder<E, L, P>
+ {
+ public Quorum(CoordinationPlanner<E, L, P> planner)
+ {
+ super(planner);
+ }
+
+ @Override
+ ResponseTracker aggregateTrackers(Map<String, ResponseTracker> dcTrackers)
+ {
+ return new CompositeTracker(CompositeTracker.quorum(dcTrackers.size()), new ArrayList<>(dcTrackers.values()));
+ }
+ }
+
+ static class EachQuorum<E extends Endpoints<E>, L extends ReplicaLayout<E>, P extends ReplicaPlan<E, P>> extends ResponseTrackerBuilder<E, L, P>
+ {
+ public EachQuorum(CoordinationPlanner<E, L, P> planner)
+ {
+ super(planner);
+ }
+
+ @Override
+ ResponseTracker aggregateTrackers(Map<String, ResponseTracker> dcTrackers)
+ {
+ return new CompositeTracker(dcTrackers.size(), new ArrayList<>(dcTrackers.values()));
+ }
+ }
+ }
+
+ static abstract class ReadReplicaPlanBuilder<E extends Endpoints<E>, L extends ReplicaLayout<E>, P extends ReplicaPlan.ForRead<E, P>>
+ implements CoordinationPlanner.IndexReadExceptionCheck<E>
+ {
+
+ private static <E extends Endpoints<E>> Iterable<Replica> removedSupplier(Map<String, E> initial, Map<String, E> filtered)
+ {
+ return () -> initial.keySet().stream().map(dc -> {
+ Set<InetAddressAndPort> f = filtered.containsKey(dc) ? filtered.get(dc).endpoints() : Collections.emptySet();
+ return initial.get(dc).without(f);
+ }).flatMap(AbstractReplicaCollection::stream).iterator();
+ }
+
+ final CoordinationPlanner.ForRead<E, L, P> planner;
+
+ public ReadReplicaPlanBuilder(CoordinationPlanner.ForRead<E, L, P> planner)
+ {
+ this.planner = planner;
+ }
+
+ abstract boolean canSelectReplicasFromDc(String dc, int liveNodes);
+ abstract int minBlockFor();
+ abstract int eligibleCandidates(Map<String, E> candidates);
+
+ abstract boolean haveSufficientContacts(int contactedReplicas, int contactedDcs);
+ abstract boolean haveSufficientContactsForDc(String dc, int contactedReplicas, int contactedDcs, int dcContacts);
+ abstract boolean haveSufficientLiveNodes(int contactedReplicas, int contactedDcs);
+
+ private E allLiveAndDown()
+ {
+ E liveAndDownEndpoints = planner.liveAndDownLayout.all();
+ ReplicaCollection.Builder<E> liveAndDownBuilder = liveAndDownEndpoints.newBuilder(liveAndDownEndpoints.size());
+
+ for (String dc : planner.dcs)
+ {
+ E endpoints = planner.liveAndDownEndpoints.get(dc);
+ if (endpoints == null)
+ continue;
+
+ liveAndDownBuilder.addAll(endpoints);
+ }
+
+ return liveAndDownBuilder.build();
+ }
+
+ private E allCandidates(Map<String, E> candidates)
+ {
+ E liveEndpoints = planner.liveLayout.all();
+ ReplicaCollection.Builder<E> candidateBuilder = liveEndpoints.newBuilder(liveEndpoints.size());
+
+ for (String dc : planner.dcs)
+ {
+ // TODO: make sure the full replica is added if this is the dc it's from
+ E dcCandidates = candidates.get(dc);
+ if (dcCandidates == null || !canSelectReplicasFromDc(dc, dcCandidates.size()))
+ continue;
+
+ candidateBuilder.addAll(dcCandidates);
+ }
+
+ return candidateBuilder.build();
+ }
+
+ private E contacts(Map<String, E> candidates)
+ {
+ E liveEndpoints = planner.liveLayout.all();
+
+ ReplicaCollection.Builder<E> contactBuilder = liveEndpoints.newBuilder(liveEndpoints.size());
+
+ // find full replica
+ Replica fullReplica = null;
+ String fullDc = null;
+ for (String dc : planner.dcs)
+ {
+ E dcCandidates = candidates.get(dc);
+ if (dcCandidates == null || !canSelectReplicasFromDc(dc, dcCandidates.size()))
+ continue;
+
+ for (Replica replica : dcCandidates)
+ {
+ if (replica.isFull())
+ {
+ fullReplica = replica;
+ fullDc = dc;
+ break;
+ }
+ }
+
+ if (fullReplica != null)
+ break;
+ }
+
+ if (fullReplica == null)
+ {
+ // No full replicas available - throw error similar to assureSufficientLiveReplicas
+ throw new UnavailableException("No full replicas available", planner.cl, 1, eligibleCandidates(candidates));
+ }
+
+ int totalContacts = 0;
+ int totalContactedDcs = 0;
+
+ // add dc with full replicas first
+ {
+ E dcCandidates = candidates.get(fullDc);
+ Preconditions.checkNotNull(dcCandidates);
+
+ contactBuilder.add(fullReplica);
+ totalContacts++;
+ int dcContacts = 1;
+
+ if (!haveSufficientContacts(totalContacts, totalContactedDcs))
+ {
+ for (Replica replica : dcCandidates)
+ {
+ if (replica.equals(fullReplica))
+ continue;
+
+ if (haveSufficientContactsForDc(fullDc, totalContacts, totalContactedDcs, dcContacts))
+ break;
+
+ contactBuilder.add(replica);
+ totalContacts++;
+ dcContacts++;
+ }
+ }
+ totalContactedDcs++;
+ }
+
+ for (String dc : planner.dcs)
+ {
+ if (dc.equals(fullDc))
+ continue;
+
+ E dcCandidates = candidates.get(dc);
+ if (dcCandidates == null || dcCandidates.isEmpty())
+ continue;
+
+ if (haveSufficientContacts(totalContacts, totalContactedDcs))
+ break;
+
+ int dcContacts = 0;
+ for (Replica replica : dcCandidates)
+ {
+ if (haveSufficientContactsForDc(dc, totalContacts, totalContactedDcs, dcContacts))
+ break;
+
+ contactBuilder.add(replica);
+ totalContacts++;
+ dcContacts++;
+ }
+
+ totalContactedDcs++;
+ }
+
+ if (!haveSufficientLiveNodes(totalContacts, totalContactedDcs))
+ {
+ int fullReplicas = 0;
+ for (String dc : planner.dcs)
+ {
+ E dcCandidates = candidates.get(dc);
+ if (dcCandidates == null || !canSelectReplicasFromDc(dc, dcCandidates.size()))
+ continue;
+ for (Replica replica : dcCandidates)
+ if (replica.isFull())
+ fullReplicas++;
+ }
+
+ throw new UnavailableException("Unable to reach quorum of quorum", planner.cl, minBlockFor(), fullReplicas);
+ }
+
+ E result = contactBuilder.build();
+
+ if (ADDL_CHECKS_ENABLED)
+ {
+ // check for duplicate contacts
+ Set<InetAddressAndPort> seen = new HashSet<>();
+ for (Replica r : result)
+ Preconditions.checkState(seen.add(r.endpoint()), "Duplicate contact: %s", r.endpoint());
+ }
+
+ return result;
+ }
+
+ public P build()
+ {
+ Map<String, E> candidates = planner.candidates(this);
+
+ E allEndpoints = allLiveAndDown();
+ E allCandidates = allCandidates(candidates);
+ E allContacts = contacts(candidates);
+
+ if (ADDL_CHECKS_ENABLED)
+ {
+ // contacts must be a subset of candidates
+ Set<InetAddressAndPort> candidateSet = allCandidates.endpoints();
+ for (Replica contact : allContacts)
+ Preconditions.checkState(candidateSet.contains(contact.endpoint()), "Contact %s not in candidates", contact.endpoint());
+ }
+
+ return planner.replicaPlanBuilder(allCandidates, allContacts, allEndpoints, minBlockFor());
+ }
+
+ static class Simple<E extends Endpoints<E>, L extends ReplicaLayout<E>, P extends ReplicaPlan.ForRead<E, P>> extends ReadReplicaPlanBuilder<E, L, P>
+ {
+ private final int blockFor;
+ private final int targetContacts;
+
+ public Simple(CoordinationPlanner.ForRead<E, L, P> planner)
+ {
+ super(planner);
+
+ if (planner.cl == ConsistencyLevel.ALL)
+ {
+ int required = 0;
+ for (E replicas : planner.liveAndDownEndpoints.values())
+ required += replicas.size();
+
+ this.blockFor = required;
+ }
+ else
+ {
+ this.blockFor = planner.cl.blockFor(planner.strategy);
+ }
+
+ this.targetContacts = planner.alwaysSpeculate ? blockFor + 1 : blockFor;
+ }
+
+ @Override
+ public void maybeThrowIndexReadException(int initial, int filtered, Map<String, E> candidates, Map<InetAddressAndPort, Index.Status> indexStatusMap)
+ {
+ if (blockFor <= initial && blockFor > filtered)
+ {
+ IndexStatusManager.instance.readFailureException(filtered, removedSupplier(planner.liveAndDownEndpoints, candidates), indexStatusMap, planner.cl, blockFor);
+ }
+ }
+
+ @Override
+ boolean canSelectReplicasFromDc(String dc, int liveNodes)
+ {
+ return true;
+ }
+
+ @Override
+ int minBlockFor()
+ {
+ return blockFor;
+ }
+
+ @Override
+ int eligibleCandidates(Map<String, E> candidates)
+ {
+ int eligible = 0;
+ for (Map.Entry<String, E> entry : candidates.entrySet())
+ {
+ eligible += entry.getValue().size();
+ }
+ return eligible;
+ }
+
+ @Override
+ boolean haveSufficientContacts(int contactedReplicas, int contactedDcs)
+ {
+ return contactedReplicas >= targetContacts;
+ }
+
+ @Override
+ boolean haveSufficientContactsForDc(String dc, int contactedReplicas, int contactedDcs, int dcContacts)
+ {
+ return haveSufficientContacts(contactedReplicas, contactedDcs);
+ }
+
+ @Override
+ boolean haveSufficientLiveNodes(int contactedReplicas, int contactedDcs)
+ {
+ return contactedReplicas >= blockFor;
+ }
+ }
+
+ static class DcQuorum<E extends Endpoints<E>, L extends ReplicaLayout<E>, P extends ReplicaPlan.ForRead<E, P>> extends ReadReplicaPlanBuilder<E, L, P>
+ {
+ private final int blockForDcs;
+ private final Object2IntHashMap<String> dcQuorums = new Object2IntHashMap<>(-1);
+ private boolean approvedSpeculativeReplica = false;
+
+ public DcQuorum(CoordinationPlanner.ForRead<E, L, P> planner)
+ {
+ super(planner);
+ this.blockForDcs = planner.cl != ConsistencyLevel.EACH_QUORUM ? (planner.dcs.size() / 2) + 1 : planner.dcs.size();
+ for (String dc : planner.dcs)
+ dcQuorums.put(dc, planner.strategy.calculateQuorum(dc));
+ }
+ private int dcQuorum(String dc)
+ {
+ int v = dcQuorums.get(dc);
+ Preconditions.checkArgument(v >= 0, "dcQuorum must be >= 0");
+ return v;
+ }
+
+ @Override
+ public void maybeThrowIndexReadException(int initial, int filtered, Map<String, E> candidates, Map<InetAddressAndPort, Index.Status> indexStatusMap)
+ {
+ int required = 0;
+ int available = 0;
+ int liveDcs = 0;
+ for (Map.Entry<String, E> entry : candidates.entrySet())
+ {
+ int dcQuorum = dcQuorum(entry.getKey());
+ int dcLive = entry.getValue().size();
+ if (dcLive >= dcQuorum)
+ liveDcs++;
+
+ required += dcQuorum;
+ available += entry.getValue().size();
+ }
+
+ if (liveDcs <= blockForDcs)
+ {
+ IndexStatusManager.instance.readFailureException(available, removedSupplier(planner.liveAndDownEndpoints, candidates), indexStatusMap, planner.cl, required);
+ }
+ }
+
+ @Override
+ boolean canSelectReplicasFromDc(String dc, int liveNodes)
+ {
+ int dcQuorum = dcQuorum(dc);
+ return liveNodes >= dcQuorum;
+ }
+
+ @Override
+ int minBlockFor()
+ {
+ int[] quorums = new int[planner.dcs.size()];
+ for (int i = 0; i < quorums.length; i++)
+ quorums[i] = dcQuorum(planner.dcs.get(i));
+
+ Arrays.sort(quorums);
+
+ int blockFor = 0;
+ for (int i=0; i<blockForDcs; i++)
+ blockFor += quorums[i];
+ return blockFor;
+ }
+
+ @Override
+ int eligibleCandidates(Map<String, E> candidates)
+ {
+ int eligible = 0;
+ for (Map.Entry<String, E> entry : candidates.entrySet())
+ {
+ int dcLive = entry.getValue().size();
+ int dcQuorum = dcQuorum(entry.getKey());
+ if (dcLive >= dcQuorum)
+ eligible += dcLive;
+ }
+
+ return eligible;
+ }
+
+ @Override
+ boolean haveSufficientContacts(int contactedReplicas, int contactedDcs)
+ {
+ return contactedDcs >= blockForDcs;
+ }
+
+ @Override
+ boolean haveSufficientContactsForDc(String dc, int contactedReplicas, int contactedDcs, int dcContacts)
+ {
+ int dcQuorum = dcQuorum(dc);
+
+ if (dcContacts < dcQuorum)
+ return false;
+
+ if (dcContacts == dcQuorum && planner.alwaysSpeculate && !approvedSpeculativeReplica)
+ {
+ approvedSpeculativeReplica = true;
+ return false;
+ }
+
+ return true;
+ }
+
+ @Override
+ boolean haveSufficientLiveNodes(int contactedReplicas, int contactedDcs)
+ {
+ return contactedDcs >= blockForDcs;
+ }
+ }
+
+ }
+
+ @Override
+ protected CoordinationPlan.ForWrite planForWriteInternal(ClusterMetadata metadata,
+ Keyspace keyspace,
+ ConsistencyLevel consistencyLevel,
+ Function<ClusterMetadata, ReplicaLayout.ForTokenWrite> liveAndDown,
+ ReplicaPlans.Selector selector)
+ {
+ // writes don't need any special handling during primary failover. We just write to the current
+ // primary DC, regardless of any other failover status
+ ReplicaLayout.ForTokenWrite liveAndDownLayout = liveAndDown.apply(metadata);
+
+ CoordinationPlanner.ForWrite planner = new CoordinationPlanner.ForWrite(metadata, keyspace, consistencyLevel, this, primaryDC, liveAndDownLayout, selector);
+
+ return new CoordinationPlan.ForWrite(planner.createReplicaPlan(), planner.createResponseTracker());
+ }
+
+ /**
+ * Holds the information needed to send satellite commit mutations alongside a paxos commit.
+ * The tracker has the primary DC pre-completed, so only satellite/secondary DC quorums
+ * need actual responses.
+ */
+ static class SatelliteCommitPlan
+ {
+ final EndpointsForToken liveEndpoints;
+ final EndpointsForToken downEndpoints;
+ final ResponseTracker tracker;
+
+ SatelliteCommitPlan(EndpointsForToken liveEndpoints, EndpointsForToken downEndpoints, ResponseTracker tracker)
+ {
+ this.liveEndpoints = liveEndpoints;
+ this.downEndpoints = downEndpoints;
+ this.tracker = tracker;
+ }
+ }
+
+ SatelliteCommitPlan createSatelliteCommitPlan(ClusterMetadata metadata, Keyspace keyspace, Token token)
+ {
+ ReplicaLayout.ForTokenWrite liveAndDownLayout = ReplicaLayout.forTokenWriteLiveAndDown(metadata, keyspace, token);
+ CoordinationPlanner.ForWrite planner = new CoordinationPlanner.ForWrite(metadata, keyspace, ConsistencyLevel.QUORUM,
+ this, primaryDC, liveAndDownLayout, ReplicaPlans.writeAll);
+ ResponseTracker tracker = planner.createResponseTracker();
+
+ // paxos is handling the consensus/commit acks in the primary DC, we just need to worry about the witness DCs
+ EndpointsForToken primaryEndpoints = planner.liveAndDownLayouts.get(primaryDC).all();
+ if (primaryEndpoints != null)
+ {
+ for (int i = 0; i < primaryEndpoints.size(); i++)
+ tracker.onResponse(primaryEndpoints.endpoint(i));
+ }
+
+ // Collect satellite/secondary DC endpoints from the filtered layout.
+ // planner.liveAndDownLayout is already filtered to only include DCs in the coordination plan
+ // (primary DC + its satellite + other full DCs), excluding satellites of non-primary DCs.
+ Predicate<Replica> notPrimary = rp -> !metadata.locator.location(rp.endpoint()).datacenter.equals(primaryDC);
+ EndpointsForToken allSatellite = planner.liveAndDownLayout.all().filter(notPrimary);
+ EndpointsForToken liveSatellite = allSatellite.filter(FailureDetector.isReplicaAlive);
+ EndpointsForToken downSatellite = allSatellite.filter(rp -> !FailureDetector.isReplicaAlive.test(rp));
+
+ return new SatelliteCommitPlan(liveSatellite, downSatellite, tracker);
+ }
+
+ /**
+ * Sends plain mutations to satellite/secondary DCs alongside a paxos commit.
+ *
+ * Paxos consensus handles the primary DC writes. This method sends the committed mutation to satellite DCs using
+ * the standard SRS write coordination. The primary DC group is pre-completed in the tracker so only satellite DC
+ * quorums need actual responses.
+ *
+ * @return a future that completes when satellite DC quorum requirements are met (or failed)
+ */
+ @Override
+ public Future<Void> sendPaxosCommitMutations(Agreed commit, boolean isUrgent)
+ {
+ ClusterMetadata metadata = ClusterMetadata.current();
+ Keyspace keyspace = Keyspace.open(commit.metadata().keyspace);
+ Token token = commit.partitionKey().getToken();
+
+ // Reject satellite commit writes during TRANSITION_ACK. Paxos operations should not be
+ // in progress during this state, but check defensively to avoid propagating stale commits.
+ SatelliteFailoverState.FailoverInfo failoverInfo = getFailoverInfo(token, metadata);
+ if (failoverInfo.getState() == SatelliteFailoverState.State.TRANSITION_ACK)
+ throw new UnavailableException("Paxos commit rejected during TRANSITION_ACK failover state",
+ ConsistencyLevel.SERIAL, 1, 0);
+
+ SatelliteCommitPlan plan = createSatelliteCommitPlan(metadata, keyspace, token);
+ ResponseTracker tracker = plan.tracker;
+ MutationId mutationId = commit.mutation.id();
+ Preconditions.checkState(!mutationId.isNone());
+
+ // Register satellite replicas with mutation tracking service
+ if (!plan.liveEndpoints.isEmpty() || !plan.downEndpoints.isEmpty())
+ {
+ IntHashSet satelliteHostIds = new IntHashSet();
+ for (int i = 0; i < plan.liveEndpoints.size(); i++)
+ satelliteHostIds.add(metadata.directory.peerId(plan.liveEndpoints.endpoint(i)).id());
+ for (int i = 0; i < plan.downEndpoints.size(); i++)
+ satelliteHostIds.add(metadata.directory.peerId(plan.downEndpoints.endpoint(i)).id());
+ if (!satelliteHostIds.isEmpty())
+ MutationTrackingService.instance().sentWriteRequest(commit.mutation, satelliteHostIds);
+ }
+
+ // Create promise that resolves when satellite tracker completes
+ AsyncPromise<Void> promise = new AsyncPromise<>();
+
+ // Send plain mutations to live satellite endpoints
+ Message<Mutation> satelliteMessage = Message.out(Verb.PAXOS2_COMMIT_REMOTE_REQ, commit.makeMutation(), isUrgent);
+ for (int i = 0, mi = plan.liveEndpoints.size(); i < mi; ++i)
+ {
+ InetAddressAndPort endpoint = plan.liveEndpoints.endpoint(i);
+ logger.trace("Sending satellite commit mutation for {} to {}", commit.partitionKey(), endpoint);
+ MessagingService.instance().sendWithCallback(satelliteMessage, endpoint, new RequestCallback<NoPayload>()
+ {
+ @Override
+ public boolean invokeOnFailure()
+ {
+ return true;
+ }
+
+ @Override
+ public void onResponse(Message<NoPayload> msg)
+ {
+ MutationTrackingService.instance().receivedWriteResponse(mutationId, msg.from());
+
+ tracker.onResponse(msg.from());
+ if (tracker.isComplete())
+ resolvePromise(promise, tracker);
+ }
+
+ @Override
+ public void onFailure(InetAddressAndPort from, RequestFailure failure)
+ {
+ MutationTrackingService.instance().retryFailedWrite(mutationId, from, failure);
+
+ tracker.onFailure(from);
+ if (tracker.isComplete())
+ resolvePromise(promise, tracker);
+ }
+ });
+ }
+
+ // Mark down satellite endpoints as failed
+ for (int i = 0, mi = plan.downEndpoints.size(); i < mi; ++i)
+ {
+ InetAddressAndPort endpoint = plan.downEndpoints.endpoint(i);
+ MutationTrackingService.instance().retryFailedWrite(mutationId, endpoint, RequestFailure.NODE_DOWN);
+ tracker.onFailure(endpoint);
+ }
+
+ if (tracker.isComplete())
+ resolvePromise(promise, tracker);
+
+ return promise;
+ }
+
+ private static void resolvePromise(AsyncPromise<Void> promise, ResponseTracker tracker)
+ {
+ if (tracker.isSuccessful())
+ promise.trySuccess(null);
+ else
+ promise.tryFailure(new RuntimeException("Satellite DC quorum not met. Responses required: " + tracker.required() + ", responses received: " + tracker.received()));
+ }
+
+ private CoordinationPlan.ForTokenRead planForTokenReadPrimary(ClusterMetadata metadata,
+ String primary,
+ Keyspace keyspace,
+ TableId tableId,
+ Token token,
+ @Nullable Index.QueryPlan indexQueryPlan,
+ ConsistencyLevel consistencyLevel,
+ SpeculativeRetryPolicy retry,
+ ReadCoordinator coordinator)
+ {
+ ReplicaLayout.ForTokenRead liveAndDownLayout = ReplicaLayout.forTokenReadSorted(metadata, keyspace, this, tableId, token, coordinator);
+
+ CoordinationPlanner.ForTokenRead planner = new CoordinationPlanner.ForTokenRead(metadata, keyspace, token, consistencyLevel, this, primary, liveAndDownLayout, indexQueryPlan, retry, tableId, coordinator);
+
+ return new CoordinationPlan.ForTokenRead(ReplicaPlan.shared(planner.createReplicaPlan()), planner.createResponseTracker());
+ }
+
+ @Override
+ public CoordinationPlan.ForTokenRead planForTokenRead(ClusterMetadata metadata, Keyspace keyspace, TableId tableId, Token token, @Nullable Index.QueryPlan indexQueryPlan, ConsistencyLevel consistencyLevel, SpeculativeRetryPolicy retry, ReadCoordinator coordinator)
+ {
+ CoordinationPlan.ForTokenRead primaryPlan = planForTokenReadPrimary(metadata, primaryDC, keyspace, tableId, token, indexQueryPlan, consistencyLevel, retry, coordinator);
+
+ SatelliteFailoverState.FailoverInfo failoverInfo = getFailoverInfo(token, metadata);
+ if (!failoverInfo.isTransitioning())
+ return primaryPlan;
+
+ CoordinationPlan.ForTokenRead previousPlan = planForTokenReadPrimary(metadata, failoverInfo.getFromDC(), keyspace, tableId, token, indexQueryPlan, consistencyLevel, retry, coordinator);
+
+ return mergeTokenReadPlans(primaryPlan, previousPlan, cm -> strategy(cm).planForTokenRead(cm, keyspace, tableId, token, indexQueryPlan, consistencyLevel, retry, coordinator).replicas());
+ }
+
+ /**
+ * Merge two ForTokenRead coordination plans during failover transition. The primary plan's replicas are added
+ * first to preserve proximity ordering for data vs digest request selection. Replicas from the previous plan
+ * are appended if not already present in the primary plan.
+ */
+ private static CoordinationPlan.ForTokenRead mergeTokenReadPlans(CoordinationPlan.ForTokenRead primaryPlan,
+ CoordinationPlan.ForTokenRead previousPlan,
+ Function<ClusterMetadata, ReplicaPlan.ForTokenRead> recompute)
+ {
+ ReplicaPlan.ForTokenRead primary = primaryPlan.replicas();
+ ReplicaPlan.ForTokenRead previous = previousPlan.replicas();
+
+ EndpointsForToken candidates = mergeEndpoints(primary.readCandidates(), previous.readCandidates());
+ EndpointsForToken contacts = mergeEndpoints(primary.contacts(), previous.contacts());
+ EndpointsForToken liveAndDown = mergeEndpoints(primary.liveAndDown(), previous.liveAndDown());
+ int readQuorum = Math.max(primary.readQuorum(), previous.readQuorum());
+
+ ReplicaPlan.ForTokenRead merged = new ReplicaPlan.ForTokenRead(primary.keyspace(),
+ primary.replicationStrategy(),
+ primary.consistencyLevel(),
+ candidates,
+ contacts,
+ liveAndDown,
+ recompute,
+ CoordinationPlanner.ForTokenRead.READ_REPAIR_PLAN,
+ primary.epoch(),
+ readQuorum);
+
+ ResponseTracker tracker = new CompositeTracker(2, primaryPlan.responses(), previousPlan.responses());
+
+ return new CoordinationPlan.ForTokenRead(ReplicaPlan.shared(merged), tracker);
+ }
+
+ private static <E extends Endpoints<E>> E mergeEndpoints(E first, E second)
+ {
+ ReplicaCollection.Builder<E> builder = first.newBuilder(first.size() + second.size());
+ builder.addAll(first);
+ builder.addAll(second, ReplicaCollection.Builder.Conflict.DUPLICATE);
+ return builder.build();
+ }
+
+ private CoordinationPlan.ForRangeRead planForRangeReadPrimary(ClusterMetadata metadata,
+ String primary,
+ Keyspace keyspace,
+ TableId tableId,
+ AbstractBounds<PartitionPosition> range,
+ int vnodeCount,
+ @Nullable Index.QueryPlan indexQueryPlan,
+ ConsistencyLevel consistencyLevel)
+ {
+
+ ReplicaLayout.ForRangeRead liveAndDownLayout = ReplicaLayout.forRangeReadSorted(metadata, keyspace, this, range);
+
+ CoordinationPlanner.ForRangeRead planner = new CoordinationPlanner.ForRangeRead(metadata, keyspace, range, vnodeCount, consistencyLevel, this, primary, liveAndDownLayout, indexQueryPlan, tableId);
+
+ return new CoordinationPlan.ForRangeRead(ReplicaPlan.shared(planner.createReplicaPlan()), planner.createResponseTracker());
+ }
+
+
+ @Override
+ public CoordinationPlan.ForRangeRead planForRangeRead(ClusterMetadata metadata, Keyspace keyspace, TableId tableId, @Nullable Index.QueryPlan indexQueryPlan, ConsistencyLevel consistencyLevel, AbstractBounds<PartitionPosition> range, int vnodeCount)
+ {
+ CoordinationPlan.ForRangeRead primaryPlan = planForRangeReadPrimary(metadata, primaryDC, keyspace, tableId, range, vnodeCount, indexQueryPlan, consistencyLevel);
+
+ SatelliteFailoverState.FailoverInfo failoverInfo = getFailoverInfo(range, metadata);
+ if (!failoverInfo.isTransitioning())
+ return primaryPlan;
+
+ CoordinationPlan.ForRangeRead previousPlan = planForRangeReadPrimary(metadata, failoverInfo.getFromDC(), keyspace, tableId, range, vnodeCount, indexQueryPlan, consistencyLevel);
+
+ return mergeRangeReadPlans(primaryPlan, previousPlan, cm -> strategy(cm).planForRangeRead(cm, keyspace, tableId, indexQueryPlan, consistencyLevel, range, vnodeCount).replicas());
+ }
+
+ private static CoordinationPlan.ForRangeRead mergeRangeReadPlans(CoordinationPlan.ForRangeRead primaryPlan,
+ CoordinationPlan.ForRangeRead previousPlan,
+ Function<ClusterMetadata, ReplicaPlan.ForRangeRead> recompute)
+ {
+ ReplicaPlan.ForRangeRead primary = primaryPlan.replicas();
+ ReplicaPlan.ForRangeRead previous = previousPlan.replicas();
+
+ EndpointsForRange candidates = mergeEndpoints(primary.readCandidates(), previous.readCandidates());
+ EndpointsForRange contacts = mergeEndpoints(primary.contacts(), previous.contacts());
+ EndpointsForRange liveAndDown = mergeEndpoints(primary.liveAndDown(), previous.liveAndDown());
+ int readQuorum = Math.max(primary.readQuorum(), previous.readQuorum());
+
+ ReplicaPlan.ForRangeRead merged = new ReplicaPlan.ForRangeRead(primary.keyspace(),
+ primary.replicationStrategy(),
+ primary.consistencyLevel(),
+ primary.range(),
+ candidates,
+ contacts,
+ liveAndDown,
+ primary.vnodeCount(),
+ recompute,
+ CoordinationPlanner.ForRangeRead.READ_REPAIR_PLAN,
+ primary.epoch(),
+ readQuorum);
+
+ ResponseTracker tracker = new CompositeTracker(2, primaryPlan.responses(), previousPlan.responses());
+
+ return new CoordinationPlan.ForRangeRead(ReplicaPlan.shared(merged), tracker);
+ }
+
+ @Override
+ public CoordinationPlan.ForRangeRead planForSingleReplicaRangeRead(Keyspace keyspace, AbstractBounds<PartitionPosition> range, Replica replica, int vnodeCount)
+ {
+ throw new IllegalStateException("planForSingleReplicaRangeRead not supported by SatelliteReplicationStrategy");
+ }
+
+ @Override
+ public CoordinationPlan.ForTokenRead planForSingleReplicaTokenRead(Keyspace keyspace, Token token, Replica replica)
+ {
+ throw new IllegalStateException("planForSingleReplicaTokenRead not supported by SatelliteReplicationStrategy");
+ }
+
+ @Override
+ public CoordinationPlan.ForRangeRead maybeMergeRangeReads(ClusterMetadata metadata, Keyspace keyspace, TableId tableId, ConsistencyLevel consistencyLevel, ReplicaPlan.ForRangeRead left, ReplicaPlan.ForRangeRead right)
+ {
+ return null;
+ }
+
+ private String getSatelliteForDC(String dc)
+ {
+ for (SatelliteInfo sat : satellites.values())
+ if (sat.parentDC.equals(dc))
+ return sat.name;
+ return null;
+ }
+
+ private int calculateQuorum(String dc)
+ {
+ ReplicationFactor rf = fullDCs.get(dc);
+ if (rf != null)
+ return rf.fullReplicas / 2 + 1;
+
+ SatelliteInfo sat = satellites.get(dc);
+ return sat != null ? sat.rf.allReplicas / 2 + 1 : 0;
+ }
+
+ public SatelliteFailoverState.FailoverInfo getFailoverInfo(Token token, ClusterMetadata metadata)
+ {
+ Range<Token> range = TokenRingUtils.getRange(
+ metadata.tokenMap.tokens(), token);
+ return failoverState.getFailoverInfo(range);
+ }
+
+ public SatelliteFailoverState.FailoverInfo getFailoverInfo(AbstractBounds<PartitionPosition> range, ClusterMetadata metadata)
+ {
+ return failoverState.getFailoverInfo(range.right.getToken());
+ }
+
+ /**
+ * Reject paxos consensus operations during TRANSITION_ACK or when the local node is not in the primary DC.
+ *
+ * Paxos consensus operates entirely within the primary DC. Nodes in satellite or secondary DCs should
+ * never process paxos consensus messages. During TRANSITION_ACK, all paxos operations must be blocked
+ * to prevent conflicting operations across different full DCs.
+ */
+ @Override
+ public boolean shouldRejectPaxos(Token token)
+ {
+ ClusterMetadata metadata = ClusterMetadata.current();
+
+ SatelliteFailoverState.FailoverInfo failoverInfo = getFailoverInfo(token, metadata);
+ if (failoverInfo.getState() == SatelliteFailoverState.State.TRANSITION_ACK)
+ return true;
+
+ String localDC = metadata.locator.location(FBUtilities.getBroadcastAddressAndPort()).datacenter;
+ return !primaryDC.equals(localDC);
+ }
+
+ @VisibleForTesting
+ void setFailoverState(SatelliteFailoverState.FailoverStateMap state)
+ {
+ this.failoverState = state;
+ }
}
diff --git a/src/java/org/apache/cassandra/metrics/PaxosMetrics.java b/src/java/org/apache/cassandra/metrics/PaxosMetrics.java
index 45088a1d17..0d5698b067 100644
--- a/src/java/org/apache/cassandra/metrics/PaxosMetrics.java
+++ b/src/java/org/apache/cassandra/metrics/PaxosMetrics.java
@@ -29,6 +29,7 @@ public class PaxosMetrics
private static final MetricNameFactory factory = new DefaultNameFactory(TYPE_NAME);
public static final Counter linearizabilityViolations = Metrics.counter(factory.createMetricName("LinearizabilityViolations"));
public static final Meter repairPaxosTopologyRetries = Metrics.meter(factory.createMetricName("RepairPaxosTopologyRetries"));
+ public static final Meter rejectedOperations = Metrics.meter(factory.createMetricName("RejectedOperations"));
public static void initialize() {}
}
diff --git a/src/java/org/apache/cassandra/net/Verb.java b/src/java/org/apache/cassandra/net/Verb.java
index 56733e3123..2afbf1ccde 100644
--- a/src/java/org/apache/cassandra/net/Verb.java
+++ b/src/java/org/apache/cassandra/net/Verb.java
@@ -36,6 +36,7 @@ import org.apache.cassandra.db.CounterMutation;
import org.apache.cassandra.db.CounterMutationVerbHandler;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.MutationVerbHandler;
+import org.apache.cassandra.db.PaxosCommitRemoteMutationVerbHandler;
import org.apache.cassandra.db.ReadCommand;
import org.apache.cassandra.db.ReadCommandVerbHandler;
import org.apache.cassandra.db.ReadRepairVerbHandler;
@@ -304,7 +305,7 @@ public enum Verb
SNAPSHOT_RSP (87, P0, rpcTimeout, MISC, () -> NoPayload.serializer, RESPONSE_HANDLER ),
SNAPSHOT_REQ (27, P0, rpcTimeout, MISC, () -> SnapshotCommand.serializer, () -> SnapshotVerbHandler.instance, SNAPSHOT_RSP ),
- PAXOS2_COMMIT_REMOTE_REQ (38, P2, writeTimeout, MUTATION, () -> Mutation.serializer, () -> MutationVerbHandler.instance, MUTATION_RSP ),
+ PAXOS2_COMMIT_REMOTE_REQ (38, P2, writeTimeout, MUTATION, () -> Mutation.serializer, () -> PaxosCommitRemoteMutationVerbHandler.instance, MUTATION_RSP ),
PAXOS2_COMMIT_REMOTE_RSP (39, P2, writeTimeout, REQUEST_RESPONSE, () -> NoPayload.serializer, RESPONSE_HANDLER ),
PAXOS2_PREPARE_RSP (50, P2, writeTimeout, REQUEST_RESPONSE, () -> PaxosPrepare.responseSerializer, RESPONSE_HANDLER ),
PAXOS2_PREPARE_REQ (40, P2, writeTimeout, MUTATION, () -> PaxosPrepare.requestSerializer, () -> PaxosPrepare.requestHandler, PAXOS2_PREPARE_RSP ),
diff --git a/src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java b/src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java
index 176124517f..9e941e40ab 100644
--- a/src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java
+++ b/src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java
@@ -31,7 +31,6 @@ import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
-import org.apache.cassandra.locator.CoordinationPlan;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -46,6 +45,7 @@ import org.apache.cassandra.exceptions.RequestFailureReason;
import org.apache.cassandra.exceptions.RetryOnDifferentSystemException;
import org.apache.cassandra.exceptions.WriteFailureException;
import org.apache.cassandra.exceptions.WriteTimeoutException;
+import org.apache.cassandra.locator.CoordinationPlan;
import org.apache.cassandra.locator.EndpointsForToken;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.ReplicaPlan.ForWrite;
diff --git a/src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java b/src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java
index aef1c1ea7b..53a68cde1e 100644
--- a/src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java
+++ b/src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java
@@ -35,7 +35,7 @@ import org.apache.cassandra.utils.FBUtilities;
/**
* This class blocks for a quorum of responses _in all datacenters_ (CL.EACH_QUORUM).
*
- * The PerDcResponseTracker handles per-datacenter counting.
+ * The CompositeTracker handles per-datacenter counting.
*/
public class DatacenterSyncWriteResponseHandler<T> extends AbstractWriteResponseHandler<T>
{
diff --git a/src/java/org/apache/cassandra/service/paxos/Paxos.java b/src/java/org/apache/cassandra/service/paxos/Paxos.java
index 3f6fcc42fc..3791db6cc6 100644
--- a/src/java/org/apache/cassandra/service/paxos/Paxos.java
+++ b/src/java/org/apache/cassandra/service/paxos/Paxos.java
@@ -46,6 +46,7 @@ import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.IReadResponse;
import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.ReadCommand;
import org.apache.cassandra.db.ReadKind;
import org.apache.cassandra.db.ReadResponse;
import org.apache.cassandra.db.SinglePartitionReadCommand;
@@ -86,6 +87,7 @@ import org.apache.cassandra.locator.ReplicaLayout.ForTokenWrite;
import org.apache.cassandra.locator.ReplicaPlan.ForRead;
import org.apache.cassandra.metrics.ClientRequestMetrics;
import org.apache.cassandra.metrics.ClientRequestSizeMetrics;
+import org.apache.cassandra.metrics.PaxosMetrics;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.schema.TableMetadata;
@@ -101,6 +103,7 @@ import org.apache.cassandra.service.reads.DataResolver;
import org.apache.cassandra.service.reads.ReadCoordinator;
import org.apache.cassandra.service.reads.repair.NoopReadRepair;
import org.apache.cassandra.service.reads.tracked.TrackedDataResponse;
+import org.apache.cassandra.service.reads.tracked.TrackedRead;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.membership.NodeId;
@@ -573,6 +576,28 @@ public class Paxos
{
return keyspace.getMetadata().params.replication.isMeta();
}
+
+ /**
+ * Hook called after electorate messages are sent during a tracked prepare phase.
+ * Base implementation is a no-op. SRS overrides this to fire satellite summary
+ * read requests in parallel with electorate prepare messages.
+ */
+ public void onPrepareStarted(TrackedRead.Id readId, int dataNodeId, int[] summaryHostIds, ReadCommand readCommand)
+ {
+ }
+
+ /**
+ * Returns additional summary host IDs to include in tracked read reconciliation.
+ * These are nodes that should participate in the ReadReconciliations protocol
+ * but are not part of the paxos electorate (e.g., satellite DC endpoints for SRS).
+ * Base implementation returns an empty array.
+ */
+ public int[] additionalSummaryHostIds(ClusterMetadata metadata)
+ {
+ return EMPTY_HOST_IDS;
+ }
+
+ private static final int[] EMPTY_HOST_IDS = new int[0];
}
/**
@@ -1147,7 +1172,7 @@ public class Paxos
}
else
{
- DataResolver<EndpointsForToken, Participants> resolver = new DataResolver<>(ReadCoordinator.DEFAULT, query, () -> success.participants, NoopReadRepair.instance, requestTime);
+ DataResolver<EndpointsForToken, Participants> resolver = new DataResolver<>(ReadCoordinator.DEFAULT, query, success::participants, NoopReadRepair.instance, requestTime);
for (int i = 0 ; i < responses.size() ; ++i)
{
@@ -1207,6 +1232,13 @@ public class Paxos
// replicas using the supplied token as this can actually be of the incorrect type (for example when
// performing Paxos repair).
Token token = table.partitioner == MetaStrategy.partitioner ? MetaStrategy.entireRange.right : key.getToken();
+
+ if (keyspace.getReplicationStrategy().shouldRejectPaxos(token))
+ {
+ PaxosMetrics.rejectedOperations.mark();
+ return false;
+ }
+
return (includesRead ? EndpointsForToken.natural(keyspace, token).get()
: ReplicaLayout.forTokenWriteLiveAndDown(keyspace, token).all()
).contains(getBroadcastAddressAndPort());
diff --git a/src/java/org/apache/cassandra/service/paxos/PaxosCommit.java b/src/java/org/apache/cassandra/service/paxos/PaxosCommit.java
index b88cdd0528..ab7e9e08dd 100644
--- a/src/java/org/apache/cassandra/service/paxos/PaxosCommit.java
+++ b/src/java/org/apache/cassandra/service/paxos/PaxosCommit.java
@@ -19,11 +19,14 @@
package org.apache.cassandra.service.paxos;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+import java.util.function.BiFunction;
import java.util.function.Consumer;
import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
import org.agrona.collections.IntHashSet;
import org.slf4j.Logger;
@@ -33,9 +36,11 @@ import org.apache.cassandra.concurrent.ExecutorPlus;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.RequestFailure;
+import org.apache.cassandra.locator.AbstractReplicationStrategy;
import org.apache.cassandra.locator.EndpointsForToken;
import org.apache.cassandra.locator.InOurDc;
import org.apache.cassandra.locator.InetAddressAndPort;
@@ -54,6 +59,7 @@ import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.concurrent.ConditionAsConsumer;
+import org.apache.cassandra.utils.concurrent.Future;
import static com.google.common.base.Preconditions.checkState;
import static java.util.Collections.emptyMap;
@@ -118,6 +124,150 @@ public class PaxosCommit<OnDone extends Consumer<? super PaxosCommit.Status>> ex
@Nullable
final IntHashSet remoteReplicas;
+ /**
+ * Handles composition of paxos and additional commit mutations for replication strategies that
+ * need to send additional mutations (SRS).
+ *
+ * @param <OnDone>
+ */
+ static class AugmentedCommit<OnDone extends Consumer<? super PaxosCommit.Status>>
+ {
+ private static final AtomicReferenceFieldUpdater<AugmentedCommit, State> stateUpdater = AtomicReferenceFieldUpdater.newUpdater(AugmentedCommit.class, AugmentedCommit.State.class, "state");
+
+ static abstract class State
+ {
+ abstract State onPaxosComplete(Status status);
+ abstract State onMutationComplete(Status status);
+ boolean isComplete()
+ {
+ return false;
+ }
+ Complete asComplete()
+ {
+ throw new IllegalStateException();
+ }
+ }
+
+ static class Pending extends State
+ {
+ final Status commit;
+ final Status addl;
+
+ public Pending(Status commit, Status addl)
+ {
+ this.commit = commit;
+ this.addl = addl;
+ }
+
+ public Pending()
+ {
+ this(null, null);
+ }
+
+ private static State next(Status commit, Status addl)
+ {
+ if (commit != null && !commit.isSuccess())
+ return new Complete(commit);
+
+ if (addl != null && !addl.isSuccess())
+ return new Complete(addl);
+
+ if (commit == null || addl == null)
+ return new Pending(commit, addl);
+
+ return new Complete(commit);
+ }
+
+ @Override
+ State onPaxosComplete(Status status)
+ {
+ Preconditions.checkState(commit == null);
+ return next(status, addl);
+ }
+
+ @Override
+ State onMutationComplete(Status status)
+ {
+ Preconditions.checkState(addl == null);
+ return next(commit, status);
+ }
+ }
+
+ static class Complete extends State
+ {
+ final Status result;
+
+ public Complete(Status result)
+ {
+ this.result = result;
+ }
+
+ @Override
+ State onPaxosComplete(Status status)
+ {
+ return this;
+ }
+
+ @Override
+ State onMutationComplete(Status status)
+ {
+ return this;
+ }
+
+ @Override
+ boolean isComplete()
+ {
+ return true;
+ }
+
+ @Override
+ Complete asComplete()
+ {
+ return this;
+ }
+ }
+
+ volatile State state = new Pending();
+
+ final OnDone onDone;
+
+ public AugmentedCommit(OnDone onDone)
+ {
+ this.onDone = onDone;
+ }
+
+ private void onCompletion(BiFunction<State, Status, State> update, Status status)
+ {
+ for (;;)
+ {
+ State current = state;
+ if (current.isComplete())
+ return;
+
+ State next = update.apply(current, status);
+ if (stateUpdater.compareAndSet(this, current, next))
+ {
+ if (next.isComplete())
+ onDone.accept(next.asComplete().result);
+ return;
+ }
+ }
+ }
+
+ void onPaxosComplete(Status status)
+ {
+ onCompletion(State::onPaxosComplete, status);
+ }
+
+ void onMutationComplete(Status status)
+ {
+ onCompletion(State::onMutationComplete, status);
+ }
+ }
+
+ @Nullable
+ volatile AugmentedCommit<OnDone> augmentedCommit = null;
+
/**
* packs two 32-bit integers;
* bit 00-31: accepts
@@ -274,6 +424,24 @@ public class PaxosCommit<OnDone extends Consumer<? super PaxosCommit.Status>> ex
consistencyForConsensus, consistencyForCommit, allowHints, onDone);
}
+ void setAugmentedCommitFuture(Future<Void> future)
+ {
+ if (future != null)
+ {
+ augmentedCommit = new AugmentedCommit<>(onDone);
+ future.addCallback((result, failure) -> {
+ if (failure != null)
+ {
+ augmentedCommit.onMutationComplete(new Status(new Paxos.MaybeFailure(true, replicas.size(), required, accepts(responses), emptyMap())));
+ }
+ else
+ {
+ augmentedCommit.onMutationComplete(success);
+ }
+ });
+ }
+ }
+
/**
* Send commit messages to peers (or self)
*/
@@ -293,6 +461,19 @@ public class PaxosCommit<OnDone extends Consumer<? super PaxosCommit.Status>> ex
boolean localExecutedSynchronously = false;
InetAddressAndPort localEndpoint = FBUtilities.getBroadcastAddressAndPort();
+ // Set up additional commit work from the replication strategy (e.g., satellite writes for SRS).
+ // This needs to happen before executeOnSelf() can trigger onPaxosDecision(), so that
+ // additionalCommitFuture is set before it's read. The base strategy returns an
+ // already-completed future, so this is a no-op for non-SRS keyspaces.
+ // For SRS, satellite messages are sent here (in parallel with local execution below).
+ // MutationTrackingService.retryFailedWrite for down satellite endpoints schedules async retries,
+ // which will find the mutation in the journal after executeOnSelf() completes below.
+ if (isTrackedKeyspace)
+ {
+ AbstractReplicationStrategy strategy = Keyspace.open(commit.metadata().keyspace).getReplicationStrategy();
+ setAugmentedCommitFuture(strategy.sendPaxosCommitMutations(commit, isUrgent));
+ }
+
if (isTrackedKeyspace)
{
// For tracked keyspaces, we MUST execute locally synchronously, regardless of USE_SELF_EXECUTION setting.
@@ -341,7 +522,7 @@ public class PaxosCommit<OnDone extends Consumer<? super PaxosCommit.Status>> ex
setAugmentedCommitFuture(strategy.sendPaxosCommitMutations(commit, isUrgent));
}
- // Now send to remote replicas (and record local execution for non-tracked keyspaces)
+ // Now send to remote replicas in the electorate (and record local execution for non-tracked keyspaces)
boolean executeOnSelf = false;
for (int i = 0, mi = allLive.size(); i < mi ; ++i)
{
@@ -501,9 +682,21 @@ public class PaxosCommit<OnDone extends Consumer<? super PaxosCommit.Status>> ex
long responses = responsesUpdater.addAndGet(this, success ? 0x1L : 0x100000000L);
// next two clauses mutually exclusive to ensure we only invoke onDone once, when either failed or succeeded
if (accepts(responses) == required) // if we have received _precisely_ the required accepts, we have succeeded
- onDone.accept(status());
+ onPaxosDecision();
else if (replicas.size() - failures(responses) == required - 1) // if we are _unable_ to receive the required accepts, we have failed
+ onPaxosDecision();
+ }
+
+ /**
+ * Called exactly once when the paxos consensus decision (success or failure) is made.
+ * If an additional commit future exists, onDone is deferred until it also completes.
+ */
+ private void onPaxosDecision()
+ {
+ if (augmentedCommit == null)
onDone.accept(status());
+ else
+ augmentedCommit.onPaxosComplete(status());
}
/**
diff --git a/src/java/org/apache/cassandra/service/paxos/PaxosPrepare.java b/src/java/org/apache/cassandra/service/paxos/PaxosPrepare.java
index e7efe2c440..161c23e980 100644
--- a/src/java/org/apache/cassandra/service/paxos/PaxosPrepare.java
+++ b/src/java/org/apache/cassandra/service/paxos/PaxosPrepare.java
@@ -31,10 +31,11 @@ import java.util.stream.Collectors;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
+import com.google.common.annotations.VisibleForTesting;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.annotations.VisibleForTesting;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.ConsistencyLevel;
@@ -42,6 +43,7 @@ import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.EmbeddableSinglePartitionReadCommand;
import org.apache.cassandra.db.IReadResponse;
import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.ReadCommand;
import org.apache.cassandra.db.ReadExecutionController;
import org.apache.cassandra.db.ReadResponse;
import org.apache.cassandra.db.SinglePartitionReadCommand;
@@ -192,6 +194,7 @@ public class PaxosPrepare extends PaxosRequestCallback<PaxosPrepare.Response> im
FoundIncompleteAccepted incompleteAccepted() { return (FoundIncompleteAccepted) this; }
FoundIncompleteCommitted incompleteCommitted() { return (FoundIncompleteCommitted) this; }
Paxos.MaybeFailure maybeFailure() { return ((MaybeFailure) this).info; }
+ Participants participants() { return participants; }
}
static class Success extends WithRequestedBallot
@@ -471,6 +474,16 @@ public class PaxosPrepare extends PaxosRequestCallback<PaxosPrepare.Response> im
summaryHostIds[summaryIndex++] = metadata.directory.peerId(replica.endpoint()).id();
}
+ // Merge additional summary host IDs (for SRS)
+ int[] additionalIds = participants.additionalSummaryHostIds(metadata);
+ if (additionalIds.length > 0 || summaryIndex < summaryHostIds.length)
+ {
+ int[] merged = new int[summaryIndex + additionalIds.length];
+ System.arraycopy(summaryHostIds, 0, merged, 0, summaryIndex);
+ System.arraycopy(additionalIds, 0, merged, summaryIndex, additionalIds.length);
+ summaryHostIds = merged;
+ }
+
for (int i = 0, size = participants.sizeOfPoll() ; i < size ; ++i)
{
Replica replica = participants.voterReplica(i);
@@ -500,6 +513,8 @@ public class PaxosPrepare extends PaxosRequestCallback<PaxosPrepare.Response> im
Message<R> selfMessageFinal = selfMessage;
send.verb().stage.execute(() -> prepare.executeOnSelfAsync(selfMessageFinal.payload, new RequestTime(selfMessageFinal.createdAtNanos()), selfHandler));
}
+
+ participants.onPrepareStarted(readId, dataNodeId, summaryHostIds, (ReadCommand) prepare.request.read);
}
private static <R extends AbstractRequest<R>> void startUntracked(PaxosPrepare prepare, Participants participants, Message<R> send, BiFunction<R, RequestTime, Future<Response>> selfHandler)
diff --git a/src/java/org/apache/cassandra/service/paxos/SatellitePaxosParticipants.java b/src/java/org/apache/cassandra/service/paxos/SatellitePaxosParticipants.java
new file mode 100644
index 0000000000..1dc0e4eec6
--- /dev/null
+++ b/src/java/org/apache/cassandra/service/paxos/SatellitePaxosParticipants.java
@@ -0,0 +1,98 @@
+/*
+ * 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.service.paxos;
+
+import java.util.function.Function;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.ReadCommand;
+import org.apache.cassandra.locator.EndpointsForToken;
+import org.apache.cassandra.locator.Replica;
+import org.apache.cassandra.locator.ReplicaLayout;
+import org.apache.cassandra.net.Message;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.net.Verb;
+import org.apache.cassandra.service.reads.tracked.TrackedRead;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.Epoch;
+
+/**
+ * Paxos participants for SatelliteReplicationStrategy.
+ *
+ * Paxos consensus operates entirely within the primary DC. Satellites and secondary DCs do not participate in propose
+ * / accept. However paxos reads and writes do need QoQ consistency to support quick failover. This class adds the
+ * additional summary nodes to the prepare/read stage and instructs them to send summaries to the read coordinator.
+ * Additional commit mutations are handled by the replication strategy itself.
+ */
+public class SatellitePaxosParticipants extends Paxos.Participants
+{
+ private static final Logger logger = LoggerFactory.getLogger(SatellitePaxosParticipants.class);
+
+ /** Endpoints in satellite/secondary DCs that receive reads during prepare and writes during commit */
+ private final EndpointsForToken additionalSummaryEndpoints;
+
+ public SatellitePaxosParticipants(Epoch epoch,
+ Keyspace keyspace,
+ ConsistencyLevel consistencyForConsensus,
+ ReplicaLayout.ForTokenWrite all,
+ ReplicaLayout.ForTokenWrite electorate,
+ EndpointsForToken live,
+ Function<ClusterMetadata, Paxos.Participants> recompute,
+ EndpointsForToken additionalSummaryEndpoints)
+ {
+ super(epoch, keyspace, consistencyForConsensus, all, electorate, live, recompute);
+ this.additionalSummaryEndpoints = additionalSummaryEndpoints;
+ }
+
+ public EndpointsForToken getAdditionalSummaryEndpoints()
+ {
+ return additionalSummaryEndpoints;
+ }
+
+ @Override
+ public int[] additionalSummaryHostIds(ClusterMetadata metadata)
+ {
+ if (additionalSummaryEndpoints.isEmpty())
+ return super.additionalSummaryHostIds(metadata);
+
+ int[] ids = new int[additionalSummaryEndpoints.size()];
+ for (int i = 0; i < additionalSummaryEndpoints.size(); i++)
+ ids[i] = metadata.directory.peerId(additionalSummaryEndpoints.endpoint(i)).id();
+ return ids;
+ }
+
+ @Override
+ public void onPrepareStarted(TrackedRead.Id readId, int dataNodeId, int[] summaryHostIds, ReadCommand readCommand)
+ {
+ if (additionalSummaryEndpoints.isEmpty() || readCommand == null)
+ return;
+
+ // Send standalone TRACKED_SUMMARY_REQ to each additional satellite endpoint.
+ TrackedRead.SummaryRequest summaryRequest = new TrackedRead.SummaryRequest(readId, readCommand, dataNodeId, summaryHostIds);
+ Message<TrackedRead.SummaryRequest> summaryMessage = Message.out(Verb.TRACKED_SUMMARY_REQ, summaryRequest);
+ for (Replica replica : additionalSummaryEndpoints)
+ {
+ logger.trace("Sending satellite summary request for {} to {}", readId, replica.endpoint());
+ MessagingService.instance().send(summaryMessage, replica.endpoint());
+ }
+ }
+}
diff --git a/src/java/org/apache/cassandra/service/reads/range/RangeCommandIterator.java b/src/java/org/apache/cassandra/service/reads/range/RangeCommandIterator.java
index 4f4dd16fd1..2887a132e8 100644
--- a/src/java/org/apache/cassandra/service/reads/range/RangeCommandIterator.java
+++ b/src/java/org/apache/cassandra/service/reads/range/RangeCommandIterator.java
@@ -26,9 +26,6 @@ import java.util.function.Function;
import com.google.common.annotations.VisibleForTesting;
-import org.apache.cassandra.db.*;
-import org.apache.cassandra.locator.CoordinationPlan;
-import org.apache.cassandra.service.reads.DataResolver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -48,6 +45,7 @@ import org.apache.cassandra.exceptions.ReadFailureException;
import org.apache.cassandra.exceptions.ReadTimeoutException;
import org.apache.cassandra.exceptions.RetryOnDifferentSystemException;
import org.apache.cassandra.exceptions.UnavailableException;
+import org.apache.cassandra.locator.CoordinationPlan;
import org.apache.cassandra.locator.EndpointsForRange;
import org.apache.cassandra.locator.Replica;
import org.apache.cassandra.locator.ReplicaPlan;
@@ -59,6 +57,7 @@ import org.apache.cassandra.service.accord.txn.TxnResult;
import org.apache.cassandra.service.consensus.migration.ConsensusRequestRouter;
import org.apache.cassandra.service.consensus.migration.ConsensusRequestRouter.RangeReadTarget;
import org.apache.cassandra.service.consensus.migration.ConsensusRequestRouter.RangeReadWithTarget;
+import org.apache.cassandra.service.reads.DataResolver;
import org.apache.cassandra.service.reads.ReadCallback;
import org.apache.cassandra.service.reads.ReadCoordinator;
import org.apache.cassandra.service.reads.repair.ReadRepair;
diff --git a/src/java/org/apache/cassandra/service/reads/tracked/ReadReconciliations.java b/src/java/org/apache/cassandra/service/reads/tracked/ReadReconciliations.java
index 8e3f2e1fca..fb3d36f995 100644
--- a/src/java/org/apache/cassandra/service/reads/tracked/ReadReconciliations.java
+++ b/src/java/org/apache/cassandra/service/reads/tracked/ReadReconciliations.java
@@ -161,6 +161,8 @@ public class ReadReconciliations implements ExpiredStatePurger.Expireable
{
private static final Logger logger = LoggerFactory.getLogger(Coordinator.class);
+ // FIXME: this will probably break per-DC consistency semantics of SatelliteReplicationStrategy
+ // once read speculation is implemented
private static final AtomicLongFieldUpdater<Coordinator> remainingUpdater =
AtomicLongFieldUpdater.newUpdater(Coordinator.class, "remaining");
private volatile long remaining; // three values packed into one atomic long
@@ -190,6 +192,24 @@ public class ReadReconciliations implements ExpiredStatePurger.Expireable
summaries = new Accumulator<>(remainingSummaries);
}
+ /**
+ * Confirm that we're only counting responses from nodes initially chosen by the read coordinator
+ * This is to prevent the implementation of tracked read speculation (doesn't exist yet) from breaking
+ * the per-dc consistency semantics of SatelliteReplicationStrategy because of the simple count completion
+ * mechanics this class uses for tracking received summarys / syncAcks
+ */
+ private void checkNodeIsExpected(int check)
+ {
+ if (check == dataNode)
+ return;
+
+ for (int node : summaryNodes)
+ if (check == node)
+ return;
+
+ throw new IllegalStateException("Not expecting response from node " + check);
+ }
+
/**
* For all the logs in the summary that are owned by us, preemptively prioritise delivery of
* any mutations that are absent from other participating nodes according to our primary coordinator
@@ -197,6 +217,7 @@ public class ReadReconciliations implements ExpiredStatePurger.Expireable
*/
boolean acceptLocalSummary(MutationSummary summary)
{
+ checkNodeIsExpected(LOCAL_NODE);
IntArrayList remoteNodes = new IntArrayList(summaryNodes.length, Integer.MIN_VALUE);
if (dataNode != LOCAL_NODE)
remoteNodes.addInt(dataNode);
@@ -238,6 +259,7 @@ public class ReadReconciliations implements ExpiredStatePurger.Expireable
*/
boolean acceptRemoteSummary(MutationSummary summary, int remoteNode)
{
+ checkNodeIsExpected(remoteNode);
Log2OffsetsMap.Mutable missingMutations = new Log2OffsetsMap.Mutable();
MutationTrackingService.instance().collectLocallyMissingMutations(summary, missingMutations);
@@ -257,8 +279,9 @@ public class ReadReconciliations implements ExpiredStatePurger.Expireable
return updateRemainingAndMaybeComplete(missingCount, -1, 0);
}
- boolean acceptSyncAck(int ignoredSyncId)
+ boolean acceptSyncAck(int node)
{
+ checkNodeIsExpected(node);
return updateRemainingAndMaybeComplete(0, 0, -1);
}
@@ -355,12 +378,12 @@ public class ReadReconciliations implements ExpiredStatePurger.Expireable
return next;
}
- protected boolean updateRemainingAndMaybeComplete(int mutationsDelta, int summariesDelta, int syncAcksDelta)
+ boolean updateRemainingAndMaybeComplete(int mutationsDelta, int summariesDelta, int syncAcksDelta)
{
return updateRemaining(mutationsDelta, summariesDelta, syncAcksDelta) == 0 && complete();
}
- protected boolean complete()
+ private boolean complete()
{
if (isDataNode())
MutationTrackingService.instance().localReads().acknowledgeReconcile(id, augmentingOffsets());
diff --git a/test/unit/org/apache/cassandra/ServerTestUtils.java b/test/unit/org/apache/cassandra/ServerTestUtils.java
index 448ac0468e..f9dd5fd34c 100644
--- a/test/unit/org/apache/cassandra/ServerTestUtils.java
+++ b/test/unit/org/apache/cassandra/ServerTestUtils.java
@@ -155,9 +155,6 @@ public final class ServerTestUtils
{
daemonInitialization();
- // Need to happen after daemonInitialization for config to be set, but before CFS initialization
- MutationJournal.start();
-
if (isServerPrepared)
return;
@@ -174,6 +171,10 @@ public final class ServerTestUtils
throw new RuntimeException(e);
}
+ // Need to happen after daemonInitialization for config to be set and after
+ // cleanupAndLeaveDirs for directories to exist, but before CFS initialization
+ MutationJournal.start();
+
try
{
remoteAddrs.add(InetAddressAndPort.getByName("127.0.0.4"));
diff --git a/test/unit/org/apache/cassandra/db/virtual/MutationJournalTableTest.java b/test/unit/org/apache/cassandra/db/virtual/MutationJournalTableTest.java
index 9bfd68e2b9..44a7557191 100644
--- a/test/unit/org/apache/cassandra/db/virtual/MutationJournalTableTest.java
+++ b/test/unit/org/apache/cassandra/db/virtual/MutationJournalTableTest.java
@@ -50,7 +50,8 @@ public class MutationJournalTableTest extends CQLTester
@Before
public void setUp()
{
- schemaChange("CREATE TABLE " + KEYSPACE + ".tbl(pk int PRIMARY KEY, v int)");
+ schemaChange("CREATE KEYSPACE ks WITH replication={'class':'SimpleStrategy', 'replication_factor':1} AND replication_type='tracked'");
+ schemaChange("CREATE TABLE ks.tbl(pk int PRIMARY KEY, v int)");
}
@Test
@@ -58,11 +59,12 @@ public class MutationJournalTableTest extends CQLTester
{
// Start the mutation journal
MutationJournal.start();
+ enableCoordinatorExecution();
// Write data to trigger journal writes
for (int i = 0; i < 100; i++)
{
- execute("INSERT INTO " + KEYSPACE + ".tbl(pk, v) VALUES (?, ?)", i, i);
+ execute("INSERT INTO ks.tbl(pk, v) VALUES (?, ?)", i, i);
}
// Query the virtual table
diff --git a/test/unit/org/apache/cassandra/index/IndexStatusManagerTest.java b/test/unit/org/apache/cassandra/index/IndexStatusManagerTest.java
index 8b28c3e464..4de1c7aeb4 100644
--- a/test/unit/org/apache/cassandra/index/IndexStatusManagerTest.java
+++ b/test/unit/org/apache/cassandra/index/IndexStatusManagerTest.java
@@ -393,7 +393,7 @@ public class IndexStatusManagerTest
Index.QueryPlan qp = mockedQueryPlan(indexes);
ConsistencyLevel cl = mockedConsistencyLevel(testcase.numRequired);
- EndpointsForRange actual = IndexStatusManager.instance.filterForQuery(endpoints, ks, qp, cl);
+ EndpointsForRange actual = IndexStatusManager.instance.filterForQueryOrThrow(endpoints, ks, qp, cl);
assertArrayEquals(
testcase.expected.stream().toArray(),
diff --git a/test/unit/org/apache/cassandra/locator/CompositeTrackerTest.java b/test/unit/org/apache/cassandra/locator/CompositeTrackerTest.java
new file mode 100644
index 0000000000..6c8d7fc000
--- /dev/null
+++ b/test/unit/org/apache/cassandra/locator/CompositeTrackerTest.java
@@ -0,0 +1,427 @@
+/*
+ * 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.locator;
+
+import java.net.UnknownHostException;
+import java.util.Arrays;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Test;
+
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.*;
+
+public class CompositeTrackerTest
+{
+ private InetAddressAndPort endpoint(String ip) throws UnknownHostException
+ {
+ return InetAddressAndPort.getByName(ip);
+ }
+
+ private ResponseTracker createMockTracker(boolean successful, boolean complete)
+ {
+ ResponseTracker tracker = mock(ResponseTracker.class);
+ when(tracker.isSuccessful()).thenReturn(successful);
+ when(tracker.isComplete()).thenReturn(complete);
+ when(tracker.required()).thenReturn(2);
+ when(tracker.received()).thenReturn(successful ? 2 : 0);
+ when(tracker.failures()).thenReturn(successful ? 0 : 2);
+ return tracker;
+ }
+
+ @Test
+ public void testQuorumCalculation()
+ {
+ assertEquals(1, CompositeTracker.quorum(1));
+ assertEquals(2, CompositeTracker.quorum(2));
+ assertEquals(2, CompositeTracker.quorum(3));
+ assertEquals(3, CompositeTracker.quorum(4));
+ assertEquals(3, CompositeTracker.quorum(5));
+ }
+
+ @Test
+ public void testQuorumSuccessAndFailure()
+ {
+ // N=1: quorum=1
+ assertTrue(new CompositeTracker(CompositeTracker.quorum(1), Arrays.asList(
+ createMockTracker(true, true)
+ )).isSuccessful());
+
+ // N=2: quorum=2 (both required)
+ assertFalse(new CompositeTracker(CompositeTracker.quorum(2), Arrays.asList(
+ createMockTracker(true, true), createMockTracker(false, false)
+ )).isSuccessful());
+
+ assertTrue(new CompositeTracker(CompositeTracker.quorum(2), Arrays.asList(
+ createMockTracker(true, true), createMockTracker(true, true)
+ )).isSuccessful());
+
+ // N=3: quorum=2, last child fails
+ assertTrue(new CompositeTracker(CompositeTracker.quorum(3), Arrays.asList(
+ createMockTracker(true, true), createMockTracker(true, true), createMockTracker(false, false)
+ )).isSuccessful());
+
+ // N=3: quorum=2, first child fails (any child can be the failure)
+ assertTrue(new CompositeTracker(CompositeTracker.quorum(3), Arrays.asList(
+ createMockTracker(false, true), createMockTracker(true, true), createMockTracker(true, true)
+ )).isSuccessful());
+
+ // N=3: only 1 succeeds → not successful
+ assertFalse(new CompositeTracker(CompositeTracker.quorum(3), Arrays.asList(
+ createMockTracker(true, true), createMockTracker(false, true), createMockTracker(false, false)
+ )).isSuccessful());
+
+ // N=4: quorum=3
+ assertFalse(new CompositeTracker(CompositeTracker.quorum(4), Arrays.asList(
+ createMockTracker(true, true), createMockTracker(true, true), createMockTracker(false, false), createMockTracker(false, false)
+ )).isSuccessful());
+
+ assertTrue(new CompositeTracker(CompositeTracker.quorum(4), Arrays.asList(
+ createMockTracker(true, true), createMockTracker(true, true), createMockTracker(true, true), createMockTracker(false, false)
+ )).isSuccessful());
+
+ // N=5: quorum=3
+ assertTrue(new CompositeTracker(CompositeTracker.quorum(5), Arrays.asList(
+ createMockTracker(true, true), createMockTracker(true, true), createMockTracker(true, true), createMockTracker(false, false), createMockTracker(false, false)
+ )).isSuccessful());
+ }
+
+ @Test
+ public void testAllSuccessAndFailure()
+ {
+ // All succeed
+ CompositeTracker tracker = new CompositeTracker(2,
+ createMockTracker(true, true),
+ createMockTracker(true, true)
+ );
+ assertTrue(tracker.isSuccessful());
+ assertTrue(tracker.isComplete());
+
+ // First fails
+ assertFalse(new CompositeTracker(2,
+ createMockTracker(false, true),
+ createMockTracker(true, true)
+ ).isSuccessful());
+
+ // Second fails
+ assertFalse(new CompositeTracker(2,
+ createMockTracker(true, true),
+ createMockTracker(false, true)
+ ).isSuccessful());
+
+ // Both fail
+ tracker = new CompositeTracker(2,
+ createMockTracker(false, true),
+ createMockTracker(false, true)
+ );
+ assertFalse(tracker.isSuccessful());
+ assertTrue(tracker.isComplete());
+
+ // Any single failure in N children → overall failure
+ assertFalse(new CompositeTracker(3,
+ createMockTracker(true, true),
+ createMockTracker(true, true),
+ createMockTracker(false, true)
+ ).isSuccessful());
+ }
+
+ @Test
+ public void testQuorumEarlyCompletionWhenSuccessful()
+ {
+ CompositeTracker tracker = new CompositeTracker(CompositeTracker.quorum(3), Arrays.asList(
+ createMockTracker(true, true),
+ createMockTracker(true, true),
+ createMockTracker(false, false)
+ ));
+
+ assertTrue(tracker.isComplete());
+ }
+
+ @Test
+ public void testQuorumEarlyCompletionWhenImpossible()
+ {
+ // 4 children, 2 failed → max possible = 2 < quorum(4) → complete
+ CompositeTracker tracker = new CompositeTracker(CompositeTracker.quorum(4), Arrays.asList(
+ createMockTracker(true, false),
+ createMockTracker(false, true),
+ createMockTracker(false, true),
+ createMockTracker(false, false)
+ ));
+
+ assertFalse(tracker.isSuccessful());
+ assertTrue(tracker.isComplete());
+ }
+
+ @Test
+ public void testQuorumNotCompleteWhenStillPossible()
+ {
+ // 4 children: 1 succeeded, 1 failed, 2 pending → max possible = 3 >= quorum(4)
+ CompositeTracker tracker = new CompositeTracker(CompositeTracker.quorum(4), Arrays.asList(
+ createMockTracker(true, true),
+ createMockTracker(false, true),
+ createMockTracker(false, false),
+ createMockTracker(false, false)
+ ));
+
+ assertFalse(tracker.isComplete());
+ }
+
+ @Test
+ public void testAllEarlyCompletionOnAnyFailure()
+ {
+ CompositeTracker tracker = new CompositeTracker(3,
+ createMockTracker(true, true),
+ createMockTracker(false, true), // Failed
+ createMockTracker(false, false) // Still pending
+ );
+
+ // One child has definitively failed → can't all succeed
+ assertTrue(tracker.isComplete());
+ assertFalse(tracker.isSuccessful());
+ }
+
+ @Test
+ public void testAllNotCompleteWhenPending()
+ {
+ CompositeTracker tracker = new CompositeTracker(2,
+ createMockTracker(true, true),
+ createMockTracker(false, false) // Pending, not failed
+ );
+
+ assertFalse(tracker.isComplete());
+ assertFalse(tracker.isSuccessful());
+ }
+
+ @Test
+ public void testOnResponseDelegatesToAll() throws Exception
+ {
+ ResponseTracker c0 = mock(ResponseTracker.class);
+ ResponseTracker c1 = mock(ResponseTracker.class);
+ ResponseTracker c2 = mock(ResponseTracker.class);
+
+ CompositeTracker tracker = new CompositeTracker(CompositeTracker.quorum(3), c0, c1, c2);
+
+ InetAddressAndPort ep = endpoint("127.0.0.1");
+ tracker.onResponse(ep);
+
+ verify(c0).onResponse(ep);
+ verify(c1).onResponse(ep);
+ verify(c2).onResponse(ep);
+ }
+
+ @Test
+ public void testOnFailureDelegatesToAll() throws Exception
+ {
+ ResponseTracker c0 = mock(ResponseTracker.class);
+ ResponseTracker c1 = mock(ResponseTracker.class);
+ ResponseTracker c2 = mock(ResponseTracker.class);
+
+ CompositeTracker tracker = new CompositeTracker(3, c0, c1, c2);
+
+ InetAddressAndPort ep = endpoint("127.0.0.1");
+ tracker.onFailure(ep);
+
+ verify(c0).onFailure(ep);
+ verify(c1).onFailure(ep);
+ verify(c2).onFailure(ep);
+ }
+
+ @Test
+ public void testAggregatesSums()
+ {
+ ResponseTracker c0 = mock(ResponseTracker.class);
+ when(c0.received()).thenReturn(2);
+ ResponseTracker c1 = mock(ResponseTracker.class);
+ when(c1.received()).thenReturn(1);
+ ResponseTracker c2 = mock(ResponseTracker.class);
+ when(c2.received()).thenReturn(3);
+
+ CompositeTracker tracker = new CompositeTracker(CompositeTracker.quorum(3), c0, c1, c2);
+
+ assertEquals(6, tracker.received());
+ }
+
+ @Test
+ public void testFailuresSum()
+ {
+ ResponseTracker c0 = mock(ResponseTracker.class);
+ when(c0.failures()).thenReturn(1);
+ ResponseTracker c1 = mock(ResponseTracker.class);
+ when(c1.failures()).thenReturn(2);
+ ResponseTracker c2 = mock(ResponseTracker.class);
+ when(c2.failures()).thenReturn(0);
+
+ CompositeTracker tracker = new CompositeTracker(3, c0, c1, c2);
+
+ assertEquals(3, tracker.failures());
+ }
+
+ @Test
+ public void testCountsTowardQuorumFromAny() throws Exception
+ {
+ ResponseTracker c0 = mock(ResponseTracker.class);
+ when(c0.countsTowardQuorum(any())).thenReturn(true);
+ ResponseTracker c1 = mock(ResponseTracker.class);
+ when(c1.countsTowardQuorum(any())).thenReturn(false);
+
+ CompositeTracker tracker = new CompositeTracker(2, c0, c1);
+
+ assertTrue(tracker.countsTowardQuorum(endpoint("127.0.0.1")));
+ }
+
+ @Test
+ public void testCountsTowardQuorumFromNone() throws Exception
+ {
+ ResponseTracker c0 = mock(ResponseTracker.class);
+ when(c0.countsTowardQuorum(any())).thenReturn(false);
+ ResponseTracker c1 = mock(ResponseTracker.class);
+ when(c1.countsTowardQuorum(any())).thenReturn(false);
+
+ CompositeTracker tracker = new CompositeTracker(CompositeTracker.quorum(2), c0, c1);
+
+ assertFalse(tracker.countsTowardQuorum(endpoint("127.0.0.1")));
+ }
+
+ @Test
+ public void testNestedComposition()
+ {
+ // CompositeTracker(all) containing CompositeTracker(quorum)s
+ CompositeTracker inner1 = new CompositeTracker(CompositeTracker.quorum(3),
+ createMockTracker(true, true),
+ createMockTracker(true, true),
+ createMockTracker(false, false)
+ );
+
+ CompositeTracker inner2 = new CompositeTracker(CompositeTracker.quorum(2),
+ createMockTracker(true, true),
+ createMockTracker(true, true)
+ );
+
+ CompositeTracker outer = new CompositeTracker(2, inner1, inner2);
+
+ assertTrue(outer.isSuccessful());
+ assertTrue(outer.isComplete());
+ }
+
+ @Test
+ public void testNestedCompositionWithFailure()
+ {
+ CompositeTracker inner1 = new CompositeTracker(CompositeTracker.quorum(2),
+ createMockTracker(true, true),
+ createMockTracker(true, true)
+ );
+
+ CompositeTracker inner2 = new CompositeTracker(CompositeTracker.quorum(2),
+ createMockTracker(false, true),
+ createMockTracker(false, true)
+ );
+
+ CompositeTracker outer = new CompositeTracker(2, inner1, inner2);
+
+ assertFalse(outer.isSuccessful());
+ assertTrue(outer.isComplete());
+ }
+
+ @Test
+ public void testConcurrentResponses() throws Exception
+ {
+ SimpleResponseTracker c0 = new SimpleResponseTracker(2, 3);
+ SimpleResponseTracker c1 = new SimpleResponseTracker(2, 3);
+ SimpleResponseTracker c2 = new SimpleResponseTracker(2, 3);
+
+ CompositeTracker tracker = new CompositeTracker(CompositeTracker.quorum(3), c0, c1, c2);
+
+ ExecutorService executor = Executors.newFixedThreadPool(10);
+ CountDownLatch startLatch = new CountDownLatch(1);
+ CountDownLatch doneLatch = new CountDownLatch(9);
+
+ try
+ {
+ for (int i = 0; i < 9; i++)
+ {
+ final int index = i;
+ executor.submit(() -> {
+ try
+ {
+ startLatch.await();
+ tracker.onResponse(endpoint("127.0.0." + index));
+ doneLatch.countDown();
+ }
+ catch (Exception e)
+ {
+ throw new RuntimeException(e);
+ }
+ });
+ }
+
+ startLatch.countDown();
+ assertTrue(doneLatch.await(10, TimeUnit.SECONDS));
+
+ assertTrue(tracker.isSuccessful());
+ assertTrue(tracker.isComplete());
+ assertEquals(27, tracker.received());
+ }
+ finally
+ {
+ executor.shutdownNow();
+ }
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testNullChildren()
+ {
+ new CompositeTracker(1, (ResponseTracker[]) null);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testEmptyChildren()
+ {
+ new CompositeTracker(1);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testBlockForZero()
+ {
+ new CompositeTracker(0, mock(ResponseTracker.class));
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testBlockForExceedsChildren()
+ {
+ new CompositeTracker(3, mock(ResponseTracker.class), mock(ResponseTracker.class));
+ }
+
+ @Test
+ public void testToString()
+ {
+ CompositeTracker tracker = new CompositeTracker(CompositeTracker.quorum(3),
+ mock(ResponseTracker.class),
+ mock(ResponseTracker.class),
+ mock(ResponseTracker.class)
+ );
+
+ String str = tracker.toString();
+ assertTrue(str.contains("CompositeTracker"));
+ assertTrue(str.contains("children=3"));
+ assertTrue(str.contains("blockFor=2"));
+ }
+}
diff --git a/test/unit/org/apache/cassandra/locator/PerDcResponseTrackerTest.java b/test/unit/org/apache/cassandra/locator/PerDcResponseTrackerTest.java
deleted file mode 100644
index 0af71cdcea..0000000000
--- a/test/unit/org/apache/cassandra/locator/PerDcResponseTrackerTest.java
+++ /dev/null
@@ -1,401 +0,0 @@
-/*
- * 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.locator;
-
-import java.net.UnknownHostException;
-import java.util.HashMap;
-import java.util.Map;
-
-import org.junit.Test;
-
-import org.apache.cassandra.exceptions.RequestFailureReason;
-import org.apache.cassandra.tcm.membership.Location;
-
-import static org.junit.Assert.*;
-import static org.mockito.Mockito.*;
-
-public class PerDcResponseTrackerTest
-{
- private static final RequestFailureReason TIMEOUT = RequestFailureReason.TIMEOUT;
-
- private InetAddressAndPort endpoint(String ip) throws UnknownHostException
- {
- return InetAddressAndPort.getByName(ip);
- }
-
- private Locator mockLocator(Map<InetAddressAndPort, String> endpointToDc) throws Exception
- {
- Locator locator = mock(Locator.class);
- for (Map.Entry<InetAddressAndPort, String> entry : endpointToDc.entrySet())
- {
- when(locator.location(entry.getKey())).thenReturn(new Location(entry.getValue(), "rack1"));
- }
- return locator;
- }
-
- /**
- * Helper to create a map of SimpleResponseTrackers from blockFor/totalReplicas config.
- */
- private Map<String, ResponseTracker> simpleTrackers(Object... dcBlockForTotal)
- {
- Map<String, ResponseTracker> trackers = new HashMap<>();
- for (int i = 0; i < dcBlockForTotal.length; i += 3)
- {
- String dc = (String) dcBlockForTotal[i];
- int blockFor = (Integer) dcBlockForTotal[i + 1];
- int totalReplicas = (Integer) dcBlockForTotal[i + 2];
- trackers.put(dc, new SimpleResponseTracker(blockFor, totalReplicas));
- }
- return trackers;
- }
-
- @Test
- public void testAllDcsReachQuorum() throws Exception
- {
- Map<String, ResponseTracker> trackers = simpleTrackers(
- "DC1", 2, 3,
- "DC2", 2, 3
- );
-
- Map<InetAddressAndPort, String> endpointToDc = new HashMap<>();
- endpointToDc.put(endpoint("127.0.0.1"), "DC1");
- endpointToDc.put(endpoint("127.0.0.2"), "DC1");
- endpointToDc.put(endpoint("127.0.0.3"), "DC1");
- endpointToDc.put(endpoint("192.168.1.1"), "DC2");
- endpointToDc.put(endpoint("192.168.1.2"), "DC2");
- endpointToDc.put(endpoint("192.168.1.3"), "DC2");
-
- Locator locator = mockLocator(endpointToDc);
- PerDcResponseTracker tracker = new PerDcResponseTracker(trackers, locator);
-
- // DC1 reaches quorum
- tracker.onResponse(endpoint("127.0.0.1"));
- tracker.onResponse(endpoint("127.0.0.2"));
- assertFalse(tracker.isComplete()); // DC2 not done yet
-
- // DC2 reaches quorum
- tracker.onResponse(endpoint("192.168.1.1"));
- tracker.onResponse(endpoint("192.168.1.2"));
-
- assertTrue(tracker.isComplete());
- assertTrue(tracker.isSuccessful());
- assertEquals(4, tracker.received());
- assertEquals(4, tracker.required()); // 2 + 2
- }
-
- @Test
- public void testOneDcFails() throws Exception
- {
- Map<String, ResponseTracker> trackers = simpleTrackers(
- "DC1", 2, 3,
- "DC2", 2, 3
- );
-
- Map<InetAddressAndPort, String> endpointToDc = new HashMap<>();
- endpointToDc.put(endpoint("127.0.0.1"), "DC1");
- endpointToDc.put(endpoint("127.0.0.2"), "DC1");
- endpointToDc.put(endpoint("127.0.0.3"), "DC1");
- endpointToDc.put(endpoint("192.168.1.1"), "DC2");
- endpointToDc.put(endpoint("192.168.1.2"), "DC2");
- endpointToDc.put(endpoint("192.168.1.3"), "DC2");
-
- Locator locator = mockLocator(endpointToDc);
- PerDcResponseTracker tracker = new PerDcResponseTracker(trackers, locator);
-
- // DC1 reaches quorum
- tracker.onResponse(endpoint("127.0.0.1"));
- tracker.onResponse(endpoint("127.0.0.2"));
-
- // DC2 fails (all fail)
- tracker.onFailure(endpoint("192.168.1.1"), TIMEOUT);
- tracker.onFailure(endpoint("192.168.1.2"), TIMEOUT);
- tracker.onFailure(endpoint("192.168.1.3"), TIMEOUT);
-
- assertTrue(tracker.isComplete()); // DC2 reached definite failure
- assertFalse(tracker.isSuccessful()); // DC2 failed
- assertEquals(2, tracker.received());
- assertEquals(3, tracker.failures());
- }
-
- @Test
- public void testPartialDcProgress() throws Exception
- {
- Map<String, ResponseTracker> trackers = simpleTrackers(
- "DC1", 2, 3,
- "DC2", 2, 3
- );
-
- Map<InetAddressAndPort, String> endpointToDc = new HashMap<>();
- endpointToDc.put(endpoint("127.0.0.1"), "DC1");
- endpointToDc.put(endpoint("192.168.1.1"), "DC2");
-
- Locator locator = mockLocator(endpointToDc);
- PerDcResponseTracker tracker = new PerDcResponseTracker(trackers, locator);
-
- // DC1 gets one response
- tracker.onResponse(endpoint("127.0.0.1"));
-
- assertFalse(tracker.isComplete());
- assertFalse(tracker.isSuccessful());
- assertEquals(1, tracker.received());
- assertEquals(4, tracker.required()); // 2 + 2
- }
-
- @Test
- public void testIgnoresUnknownDc() throws Exception
- {
- Map<String, ResponseTracker> trackers = simpleTrackers(
- "DC1", 2, 3
- );
-
- Map<InetAddressAndPort, String> endpointToDc = new HashMap<>();
- endpointToDc.put(endpoint("127.0.0.1"), "DC1");
- endpointToDc.put(endpoint("127.0.0.2"), "DC1");
- endpointToDc.put(endpoint("192.168.1.1"), "DC2"); // DC2 not in config
-
- Locator locator = mockLocator(endpointToDc);
- PerDcResponseTracker tracker = new PerDcResponseTracker(trackers, locator);
-
- // DC2 response is ignored
- tracker.onResponse(endpoint("192.168.1.1"));
- assertFalse(tracker.countsTowardQuorum(endpoint("192.168.1.1")));
- assertEquals(0, tracker.received());
-
- // DC1 responses count
- tracker.onResponse(endpoint("127.0.0.1"));
- tracker.onResponse(endpoint("127.0.0.2"));
-
- assertTrue(tracker.isComplete());
- assertTrue(tracker.isSuccessful());
- assertEquals(2, tracker.received());
- }
-
- @Test
- public void testAsymmetricRequirements() throws Exception
- {
- Map<String, ResponseTracker> trackers = simpleTrackers(
- "DC1", 3, 5, // Need 3 of 5
- "DC2", 2, 3 // Need 2 of 3
- );
-
- Map<InetAddressAndPort, String> endpointToDc = new HashMap<>();
- endpointToDc.put(endpoint("127.0.0.1"), "DC1");
- endpointToDc.put(endpoint("127.0.0.2"), "DC1");
- endpointToDc.put(endpoint("127.0.0.3"), "DC1");
- endpointToDc.put(endpoint("192.168.1.1"), "DC2");
- endpointToDc.put(endpoint("192.168.1.2"), "DC2");
-
- Locator locator = mockLocator(endpointToDc);
- PerDcResponseTracker tracker = new PerDcResponseTracker(trackers, locator);
-
- // DC2 reaches quorum (2 of 3)
- tracker.onResponse(endpoint("192.168.1.1"));
- tracker.onResponse(endpoint("192.168.1.2"));
- assertFalse(tracker.isComplete()); // DC1 not done
-
- // DC1 reaches quorum (3 of 5)
- tracker.onResponse(endpoint("127.0.0.1"));
- tracker.onResponse(endpoint("127.0.0.2"));
- tracker.onResponse(endpoint("127.0.0.3"));
-
- assertTrue(tracker.isComplete());
- assertTrue(tracker.isSuccessful());
- assertEquals(5, tracker.received());
- assertEquals(5, tracker.required()); // 3 + 2
- }
-
- // Aggregation tests
-
- @Test
- public void testRequiredSum() throws Exception
- {
- Map<String, ResponseTracker> trackers = simpleTrackers(
- "DC1", 3, 5,
- "DC2", 2, 3,
- "DC3", 1, 2
- );
-
- Map<InetAddressAndPort, String> endpointToDc = new HashMap<>();
- Locator locator = mockLocator(endpointToDc);
-
- PerDcResponseTracker tracker = new PerDcResponseTracker(trackers, locator);
-
- assertEquals(6, tracker.required()); // 3 + 2 + 1
- }
-
- @Test
- public void testReceivedSum() throws Exception
- {
- Map<String, ResponseTracker> trackers = simpleTrackers(
- "DC1", 2, 3,
- "DC2", 2, 3
- );
-
- Map<InetAddressAndPort, String> endpointToDc = new HashMap<>();
- endpointToDc.put(endpoint("127.0.0.1"), "DC1");
- endpointToDc.put(endpoint("127.0.0.2"), "DC1");
- endpointToDc.put(endpoint("192.168.1.1"), "DC2");
-
- Locator locator = mockLocator(endpointToDc);
- PerDcResponseTracker tracker = new PerDcResponseTracker(trackers, locator);
-
- tracker.onResponse(endpoint("127.0.0.1"));
- tracker.onResponse(endpoint("127.0.0.2"));
- tracker.onResponse(endpoint("192.168.1.1"));
-
- assertEquals(3, tracker.received()); // 2 from DC1 + 1 from DC2
- }
-
- @Test
- public void testFailuresSum() throws Exception
- {
- Map<String, ResponseTracker> trackers = simpleTrackers(
- "DC1", 2, 3,
- "DC2", 2, 3
- );
-
- Map<InetAddressAndPort, String> endpointToDc = new HashMap<>();
- endpointToDc.put(endpoint("127.0.0.1"), "DC1");
- endpointToDc.put(endpoint("192.168.1.1"), "DC2");
- endpointToDc.put(endpoint("192.168.1.2"), "DC2");
-
- Locator locator = mockLocator(endpointToDc);
- PerDcResponseTracker tracker = new PerDcResponseTracker(trackers, locator);
-
- tracker.onFailure(endpoint("127.0.0.1"), TIMEOUT);
- tracker.onFailure(endpoint("192.168.1.1"), TIMEOUT);
- tracker.onFailure(endpoint("192.168.1.2"), TIMEOUT);
-
- assertEquals(3, tracker.failures()); // 1 from DC1 + 2 from DC2
- }
-
- // Composition tests
-
- @Test
- public void testCountsTowardQuorum() throws Exception
- {
- Map<String, ResponseTracker> trackers = simpleTrackers(
- "DC1", 2, 3
- );
-
- Map<InetAddressAndPort, String> endpointToDc = new HashMap<>();
- endpointToDc.put(endpoint("127.0.0.1"), "DC1");
- endpointToDc.put(endpoint("192.168.1.1"), "DC2");
-
- Locator locator = mockLocator(endpointToDc);
- PerDcResponseTracker tracker = new PerDcResponseTracker(trackers, locator);
-
- assertTrue(tracker.countsTowardQuorum(endpoint("127.0.0.1"))); // DC1 tracked
- assertFalse(tracker.countsTowardQuorum(endpoint("192.168.1.1"))); // DC2 not tracked
- }
-
- @Test
- public void testGetTrackerForDc() throws Exception
- {
- Map<String, ResponseTracker> trackers = simpleTrackers(
- "DC1", 2, 3,
- "DC2", 1, 2
- );
-
- Locator locator = mock(Locator.class);
- PerDcResponseTracker tracker = new PerDcResponseTracker(trackers, locator);
-
- assertNotNull(tracker.getTrackerForDc("DC1"));
- assertNotNull(tracker.getTrackerForDc("DC2"));
- assertNull(tracker.getTrackerForDc("DC3"));
- assertEquals(2, tracker.getTrackerForDc("DC1").required());
- assertEquals(1, tracker.getTrackerForDc("DC2").required());
- }
-
- @Test
- public void testWithWriteResponseTrackers() throws Exception
- {
- // Test that PerDcResponseTracker works with WriteResponseTrackers (double-count model)
- Map<String, ResponseTracker> trackers = new HashMap<>();
- // DC1: baseBlockFor=2, totalBlockFor=3, committed=3, pending=1
- trackers.put("DC1", new WriteResponseTracker(2, 3, 3, 1,
- addr -> addr.getHostAddress(false).equals("127.0.0.4"))); // .4 is pending
- // DC2: no pending, degenerates to simple case
- trackers.put("DC2", new SimpleResponseTracker(2, 3));
-
- Map<InetAddressAndPort, String> endpointToDc = new HashMap<>();
- endpointToDc.put(endpoint("127.0.0.1"), "DC1");
- endpointToDc.put(endpoint("127.0.0.2"), "DC1");
- endpointToDc.put(endpoint("127.0.0.3"), "DC1");
- endpointToDc.put(endpoint("127.0.0.4"), "DC1"); // pending
- endpointToDc.put(endpoint("192.168.1.1"), "DC2");
- endpointToDc.put(endpoint("192.168.1.2"), "DC2");
- endpointToDc.put(endpoint("192.168.1.3"), "DC2");
-
- Locator locator = mockLocator(endpointToDc);
- PerDcResponseTracker tracker = new PerDcResponseTracker(trackers, locator);
-
- // DC1: 2 committed successes (meets base requirement but not total)
- tracker.onResponse(endpoint("127.0.0.1"));
- tracker.onResponse(endpoint("127.0.0.2"));
- assertFalse(tracker.isComplete());
-
- // DC2: 2 successes (meets requirement)
- tracker.onResponse(endpoint("192.168.1.1"));
- tracker.onResponse(endpoint("192.168.1.2"));
- assertFalse(tracker.isComplete()); // DC1 still needs pending
-
- // DC1: 1 pending success (now meets total requirement)
- tracker.onResponse(endpoint("127.0.0.4"));
-
- assertTrue(tracker.isComplete());
- assertTrue(tracker.isSuccessful());
- }
-
- // Validation tests
-
- @Test(expected = IllegalArgumentException.class)
- public void testNullTrackers()
- {
- Locator locator = mock(Locator.class);
- new PerDcResponseTracker(null, locator);
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void testEmptyTrackers()
- {
- Locator locator = mock(Locator.class);
- new PerDcResponseTracker(new HashMap<>(), locator);
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void testNullLocator()
- {
- Map<String, ResponseTracker> trackers = simpleTrackers("DC1", 2, 3);
- new PerDcResponseTracker(trackers, null);
- }
-
- @Test
- public void testToString() throws Exception
- {
- Map<String, ResponseTracker> trackers = simpleTrackers("DC1", 2, 3);
-
- Locator locator = mock(Locator.class);
- PerDcResponseTracker tracker = new PerDcResponseTracker(trackers, locator);
-
- String str = tracker.toString();
- assertTrue(str.contains("PerDcResponseTracker"));
- assertTrue(str.contains("DC1"));
- }
-}
diff --git a/test/unit/org/apache/cassandra/locator/SatelliteCommitPlanTest.java b/test/unit/org/apache/cassandra/locator/SatelliteCommitPlanTest.java
new file mode 100644
index 0000000000..ef12ed3d43
--- /dev/null
+++ b/test/unit/org/apache/cassandra/locator/SatelliteCommitPlanTest.java
@@ -0,0 +1,199 @@
+/*
+ * 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.locator;
+
+import java.net.UnknownHostException;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.junit.After;
+import org.junit.Test;
+
+import org.apache.cassandra.CassandraTestBase;
+import org.apache.cassandra.CassandraTestBase.UseMurmur3Partitioner;
+import org.apache.cassandra.ServerTestUtils;
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.marshal.AsciiType;
+import org.apache.cassandra.dht.Murmur3Partitioner.LongToken;
+import org.apache.cassandra.distributed.test.log.ClusterMetadataTestHelper;
+import org.apache.cassandra.locator.AbstractReplicaCollection.ReplicaList;
+import org.apache.cassandra.schema.KeyspaceMetadata;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.service.paxos.Paxos;
+import org.apache.cassandra.service.paxos.SatellitePaxosParticipants;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.membership.Location;
+
+import static org.apache.cassandra.CassandraTestBase.DisableMBeanRegistration;
+import static org.apache.cassandra.CassandraTestBase.PrepareServerNoRegister;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+@PrepareServerNoRegister
+@DisableMBeanRegistration
+@UseMurmur3Partitioner
+public class SatelliteCommitPlanTest extends CassandraTestBase
+{
+ private static final String KEYSPACE = "scp_test";
+ private static final LongToken TOKEN = new LongToken(150);
+
+ @After
+ public void teardown()
+ {
+ ServerTestUtils.resetCMS();
+ }
+
+ private void addToken(long token, String address, Location location) throws UnknownHostException
+ {
+ InetAddressAndPort addr = InetAddressAndPort.getByName(address);
+ ClusterMetadataTestHelper.addEndpoint(addr, new LongToken(token), location);
+ }
+
+ private void setupTopology() throws UnknownHostException
+ {
+ DatabaseDescriptor.setPaxosVariant(Config.PaxosVariant.v2);
+
+ Location dc1 = new Location("dc1", "rack1");
+ Location dc2 = new Location("dc2", "rack1");
+ Location sat1 = new Location("sat1", "rack1");
+ Location sat2 = new Location("sat2", "rack1");
+
+ addToken(100, "10.0.0.10", dc1);
+ addToken(200, "10.0.0.11", dc1);
+ addToken(300, "10.0.0.12", dc1);
+
+ addToken(400, "10.1.0.10", dc2);
+ addToken(500, "10.1.0.11", dc2);
+ addToken(600, "10.1.0.12", dc2);
+
+ addToken(700, "10.2.0.10", sat1);
+ addToken(800, "10.2.0.11", sat1);
+ addToken(1100, "10.2.0.12", sat1);
+
+ addToken(900, "10.3.0.10", sat2);
+ addToken(1000, "10.3.0.11", sat2);
+ addToken(1200, "10.3.0.12", sat2);
+ }
+
+ private void createDualDCKeyspace() throws Exception
+ {
+ String cql = "CREATE KEYSPACE " + KEYSPACE + " WITH replication = {" +
+ "'class': 'SatelliteReplicationStrategy', " +
+ "'dc1': '3', " +
+ "'dc1.satellite.sat1': '3/3', " +
+ "'dc2': '3', " +
+ "'dc2.satellite.sat2': '3/3', " +
+ "'primary': 'dc1'" +
+ "} AND replication_type = 'tracked'";
+ ClusterMetadataTestHelper.createKeyspace(cql);
+ }
+
+ private SatelliteReplicationStrategy getSRS()
+ {
+ KeyspaceMetadata ksm = ClusterMetadata.current().schema.getKeyspaces().getNullable(KEYSPACE);
+ return (SatelliteReplicationStrategy) ksm.replicationStrategy;
+ }
+
+ private SatelliteReplicationStrategy.SatelliteCommitPlan createPlan() throws Exception
+ {
+ ClusterMetadata metadata = ClusterMetadata.current();
+ SatelliteReplicationStrategy srs = getSRS();
+ Keyspace keyspace = Keyspace.mockKS(metadata.schema.getKeyspaces().getNullable(KEYSPACE));
+ return srs.createSatelliteCommitPlan(metadata, keyspace, TOKEN);
+ }
+
+ private Set<String> collectDCs(AbstractReplicaCollection<?> endpoints)
+ {
+ ClusterMetadata metadata = ClusterMetadata.current();
+ Set<String> dcs = new HashSet<>();
+ ReplicaList epList = endpoints.list;
+ for (int i = 0; i < endpoints.size(); i++)
+ dcs.add(metadata.locator.location(epList.get(i).endpoint()).datacenter);
+ return dcs;
+ }
+
+ private void assertExpectedDCs(Set<String> dcs)
+ {
+ assertTrue("Should include sat1 (primary's satellite)", dcs.contains("sat1"));
+ assertTrue("Should include dc2 (other full DC)", dcs.contains("dc2"));
+ assertFalse("Should NOT include sat2 (dc2's satellite, not primary's)", dcs.contains("sat2"));
+ assertFalse("Should NOT include dc1 (primary, handled by paxos)", dcs.contains("dc1"));
+ }
+
+ /**
+ * Both createSatelliteCommitPlan and paxosParticipants should include only
+ * the primary DC's satellite (sat1) and other full DCs (dc2), excluding
+ * the primary DC itself (dc1) and non-primary satellites (sat2).
+ */
+ @Test
+ public void testEndpointDCSelection() throws Exception
+ {
+ setupTopology();
+ createDualDCKeyspace();
+
+ // check commit plan endpoint selection
+ SatelliteReplicationStrategy.SatelliteCommitPlan plan = createPlan();
+ assertExpectedDCs(collectDCs(plan.liveEndpoints));
+
+ // check paxos participant endpoint selection
+ ClusterMetadata metadata = ClusterMetadata.current();
+ SatelliteReplicationStrategy srs = getSRS();
+ TableMetadata table = TableMetadata.builder(KEYSPACE, "test_table")
+ .addPartitionKeyColumn("key", AsciiType.instance)
+ .build();
+
+ Paxos.Participants participants = srs.paxosParticipants(metadata, table,
+ TOKEN,
+ ConsistencyLevel.SERIAL,
+ r -> true);
+
+ assertTrue(participants instanceof SatellitePaxosParticipants);
+ SatellitePaxosParticipants spp = (SatellitePaxosParticipants) participants;
+ assertExpectedDCs(collectDCs(spp.getAdditionalSummaryEndpoints()));
+ }
+
+ /**
+ * The tracker should not be complete with only the pre-completed primary DC,
+ * but should complete once a quorum of groups has responded (dc1 pre-completed + sat1).
+ */
+ @Test
+ public void testTrackerCompletesWithQuorumOfGroups() throws Exception
+ {
+ setupTopology();
+ createDualDCKeyspace();
+
+ SatelliteReplicationStrategy.SatelliteCommitPlan plan = createPlan();
+ ClusterMetadata metadata = ClusterMetadata.current();
+
+ assertFalse("Tracker should not be complete with only primary DC pre-completed", plan.tracker.isComplete());
+
+ // meet quorum in sat1
+ for (int i = 0; i < plan.liveEndpoints.size(); i++)
+ {
+ InetAddressAndPort ep = plan.liveEndpoints.endpoint(i);
+ if (metadata.locator.location(ep).datacenter.equals("sat1"))
+ plan.tracker.onResponse(ep);
+ }
+
+ assertTrue("Should be complete with quorum of groups", plan.tracker.isComplete());
+ assertTrue("Should be successful", plan.tracker.isSuccessful());
+ }
+}
diff --git a/test/unit/org/apache/cassandra/locator/SatellitePaxosFailoverTest.java b/test/unit/org/apache/cassandra/locator/SatellitePaxosFailoverTest.java
new file mode 100644
index 0000000000..603d218e61
--- /dev/null
+++ b/test/unit/org/apache/cassandra/locator/SatellitePaxosFailoverTest.java
@@ -0,0 +1,394 @@
+/*
+ * 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.locator;
+
+import java.net.InetAddress;
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.BufferDecoratedKey;
+import org.apache.cassandra.db.SinglePartitionReadCommand;
+import org.apache.cassandra.db.marshal.AsciiType;
+import org.apache.cassandra.db.partitions.PartitionUpdate;
+import org.apache.cassandra.dht.Murmur3Partitioner.LongToken;
+import org.apache.cassandra.distributed.test.log.ClusterMetadataTestHelper;
+import org.apache.cassandra.exceptions.UnavailableException;
+import org.apache.cassandra.gms.Gossiper;
+import org.apache.cassandra.net.Message;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.net.NoPayload;
+import org.apache.cassandra.net.Verb;
+import org.apache.cassandra.replication.MutationId;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.service.paxos.Ballot;
+import org.apache.cassandra.service.paxos.Commit;
+import org.apache.cassandra.service.paxos.Paxos;
+import org.apache.cassandra.service.paxos.SatellitePaxosParticipants;
+import org.apache.cassandra.service.reads.tracked.TrackedRead;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.concurrent.Future;
+
+import static org.apache.cassandra.utils.ByteBufferUtil.bytes;
+import static org.junit.Assert.*;
+
+public class SatellitePaxosFailoverTest extends SatelliteReplicationStrategyTestBase
+{
+ private static final LongToken TOKEN = new LongToken(150);
+
+ @Before
+ public void registerLocalNode() throws Exception
+ {
+ // Register the local broadcast address in dc1 so shouldRejectPaxos can resolve the local DC
+ InetAddress localAddr = InetAddress.getByName("127.0.0.1");
+ DatabaseDescriptor.setBroadcastAddress(localAddr);
+ InetAddressAndPort localEndpoint = InetAddressAndPort.getByAddress(localAddr);
+ ClusterMetadataTestHelper.register(localEndpoint, "dc1", "rack1");
+ }
+
+ @After
+ public void clearSinks()
+ {
+ MessagingService.instance().outboundSink.clear();
+ }
+
+ @Test
+ public void testShouldRejectPaxosReturnsTrueDuringTransitionAck() throws Exception
+ {
+ createDualDCKeyspace("dc1");
+ SatelliteReplicationStrategy strategy = getSRS(DUAL_DC_KEYSPACE);
+
+ strategy.setFailoverState(SatelliteFailoverState.FailoverStateMap.allRanges(
+ SatelliteFailoverState.FailoverInfo.transitionAck("dc1")));
+
+ assertTrue("Should reject paxos during TRANSITION_ACK", strategy.shouldRejectPaxos(TOKEN));
+ }
+
+ @Test
+ public void testShouldRejectPaxosReturnsTrueWhenNotInPrimaryDC() throws Exception
+ {
+ // Local node is in dc1, but primary is dc2 — should reject
+ createDualDCKeyspace("dc2");
+ SatelliteReplicationStrategy strategy = getSRS(DUAL_DC_KEYSPACE);
+
+ assertTrue("Should reject paxos when local node is not in primary DC", strategy.shouldRejectPaxos(TOKEN));
+ }
+
+ @Test
+ public void testShouldRejectPaxosReturnsFalseInNormalState() throws Exception
+ {
+ // Local node is in dc1 and dc1 is primary — should allow
+ createDualDCKeyspace("dc1");
+ SatelliteReplicationStrategy strategy = getSRS(DUAL_DC_KEYSPACE);
+
+ assertFalse("Should not reject paxos in NORMAL state when in primary DC", strategy.shouldRejectPaxos(TOKEN));
+ }
+
+ @Test
+ public void testShouldRejectPaxosReturnsFalseDuringTransition() throws Exception
+ {
+ // Local node is in dc1 and dc1 is the new primary during TRANSITION — should allow
+ createDualDCKeyspace("dc1");
+ SatelliteReplicationStrategy strategy = getSRS(DUAL_DC_KEYSPACE);
+
+ strategy.setFailoverState(SatelliteFailoverState.FailoverStateMap.allRanges(
+ SatelliteFailoverState.FailoverInfo.transition("dc2")));
+
+ assertFalse("Should not reject paxos during TRANSITION when in primary DC", strategy.shouldRejectPaxos(TOKEN));
+ }
+
+ private TableMetadata tableMetadata(String keyspace)
+ {
+ return TableMetadata.builder(keyspace, "test_table")
+ .addPartitionKeyColumn("key", AsciiType.instance)
+ .build();
+ }
+
+ @Test
+ public void testPaxosParticipantsRejectedDuringTransitionAck() throws Exception
+ {
+ createDualDCKeyspace("dc2");
+ SatelliteReplicationStrategy strategy = getSRS(DUAL_DC_KEYSPACE);
+
+ strategy.setFailoverState(SatelliteFailoverState.FailoverStateMap.allRanges(
+ SatelliteFailoverState.FailoverInfo.transitionAck("dc1")));
+
+ try
+ {
+ strategy.paxosParticipants(ClusterMetadata.current(), tableMetadata(DUAL_DC_KEYSPACE),
+ TOKEN, ConsistencyLevel.SERIAL, r -> true);
+ fail("paxosParticipants should throw UnavailableException during TRANSITION_ACK");
+ }
+ catch (UnavailableException e)
+ {
+ // expected
+ }
+ }
+
+ @Test
+ public void testPaxosParticipantsAllowedInNormalState() throws Exception
+ {
+ createDualDCKeyspace("dc1");
+ SatelliteReplicationStrategy strategy = getSRS(DUAL_DC_KEYSPACE);
+
+ Paxos.Participants participants = strategy.paxosParticipants(
+ ClusterMetadata.current(), tableMetadata(DUAL_DC_KEYSPACE),
+ TOKEN, ConsistencyLevel.SERIAL, r -> true);
+ assertNotNull(participants);
+ }
+
+ @Test
+ public void testSendPaxosCommitMutationsRejectedDuringTransitionAck() throws Exception
+ {
+ createDualDCKeyspace("dc2");
+ SatelliteReplicationStrategy strategy = getSRS(DUAL_DC_KEYSPACE);
+
+ strategy.setFailoverState(SatelliteFailoverState.FailoverStateMap.allRanges(
+ SatelliteFailoverState.FailoverInfo.transitionAck("dc1")));
+
+ TableMetadata table = tableMetadata(DUAL_DC_KEYSPACE);
+ DecoratedKey key = table.partitioner.decorateKey(bytes("test_key"));
+ PartitionUpdate update = PartitionUpdate.emptyUpdate(table, key);
+ Commit.Agreed commit = new Commit.Agreed(Ballot.none(), update);
+
+ try
+ {
+ strategy.sendPaxosCommitMutations(commit, false);
+ fail("sendPaxosCommitMutations should throw UnavailableException during TRANSITION_ACK");
+ }
+ catch (UnavailableException e)
+ {
+ // expected
+ }
+ }
+
+ @Test
+ public void testSendPaxosCommitMutationsFailsWhenSatelliteRequestsFail() throws Exception
+ {
+ createDualDCKeyspace("dc1");
+ SatelliteReplicationStrategy strategy = getSRS(DUAL_DC_KEYSPACE);
+
+ // Mark all non-primary (satellite/secondary) endpoints alive so they are contacted with a callback,
+ // rather than being pre-marked as failed via the downEndpoints path.
+ markAllEndpointsAlive();
+
+ // Capture and swallow the outbound satellite commit requests so no real network I/O happens; we drive
+ // the failure ourselves via callback expiration below.
+ List<MessageCapture> captured = new CopyOnWriteArrayList<>();
+ MessagingService.instance().outboundSink.add((message, to) -> {
+ if (message.verb() == Verb.PAXOS2_COMMIT_REMOTE_REQ)
+ captured.add(new MessageCapture(message, to));
+ return false;
+ });
+
+ Commit.Agreed commit = agreedCommit();
+
+ Future<Void> future = strategy.sendPaxosCommitMutations(commit, false);
+
+ // A quorum of satellite endpoints must have been contacted with a callback.
+ assertFalse("Expected satellite commit requests to be sent", captured.isEmpty());
+ assertFalse("Future should not complete before any responses/failures", future.isDone());
+
+ // Simulate a request failure (timeout) for every satellite endpoint. This exercises the callback's
+ // onFailure path; it only runs because invokeOnFailure() returns true.
+ for (MessageCapture cap : captured)
+ MessagingService.instance().callbacks.onExpired(cap.message, cap.to);
+
+ assertTrue("Future should fail once satellite quorum is unreachable", future.awaitUninterruptibly(30, TimeUnit.SECONDS));
+ assertTrue("Future should have failed", future.isDone() && !future.isSuccess());
+ assertNotNull("Failure cause should be set", future.cause());
+ }
+
+ @Test
+ public void testSendPaxosCommitMutationsSucceedsWhenSatelliteRequestsRespond() throws Exception
+ {
+ createDualDCKeyspace("dc1");
+ SatelliteReplicationStrategy strategy = getSRS(DUAL_DC_KEYSPACE);
+
+ markAllEndpointsAlive();
+
+ List<MessageCapture> captured = new CopyOnWriteArrayList<>();
+ MessagingService.instance().outboundSink.add((message, to) -> {
+ if (message.verb() == Verb.PAXOS2_COMMIT_REMOTE_REQ)
+ captured.add(new MessageCapture(message, to));
+ return false;
+ });
+
+ Commit.Agreed commit = agreedCommit();
+
+ Future<Void> future = strategy.sendPaxosCommitMutations(commit, false);
+
+ assertFalse("Expected satellite commit requests to be sent", captured.isEmpty());
+
+ // Deliver a successful response for every satellite endpoint via the registered callback.
+ for (MessageCapture cap : captured)
+ {
+ Message<NoPayload> response = Message.internalResponse(Verb.PAXOS2_COMMIT_REMOTE_RSP, NoPayload.noPayload)
+ .withFrom(cap.to);
+ MessagingService.instance().callbacks.removeAndRespond(cap.message.id(), cap.to, response);
+ }
+
+ assertTrue("Future should complete once satellite quorum responds",
+ future.awaitUninterruptibly(30, TimeUnit.SECONDS));
+ assertTrue("Future should have succeeded", future.isSuccess());
+ }
+
+ private Commit.Agreed agreedCommit()
+ {
+ TableMetadata table = tableMetadata(DUAL_DC_KEYSPACE);
+ // Pin the key to TOKEN (150) so it maps to the configured ring ranges set up by the test base.
+ DecoratedKey key = new BufferDecoratedKey(TOKEN, bytes("test_key"));
+ PartitionUpdate update = PartitionUpdate.emptyUpdate(table, key);
+ // A non-none mutation id is required by sendPaxosCommitMutations (it registers with mutation tracking).
+ return new Commit.Agreed(Ballot.none(), update).withMutationId(new MutationId(1L, 1L));
+ }
+
+ private void markAllEndpointsAlive()
+ {
+ for (InetAddressAndPort endpoint : ClusterMetadata.current().directory.allAddresses())
+ Gossiper.instance.initializeNodeUnsafe(endpoint, UUID.randomUUID(), 1);
+ }
+
+ private SatellitePaxosParticipants getParticipants(String keyspace) throws Exception
+ {
+ SatelliteReplicationStrategy strategy = getSRS(keyspace);
+ ClusterMetadata metadata = ClusterMetadata.current();
+ Paxos.Participants participants = strategy.paxosParticipants(
+ metadata, tableMetadata(keyspace), TOKEN, ConsistencyLevel.SERIAL, r -> true);
+ assertTrue("SRS should return SatellitePaxosParticipants",
+ participants instanceof SatellitePaxosParticipants);
+ return (SatellitePaxosParticipants) participants;
+ }
+
+ @Test
+ public void testPaxosParticipantsReturnsSatelliteEndpoints() throws Exception
+ {
+ // Dual DC with dc1 primary: satellite endpoints should include sat1 (dc1's satellite) and dc2 (other full DC)
+ createDualDCKeyspace("dc1");
+ SatellitePaxosParticipants spp = getParticipants(DUAL_DC_KEYSPACE);
+
+ EndpointsForToken satelliteEndpoints = spp.getAdditionalSummaryEndpoints();
+ ClusterMetadata metadata = ClusterMetadata.current();
+ Set<String> dcs = replicaDCs(satelliteEndpoints, metadata);
+
+ assertTrue("Should include sat1 (primary's satellite)", dcs.contains("sat1"));
+ assertTrue("Should include dc2 (other full DC)", dcs.contains("dc2"));
+ assertFalse("Should not include dc1 (primary DC)", dcs.contains("dc1"));
+ assertFalse("Should not include sat2 (other DC's satellite)", dcs.contains("sat2"));
+ }
+
+ @Test
+ public void testPaxosParticipantsSingleDCHasSatelliteOnly() throws Exception
+ {
+ createSingleDCKeyspace();
+ SatellitePaxosParticipants spp = getParticipants(SINGLE_DC_KEYSPACE);
+
+ EndpointsForToken satelliteEndpoints = spp.getAdditionalSummaryEndpoints();
+ ClusterMetadata metadata = ClusterMetadata.current();
+ Set<String> dcs = replicaDCs(satelliteEndpoints, metadata);
+
+ assertTrue("Should include sat1", dcs.contains("sat1"));
+ assertEquals("Should only have sat1", 1, dcs.size());
+ }
+
+ @Test
+ public void testAdditionalSummaryHostIdsMatchesSatelliteEndpoints() throws Exception
+ {
+ createDualDCKeyspace("dc1");
+ SatellitePaxosParticipants spp = getParticipants(DUAL_DC_KEYSPACE);
+
+ ClusterMetadata metadata = ClusterMetadata.current();
+ EndpointsForToken satelliteEndpoints = spp.getAdditionalSummaryEndpoints();
+ int[] additionalIds = spp.additionalSummaryHostIds(metadata);
+
+ assertEquals(satelliteEndpoints.size(), additionalIds.length);
+ for (int i = 0; i < satelliteEndpoints.size(); i++)
+ {
+ int expectedId = metadata.directory.peerId(satelliteEndpoints.endpoint(i)).id();
+ assertEquals(expectedId, additionalIds[i]);
+ }
+ }
+
+ @Test
+ public void testOnPrepareStartedSendsSummaryRequestToSatellites() throws Exception
+ {
+ createDualDCKeyspace("dc1");
+ SatellitePaxosParticipants spp = getParticipants(DUAL_DC_KEYSPACE);
+
+ List<MessageCapture> captured = new CopyOnWriteArrayList<>();
+ MessagingService.instance().outboundSink.add((message, to) -> {
+ captured.add(new MessageCapture(message, to));
+ return false;
+ });
+
+ TrackedRead.Id readId = new TrackedRead.Id(1, 100L);
+ TableMetadata table = tableMetadata(DUAL_DC_KEYSPACE);
+ SinglePartitionReadCommand readCommand = SinglePartitionReadCommand.fullPartitionRead(table, 0, ByteBufferUtil.bytes(0));
+
+ spp.onPrepareStarted(readId, 42, new int[] { 1, 2, 3 }, readCommand);
+
+ EndpointsForToken satelliteEndpoints = spp.getAdditionalSummaryEndpoints();
+ assertEquals(satelliteEndpoints.size(), captured.size());
+
+ Set<InetAddressAndPort> sentTo = captured.stream().map(c -> c.to).collect(Collectors.toSet());
+ for (MessageCapture cap : captured)
+ assertEquals(Verb.TRACKED_SUMMARY_REQ, cap.message.verb());
+ for (int i = 0; i < satelliteEndpoints.size(); i++)
+ assertTrue("Should send to satellite endpoint", sentTo.contains(satelliteEndpoints.endpoint(i)));
+ }
+
+ @Test
+ public void testOnPrepareStartedNoOpWhenReadCommandNull() throws Exception
+ {
+ createDualDCKeyspace("dc1");
+ SatellitePaxosParticipants spp = getParticipants(DUAL_DC_KEYSPACE);
+
+ List<MessageCapture> captured = new CopyOnWriteArrayList<>();
+ MessagingService.instance().outboundSink.add((message, to) -> {
+ captured.add(new MessageCapture(message, to));
+ return false;
+ });
+
+ spp.onPrepareStarted(new TrackedRead.Id(1, 100L), 42, new int[] { 1, 2, 3 }, null);
+
+ assertEquals(0, captured.size());
+ }
+
+ private static class MessageCapture
+ {
+ final Message<?> message;
+ final InetAddressAndPort to;
+
+ MessageCapture(Message<?> message, InetAddressAndPort to)
+ {
+ this.message = message;
+ this.to = to;
+ }
+ }
+}
diff --git a/test/unit/org/apache/cassandra/locator/SatelliteReadPlanTest.java b/test/unit/org/apache/cassandra/locator/SatelliteReadPlanTest.java
new file mode 100644
index 0000000000..33eec30e5f
--- /dev/null
+++ b/test/unit/org/apache/cassandra/locator/SatelliteReadPlanTest.java
@@ -0,0 +1,291 @@
+/*
+ * 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.locator;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.junit.Assume;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.dht.Murmur3Partitioner.LongToken;
+import org.apache.cassandra.dht.Range;
+import org.apache.cassandra.schema.KeyspaceMetadata;
+import org.apache.cassandra.service.reads.AlwaysSpeculativeRetryPolicy;
+import org.apache.cassandra.service.reads.ReadCoordinator;
+import org.apache.cassandra.tcm.ClusterMetadata;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+@RunWith(Parameterized.class)
+public class SatelliteReadPlanTest extends SatelliteReplicationStrategyTestBase
+{
+ enum ReadType
+ {
+ TOKEN, RANGE
+ }
+
+ @Parameterized.Parameters(name = "{0}/{1}")
+ public static Collection<Object[]> params()
+ {
+ return Arrays.asList(new Object[][] {
+ { ReadType.TOKEN, SatelliteFailoverState.State.NORMAL },
+ { ReadType.TOKEN, SatelliteFailoverState.State.TRANSITION_ACK },
+ { ReadType.TOKEN, SatelliteFailoverState.State.TRANSITION },
+ { ReadType.RANGE, SatelliteFailoverState.State.NORMAL },
+ { ReadType.RANGE, SatelliteFailoverState.State.TRANSITION_ACK },
+ { ReadType.RANGE, SatelliteFailoverState.State.TRANSITION },
+ });
+ }
+
+ private final ReadType readType;
+ private final SatelliteFailoverState.State failoverState;
+
+ public SatelliteReadPlanTest(ReadType readType, SatelliteFailoverState.State failoverState)
+ {
+ this.readType = readType;
+ this.failoverState = failoverState;
+ }
+
+ private boolean isTransition()
+ {
+ return failoverState != SatelliteFailoverState.State.NORMAL;
+ }
+
+ private SatelliteFailoverState.FailoverInfo failoverInfo()
+ {
+ switch (failoverState)
+ {
+ case TRANSITION_ACK: return SatelliteFailoverState.FailoverInfo.transitionAck("dc1");
+ case TRANSITION: return SatelliteFailoverState.FailoverInfo.transition("dc1");
+ default: throw new IllegalStateException("No failover info for NORMAL");
+ }
+ }
+
+ private CoordinationPlan.ForRead<?, ?> createPlan(SatelliteReplicationStrategy strategy, String keyspaceName) throws Exception
+ {
+ ClusterMetadata metadata = ClusterMetadata.current();
+ KeyspaceMetadata ksm = metadata.schema.getKeyspaces().getNullable(keyspaceName);
+ Keyspace keyspace = Keyspace.mockKS(ksm);
+
+ switch (readType)
+ {
+ case TOKEN:
+ return strategy.planForTokenRead(metadata, keyspace, TABLE_ID,
+ new LongToken(150), null,
+ ConsistencyLevel.QUORUM,
+ AlwaysSpeculativeRetryPolicy.INSTANCE,
+ ReadCoordinator.DEFAULT);
+ case RANGE:
+ return strategy.planForRangeRead(metadata, keyspace, TABLE_ID, null,
+ ConsistencyLevel.QUORUM,
+ Range.makeRowRange(new LongToken(100),
+ new LongToken(200)),
+ 1);
+ default:
+ throw new IllegalStateException();
+ }
+ }
+
+ private ReplicaPlan.ForRead<?, ?> replicas(CoordinationPlan.ForRead<?, ?> plan)
+ {
+ return (ReplicaPlan.ForRead<?, ?>) plan.replicas();
+ }
+
+ private void assertNoDuplicateEndpoints(String label, Iterable<Replica> replicas)
+ {
+ Set<InetAddressAndPort> seen = new HashSet<>();
+ for (Replica r : replicas)
+ assertTrue(label + " contains duplicate endpoint: " + r.endpoint(),
+ seen.add(r.endpoint()));
+ }
+
+ @Test
+ public void testReadPlanDualDC() throws Exception
+ {
+ createDualDCKeyspace(isTransition() ? "dc2" : "dc1");
+ SatelliteReplicationStrategy strategy = getSRS(DUAL_DC_KEYSPACE);
+ ClusterMetadata metadata = ClusterMetadata.current();
+
+ if (isTransition())
+ strategy.setFailoverState(SatelliteFailoverState.FailoverStateMap.allRanges(failoverInfo()));
+
+ CoordinationPlan.ForRead<?, ?> plan = createPlan(strategy, DUAL_DC_KEYSPACE);
+
+ Set<String> contactDCs = replicaDCs(plan.replicas().contacts(), metadata);
+ if (isTransition())
+ {
+ assertTrue("Should include dc2 (new primary)", contactDCs.contains("dc2"));
+ assertTrue("Should include dc1 (old primary)", contactDCs.contains("dc1"));
+ }
+ else
+ {
+ assertTrue("Should include dc1", contactDCs.contains("dc1"));
+ assertTrue("Should include sat1 (dc1's satellite)", contactDCs.contains("sat1"));
+ assertFalse("Should NOT include sat2 (dc2's satellite)", contactDCs.contains("sat2"));
+ }
+ }
+
+ @Test
+ public void testReadPlanSingleDC() throws Exception
+ {
+ createSingleDCKeyspace();
+ SatelliteReplicationStrategy strategy = getSRS(SINGLE_DC_KEYSPACE);
+ ClusterMetadata metadata = ClusterMetadata.current();
+
+ CoordinationPlan.ForRead<?, ?> plan = createPlan(strategy, SINGLE_DC_KEYSPACE);
+
+ Set<String> contactDCs = replicaDCs(plan.replicas().contacts(), metadata);
+ assertTrue("Should include dc1", contactDCs.contains("dc1"));
+ assertTrue("Should include sat1", contactDCs.contains("sat1"));
+ assertFalse("Should NOT include dc2", contactDCs.contains("dc2"));
+ assertFalse("Should NOT include sat2", contactDCs.contains("sat2"));
+ }
+
+ @Test
+ public void testReadPlanExcludesDisabledDC() throws Exception
+ {
+ createDisabledDCKeyspace();
+ SatelliteReplicationStrategy strategy = getSRS(DISABLED_DC_KEYSPACE);
+ ClusterMetadata metadata = ClusterMetadata.current();
+
+ CoordinationPlan.ForRead<?, ?> plan = createPlan(strategy, DISABLED_DC_KEYSPACE);
+
+ Set<String> contactDCs = replicaDCs(plan.replicas().contacts(), metadata);
+ assertTrue("Should include dc1", contactDCs.contains("dc1"));
+ assertTrue("Should include sat1 (dc1's satellite)", contactDCs.contains("sat1"));
+ assertFalse("Should NOT include dc2 (disabled)", contactDCs.contains("dc2"));
+ assertFalse("Should NOT include sat2 (disabled dc2's satellite)", contactDCs.contains("sat2"));
+ }
+
+ @Test
+ public void testReadPlanPrimaryDCFirst() throws Exception
+ {
+ createDualDCKeyspace(isTransition() ? "dc2" : "dc1");
+ SatelliteReplicationStrategy strategy = getSRS(DUAL_DC_KEYSPACE);
+ ClusterMetadata metadata = ClusterMetadata.current();
+
+ if (isTransition())
+ strategy.setFailoverState(SatelliteFailoverState.FailoverStateMap.allRanges(failoverInfo()));
+
+ CoordinationPlan.ForRead<?, ?> plan = createPlan(strategy, DUAL_DC_KEYSPACE);
+
+ String expectedPrimary = isTransition() ? "dc2" : "dc1";
+ Replica first = plan.replicas().contacts().iterator().next();
+ assertEquals("First contact should be in primary DC",
+ expectedPrimary, metadata.locator.location(first.endpoint()).datacenter);
+
+ if (isTransition())
+ {
+ // dc2 contacts should appear before dc1 contacts
+ boolean seenDc1 = false;
+ for (Replica r : plan.replicas().contacts())
+ {
+ String dc = metadata.locator.location(r.endpoint()).datacenter;
+ if (dc.equals("dc1"))
+ seenDc1 = true;
+ if (dc.equals("dc2") && seenDc1)
+ fail("dc2 contact appeared after dc1 contact — primary should come first");
+ }
+ }
+ }
+
+ @Test
+ public void testReadPlanNoMerge() throws Exception
+ {
+ createDualDCKeyspace("dc1");
+ SatelliteReplicationStrategy strategy = getSRS(DUAL_DC_KEYSPACE);
+ ClusterMetadata metadata = ClusterMetadata.current();
+
+ CoordinationPlan.ForRead<?, ?> plan = createPlan(strategy, DUAL_DC_KEYSPACE);
+
+ Set<String> candidateDCs = replicaDCs(replicas(plan).readCandidates(), metadata);
+ assertTrue("Should have dc1 candidates", candidateDCs.contains("dc1"));
+ }
+
+ @Test
+ public void testTransitionReadPlanMergesDCs() throws Exception
+ {
+ Assume.assumeTrue(isTransition());
+
+ createDualDCKeyspace("dc2");
+ SatelliteReplicationStrategy strategy = getSRS(DUAL_DC_KEYSPACE);
+ ClusterMetadata metadata = ClusterMetadata.current();
+
+ strategy.setFailoverState(SatelliteFailoverState.FailoverStateMap.allRanges(failoverInfo()));
+
+ CoordinationPlan.ForRead<?, ?> plan = createPlan(strategy, DUAL_DC_KEYSPACE);
+
+ Set<String> candidateDCs = replicaDCs(replicas(plan).readCandidates(), metadata);
+ assertTrue("Merged candidates should include dc2", candidateDCs.contains("dc2"));
+ assertTrue("Merged candidates should include dc1", candidateDCs.contains("dc1"));
+
+ Set<String> liveAndDownDCs = replicaDCs(plan.replicas().liveAndDown(), metadata);
+ assertTrue("Merged liveAndDown should include dc2", liveAndDownDCs.contains("dc2"));
+ assertTrue("Merged liveAndDown should include dc1", liveAndDownDCs.contains("dc1"));
+ }
+
+ @Test
+ public void testTransitionReadPlanNoDuplicates() throws Exception
+ {
+ Assume.assumeTrue(isTransition());
+
+ createDualDCKeyspace("dc2");
+ SatelliteReplicationStrategy strategy = getSRS(DUAL_DC_KEYSPACE);
+
+ strategy.setFailoverState(SatelliteFailoverState.FailoverStateMap.allRanges(failoverInfo()));
+
+ CoordinationPlan.ForRead<?, ?> plan = createPlan(strategy, DUAL_DC_KEYSPACE);
+ ReplicaPlan.ForRead<?, ?> replicas = replicas(plan);
+
+ assertNoDuplicateEndpoints("contacts", replicas.contacts());
+ assertNoDuplicateEndpoints("candidates", replicas.readCandidates());
+ assertNoDuplicateEndpoints("liveAndDown", replicas.liveAndDown());
+ }
+
+ @Test
+ public void testTransitionReadPlanQuorum() throws Exception
+ {
+ Assume.assumeTrue(isTransition());
+
+ createDualDCKeyspace("dc2");
+ SatelliteReplicationStrategy strategy = getSRS(DUAL_DC_KEYSPACE);
+
+ // Get the individual plan quorum before merging
+ CoordinationPlan.ForRead<?, ?> primaryOnly = createPlan(strategy, DUAL_DC_KEYSPACE);
+ int primaryQuorum = replicas(primaryOnly).readQuorum();
+
+ // Now set transition state and get merged plan
+ strategy.setFailoverState(SatelliteFailoverState.FailoverStateMap.allRanges(failoverInfo()));
+
+ CoordinationPlan.ForRead<?, ?> merged = createPlan(strategy, DUAL_DC_KEYSPACE);
+ int mergedQuorum = replicas(merged).readQuorum();
+
+ assertTrue("Merged quorum (" + mergedQuorum + ") should be >= primary quorum (" + primaryQuorum + ")",
+ mergedQuorum >= primaryQuorum);
+ }
+}
diff --git a/test/unit/org/apache/cassandra/locator/SatelliteReplicationStrategyTest.java b/test/unit/org/apache/cassandra/locator/SatelliteReplicationStrategyTest.java
index fc2a81e2e1..03ee31377e 100644
--- a/test/unit/org/apache/cassandra/locator/SatelliteReplicationStrategyTest.java
+++ b/test/unit/org/apache/cassandra/locator/SatelliteReplicationStrategyTest.java
@@ -17,87 +17,29 @@
*/
package org.apache.cassandra.locator;
-import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.Map;
-import org.junit.After;
import org.junit.Test;
-import org.apache.cassandra.CassandraTestBase;
-import org.apache.cassandra.CassandraTestBase.UseMurmur3Partitioner;
-import org.apache.cassandra.ServerTestUtils;
-import org.apache.cassandra.dht.Murmur3Partitioner.LongToken;
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.distributed.test.log.ClusterMetadataTestHelper;
+import org.apache.cassandra.dht.Murmur3Partitioner.LongToken;
import org.apache.cassandra.exceptions.ConfigurationException;
-import org.apache.cassandra.schema.KeyspaceMetadata;
import org.apache.cassandra.schema.ReplicationType;
import org.apache.cassandra.tcm.ClusterMetadata;
-import org.apache.cassandra.tcm.membership.Location;
-import static org.apache.cassandra.CassandraTestBase.DisableMBeanRegistration;
-import static org.apache.cassandra.CassandraTestBase.PrepareServerNoRegister;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
-@PrepareServerNoRegister
-@DisableMBeanRegistration
-@UseMurmur3Partitioner
-public class SatelliteReplicationStrategyTest extends CassandraTestBase
+public class SatelliteReplicationStrategyTest extends SatelliteReplicationStrategyTestBase
{
- private static final String KEYSPACE = "test";
-
- @After
- public void teardown()
- {
- ServerTestUtils.resetCMS();
- }
-
- private void addToken(long token, String address, Location location) throws UnknownHostException
- {
- InetAddressAndPort addr = InetAddressAndPort.getByName(address);
- ClusterMetadataTestHelper.addEndpoint(addr, new LongToken(token), location);
- }
-
- private void setupDCs() throws UnknownHostException
- {
- Location dc1 = new Location("dc1", "rack1");
- Location dc2 = new Location("dc2", "rack1");
- Location sat1 = new Location("sat1", "rack1");
- Location sat2 = new Location("sat2", "rack1");
-
- // DC1
- addToken(100, "10.0.0.10", dc1);
- addToken(200, "10.0.0.11", dc1);
- addToken(300, "10.0.0.12", dc1);
-
- // DC2
- addToken(400, "10.1.0.10", dc2);
- addToken(500, "10.1.0.11", dc2);
- addToken(600, "10.1.0.12", dc2);
-
- // SAT1
- addToken(700, "10.2.0.10", sat1);
- addToken(800, "10.2.0.11", sat1);
-
- // SAT2
- addToken(900, "10.3.0.10", sat2);
- addToken(1000, "10.3.0.11", sat2);
- }
-
- private static SatelliteReplicationStrategy getSRS(String keyspace)
- {
- KeyspaceMetadata ksm = ClusterMetadata.current().schema.getKeyspaces().getNullable(keyspace);
- return (SatelliteReplicationStrategy) ksm.replicationStrategy;
- }
-
@Test
public void testValidSingleDCWithSatellite() throws Exception
{
- setupDCs();
-
String cql = "CREATE KEYSPACE " + KEYSPACE + " WITH replication = {" +
"'class': 'SatelliteReplicationStrategy', " +
"'dc1': '3', " +
@@ -119,8 +61,6 @@ public class SatelliteReplicationStrategyTest extends CassandraTestBase
@Test
public void testValidMultipleDCsWithSatellites() throws Exception
{
- setupDCs();
-
String cql = "CREATE KEYSPACE " + KEYSPACE + " WITH replication = {" +
"'class': 'SatelliteReplicationStrategy', " +
"'dc1': '3', " +
@@ -139,10 +79,8 @@ public class SatelliteReplicationStrategyTest extends CassandraTestBase
assertEquals(2, strategy.getSatellites().size());
}
- private void testConfigurationException(Map<String, String> options, String messageContains) throws UnknownHostException
+ private void testConfigurationException(Map<String, String> options, String messageContains)
{
- setupDCs();
-
try
{
new SatelliteReplicationStrategy(KEYSPACE, options, ReplicationType.tracked);
@@ -176,8 +114,6 @@ public class SatelliteReplicationStrategyTest extends CassandraTestBase
@Test
public void testUntrackedReplicationFails() throws Exception
{
- setupDCs();
-
Map<String, String> options = new HashMap<>();
options.put("dc1", "3");
options.put("primary", "dc1");
@@ -196,6 +132,33 @@ public class SatelliteReplicationStrategyTest extends CassandraTestBase
}
}
+ @Test
+ public void testPaxosV1Fails() throws Exception
+ {
+ Map<String, String> options = new HashMap<>();
+ options.put("dc1", "3");
+ options.put("primary", "dc1");
+
+ SatelliteReplicationStrategy strategy = new SatelliteReplicationStrategy(
+ KEYSPACE, options, ReplicationType.tracked);
+
+ Config.PaxosVariant prev = DatabaseDescriptor.getPaxosVariant();
+ try
+ {
+ DatabaseDescriptor.setPaxosVariant(Config.PaxosVariant.v1);
+ strategy.validateExpectedOptions(ClusterMetadata.current());
+ fail("ConfigurationException expected");
+ }
+ catch (ConfigurationException e)
+ {
+ assertTrue(e.getMessage().contains("requires paxos_variant=v2"));
+ }
+ finally
+ {
+ DatabaseDescriptor.setPaxosVariant(prev);
+ }
+ }
+
@Test
public void testDotsInDCNamesFails() throws Exception
{
@@ -242,12 +205,10 @@ public class SatelliteReplicationStrategyTest extends CassandraTestBase
@Test
public void testReplicaCalculationWithSatellites() throws Exception
{
- setupDCs();
-
String cql = "CREATE KEYSPACE " + KEYSPACE + " WITH replication = {" +
"'class': 'SatelliteReplicationStrategy', " +
"'dc1': '3', " +
- "'dc1.satellite.sat1': '2/2', " +
+ "'dc1.satellite.sat1': '3/3', " +
"'primary': 'dc1'" +
"} AND replication_type = 'tracked'";
@@ -258,8 +219,8 @@ public class SatelliteReplicationStrategyTest extends CassandraTestBase
EndpointsForRange replicas = strategy.calculateNaturalReplicas(
new LongToken(150), ClusterMetadata.current());
- // Should have 3 full replicas from dc1 + 2 satellite replicas from sat1
- assertEquals(5, replicas.size());
+ // Should have 3 full replicas from dc1 + 3 satellite replicas from sat1
+ assertEquals(6, replicas.size());
int fullCount = 0;
int witnessCount = 0;
@@ -272,14 +233,12 @@ public class SatelliteReplicationStrategyTest extends CassandraTestBase
}
assertEquals(3, fullCount);
- assertEquals(2, witnessCount);
+ assertEquals(3, witnessCount);
}
@Test
public void testDisableNonPrimaryDC() throws Exception
{
- setupDCs();
-
Map<String, String> options = new HashMap<>();
options.put("dc1", "3");
options.put("dc2", "3");
@@ -331,14 +290,12 @@ public class SatelliteReplicationStrategyTest extends CassandraTestBase
@Test
public void testDisabledDCSatelliteStillGetsReplicas() throws Exception
{
- setupDCs();
-
String cql = "CREATE KEYSPACE " + KEYSPACE + " WITH replication = {" +
"'class': 'SatelliteReplicationStrategy', " +
"'dc1': '3', " +
- "'dc1.satellite.sat1': '2/2', " +
+ "'dc1.satellite.sat1': '3/3', " +
"'dc2': '3', " +
- "'dc2.satellite.sat2': '2/2', " +
+ "'dc2.satellite.sat2': '3/3', " +
"'dc2.disabled': 'true', " +
"'primary': 'dc1'" +
"} AND replication_type = 'tracked'";
@@ -351,8 +308,8 @@ public class SatelliteReplicationStrategyTest extends CassandraTestBase
new LongToken(150), ClusterMetadata.current());
// Disabled does not affect placement — all DCs and satellites still get replicas
- // 3 full from dc1 + 3 full from dc2 + 2 witness from sat1 + 2 witness from sat2
- assertEquals(10, replicas.size());
+ // 3 full from dc1 + 3 full from dc2 + 3 witness from sat1 + 3 witness from sat2
+ assertEquals(12, replicas.size());
}
@Test
@@ -370,8 +327,6 @@ public class SatelliteReplicationStrategyTest extends CassandraTestBase
@Test
public void testHasSameSettingsWithDisabled() throws Exception
{
- setupDCs();
-
Map<String, String> optionsA = new HashMap<>();
optionsA.put("dc1", "3");
optionsA.put("dc2", "3");
diff --git a/test/unit/org/apache/cassandra/locator/SatelliteReplicationStrategyTestBase.java b/test/unit/org/apache/cassandra/locator/SatelliteReplicationStrategyTestBase.java
new file mode 100644
index 0000000000..8201265f4c
--- /dev/null
+++ b/test/unit/org/apache/cassandra/locator/SatelliteReplicationStrategyTestBase.java
@@ -0,0 +1,166 @@
+/*
+ * 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.locator;
+
+import java.net.UnknownHostException;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+
+import org.apache.cassandra.ServerTestUtils;
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.distributed.test.log.ClusterMetadataTestHelper;
+import org.apache.cassandra.dht.Murmur3Partitioner.LongToken;
+import org.apache.cassandra.schema.KeyspaceMetadata;
+import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.membership.Location;
+
+import static org.apache.cassandra.config.CassandraRelevantProperties.ORG_APACHE_CASSANDRA_DISABLE_MBEAN_REGISTRATION;
+
+public abstract class SatelliteReplicationStrategyTestBase
+{
+ protected static final String KEYSPACE = "test";
+ protected static final TableId TABLE_ID = TableId.generate();
+ protected static final String DUAL_DC_KEYSPACE = "dual_dc_test";
+ protected static final String SINGLE_DC_KEYSPACE = "single_dc_test";
+ protected static final String DISABLED_DC_KEYSPACE = "disabled_dc_test";
+
+ @BeforeClass
+ public static void setUpClass()
+ {
+ ORG_APACHE_CASSANDRA_DISABLE_MBEAN_REGISTRATION.setBoolean(true);
+ ServerTestUtils.daemonInitialization();
+ StorageService.instance.setPartitionerUnsafe(Murmur3Partitioner.instance);
+ DatabaseDescriptor.setPaxosVariant(Config.PaxosVariant.v2);
+ ServerTestUtils.prepareServerNoRegister();
+ }
+
+ @Before
+ public void setup() throws UnknownHostException
+ {
+ setupDCs();
+ }
+
+ @After
+ public void teardown()
+ {
+ ServerTestUtils.resetCMS();
+ }
+
+ private void addToken(long token, String address, Location location) throws UnknownHostException
+ {
+ InetAddressAndPort addr = InetAddressAndPort.getByName(address);
+ ClusterMetadataTestHelper.addEndpoint(addr, new LongToken(token), location);
+ }
+
+ private void setupDCs() throws UnknownHostException
+ {
+ Location dc1 = new Location("dc1", "rack1");
+ Location dc2 = new Location("dc2", "rack1");
+ Location sat1 = new Location("sat1", "rack1");
+ Location sat2 = new Location("sat2", "rack1");
+
+ // DC1
+ addToken(100, "10.0.0.10", dc1);
+ addToken(200, "10.0.0.11", dc1);
+ addToken(300, "10.0.0.12", dc1);
+
+ // DC2
+ addToken(400, "10.1.0.10", dc2);
+ addToken(500, "10.1.0.11", dc2);
+ addToken(600, "10.1.0.12", dc2);
+
+ // SAT1
+ addToken(700, "10.2.0.10", sat1);
+ addToken(800, "10.2.0.11", sat1);
+ addToken(900, "10.2.0.12", sat1);
+
+ // SAT2
+ addToken(1000, "10.3.0.10", sat2);
+ addToken(1100, "10.3.0.11", sat2);
+ addToken(1200, "10.3.0.12", sat2);
+ }
+
+ protected static SatelliteReplicationStrategy getSRS(String keyspace)
+ {
+ KeyspaceMetadata ksm = ClusterMetadata.current().schema.getKeyspaces().getNullable(keyspace);
+ return (SatelliteReplicationStrategy) ksm.replicationStrategy;
+ }
+
+ protected void createDualDCKeyspace(String primary) throws Exception
+ {
+ String cql = "CREATE KEYSPACE " + DUAL_DC_KEYSPACE + " WITH replication = {" +
+ "'class': 'SatelliteReplicationStrategy', " +
+ "'dc1': '3', " +
+ "'dc1.satellite.sat1': '3/3', " +
+ "'dc2': '3', " +
+ "'dc2.satellite.sat2': '3/3', " +
+ "'primary': '" + primary + "'" +
+ "} AND replication_type = 'tracked'";
+ ClusterMetadataTestHelper.createKeyspace(cql);
+ }
+
+ protected void createSingleDCKeyspace() throws Exception
+ {
+ String cql = "CREATE KEYSPACE " + SINGLE_DC_KEYSPACE + " WITH replication = {" +
+ "'class': 'SatelliteReplicationStrategy', " +
+ "'dc1': '3', " +
+ "'dc1.satellite.sat1': '3/3', " +
+ "'primary': 'dc1'" +
+ "} AND replication_type = 'tracked'";
+ ClusterMetadataTestHelper.createKeyspace(cql);
+ }
+
+ protected void createDisabledDCKeyspace() throws Exception
+ {
+ String cql = "CREATE KEYSPACE " + DISABLED_DC_KEYSPACE + " WITH replication = {" +
+ "'class': 'SatelliteReplicationStrategy', " +
+ "'dc1': '3', " +
+ "'dc1.satellite.sat1': '3/3', " +
+ "'dc2': '3', " +
+ "'dc2.satellite.sat2': '3/3', " +
+ "'dc2.disabled': 'true', " +
+ "'primary': 'dc1'" +
+ "} AND replication_type = 'tracked'";
+ ClusterMetadataTestHelper.createKeyspace(cql);
+ }
+
+ protected Set<String> replicaDCs(Iterable<Replica> replicas, ClusterMetadata metadata)
+ {
+ Set<String> dcs = new HashSet<>();
+ for (Replica r : replicas)
+ dcs.add(metadata.locator.location(r.endpoint()).datacenter);
+ return dcs;
+ }
+
+ protected Set<InetAddressAndPort> replicasInDC(Iterable<Replica> replicas, String dc, ClusterMetadata metadata)
+ {
+ Set<InetAddressAndPort> eps = new HashSet<>();
+ for (Replica r : replicas)
+ if (metadata.locator.location(r.endpoint()).datacenter.equals(dc))
+ eps.add(r.endpoint());
+ return eps;
+ }
+}
diff --git a/test/unit/org/apache/cassandra/locator/SatelliteWritePlanTest.java b/test/unit/org/apache/cassandra/locator/SatelliteWritePlanTest.java
new file mode 100644
index 0000000000..68bd757352
--- /dev/null
+++ b/test/unit/org/apache/cassandra/locator/SatelliteWritePlanTest.java
@@ -0,0 +1,153 @@
+/*
+ * 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.locator;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Set;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.dht.Murmur3Partitioner.LongToken;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.schema.KeyspaceMetadata;
+import org.apache.cassandra.tcm.ClusterMetadata;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+@RunWith(Parameterized.class)
+public class SatelliteWritePlanTest extends SatelliteReplicationStrategyTestBase
+{
+ @Parameterized.Parameters(name = "{0}")
+ public static Collection<Object[]> params()
+ {
+ return Arrays.asList(new Object[][] {
+ { SatelliteFailoverState.State.NORMAL },
+ { SatelliteFailoverState.State.TRANSITION_ACK },
+ { SatelliteFailoverState.State.TRANSITION },
+ });
+ }
+
+ private final SatelliteFailoverState.State failoverState;
+
+ public SatelliteWritePlanTest(SatelliteFailoverState.State failoverState)
+ {
+ this.failoverState = failoverState;
+ }
+
+ private boolean isTransition()
+ {
+ return failoverState != SatelliteFailoverState.State.NORMAL;
+ }
+
+ private void applyFailoverState(SatelliteReplicationStrategy strategy)
+ {
+ if (!isTransition())
+ return;
+
+ SatelliteFailoverState.FailoverInfo info;
+ switch (failoverState)
+ {
+ case TRANSITION_ACK: info = SatelliteFailoverState.FailoverInfo.transitionAck("dc1"); break;
+ case TRANSITION: info = SatelliteFailoverState.FailoverInfo.transition("dc1"); break;
+ default: throw new IllegalStateException();
+ }
+ strategy.setFailoverState(SatelliteFailoverState.FailoverStateMap.allRanges(info));
+ }
+
+ private CoordinationPlan.ForWrite callPlanForWrite(SatelliteReplicationStrategy strategy,
+ String keyspaceName, Token token)
+ {
+ ClusterMetadata metadata = ClusterMetadata.current();
+ KeyspaceMetadata ksm = metadata.schema.getKeyspaces().getNullable(keyspaceName);
+ Keyspace keyspace = Keyspace.mockKS(ksm);
+ return strategy.planForWriteInternal(metadata, keyspace, ConsistencyLevel.QUORUM,
+ (cm) -> ReplicaLayout.forTokenWriteLiveAndDown(cm, keyspace, token),
+ ReplicaPlans.writeAll);
+ }
+
+ @Test
+ public void testWriteContactsExcludeOtherSatellite() throws Exception
+ {
+ createDualDCKeyspace("dc1");
+ SatelliteReplicationStrategy strategy = getSRS(DUAL_DC_KEYSPACE);
+ applyFailoverState(strategy);
+ ClusterMetadata metadata = ClusterMetadata.current();
+
+ CoordinationPlan.ForWrite plan = callPlanForWrite(strategy, DUAL_DC_KEYSPACE, new LongToken(150));
+
+ Set<String> dcs = replicaDCs(plan.replicas().contacts(), metadata);
+ assertTrue("Should include dc1", dcs.contains("dc1"));
+ assertTrue("Should include dc2", dcs.contains("dc2"));
+ assertTrue("Should include sat1 (dc1's satellite)", dcs.contains("sat1"));
+ assertFalse("Should NOT include sat2 (dc2's satellite)", dcs.contains("sat2"));
+ }
+
+ @Test
+ public void testWriteContactsExcludeDisabledDC() throws Exception
+ {
+ createDisabledDCKeyspace();
+ SatelliteReplicationStrategy strategy = getSRS(DISABLED_DC_KEYSPACE);
+ applyFailoverState(strategy);
+ ClusterMetadata metadata = ClusterMetadata.current();
+
+ CoordinationPlan.ForWrite plan = callPlanForWrite(strategy, DISABLED_DC_KEYSPACE, new LongToken(150));
+
+ Set<String> dcs = replicaDCs(plan.replicas().contacts(), metadata);
+ assertTrue("Should include dc1", dcs.contains("dc1"));
+ assertTrue("Should include sat1 (dc1's satellite)", dcs.contains("sat1"));
+ assertFalse("Should NOT include dc2 (disabled)", dcs.contains("dc2"));
+ assertFalse("Should NOT include sat2 (disabled dc2's satellite)", dcs.contains("sat2"));
+ }
+
+ @Test
+ public void testWriteTrackerComposition() throws Exception
+ {
+ createDualDCKeyspace("dc1");
+ SatelliteReplicationStrategy strategy = getSRS(DUAL_DC_KEYSPACE);
+ applyFailoverState(strategy);
+ ClusterMetadata metadata = ClusterMetadata.current();
+
+ CoordinationPlan.ForWrite plan = callPlanForWrite(strategy, DUAL_DC_KEYSPACE, new LongToken(150));
+ ResponseTracker tracker = plan.responses();
+
+ assertTrue("Write should use CompositeTracker",
+ tracker instanceof CompositeTracker);
+
+ Set<InetAddressAndPort> dc1Contacts = replicasInDC(plan.replicas().contacts(), "dc1", metadata);
+ Set<InetAddressAndPort> sat1Contacts = replicasInDC(plan.replicas().contacts(), "sat1", metadata);
+
+ int count = 0;
+ for (InetAddressAndPort ep : dc1Contacts)
+ {
+ tracker.onResponse(ep);
+ if (++count >= 2) break;
+ }
+ assertFalse("dc1 quorum alone should not suffice (1 of 3 groups)", tracker.isSuccessful());
+
+ for (InetAddressAndPort ep : sat1Contacts)
+ tracker.onResponse(ep);
+
+ assertTrue("Should succeed with primary + satellite quorums (2 of 3 groups)", tracker.isSuccessful());
+ }
+}
diff --git a/test/unit/org/apache/cassandra/service/paxos/AugmentedCommitTest.java b/test/unit/org/apache/cassandra/service/paxos/AugmentedCommitTest.java
new file mode 100644
index 0000000000..02de9973e4
--- /dev/null
+++ b/test/unit/org/apache/cassandra/service/paxos/AugmentedCommitTest.java
@@ -0,0 +1,217 @@
+/*
+ * 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.service.paxos;
+
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
+
+import org.junit.Test;
+
+import static java.util.Collections.emptyMap;
+import static org.junit.Assert.*;
+
+public class AugmentedCommitTest
+{
+ private static final PaxosCommit.Status SUCCESS = new PaxosCommit.Status(null);
+ private static final PaxosCommit.Status FAILURE = new PaxosCommit.Status(
+ new Paxos.MaybeFailure(true, 3, 2, 0, emptyMap()));
+
+ private static PaxosCommit.AugmentedCommit<Consumer<PaxosCommit.Status>> create(AtomicReference<PaxosCommit.Status> capture)
+ {
+ return new PaxosCommit.AugmentedCommit<>(capture::set);
+ }
+
+ // ========================================
+ // Both succeed
+ // ========================================
+
+ @Test
+ public void testBothSucceed_paxosFirst()
+ {
+ AtomicReference<PaxosCommit.Status> result = new AtomicReference<>();
+ var ac = create(result);
+
+ ac.onPaxosComplete(SUCCESS);
+ assertNull("Should not complete with only paxos", result.get());
+
+ ac.onMutationComplete(SUCCESS);
+ assertNotNull("Should complete when both done", result.get());
+ assertTrue("Should be success", result.get().isSuccess());
+ }
+
+ @Test
+ public void testBothSucceed_mutationFirst()
+ {
+ AtomicReference<PaxosCommit.Status> result = new AtomicReference<>();
+ var ac = create(result);
+
+ ac.onMutationComplete(SUCCESS);
+ assertNull("Should not complete with only mutation", result.get());
+
+ ac.onPaxosComplete(SUCCESS);
+ assertNotNull("Should complete when both done", result.get());
+ assertTrue("Should be success", result.get().isSuccess());
+ }
+
+ // ========================================
+ // Paxos fails
+ // ========================================
+
+ @Test
+ public void testPaxosFails_immediateCompletion()
+ {
+ AtomicReference<PaxosCommit.Status> result = new AtomicReference<>();
+ var ac = create(result);
+
+ ac.onPaxosComplete(FAILURE);
+ assertNotNull("Should complete immediately on paxos failure", result.get());
+ assertFalse("Should report failure", result.get().isSuccess());
+ }
+
+ @Test
+ public void testPaxosFails_afterMutationSucceeds()
+ {
+ AtomicReference<PaxosCommit.Status> result = new AtomicReference<>();
+ var ac = create(result);
+
+ ac.onMutationComplete(SUCCESS);
+ assertNull(result.get());
+
+ ac.onPaxosComplete(FAILURE);
+ assertNotNull("Should complete on paxos failure", result.get());
+ assertFalse("Should report failure", result.get().isSuccess());
+ }
+
+ // ========================================
+ // Mutation fails
+ // ========================================
+
+ @Test
+ public void testMutationFails_immediateCompletion()
+ {
+ AtomicReference<PaxosCommit.Status> result = new AtomicReference<>();
+ var ac = create(result);
+
+ ac.onMutationComplete(FAILURE);
+ assertNotNull("Should complete immediately on mutation failure", result.get());
+ assertFalse("Should report failure", result.get().isSuccess());
+ }
+
+ @Test
+ public void testMutationFails_afterPaxosSucceeds()
+ {
+ AtomicReference<PaxosCommit.Status> result = new AtomicReference<>();
+ var ac = create(result);
+
+ ac.onPaxosComplete(SUCCESS);
+ assertNull(result.get());
+
+ ac.onMutationComplete(FAILURE);
+ assertNotNull("Should complete on mutation failure", result.get());
+ assertFalse("Should report failure", result.get().isSuccess());
+ }
+
+ // ========================================
+ // Both fail
+ // ========================================
+
+ @Test
+ public void testBothFail_paxosFirst()
+ {
+ AtomicReference<PaxosCommit.Status> result = new AtomicReference<>();
+ var ac = create(result);
+
+ ac.onPaxosComplete(FAILURE);
+ assertNotNull("Should complete immediately", result.get());
+ assertFalse(result.get().isSuccess());
+
+ // Second failure is a no-op
+ ac.onMutationComplete(FAILURE);
+ }
+
+ @Test
+ public void testBothFail_mutationFirst()
+ {
+ AtomicReference<PaxosCommit.Status> result = new AtomicReference<>();
+ var ac = create(result);
+
+ ac.onMutationComplete(FAILURE);
+ assertNotNull("Should complete immediately", result.get());
+ assertFalse(result.get().isSuccess());
+
+ // Second failure is a no-op
+ ac.onPaxosComplete(FAILURE);
+ }
+
+ // ========================================
+ // Terminal state is idempotent
+ // ========================================
+
+ @Test
+ public void testCompleteState_ignoresFurtherUpdates()
+ {
+ AtomicInteger callCount = new AtomicInteger();
+ var ac = new PaxosCommit.AugmentedCommit<Consumer<PaxosCommit.Status>>(s -> callCount.incrementAndGet());
+
+ ac.onPaxosComplete(SUCCESS);
+ ac.onMutationComplete(SUCCESS);
+ assertEquals("onDone should be called exactly once", 1, callCount.get());
+
+ // Further calls should be no-ops
+ ac.onPaxosComplete(SUCCESS);
+ ac.onMutationComplete(FAILURE);
+ ac.onPaxosComplete(FAILURE);
+ assertEquals("onDone should still be called exactly once", 1, callCount.get());
+ }
+
+ @Test
+ public void testCompleteViaFailure_ignoresFurtherUpdates()
+ {
+ AtomicInteger callCount = new AtomicInteger();
+ var ac = new PaxosCommit.AugmentedCommit<Consumer<PaxosCommit.Status>>(s -> callCount.incrementAndGet());
+
+ ac.onPaxosComplete(FAILURE);
+ assertEquals(1, callCount.get());
+
+ ac.onMutationComplete(SUCCESS);
+ ac.onMutationComplete(FAILURE);
+ ac.onPaxosComplete(SUCCESS);
+ assertEquals("onDone should still be called exactly once", 1, callCount.get());
+ }
+
+ // ========================================
+ // Duplicate calls to same side
+ // ========================================
+
+ @Test(expected = IllegalStateException.class)
+ public void testDuplicatePaxosComplete_throws()
+ {
+ var ac = create(new AtomicReference<>());
+ ac.onPaxosComplete(SUCCESS);
+ ac.onPaxosComplete(SUCCESS);
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void testDuplicateMutationComplete_throws()
+ {
+ var ac = create(new AtomicReference<>());
+ ac.onMutationComplete(SUCCESS);
+ ac.onMutationComplete(SUCCESS);
+ }
+}
diff --git a/test/unit/org/apache/cassandra/service/paxos/PaxosCommitPropertyTest.java b/test/unit/org/apache/cassandra/service/paxos/PaxosCommitPropertyTest.java
index 043b065225..d6aa71d341 100644
--- a/test/unit/org/apache/cassandra/service/paxos/PaxosCommitPropertyTest.java
+++ b/test/unit/org/apache/cassandra/service/paxos/PaxosCommitPropertyTest.java
@@ -139,7 +139,7 @@ public class PaxosCommitPropertyTest extends ResponseHandlerPropertyTestBase
* Testable subclass that overrides the DC membership check using topology knowledge,
* bypassing the InOurDc/Locator infrastructure which isn't fully initialized in unit tests.
*/
- private static class TestableCommit<T extends Consumer<? super PaxosCommit.Status>>
+ static class TestableCommit<T extends Consumer<? super PaxosCommit.Status>>
extends PaxosCommit<T>
{
private final Set<InetAddressAndPort> localEndpoints;
diff --git a/test/unit/org/apache/cassandra/service/paxos/SatellitePaxosCommitTest.java b/test/unit/org/apache/cassandra/service/paxos/SatellitePaxosCommitTest.java
new file mode 100644
index 0000000000..03d4437a10
--- /dev/null
+++ b/test/unit/org/apache/cassandra/service/paxos/SatellitePaxosCommitTest.java
@@ -0,0 +1,286 @@
+/*
+ * 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.service.paxos;
+
+import java.util.Collections;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
+
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.SchemaLoader;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.partitions.PartitionUpdate;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.exceptions.RequestFailure;
+import org.apache.cassandra.locator.EndpointsForToken;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.locator.Replica;
+import org.apache.cassandra.net.Message;
+import org.apache.cassandra.net.NoPayload;
+import org.apache.cassandra.net.Verb;
+import org.apache.cassandra.schema.KeyspaceParams;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.service.paxos.PaxosCommitPropertyTest.TestableCommit;
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.concurrent.AsyncPromise;
+import org.apache.cassandra.utils.concurrent.ImmediateFuture;
+
+import static org.apache.cassandra.net.NoPayload.noPayload;
+import static org.apache.cassandra.service.paxos.Commit.Agreed;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Tests for PaxosCommit's augmented commit composition logic.
+ *
+ * Verifies that onDone fires correctly when paxos consensus is combined with an additional
+ * commit future (from the replication strategy, e.g. satellite DC writes for SRS).
+ * Either side failing should cause immediate completion with failure.
+ */
+public class SatellitePaxosCommitTest
+{
+ private static final String KEYSPACE = "spc_test";
+
+ @BeforeClass
+ public static void setup() throws Exception
+ {
+ SchemaLoader.loadSchema();
+ SchemaLoader.createKeyspace(KEYSPACE, KeyspaceParams.simple(3),
+ SchemaLoader.standardCFMD(KEYSPACE, "Standard"));
+ }
+
+ private static TestableCommit<Consumer<PaxosCommit.Status>> createHandler(EndpointsForToken replicas,
+ int required,
+ AtomicReference<PaxosCommit.Status> statusCapture)
+ {
+ Keyspace ks = Keyspace.open(KEYSPACE);
+ TableMetadata table = ks.getColumnFamilyStores().iterator().next().metadata();
+ DecoratedKey key = table.partitioner.decorateKey(ByteBufferUtil.bytes(0));
+ Agreed commit = new Agreed(Ballot.none(), PartitionUpdate.emptyUpdate(table, key));
+ return new TestableCommit<>(commit, replicas, required,
+ ConsistencyLevel.QUORUM, statusCapture::set,
+ Collections.singleton(replicas.get(0).endpoint()));
+ }
+
+ private static EndpointsForToken threeReplicas() throws Exception
+ {
+ InetAddressAndPort ep1 = InetAddressAndPort.getByName("127.0.0.1");
+ InetAddressAndPort ep2 = InetAddressAndPort.getByName("127.0.0.2");
+ InetAddressAndPort ep3 = InetAddressAndPort.getByName("127.0.0.3");
+ Token minToken = DatabaseDescriptor.getPartitioner().getMinimumToken();
+ Token maxToken = DatabaseDescriptor.getPartitioner().getRandomToken();
+ return EndpointsForToken.of(maxToken,
+ Replica.fullReplica(ep1, minToken, maxToken),
+ Replica.fullReplica(ep2, minToken, maxToken),
+ Replica.fullReplica(ep3, minToken, maxToken));
+ }
+
+ private static void sendSuccess(PaxosCommit<?> handler, InetAddressAndPort from)
+ {
+ Message<NoPayload> msg = Message.builder(Verb.ECHO_REQ, noPayload)
+ .from(from)
+ .build();
+ handler.onResponse(msg);
+ }
+
+ // ========================================
+ // No augmented commit (default behavior)
+ // ========================================
+
+ @Test
+ public void testNoAugmentedCommit_paxosQuorumFiresOnDone() throws Exception
+ {
+ EndpointsForToken replicas = threeReplicas();
+ AtomicReference<PaxosCommit.Status> status = new AtomicReference<>();
+ PaxosCommit<?> handler = createHandler(replicas, 2, status);
+
+ sendSuccess(handler, replicas.get(0).endpoint());
+ assertNull("Should not fire after 1 response", status.get());
+
+ sendSuccess(handler, replicas.get(1).endpoint());
+ assertNotNull("Should fire after quorum", status.get());
+ assertTrue("Should be success", status.get().isSuccess());
+ }
+
+ // ========================================
+ // Already-completed futures
+ // ========================================
+
+ @Test
+ public void testCompletedSuccessFuture_paxosQuorumFiresOnDone() throws Exception
+ {
+ EndpointsForToken replicas = threeReplicas();
+ AtomicReference<PaxosCommit.Status> status = new AtomicReference<>();
+ PaxosCommit<?> handler = createHandler(replicas, 2, status);
+
+ handler.setAugmentedCommitFuture(ImmediateFuture.success(null));
+
+ sendSuccess(handler, replicas.get(0).endpoint());
+ assertNull(status.get());
+
+ sendSuccess(handler, replicas.get(1).endpoint());
+ assertNotNull("Should fire after quorum (future already done)", status.get());
+ assertTrue("Should be success", status.get().isSuccess());
+ }
+
+ @Test
+ public void testCompletedFailureFuture_failsImmediately() throws Exception
+ {
+ EndpointsForToken replicas = threeReplicas();
+ AtomicReference<PaxosCommit.Status> status = new AtomicReference<>();
+ PaxosCommit<?> handler = createHandler(replicas, 2, status);
+
+ AsyncPromise<Void> failed = new AsyncPromise<>();
+ failed.tryFailure(new RuntimeException("satellite quorum not met"));
+ handler.setAugmentedCommitFuture(failed);
+
+ // Future already failed — onDone should fire immediately
+ assertNotNull("Should fire immediately on failed future", status.get());
+ assertFalse("Should report failure", status.get().isSuccess());
+ }
+
+ // ========================================
+ // Paxos completes first
+ // ========================================
+
+ @Test
+ public void testPaxosSucceedsFirst_defersUntilFutureSucceeds() throws Exception
+ {
+ EndpointsForToken replicas = threeReplicas();
+ AtomicReference<PaxosCommit.Status> status = new AtomicReference<>();
+ PaxosCommit<?> handler = createHandler(replicas, 2, status);
+
+ AsyncPromise<Void> promise = new AsyncPromise<>();
+ handler.setAugmentedCommitFuture(promise);
+
+ sendSuccess(handler, replicas.get(0).endpoint());
+ sendSuccess(handler, replicas.get(1).endpoint());
+ assertNull("onDone should NOT fire yet (future pending)", status.get());
+
+ promise.trySuccess(null);
+ assertNotNull("onDone should fire after future resolves", status.get());
+ assertTrue("Should be success", status.get().isSuccess());
+ }
+
+ @Test
+ public void testPaxosSucceedsFirst_futureFailsCausesFailure() throws Exception
+ {
+ EndpointsForToken replicas = threeReplicas();
+ AtomicReference<PaxosCommit.Status> status = new AtomicReference<>();
+ PaxosCommit<?> handler = createHandler(replicas, 2, status);
+
+ AsyncPromise<Void> promise = new AsyncPromise<>();
+ handler.setAugmentedCommitFuture(promise);
+
+ sendSuccess(handler, replicas.get(0).endpoint());
+ sendSuccess(handler, replicas.get(1).endpoint());
+ assertNull("onDone deferred", status.get());
+
+ promise.tryFailure(new RuntimeException("satellite quorum not met"));
+ assertNotNull("onDone should fire", status.get());
+ assertFalse("Should report failure", status.get().isSuccess());
+ }
+
+ @Test
+ public void testPaxosFailsFirst_failsImmediately() throws Exception
+ {
+ EndpointsForToken replicas = threeReplicas();
+ AtomicReference<PaxosCommit.Status> status = new AtomicReference<>();
+ PaxosCommit<?> handler = createHandler(replicas, 2, status);
+
+ AsyncPromise<Void> promise = new AsyncPromise<>();
+ handler.setAugmentedCommitFuture(promise);
+
+ // Paxos fails (enough failures to make quorum impossible)
+ handler.onFailure(replicas.get(0).endpoint(), RequestFailure.UNKNOWN);
+ handler.onFailure(replicas.get(1).endpoint(), RequestFailure.UNKNOWN);
+
+ // Paxos failure should fire onDone immediately without waiting for the future
+ assertNotNull("onDone should fire immediately on paxos failure", status.get());
+ assertFalse("Should report paxos failure", status.get().isSuccess());
+ }
+
+ // ========================================
+ // Future completes first
+ // ========================================
+
+ @Test
+ public void testFutureSucceedsFirst_defersUntilPaxosSucceeds() throws Exception
+ {
+ EndpointsForToken replicas = threeReplicas();
+ AtomicReference<PaxosCommit.Status> status = new AtomicReference<>();
+ PaxosCommit<?> handler = createHandler(replicas, 2, status);
+
+ AsyncPromise<Void> promise = new AsyncPromise<>();
+ handler.setAugmentedCommitFuture(promise);
+
+ promise.trySuccess(null);
+ assertNull("onDone should NOT fire yet (paxos not done)", status.get());
+
+ sendSuccess(handler, replicas.get(0).endpoint());
+ assertNull(status.get());
+ sendSuccess(handler, replicas.get(1).endpoint());
+ assertNotNull("onDone should fire after paxos quorum", status.get());
+ assertTrue("Should be success", status.get().isSuccess());
+ }
+
+ @Test
+ public void testFutureFailsFirst_failsImmediately() throws Exception
+ {
+ EndpointsForToken replicas = threeReplicas();
+ AtomicReference<PaxosCommit.Status> status = new AtomicReference<>();
+ PaxosCommit<?> handler = createHandler(replicas, 2, status);
+
+ AsyncPromise<Void> promise = new AsyncPromise<>();
+ handler.setAugmentedCommitFuture(promise);
+
+ promise.tryFailure(new RuntimeException("satellite quorum not met"));
+
+ // Future failure should fire onDone immediately without waiting for paxos
+ assertNotNull("onDone should fire immediately on future failure", status.get());
+ assertFalse("Should report failure", status.get().isSuccess());
+ }
+
+ // ========================================
+ // Null future (no-op)
+ // ========================================
+
+ @Test
+ public void testNullFuture_behavesLikeNoAugmentedCommit() throws Exception
+ {
+ EndpointsForToken replicas = threeReplicas();
+ AtomicReference<PaxosCommit.Status> status = new AtomicReference<>();
+ PaxosCommit<?> handler = createHandler(replicas, 2, status);
+
+ handler.setAugmentedCommitFuture(null);
+
+ sendSuccess(handler, replicas.get(0).endpoint());
+ assertNull(status.get());
+
+ sendSuccess(handler, replicas.get(1).endpoint());
+ assertNotNull("Should fire after quorum", status.get());
+ assertTrue("Should be success", status.get().isSuccess());
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org