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
This commit is contained in:
Blake Eggleston 2026-04-01 16:14:38 -07:00
parent ab8fb8d9c2
commit 119b71981b
34 changed files with 4612 additions and 749 deletions

View File

@ -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"),

View File

@ -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);
}
}

View File

@ -88,20 +88,18 @@ public class IndexStatusManager
* @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()));
}
}

View File

@ -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());
}
}

View File

@ -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);
}
}

View File

@ -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);
}
}

View File

@ -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; }
@ -200,6 +201,22 @@ public interface ReplicaPlan<E extends Endpoints<E>, P extends ReplicaPlan<E, P>
{
private final Function<ReplicaPlan<?, ?>, ForWrite> 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,
int readQuorum)
{
super(keyspace, replicationStrategy, consistencyLevel, candidates, contacts, liveAndDown, recompute, epoch, readQuorum);
this.repairPlan = repairPlan;
}
public ForTokenRead(Keyspace keyspace,
AbstractReplicationStrategy replicationStrategy,
ConsistencyLevel consistencyLevel,
@ -210,13 +227,12 @@ public interface ReplicaPlan<E extends Endpoints<E>, P extends ReplicaPlan<E, P>
Function<ReplicaPlan<?, ?>, ReplicaPlan.ForWrite> repairPlan,
Epoch epoch)
{
super(keyspace, replicationStrategy, consistencyLevel, candidates, contacts, liveAndDown, recompute, epoch);
this.repairPlan = repairPlan;
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

View File

@ -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);
}
/**

View File

@ -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);
}
}
}

View File

@ -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() {}
}

View File

@ -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 ),

View File

@ -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;

View File

@ -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>
{

View File

@ -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());

View File

@ -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());
}
/**

View File

@ -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)

View File

@ -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());
}
}
}

View File

@ -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;

View File

@ -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());

View File

@ -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"));

View File

@ -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

View File

@ -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(),

View File

@ -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"));
}
}

View File

@ -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"));
}
}

View File

@ -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());
}
}

View File

@ -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;
}
}
}

View File

@ -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);
}
}

View File

@ -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");

View File

@ -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;
}
}

View File

@ -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());
}
}

View File

@ -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);
}
}

View File

@ -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;

View File

@ -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());
}
}