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

CEP-58: Add property-based tests for response tracking 

Randomized tests covering response/failure orderings and timeouts for
write response handlers, ReadCallback, paxos prepare/propose/commit, and
tracked read reconciliation. Adds the small test hooks needed to drive
them (isComplete, overridable isFromLocalDc, @VisibleForTesting ctors),
and fixes PaxosPrepare to set haveTrackedDataResponseIfNeeded in the
constructor rather than at start.

Patch by Blake Eggleston; reviewed by Francisco Guerrero and Ariel Weisberg for CASSANDRA-21106
This commit is contained in:
Blake Eggleston 2026-02-10 15:28:40 -08:00
parent 154b6a38c9
commit 550dd312fb
12 changed files with 3485 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java
View File

@ -361,6 +361,14 @@ public abstract class AbstractWriteResponseHandler<T> implements RequestCallback
callback.run();
}
/**
* Check if the handler has completed (for testing purposes).
*/
public boolean isComplete()
{
return condition.isSignalled();
}
@Override
public void onFailure(InetAddressAndPort from, RequestFailure failure)
{

10
src/java/org/apache/cassandra/service/paxos/PaxosCommit.java
View File

@ -23,6 +23,8 @@ import java.util.function.Consumer;
import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
import org.agrona.collections.IntHashSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -470,13 +472,19 @@ public class PaxosCommit<OnDone extends Consumer<? super PaxosCommit.Status>> ex
response(response != null, from);
}
@VisibleForTesting
protected boolean isFromLocalDc(InetAddressAndPort endpoint)
{
return InOurDc.endpoints().test(endpoint);
}
/**
* Record a failure or success response if {@code from} contributes to our consistency.
* If we have reached a final outcome of the commit, run {@code onDone}.
*/
private void response(boolean success, InetAddressAndPort from)
{
if (consistencyForCommit.isDatacenterLocal() && !InOurDc.endpoints().test(from))
if (consistencyForCommit.isDatacenterLocal() && !isFromLocalDc(from))
return;
long responses = responsesUpdater.addAndGet(this, success ? 0x1L : 0x100000000L);

8
src/java/org/apache/cassandra/service/paxos/PaxosPrepare.java
View File

@ -34,6 +34,7 @@ import javax.annotation.Nullable;
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;
@ -371,6 +372,7 @@ public class PaxosPrepare extends PaxosRequestCallback<PaxosPrepare.Response> im
this.withLatest = new ArrayList<>(participants.sizeOfConsensusQuorum);
this.latestAccepted = this.latestCommitted = Committed.none(request.partitionKey, request.table);
this.onDone = onDone;
this.haveTrackedDataResponseIfNeeded = request.read == null || !isTracked();
}
private boolean hasInProgressProposal()
@ -438,8 +440,6 @@ public class PaxosPrepare extends PaxosRequestCallback<PaxosPrepare.Response> im
private static <R extends AbstractRequest<R>> void startTracked(PaxosPrepare prepare, Participants participants, Message<R> send, BiFunction<R, RequestTime, Future<Response>> selfHandler)
{
if (prepare.request.read == null)
prepare.haveTrackedDataResponseIfNeeded = true;
Message<R> selfMessage = null;
Message<R> summaryMessage = null;
Id readId = Id.nextId();
@ -504,7 +504,6 @@ public class PaxosPrepare extends PaxosRequestCallback<PaxosPrepare.Response> im
private static <R extends AbstractRequest<R>> void startUntracked(PaxosPrepare prepare, Participants participants, Message<R> send, BiFunction<R, RequestTime, Future<Response>> selfHandler)
{
prepare.haveTrackedDataResponseIfNeeded = true;
Message<R> selfMessage = null;
for (int i = 0, size = participants.sizeOfPoll() ; i < size ; ++i)
@ -1119,7 +1118,8 @@ public class PaxosPrepare extends PaxosRequestCallback<PaxosPrepare.Response> im
super(ballot, electorate, read, isWrite, isForRecovery);
}
private Request(Ballot ballot, Electorate electorate, DecoratedKey partitionKey, TableMetadata table, boolean isWrite, boolean isForRecovery)
@VisibleForTesting
Request(Ballot ballot, Electorate electorate, DecoratedKey partitionKey, TableMetadata table, boolean isWrite, boolean isForRecovery)
{
super(ballot, electorate, partitionKey, table, isWrite, isForRecovery);
}

6
src/java/org/apache/cassandra/service/paxos/PaxosPropose.java
View File

@ -167,6 +167,12 @@ public class PaxosPropose<OnDone extends Consumer<? super PaxosPropose.Status>>
this.onDone = onDone;
}
@VisibleForTesting
PaxosPropose(Proposal proposal, int participants, int required, OnDone onDone)
{
this(proposal, participants, required, false, onDone);
}
/**
* Submit the proposal for commit with all replicas, and return an object that can be waited on synchronously for the result,
* or for the present status if the time elapses without a final result being reached.

6
src/java/org/apache/cassandra/service/reads/tracked/ReadReconciliations.java
View File

@ -157,7 +157,7 @@ public class ReadReconciliations implements ExpiredStatePurger.Expireable
return n;
}
private static final class Coordinator
static class Coordinator
{
private static final Logger logger = LoggerFactory.getLogger(Coordinator.class);
@ -355,12 +355,12 @@ public class ReadReconciliations implements ExpiredStatePurger.Expireable
return next;
}
private boolean updateRemainingAndMaybeComplete(int mutationsDelta, int summariesDelta, int syncAcksDelta)
protected boolean updateRemainingAndMaybeComplete(int mutationsDelta, int summariesDelta, int syncAcksDelta)
{
return updateRemaining(mutationsDelta, summariesDelta, syncAcksDelta) == 0 && complete();
}
private boolean complete()
protected boolean complete()
{
if (isDataNode())
MutationTrackingService.instance().localReads().acknowledgeReconcile(id, augmentingOffsets());

562
test/unit/org/apache/cassandra/service/ReadCallbackPropertyTest.java Normal file
View File

@ -0,0 +1,562 @@
/*
* 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;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.Supplier;
import com.google.common.collect.ImmutableList;
import org.junit.Assert;
import org.junit.Test;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.ReadResponse;
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.locator.ReplicaPlan;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.service.reads.ReadCallback;
import org.apache.cassandra.service.reads.ResponseResolver;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.transport.Dispatcher;
import static org.quicktheories.QuickTheory.qt;
/**
* Property-based tests for ReadCallback with dynamic topology generation.
* Tests incremental response behavior: applies responses one at a time and validates
* completion state after each response.
*/
public class ReadCallbackPropertyTest extends ResponseHandlerPropertyTestBase
{
private static final ImmutableList<ConsistencyLevel> consistencyLevels = ImmutableList.of(
ConsistencyLevel.ONE,
ConsistencyLevel.TWO,
ConsistencyLevel.THREE,
ConsistencyLevel.QUORUM,
ConsistencyLevel.ALL,
ConsistencyLevel.LOCAL_ONE,
ConsistencyLevel.LOCAL_QUORUM,
ConsistencyLevel.EACH_QUORUM
);
@Override
protected List<ConsistencyLevel> consistencyLevels()
{
return consistencyLevels;
}
/**
* Calculate expected outcome based on responses so far.
* Returns null if more responses needed (incomplete).
*
* Single-requirement model for reads (no pending replica complexity):
* - Reads only count responses from full (non-pending) replicas
* - blockFor does NOT include pending replicas
* - Success when: full_replica_successes >= CL@RF
* - For LOCAL_* CLs, only local DC replicas are contacted
*/
private static ExpectedOutcome calculateExpectedOutcome(TopologyConfig topology,
ConsistencyLevel cl,
List<ResponseMessage> responses,
int blockFor,
int contactedReplicas)
{
// Count successes from full replicas only (pending can't serve reads)
int fullReplicaSuccesses = 0;
int fullReplicaResponses = 0;
Map<String, Integer> successesPerDc = new HashMap<>();
Map<String, Integer> responsesPerDc = new HashMap<>();
for (ResponseMessage resp : responses)
{
// Skip pending replicas - they can't serve reads
if (resp.replicaIdx >= topology.totalReplicas)
continue;
String dc = getReplicaDatacenter(resp.replicaIdx, topology);
fullReplicaResponses++;
responsesPerDc.merge(dc, 1, Integer::sum);
if (resp.isSuccess())
{
fullReplicaSuccesses++;
successesPerDc.merge(dc, 1, Integer::sum);
}
}
// Check completion conditions based on CL type
switch (cl)
{
case ONE:
case TWO:
case THREE:
case QUORUM:
case ALL:
{
// Global CLs: need blockFor successes from full replicas
if (fullReplicaSuccesses >= blockFor)
return ExpectedOutcome.SUCCESS;
// Early failure: can't satisfy requirement
int remaining = contactedReplicas - fullReplicaResponses;
if (fullReplicaSuccesses + remaining < blockFor)
return ExpectedOutcome.FAILURE;
return null;
}
case LOCAL_ONE:
case LOCAL_QUORUM:
{
// Local CLs: only count responses from local DC (datacenter1)
String localDc = "datacenter1";
int localSuccesses = successesPerDc.getOrDefault(localDc, 0);
int localResponses = responsesPerDc.getOrDefault(localDc, 0);
if (localSuccesses >= blockFor)
return ExpectedOutcome.SUCCESS;
// Early failure: can't satisfy requirement from local DC
int localRemaining = contactedReplicas - localResponses;
if (localSuccesses + localRemaining < blockFor)
return ExpectedOutcome.FAILURE;
return null;
}
case EACH_QUORUM:
{
// Note: ReadCallback doesn't have special per-DC tracking for EACH_QUORUM.
// It just counts total successes against blockFor (sum of DC quorums).
// This is different from DatacenterSyncWriteResponseHandler which tracks per-DC.
if (fullReplicaSuccesses >= blockFor)
return ExpectedOutcome.SUCCESS;
// Early failure: can't satisfy requirement
int remaining = contactedReplicas - fullReplicaResponses;
if (fullReplicaSuccesses + remaining < blockFor)
return ExpectedOutcome.FAILURE;
return null;
}
default:
throw new IllegalArgumentException("Unsupported CL: " + cl);
}
}
/**
* A minimal ResponseResolver for testing that only tracks response counts and data presence.
* This allows testing ReadCallback's completion logic without full read infrastructure.
*/
private static class TestResponseResolver extends ResponseResolver<EndpointsForToken, ReplicaPlan.ForTokenRead>
{
private volatile boolean dataPresent = false;
public TestResponseResolver(Supplier<ReplicaPlan.ForTokenRead> replicaPlan, Dispatcher.RequestTime requestTime)
{
super(null, null, replicaPlan, requestTime);
}
@Override
public boolean isDataPresent()
{
return dataPresent;
}
@Override
public void preprocess(Message<ReadResponse> message)
{
// Add to accumulator (parent class behavior)
responses.add(message);
// First full replica response makes data present
Replica replica = replicaPlan().lookup(message.from());
if (replica != null && replica.isFull())
dataPresent = true;
}
public int responseCount()
{
return responses.size();
}
/** Expose replicaPlan for testing */
public ReplicaPlan.ForTokenRead getReplicaPlan()
{
return replicaPlan();
}
}
/**
* Creates a ReplicaPlan for testing with the given parameters.
*/
private static ReplicaPlan.SharedForTokenRead createReplicaPlan(Keyspace ks,
ConsistencyLevel cl,
EndpointsForToken contacts)
{
ReplicaPlan.ForTokenRead plan = new ReplicaPlan.ForTokenRead(
ks,
ks.getReplicationStrategy(),
cl,
contacts, // candidates
contacts, // contacts
contacts, // liveAndDown
(cm) -> null, // recompute function
(self) -> null, // repair plan function
Epoch.EMPTY
);
return ReplicaPlan.shared(plan);
}
/**
* Gets the set of replica indices that belong to the local datacenter (datacenter1).
*/
private static Set<Integer> getLocalDcReplicaIndices(TopologyConfig topology)
{
Set<Integer> localIndices = new HashSet<>();
int idx = 0;
for (Map.Entry<String, Integer> entry : topology.replicationFactors.entrySet())
{
String dc = entry.getKey();
int rf = entry.getValue();
if ("datacenter1".equals(dc))
{
for (int i = 0; i < rf; i++)
localIndices.add(idx + i);
}
idx += rf;
}
return localIndices;
}
/**
* Filters replicas to only include those from the local datacenter.
*/
private static EndpointsForToken filterToLocalDc(EndpointsForToken replicas, TopologyConfig topology)
{
Set<Integer> localIndices = getLocalDcReplicaIndices(topology);
List<Replica> localReplicas = new ArrayList<>();
for (int i = 0; i < replicas.size(); i++)
{
if (localIndices.contains(i))
localReplicas.add(replicas.get(i));
}
return EndpointsForToken.of(replicas.token(), localReplicas.toArray(new Replica[0]));
}
/**
* Bundles a handler with the set of contacted replica indices.
*/
private static class HandlerWithContacts
{
final ReadCallback<EndpointsForToken, ReplicaPlan.ForTokenRead> handler;
final Set<Integer> contactedIndices;
HandlerWithContacts(ReadCallback<EndpointsForToken, ReplicaPlan.ForTokenRead> handler, Set<Integer> contactedIndices)
{
this.handler = handler;
this.contactedIndices = contactedIndices;
}
}
/**
* Creates a fresh ReadCallback for testing.
* For LOCAL_* CLs, only local DC replicas are contacted.
*/
private static HandlerWithContacts createHandler(
Keyspace ks,
ConsistencyLevel cl,
EndpointsForToken fullReplicas,
TopologyConfig topology)
{
EndpointsForToken contacts;
Set<Integer> contactedIndices;
if (cl == ConsistencyLevel.LOCAL_ONE || cl == ConsistencyLevel.LOCAL_QUORUM)
{
// For LOCAL_* CLs, only contact local DC replicas
contacts = filterToLocalDc(fullReplicas, topology);
contactedIndices = getLocalDcReplicaIndices(topology);
}
else
{
// For global CLs, contact all full replicas
contacts = fullReplicas;
contactedIndices = new HashSet<>();
for (int i = 0; i < fullReplicas.size(); i++)
contactedIndices.add(i);
}
ReplicaPlan.SharedForTokenRead sharedPlan = createReplicaPlan(ks, cl, contacts);
Dispatcher.RequestTime requestTime = new Dispatcher.RequestTime(System.nanoTime(), System.nanoTime());
TestResponseResolver resolver = new TestResponseResolver(sharedPlan, requestTime);
ReadCallback<EndpointsForToken, ReplicaPlan.ForTokenRead> handler = new ReadCallback<>(resolver, null, sharedPlan, requestTime);
return new HandlerWithContacts(handler, contactedIndices);
}
/**
* Gets the endpoint for a replica index from the replica sets.
*/
private static InetAddressAndPort getEndpoint(int replicaIdx, ReplicaSets replicaSets)
{
if (replicaIdx < replicaSets.fullReplicas.size())
return replicaSets.fullReplicas.get(replicaIdx).endpoint();
int pendingIdx = replicaIdx - replicaSets.fullReplicas.size();
if (pendingIdx < replicaSets.pendingReplicas.size())
return replicaSets.pendingReplicas.get(pendingIdx).endpoint();
throw new IllegalArgumentException("Invalid replica index: " + replicaIdx);
}
/**
* Creates a minimal ReadResponse message for testing.
* Uses Message.synthetic() which is designed for testing and allows creating messages from arbitrary nodes.
* The TestResponseResolver doesn't access the payload, so we pass null.
*/
private static Message<ReadResponse> createResponseMessage(InetAddressAndPort from)
{
return Message.synthetic(from, org.apache.cassandra.net.Verb.READ_RSP, null);
}
/**
* Applies a single response to the handler.
* Returns true if the response was applied (i.e., from a replica in contacts).
*/
private static boolean applyResponse(ReadCallback<EndpointsForToken, ReplicaPlan.ForTokenRead> handler,
ResponseMessage response,
ReplicaSets replicaSets,
Set<Integer> contactedIndices)
{
// Skip pending replicas - they can't serve reads
if (response.replicaIdx >= replicaSets.fullReplicas.size())
return false;
// Skip replicas not in the contact set (important for LOCAL_* CLs)
if (!contactedIndices.contains(response.replicaIdx))
return false;
InetAddressAndPort endpoint = getEndpoint(response.replicaIdx, replicaSets);
if (response.isSuccess())
{
Message<ReadResponse> msg = createResponseMessage(endpoint);
handler.onResponse(msg);
}
else
{
handler.onFailure(endpoint, new RequestFailure(response.failureReason, null));
}
return true;
}
/**
* Checks if the handler has signaled completion.
*/
private static boolean isComplete(ReadCallback<EndpointsForToken, ReplicaPlan.ForTokenRead> handler)
{
// ReadCallback signals completion via its condition
// We check by attempting a zero-timeout await
return handler.await(0, java.util.concurrent.TimeUnit.MILLISECONDS);
}
/**
* Applies a pre-generated response sequence to a handler, stopping when complete.
* Returns the subset of responses that were actually applied.
*/
private static List<ResponseMessage> applyResponseSequence(
ReadCallback<EndpointsForToken, ReplicaPlan.ForTokenRead> handler,
List<ResponseMessage> responses,
ReplicaSets replicaSets,
Set<Integer> contactedIndices)
{
List<ResponseMessage> applied = new ArrayList<>();
for (ResponseMessage response : responses)
{
// Check if already complete before applying
if (isComplete(handler))
break;
if (applyResponse(handler, response, replicaSets, contactedIndices))
applied.add(response);
}
return applied;
}
/**
* Validates handler completed with expected outcome.
*/
private static void validateOutcome(ReadCallback<EndpointsForToken, ReplicaPlan.ForTokenRead> handler,
List<ResponseMessage> appliedResponses,
TopologyConfig topology,
ConsistencyLevel cl,
int contactedReplicas)
{
TestResponseResolver resolver = (TestResponseResolver) handler.resolver;
int blockFor = resolver.getReplicaPlan().readQuorum();
ExpectedOutcome expected = calculateExpectedOutcome(topology, cl, appliedResponses, blockFor, contactedReplicas);
boolean complete = isComplete(handler);
int successes = resolver.responseCount();
int failures = contactedReplicas - successes;
if (expected == null)
{
// Should not be complete yet
Assert.assertFalse(
String.format("Handler completed prematurely with %d successes, %d failures (blockFor=%d, CL=%s)",
successes, failures, blockFor, cl),
complete
);
}
else if (expected == ExpectedOutcome.SUCCESS)
{
Assert.assertTrue(
String.format("Handler should have completed successfully with %d successes (blockFor=%d, CL=%s)",
successes, blockFor, cl),
complete
);
Assert.assertTrue(
String.format("Data should be present with %d successes", successes),
resolver.isDataPresent()
);
}
else
{
// Expected failure - handler should be complete due to too many failures
Assert.assertTrue(
String.format("Handler should have completed (failed) with %d failures (blockFor=%d, contacts=%d, CL=%s)",
failures, blockFor, contactedReplicas, cl),
complete
);
}
}
// ========================================
// Property Tests
// ========================================
@Test
public void readCallbackBehavior()
{
qt()
.withExamples(250)
.forAll(testCaseGen())
.assuming(testCase -> {
// Filter out topologies where any scenario has insufficient replicas for its CL
for (CLScenario scenario : testCase.scenarios)
{
if (testCase.topology.totalReplicas < minReplicasForCL(scenario.cl))
return false;
if (!hasEnoughLocalReplicas(testCase.topology, scenario.cl))
return false;
}
return true;
})
.checkAssert(testCase -> {
try
{
Keyspace ks = getOrCreateKeyspace(testCase.topology);
ReplicaSets replicaSets = createReplicaSets(testCase.topology);
// Test all scenarios for this topology
for (CLScenario scenario : testCase.scenarios)
{
HandlerWithContacts hwc = createHandler(ks, scenario.cl, replicaSets.fullReplicas, testCase.topology);
List<ResponseMessage> appliedResponses = applyResponseSequence(
hwc.handler, scenario.responses, replicaSets, hwc.contactedIndices);
validateOutcome(hwc.handler, appliedResponses, testCase.topology, scenario.cl, hwc.contactedIndices.size());
}
}
catch (Throwable e)
{
if (e instanceof AssertionError)
throw (AssertionError) e;
throw new AssertionError("Test setup failed: " + e.getMessage(), e);
}
});
}
/**
* Minimum replicas needed for a consistency level to be valid.
* For LOCAL_* CLs, this returns the minimum needed in the local DC.
*/
private static int minReplicasForCL(ConsistencyLevel cl)
{
switch (cl)
{
case ONE:
case LOCAL_ONE:
return 1;
case TWO:
return 2;
case THREE:
return 3;
case QUORUM:
case LOCAL_QUORUM:
return 3; // Need at least 3 for quorum to be meaningful (3/2 + 1 = 2)
case EACH_QUORUM:
return 1; // Just need at least 1 replica per DC (checked separately)
case ALL:
return 1;
default:
return 1;
}
}
/**
* Check if the local DC has enough replicas for the given CL.
*/
private static boolean hasEnoughLocalReplicas(TopologyConfig topology, ConsistencyLevel cl)
{
if (cl == ConsistencyLevel.LOCAL_ONE || cl == ConsistencyLevel.LOCAL_QUORUM)
{
Integer localRf = topology.replicationFactors.get("datacenter1");
if (localRf == null)
return false;
return localRf >= minReplicasForCL(cl);
}
if (cl == ConsistencyLevel.EACH_QUORUM)
{
// EACH_QUORUM requires at least 1 replica in every DC
for (int rf : topology.replicationFactors.values())
{
if (rf < 1)
return false;
}
}
return true;
}
}

597
test/unit/org/apache/cassandra/service/ResponseHandlerPropertyTestBase.java Normal file
View File

@ -0,0 +1,597 @@
/*
* 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;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiFunction;
import java.util.stream.Stream;
import org.junit.BeforeClass;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.distributed.test.log.ClusterMetadataTestHelper;
import org.apache.cassandra.exceptions.RequestFailureReason;
import org.apache.cassandra.locator.BaseProximity;
import org.apache.cassandra.locator.EndpointsForToken;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.NodeProximity;
import org.apache.cassandra.locator.Replica;
import org.apache.cassandra.locator.ReplicaCollection;
import org.apache.cassandra.locator.ReplicaUtils;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.quicktheories.core.Gen;
import static org.quicktheories.generators.SourceDSL.arbitrary;
import static org.quicktheories.generators.SourceDSL.booleans;
import static org.quicktheories.generators.SourceDSL.integers;
import static org.quicktheories.generators.SourceDSL.lists;
public abstract class ResponseHandlerPropertyTestBase
{
protected static final Logger logger = LoggerFactory.getLogger(ResponseHandlerPropertyTestBase.class);
protected static final AtomicInteger keyspaceCounter = new AtomicInteger(0);
protected static final Map<String, Keyspace> topologyCache = new HashMap<>();
protected static final Set<InetAddressAndPort> registeredNodes = new HashSet<>();
@BeforeClass
public static void setUpClass() throws Throwable
{
// Set partitioner system property BEFORE DatabaseDescriptor initialization
System.setProperty("cassandra.partitioner", Murmur3Partitioner.class.getName());
SchemaLoader.loadSchema();
// Configure node proximity
NodeProximity sorter = new BaseProximity()
{
public <C extends ReplicaCollection<? extends C>> C sortedByProximity(InetAddressAndPort address, C replicas)
{
return replicas;
}
public int compareEndpoints(InetAddressAndPort target, Replica a1, Replica a2)
{
return 0;
}
public boolean isWorthMergingForRangeQuery(ReplicaCollection<?> merged, ReplicaCollection<?> l1, ReplicaCollection<?> l2)
{
return false;
}
};
DatabaseDescriptor.setNodeProximity(sorter);
// Set broadcast address to match first replica of datacenter1 (replicaIdx=0 → 127.1.0.255)
// This ensures InOurDc.endpoints() correctly identifies local DC replicas
InetAddress broadcastAddr = InetAddress.getByName("127.1.0.255");
DatabaseDescriptor.setBroadcastAddress(broadcastAddr);
// Register broadcast address with datacenter1 so locator knows our DC
InetAddressAndPort broadcastEndpoint = InetAddressAndPort.getByAddress(broadcastAddr);
ClusterMetadataTestHelper.register(broadcastEndpoint, "datacenter1", "rack1");
registeredNodes.add(broadcastEndpoint);
}
// ========================================
// Data Structures
// ========================================
/**
* Describes a cluster topology configuration.
*/
public static class TopologyConfig
{
public final int numDatacenters;
public final Map<String, Integer> replicationFactors; // DC name -> RF
public final Map<String, Integer> pendingReplicas; // DC name -> pending count
public final int totalReplicas;
public final int totalPending;
TopologyConfig(int numDatacenters, Map<String, Integer> replicationFactors, Map<String, Integer> pendingReplicas)
{
this.numDatacenters = numDatacenters;
this.replicationFactors = replicationFactors;
this.pendingReplicas = pendingReplicas;
this.totalReplicas = replicationFactors.values().stream().mapToInt(Integer::intValue).sum();
this.totalPending = pendingReplicas.values().stream().mapToInt(Integer::intValue).sum();
}
String signature()
{
StringBuilder sb = new StringBuilder();
sb.append("dcs=").append(numDatacenters);
replicationFactors.forEach((dc, rf) -> sb.append(",").append(dc).append(":").append(rf));
if (totalPending > 0)
{
sb.append(",pending:");
pendingReplicas.forEach((dc, count) -> {
if (count > 0)
sb.append(dc).append("=").append(count).append(";");
});
}
return sb.toString();
}
@Override
public String toString()
{
return signature();
}
}
/**
* A single response message (success or failure).
*/
public static class ResponseMessage
{
public final int replicaIdx;
public final RequestFailureReason failureReason; // null = success
public ResponseMessage(int replicaIdx, RequestFailureReason failureReason)
{
this.replicaIdx = replicaIdx;
this.failureReason = failureReason;
}
public boolean isSuccess()
{
return failureReason == null;
}
@Override
public String toString()
{
return String.format("replica=%d, %s", replicaIdx, isSuccess() ? "SUCCESS" : "FAILURE(" + failureReason + ")");
}
}
/**
* A test scenario for a specific consistency level with pre-generated responses.
*/
public static class CLScenario
{
public final ConsistencyLevel cl;
public final List<ResponseMessage> responses;
public CLScenario(ConsistencyLevel cl, List<ResponseMessage> responses)
{
this.cl = cl;
this.responses = responses;
}
@Override
public String toString()
{
return String.format("CL=%s, responses=%d", cl, responses.size());
}
}
/**
* A complete test case with topology and multiple CL scenarios.
*/
public static class TestCase
{
public final TopologyConfig topology;
public final List<CLScenario> scenarios;
public TestCase(TopologyConfig topology, List<CLScenario> scenarios)
{
this.topology = topology;
this.scenarios = scenarios;
}
@Override
public String toString()
{
return String.format("topology=%s, scenarios=%d", topology, scenarios.size());
}
}
/**
* Holds both full and pending replica sets for a topology.
*/
public static class ReplicaSets
{
public final EndpointsForToken fullReplicas;
public final EndpointsForToken pendingReplicas;
public ReplicaSets(EndpointsForToken fullReplicas, EndpointsForToken pendingReplicas)
{
this.fullReplicas = fullReplicas;
this.pendingReplicas = pendingReplicas;
}
}
/**
* Expected outcome of a response sequence
*/
public enum ExpectedOutcome
{
SUCCESS, // Handler should complete successfully
FAILURE // Handler should fail
}
// ========================================
// Generators
// ========================================
/**
* Generates varied datacenter topologies (1-7 DCs with varying RF and pending replicas).
*/
protected Gen<TopologyConfig> topologyGen()
{
return integers().between(1, 7).flatMap(numDcs -> {
return lists().of(integers().between(1, 5)).ofSize(numDcs).flatMap(rfList -> {
// Generate 0-2 pending replicas per DC
return lists().of(integers().between(0, 2)).ofSize(numDcs).map(pendingList -> {
Map<String, Integer> replicationFactors = new LinkedHashMap<>();
Map<String, Integer> pendingReplicas = new LinkedHashMap<>();
for (int i = 0; i < numDcs; i++)
{
String dcName = "datacenter" + (i + 1);
replicationFactors.put(dcName, rfList.get(i));
pendingReplicas.put(dcName, pendingList.get(i));
}
return new TopologyConfig(numDcs, replicationFactors, pendingReplicas);
});
});
});
}
// ========================================
// Topology Setup
// ========================================
/**
* Gets or creates a keyspace for the given topology.
*/
public static Keyspace getOrCreateKeyspace(TopologyConfig topology) throws Exception
{
String signature = topology.signature();
if (topologyCache.containsKey(signature))
return topologyCache.get(signature);
String keyspaceName = "PropTest" + keyspaceCounter.incrementAndGet();
// Register full replica nodes
int replicaIdx = 0;
for (Map.Entry<String, Integer> entry : topology.replicationFactors.entrySet())
{
String dcName = entry.getKey();
int rf = entry.getValue();
int dcNum = Integer.parseInt(dcName.substring("datacenter".length()));
for (int i = 0; i < rf; i++)
{
String ip = String.format("127.%d.0.%d", dcNum, 255 - replicaIdx);
InetAddressAndPort endpoint = InetAddressAndPort.getByName(ip);
if (!registeredNodes.contains(endpoint))
{
ClusterMetadataTestHelper.register(endpoint, dcName, "rack1");
registeredNodes.add(endpoint);
}
replicaIdx++;
}
}
// Pending replica nodes don't need ClusterMetadata registration
// We pass them directly to ReplicaPlan in createHandler()
// Just register them with basic info so InOurDc can identify their datacenter
for (Map.Entry<String, Integer> entry : topology.pendingReplicas.entrySet())
{
String dcName = entry.getKey();
int pending = entry.getValue();
int dcNum = Integer.parseInt(dcName.substring("datacenter".length()));
for (int i = 0; i < pending; i++)
{
String ip = String.format("127.%d.0.%d", dcNum, 255 - replicaIdx);
InetAddressAndPort endpoint = InetAddressAndPort.getByName(ip);
if (!registeredNodes.contains(endpoint))
{
// Just register for DC/rack identification - no join process needed
ClusterMetadataTestHelper.register(endpoint, dcName, "rack1");
registeredNodes.add(endpoint);
}
replicaIdx++;
}
}
// Create keyspace
Object[] dcRfPairs = topology.replicationFactors.entrySet().stream()
.flatMap(e -> Stream.of(e.getKey(), e.getValue()))
.toArray();
SchemaLoader.createKeyspace(keyspaceName, KeyspaceParams.nts(dcRfPairs),
SchemaLoader.standardCFMD(keyspaceName, "Standard"));
Keyspace ks = Keyspace.open(keyspaceName);
topologyCache.put(signature, ks);
return ks;
}
/**
* Creates full and pending replica sets for the given topology.
*/
public static ReplicaSets createReplicaSets(TopologyConfig topology) throws Exception
{
List<Replica> fullReplicas = new ArrayList<>();
List<Replica> pendingReplicas = new ArrayList<>();
int replicaIdx = 0;
// Create full replicas for all DCs first (must match getOrCreateKeyspace ordering)
for (Map.Entry<String, Integer> entry : topology.replicationFactors.entrySet())
{
int rf = entry.getValue();
int dcNum = Integer.parseInt(entry.getKey().substring("datacenter".length()));
for (int i = 0; i < rf; i++)
{
String ip = String.format("127.%d.0.%d", dcNum, 255 - replicaIdx);
fullReplicas.add(ReplicaUtils.full(InetAddressAndPort.getByName(ip)));
replicaIdx++;
}
}
// Then create pending replicas for all DCs
for (Map.Entry<String, Integer> entry : topology.pendingReplicas.entrySet())
{
int pending = entry.getValue();
int dcNum = Integer.parseInt(entry.getKey().substring("datacenter".length()));
for (int i = 0; i < pending; i++)
{
String ip = String.format("127.%d.0.%d", dcNum, 255 - replicaIdx);
pendingReplicas.add(ReplicaUtils.full(InetAddressAndPort.getByName(ip)));
replicaIdx++;
}
}
var token = Murmur3Partitioner.instance.getToken(ByteBufferUtil.bytes(0));
return new ReplicaSets(
EndpointsForToken.of(token, fullReplicas.toArray(new Replica[0])),
EndpointsForToken.of(token, pendingReplicas.toArray(new Replica[0]))
);
}
// ========================================
// Response Generation
// ========================================
/**
* Maps replica index to its datacenter.
* Indices 0..(totalReplicas-1) are full replicas, totalReplicas..(totalReplicas+totalPending-1) are pending.
*/
public static String getReplicaDatacenter(int replicaIdx, TopologyConfig topology)
{
int idx = 0;
// First check full replicas
for (Map.Entry<String, Integer> entry : topology.replicationFactors.entrySet())
{
int rf = entry.getValue();
if (replicaIdx < idx + rf)
return entry.getKey();
idx += rf;
}
// Then check pending replicas (indices continue from where full replicas left off)
int pendingStartIdx = topology.totalReplicas;
idx = pendingStartIdx;
for (Map.Entry<String, Integer> entry : topology.pendingReplicas.entrySet())
{
int pending = entry.getValue();
if (replicaIdx < idx + pending)
return entry.getKey();
idx += pending;
}
throw new IllegalArgumentException("Invalid replica index: " + replicaIdx);
}
protected abstract List<ConsistencyLevel> consistencyLevels();
/**
* Generates write-applicable consistency levels.
*/
protected Gen<ConsistencyLevel> consistencyLevelGen()
{
return arbitrary().pick(consistencyLevels());
}
/**
* Generates failure reasons for failed responses.
*/
protected Gen<RequestFailureReason> failureReasonGen()
{
return arbitrary().pick(
RequestFailureReason.TIMEOUT,
RequestFailureReason.UNKNOWN,
RequestFailureReason.INCOMPATIBLE_SCHEMA,
RequestFailureReason.COORDINATOR_BEHIND
);
}
/**
* Generates a random permutation of integers [0, n-1] using Fisher-Yates shuffle.
* This ensures deterministic reproducibility from QuickTheories' seed.
*/
protected static Gen<List<Integer>> permutationGen(int n)
{
if (n == 0) return lists().of(integers().all()).ofSize(0);
if (n == 1) return lists().of(integers().all()).ofSize(1).map(list -> {
List<Integer> result = new ArrayList<>();
result.add(0);
return result;
});
// Generate n-1 random swap indices for Fisher-Yates shuffle
return lists().of(integers().between(0, n - 1)).ofSize(n - 1).map(swaps -> {
List<Integer> result = new ArrayList<>();
for (int i = 0; i < n; i++)
result.add(i);
// Fisher-Yates shuffle using generated swap indices
for (int i = 0; i < n - 1; i++)
{
int maxSwap = n - i - 1;
int j = i + Math.min(swaps.get(i), maxSwap);
Collections.swap(result, i, j);
}
return result;
});
}
/**
* Generates a response sequence of the given size where each element gets a random type
* from the provided enum values, in a random order.
*
* @param size number of responses to generate
* @param values possible response types
* @param factory creates a response message from (index, type)
*/
protected static <T extends Enum<T>, R> Gen<List<R>> typedResponseSequenceGen(
int size, T[] values, BiFunction<Integer, T, R> factory)
{
return lists().of(arbitrary().pick(values)).ofSize(size).flatMap(types ->
permutationGen(size).map(ordering -> {
List<R> responses = new ArrayList<>();
for (int idx : ordering)
responses.add(factory.apply(idx, types.get(idx)));
return responses;
})
);
}
/**
* Creates a hardcoded sequence where all responses have the same type.
*/
protected static <T extends Enum<T>, R> List<R> allSameTypeSequence(
int size, T type, BiFunction<Integer, T, R> factory)
{
List<R> responses = new ArrayList<>();
for (int i = 0; i < size; i++)
responses.add(factory.apply(i, type));
return responses;
}
/**
* Generates a response sequence for a given topology.
* Each replica (full + pending) gets a success/failure outcome, and responses are in random order.
*/
private Gen<List<ResponseMessage>> responseSequenceGen(TopologyConfig topology)
{
int totalEndpoints = topology.totalReplicas + topology.totalPending;
// Generate success/failure for each endpoint (full + pending)
return lists().of(booleans().all()).ofSize(totalEndpoints).flatMap(successFlags -> {
// Generate failure reasons for each endpoint (used only if failure)
return lists().of(failureReasonGen()).ofSize(totalEndpoints).flatMap(failureReasons -> {
// Generate random ordering of responses
return permutationGen(totalEndpoints).map(ordering -> {
List<ResponseMessage> responses = new ArrayList<>();
for (int idx : ordering)
{
RequestFailureReason reason = successFlags.get(idx) ? null : failureReasons.get(idx);
responses.add(new ResponseMessage(idx, reason));
}
return responses;
});
});
});
}
/**
* Generates hardcoded all-success response sequence (full + pending replicas).
*/
private static List<ResponseMessage> allSuccessSequence(TopologyConfig topology)
{
List<ResponseMessage> responses = new ArrayList<>();
int totalEndpoints = topology.totalReplicas + topology.totalPending;
for (int i = 0; i < totalEndpoints; i++)
responses.add(new ResponseMessage(i, null));
return responses;
}
/**
* Generates hardcoded all-failure response sequence (full + pending replicas).
*/
private static List<ResponseMessage> allFailureSequence(TopologyConfig topology)
{
List<ResponseMessage> responses = new ArrayList<>();
int totalEndpoints = topology.totalReplicas + topology.totalPending;
for (int i = 0; i < totalEndpoints; i++)
responses.add(new ResponseMessage(i, RequestFailureReason.TIMEOUT));
return responses;
}
/**
* Generates a CL scenario (consistency level + response sequence).
*/
private Gen<CLScenario> scenarioGen(TopologyConfig topology)
{
return consistencyLevelGen().flatMap(cl ->
responseSequenceGen(topology).map(responses ->
new CLScenario(cl, responses)
)
);
}
/**
* Generates a complete test case with topology and multiple CL scenarios.
* Includes 3 random scenarios plus 2 hardcoded scenarios (all-success, all-failure).
*/
protected Gen<TestCase> testCaseGen()
{
return random -> {
TopologyConfig topology = topologyGen().generate(random);
List<CLScenario> scenarios = new ArrayList<>();
for (int i=0; i<50; i++)
scenarios.add(scenarioGen(topology).generate(random));
// Add hardcoded all-success and failure scenarios for each CL
for (ConsistencyLevel cl: consistencyLevels())
{
scenarios.add(new CLScenario(cl, allSuccessSequence(topology)));
scenarios.add(new CLScenario(cl, allFailureSequence(topology)));
}
return new TestCase(topology, scenarios);
};
}
}

648
test/unit/org/apache/cassandra/service/WriteResponseHandlerPropertyTest.java Normal file
View File

@ -0,0 +1,648 @@
/*
* 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;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import com.google.common.base.Predicates;
import com.google.common.collect.ImmutableList;
import org.junit.Test;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.WriteType;
import org.apache.cassandra.exceptions.CoordinatorBehindException;
import org.apache.cassandra.exceptions.RequestFailure;
import org.apache.cassandra.exceptions.RetryOnDifferentSystemException;
import org.apache.cassandra.exceptions.WriteFailureException;
import org.apache.cassandra.exceptions.WriteTimeoutException;
import org.apache.cassandra.locator.EndpointsForToken;
import org.apache.cassandra.locator.InOurDc;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.Replica;
import org.apache.cassandra.locator.ReplicaPlans;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.transport.Dispatcher;
import static org.apache.cassandra.net.NoPayload.noPayload;
import static org.quicktheories.QuickTheory.qt;
/**
* Property-based tests for WriteResponseHandler with dynamic topology generation.
* Tests incremental response behavior: applies responses one at a time and validates
* completion state after each response.
*/
public class WriteResponseHandlerPropertyTest extends ResponseHandlerPropertyTestBase
{
private static final ImmutableList<ConsistencyLevel> consistencyLevels = ImmutableList.of(
ConsistencyLevel.ANY,
ConsistencyLevel.ONE,
ConsistencyLevel.TWO,
ConsistencyLevel.THREE,
ConsistencyLevel.QUORUM,
ConsistencyLevel.ALL,
ConsistencyLevel.LOCAL_ONE,
ConsistencyLevel.LOCAL_QUORUM,
ConsistencyLevel.EACH_QUORUM
);
@Override
protected List<ConsistencyLevel> consistencyLevels()
{
return consistencyLevels;
}
/**
* Calculate expected outcome based on responses so far.
* Returns null if more responses needed (incomplete).
*
* Two-requirement model for all CLs with pending replicas:
* 1. Consistency: committed replicas must satisfy CL@RF
* 2. Bootstrap safety: total replicas (committed + pending) must satisfy CL@RF + pending
*/
public static ExpectedOutcome calculateExpectedOutcome(TopologyConfig topology,
ConsistencyLevel cl,
List<ResponseMessage> responses,
int blockFor)
{
// Count successes/failures by committed vs pending status
int committedSuccesses = 0;
int committedTotal = 0;
int totalSuccesses = 0;
int totalResponses = responses.size();
Map<String, Integer> committedSuccessesPerDc = new HashMap<>();
Map<String, Integer> committedTotalPerDc = new HashMap<>();
Map<String, Integer> totalSuccessesPerDc = new HashMap<>();
Map<String, Integer> totalResponsesPerDc = new HashMap<>();
for (ResponseMessage resp : responses)
{
String dc = getReplicaDatacenter(resp.replicaIdx, topology);
boolean isPending = resp.replicaIdx >= topology.totalReplicas;
totalResponsesPerDc.merge(dc, 1, Integer::sum);
if (!isPending) committedTotalPerDc.merge(dc, 1, Integer::sum);
if (resp.isSuccess())
{
totalSuccesses++;
totalSuccessesPerDc.merge(dc, 1, Integer::sum);
if (!isPending)
{
committedSuccesses++;
committedSuccessesPerDc.merge(dc, 1, Integer::sum);
}
}
if (!isPending) committedTotal++;
}
int totalEndpoints = topology.totalReplicas + topology.totalPending;
int committedEndpoints = topology.totalReplicas;
// Check completion conditions based on CL type
switch (cl)
{
case ANY:
{
// ANY only needs 1 success from anywhere (can hint otherwise)
// ANY doesn't add pending to blockFor - it just needs any 1 ack
if (totalSuccesses >= 1)
return ExpectedOutcome.SUCCESS;
// Early failure: can't get even 1 success
int totalRemaining = totalEndpoints - totalResponses;
if (totalSuccesses + totalRemaining < 1)
return ExpectedOutcome.FAILURE;
return null;
}
case ONE:
case TWO:
case THREE:
case QUORUM:
case ALL:
{
// Global CLs: base requirement from RF, total includes pending
int totalBlockFor = blockFor;
int baseBlockFor = totalBlockFor - topology.totalPending;
if (committedSuccesses >= baseBlockFor && totalSuccesses >= totalBlockFor)
return ExpectedOutcome.SUCCESS;
// Early failure: can't satisfy either requirement
int committedRemaining = committedEndpoints - committedTotal;
int totalRemaining = totalEndpoints - totalResponses;
if (committedSuccesses + committedRemaining < baseBlockFor)
return ExpectedOutcome.FAILURE;
if (totalSuccesses + totalRemaining < totalBlockFor)
return ExpectedOutcome.FAILURE;
return null;
}
case LOCAL_ONE:
case LOCAL_QUORUM:
{
// Local DC requirements
String localDc = "datacenter1";
int localRf = topology.replicationFactors.get(localDc);
int localPending = topology.pendingReplicas.get(localDc);
int baseBlockFor = (cl == ConsistencyLevel.LOCAL_ONE) ? 1 : (localRf / 2 + 1);
int totalBlockFor = baseBlockFor + localPending;
int localCommittedSuccesses = committedSuccessesPerDc.getOrDefault(localDc, 0);
int localTotalSuccesses = totalSuccessesPerDc.getOrDefault(localDc, 0);
int localCommittedResponses = committedTotalPerDc.getOrDefault(localDc, 0);
int localTotalResponses = totalResponsesPerDc.getOrDefault(localDc, 0);
if (localCommittedSuccesses >= baseBlockFor && localTotalSuccesses >= totalBlockFor)
return ExpectedOutcome.SUCCESS;
// Early failure: can't satisfy either requirement
int committedRemaining = localRf - localCommittedResponses;
int totalRemaining = (localRf + localPending) - localTotalResponses;
if (localCommittedSuccesses + committedRemaining < baseBlockFor)
return ExpectedOutcome.FAILURE;
if (localTotalSuccesses + totalRemaining < totalBlockFor)
return ExpectedOutcome.FAILURE;
return null;
}
case EACH_QUORUM:
{
// Check if all DCs satisfy both requirements
boolean allDcsSatisfied = true;
for (Map.Entry<String, Integer> entry : topology.replicationFactors.entrySet())
{
String dc = entry.getKey();
int dcRf = entry.getValue();
int dcPending = topology.pendingReplicas.get(dc);
int baseBlockFor = (dcRf / 2 + 1);
int totalBlockFor = baseBlockFor + dcPending;
int dcCommittedSuccesses = committedSuccessesPerDc.getOrDefault(dc, 0);
int dcTotalSuccesses = totalSuccessesPerDc.getOrDefault(dc, 0);
if (dcCommittedSuccesses < baseBlockFor || dcTotalSuccesses < totalBlockFor)
allDcsSatisfied = false;
}
if (allDcsSatisfied) return ExpectedOutcome.SUCCESS;
// Early failure: check if any DC can't satisfy either requirement
for (Map.Entry<String, Integer> entry : topology.replicationFactors.entrySet())
{
String dc = entry.getKey();
int dcRf = entry.getValue();
int dcPending = topology.pendingReplicas.get(dc);
int baseBlockFor = (dcRf / 2 + 1);
int totalBlockFor = baseBlockFor + dcPending;
int dcCommittedSuccesses = committedSuccessesPerDc.getOrDefault(dc, 0);
int dcTotalSuccesses = totalSuccessesPerDc.getOrDefault(dc, 0);
int dcCommittedResponses = committedTotalPerDc.getOrDefault(dc, 0);
int dcTotalResponses = totalResponsesPerDc.getOrDefault(dc, 0);
int committedRemaining = dcRf - dcCommittedResponses;
int totalRemaining = (dcRf + dcPending) - dcTotalResponses;
if (dcCommittedSuccesses + committedRemaining < baseBlockFor)
return ExpectedOutcome.FAILURE;
if (dcTotalSuccesses + totalRemaining < totalBlockFor)
return ExpectedOutcome.FAILURE;
}
return null;
}
default:
throw new IllegalArgumentException("Unsupported CL: " + cl);
}
}
/**
* Applies a pre-generated response sequence to a handler, stopping when complete.
* Returns the subset of responses that were actually applied.
*/
private static List<ResponseMessage> applyResponseSequence(AbstractWriteResponseHandler handler,
List<ResponseMessage> responses,
ReplicaSets replicaSets)
{
List<ResponseMessage> appliedResponses = new ArrayList<>();
for (ResponseMessage response : responses)
{
if (handler.isComplete())
break;
applyResponse(handler, response, replicaSets);
appliedResponses.add(response);
}
return appliedResponses;
}
// ========================================
// Testing Helpers
// ========================================
/**
* Creates a fresh handler for testing.
*/
private static AbstractWriteResponseHandler createHandler(Keyspace ks,
ConsistencyLevel cl,
EndpointsForToken targets,
EndpointsForToken pending) throws Exception
{
return ks.getReplicationStrategy().getWriteResponseHandler(
ReplicaPlans.forWrite(ks, cl, (cm) -> targets, (cm) -> pending, ClusterMetadata.current().epoch,
Predicates.alwaysTrue(), ReplicaPlans.writeAll),
null,
WriteType.SIMPLE,
null,
Dispatcher.RequestTime.forImmediateExecution());
}
/**
* Applies a single response to the handler.
*/
private static void applyResponse(AbstractWriteResponseHandler handler,
ResponseMessage response,
ReplicaSets replicaSets)
{
// Determine which replica set to use based on index
EndpointsForToken targets;
int adjustedIdx;
if (response.replicaIdx < replicaSets.fullReplicas.size())
{
// Full replica
targets = replicaSets.fullReplicas;
adjustedIdx = response.replicaIdx;
}
else
{
// Pending replica
targets = replicaSets.pendingReplicas;
adjustedIdx = response.replicaIdx - replicaSets.fullReplicas.size();
}
InetAddressAndPort endpoint = targets.get(adjustedIdx).endpoint();
if (response.isSuccess())
{
Message msg = Message.builder(Verb.ECHO_REQ, noPayload)
.from(endpoint)
.build();
handler.onResponse(msg);
}
else
{
handler.onFailure(endpoint,
RequestFailure.forReason(response.failureReason));
}
}
/**
* Known bug: candidateReplicaCount() doesn't include pending replicas for LOCAL CLs.
*
* When local DC has >1 pending replicas:
* - blockFor = baseQuorum + localPending (e.g., RF=3: blockFor = 2 + 2 = 4)
* - candidateReplicaCount = localRF only (e.g., 3 full replicas)
* - Early failure detection: blockFor + failures > candidates
* - Result: 4 + 0 > 3, handler fails immediately even with zero failures
*
* With pending=1, blockFor=candidates, so bug only triggers with failures.
* With pending>1, blockFor>candidates, so bug triggers deterministically.
*
* TODO: Remove this workaround after coordinator plan refactor fixes candidateReplicaCount
*/
public static boolean isKnownBugScenario(TopologyConfig topology, ConsistencyLevel cl, List<ResponseMessage> responses)
{
if (cl == ConsistencyLevel.LOCAL_ONE || cl == ConsistencyLevel.LOCAL_QUORUM)
{
String localDc = "datacenter1";
int localPending = topology.pendingReplicas.get(localDc);
// Bug deterministically triggers when pending > 1
return localPending > 1;
}
if (cl == ConsistencyLevel.EACH_QUORUM)
{
// Known bug: EACH_QUORUM early failure detection doesn't work with pending replicas.
// When a DC has pending replicas AND receives failures, the handler can get stuck
// incomplete instead of failing early.
//
// Example: DC1 with RF=2, 1 pending needs 3 acks total. If it gets 2 successes and
// 1 failure (all 3 nodes responded), it can't possibly succeed but handler doesn't fail.
//
// TODO: Remove after EACH_QUORUM early failure detection is fixed
for (Map.Entry<String, Integer> entry : topology.replicationFactors.entrySet())
{
String dc = entry.getKey();
int dcPending = topology.pendingReplicas.get(dc);
if (dcPending > 0)
{
// Check if this DC has any failures
boolean hasFailures = responses.stream()
.anyMatch(r -> {
String respDc = getReplicaDatacenter(r.replicaIdx, topology);
return respDc.equals(dc) && !r.isSuccess();
});
if (hasFailures)
return true;
}
}
}
return false;
}
/**
* Validates handler completed with expected outcome.
*/
private static void validateOutcome(AbstractWriteResponseHandler handler,
List<ResponseMessage> responses,
TopologyConfig topology,
ConsistencyLevel cl) throws Exception
{
// TODO: Remove after coordinator plan refactor fixes candidateReplicaCount
if (isKnownBugScenario(topology, cl, responses))
{
if (cl == ConsistencyLevel.LOCAL_ONE || cl == ConsistencyLevel.LOCAL_QUORUM)
{
logger.info("[KNOWN BUG] Skipping validation: {} with {} pending replicas in datacenter1 triggers candidateReplicaCount bug (blockFor > candidates)",
cl, topology.pendingReplicas.get("datacenter1"));
}
else if (cl == ConsistencyLevel.EACH_QUORUM)
{
logger.info("[KNOWN BUG] Skipping validation: {} with pending replicas and failures triggers early failure detection bug",
cl);
}
// Drain handler to avoid hanging futures
try { handler.get(); } catch (Exception ignored) {}
return;
}
if (!handler.isComplete())
{
// Detailed diagnostics for incomplete handler
Map<String, Integer> successesPerDc = new HashMap<>();
Map<String, Integer> totalPerDc = new HashMap<>();
int successCount = 0;
for (ResponseMessage resp : responses)
{
String dc = getReplicaDatacenter(resp.replicaIdx, topology);
totalPerDc.merge(dc, 1, Integer::sum);
if (resp.isSuccess())
{
successCount++;
successesPerDc.merge(dc, 1, Integer::sum);
}
}
StringBuilder diagnostic = new StringBuilder();
diagnostic.append(String.format("Handler not complete after %d responses\n", responses.size()));
diagnostic.append(String.format("Topology: %s, CL: %s, blockFor: %d\n", topology, cl, handler.blockFor()));
diagnostic.append(String.format("Total successes: %d\n", successCount));
diagnostic.append("Per-DC breakdown:\n");
for (String dc : topology.replicationFactors.keySet())
{
int dcRf = topology.replicationFactors.get(dc);
int dcQuorum = dcRf / 2 + 1;
int dcSuccesses = successesPerDc.getOrDefault(dc, 0);
int dcTotal = totalPerDc.getOrDefault(dc, 0);
diagnostic.append(String.format(" %s: RF=%d, quorum=%d, responses=%d, successes=%d\n",
dc, dcRf, dcQuorum, dcTotal, dcSuccesses));
}
throw new AssertionError(diagnostic.toString());
}
// Calculate what the outcome should be based on responses
ExpectedOutcome expected = calculateExpectedOutcome(topology, cl, responses, handler.blockFor());
if (expected == null)
{
// Detailed diagnostics for model error
int successCount = 0;
Map<String, Integer> successesPerDc = new HashMap<>();
for (ResponseMessage resp : responses)
{
if (resp.isSuccess())
{
successCount++;
String dc = getReplicaDatacenter(resp.replicaIdx, topology);
successesPerDc.merge(dc, 1, Integer::sum);
}
}
StringBuilder diagnostic = new StringBuilder();
diagnostic.append(String.format("Handler completed but model says incomplete - model error\n"));
diagnostic.append(String.format("Topology: %s, CL: %s, blockFor: %d\n", topology, cl, handler.blockFor()));
diagnostic.append(String.format("Responses applied: %d, total successes: %d\n", responses.size(), successCount));
if (cl == ConsistencyLevel.LOCAL_ONE || cl == ConsistencyLevel.LOCAL_QUORUM)
{
int localSuccesses = successesPerDc.getOrDefault("datacenter1", 0);
diagnostic.append(String.format("LOCAL DC (datacenter1) successes: %d\n", localSuccesses));
// Check handler type
diagnostic.append(String.format("Handler type: %s\n", handler.getClass().getSimpleName()));
diagnostic.append(String.format("Handler ackCount: %d\n", handler.ackCount()));
// Check what the locator thinks
diagnostic.append(String.format("Locator's local DC: %s\n",
DatabaseDescriptor.getLocalDataCenter()));
diagnostic.append(String.format("Broadcast address: %s\n",
DatabaseDescriptor.getBroadcastAddress()));
// Show which responses were applied and their DCs
diagnostic.append("Applied responses:\n");
// Build endpoint lookup for replica indices
List<InetAddressAndPort> allEndpoints = new ArrayList<>();
handler.replicaPlan.contacts().forEach(r -> allEndpoints.add(r.endpoint()));
handler.replicaPlan.pending().forEach(r -> allEndpoints.add(r.endpoint()));
for (int i = 0; i < Math.min(responses.size(), 10); i++)
{
ResponseMessage r = responses.get(i);
String dc = getReplicaDatacenter(r.replicaIdx, topology);
String type = r.replicaIdx < topology.totalReplicas ? "full" : "pending";
// Check what InOurDc thinks about this endpoint
InetAddressAndPort endpoint = allEndpoints.get(r.replicaIdx);
boolean inOurDc = InOurDc.isInOurDc(endpoint);
diagnostic.append(String.format(" [%d] %s, DC=%s, success=%s, InOurDc=%s\n",
r.replicaIdx, type, dc, r.isSuccess(), inOurDc));
}
if (responses.size() > 10)
diagnostic.append(String.format(" ... and %d more responses\n", responses.size() - 10));
}
diagnostic.append(String.format("Total endpoints: %d (full=%d, pending=%d)\n",
topology.totalReplicas + topology.totalPending,
topology.totalReplicas, topology.totalPending));
diagnostic.append("Per-DC successes:\n");
for (String dc : topology.replicationFactors.keySet())
{
int dcSuccesses = successesPerDc.getOrDefault(dc, 0);
diagnostic.append(String.format(" %s: %d successes (RF=%d, pending=%d)\n",
dc, dcSuccesses,
topology.replicationFactors.get(dc),
topology.pendingReplicas.get(dc)));
}
// Debug: show what the handler thinks about pending replicas
diagnostic.append(String.format("Handler's replicaPlan.pending() size: %d\n", handler.replicaPlan.pending().size()));
for (Replica replica : handler.replicaPlan.pending())
{
diagnostic.append(String.format(" Pending replica: %s\n", replica.endpoint()));
}
throw new AssertionError(diagnostic.toString());
}
try
{
handler.get();
if (expected == ExpectedOutcome.FAILURE)
{
throw new AssertionError(String.format(
"Expected failure but succeeded: topology=%s, CL=%s, responses=%d",
topology, cl, responses.size()));
}
}
catch (WriteTimeoutException | WriteFailureException |
CoordinatorBehindException | RetryOnDifferentSystemException e)
{
if (expected == ExpectedOutcome.SUCCESS)
{
throw new AssertionError(String.format(
"Expected success but failed: topology=%s, CL=%s, responses=%d, error=%s",
topology, cl, responses.size(), e.getMessage()));
}
}
}
// ========================================
// Property Tests
// ========================================
@Test
public void writeResponseHandlerBehavior()
{
qt()
.withExamples(250)
.forAll(testCaseGen())
.assuming(testCase -> {
// Filter out topologies where any scenario has insufficient replicas for its CL
for (CLScenario scenario : testCase.scenarios)
{
if (testCase.topology.totalReplicas < minReplicasForCL(scenario.cl))
return false;
// For EACH_QUORUM with pending, verify each DC has enough replicas
// to satisfy: dcQuorum = (dcRf/2+1) + dcPending
if (scenario.cl == ConsistencyLevel.EACH_QUORUM)
{
for (Map.Entry<String, Integer> entry : testCase.topology.replicationFactors.entrySet())
{
String dc = entry.getKey();
int dcRf = entry.getValue();
int dcPending = testCase.topology.pendingReplicas.get(dc);
int dcQuorum = (dcRf / 2 + 1) + dcPending;
int dcTotal = dcRf + dcPending;
// Skip if impossible to achieve quorum
if (dcTotal < dcQuorum)
return false;
}
}
}
return true;
})
.checkAssert(testCase -> {
try
{
Keyspace ks = getOrCreateKeyspace(testCase.topology);
ReplicaSets replicaSets = createReplicaSets(testCase.topology);
// Test all scenarios for this topology (3 random + 2 hardcoded)
for (CLScenario scenario : testCase.scenarios)
{
AbstractWriteResponseHandler handler = createHandler(ks, scenario.cl, replicaSets.fullReplicas, replicaSets.pendingReplicas);
List<ResponseMessage> appliedResponses = applyResponseSequence(handler, scenario.responses, replicaSets);
validateOutcome(handler, appliedResponses, testCase.topology, scenario.cl);
}
}
catch (Throwable e)
{
if (e instanceof AssertionError)
throw (AssertionError) e;
throw new AssertionError("Test setup failed: " + e.getMessage(), e);
}
});
}
/**
* Minimum replicas needed for a consistency level to be valid.
*/
private static int minReplicasForCL(ConsistencyLevel cl)
{
switch (cl)
{
case ANY:
case ONE:
case LOCAL_ONE:
return 1;
case TWO:
return 2;
case THREE:
return 3;
case QUORUM:
case LOCAL_QUORUM:
case EACH_QUORUM:
return 3;
case ALL:
return 1;
default:
return 1;
}
}
}

418
test/unit/org/apache/cassandra/service/paxos/PaxosCommitPropertyTest.java Normal file
View File

@ -0,0 +1,418 @@
/*
* 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.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import com.google.common.collect.ImmutableList;
import org.junit.Test;
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.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.TableMetadata;
import org.apache.cassandra.service.ResponseHandlerPropertyTestBase;
import org.apache.cassandra.utils.ByteBufferUtil;
import static org.apache.cassandra.net.NoPayload.noPayload;
import static org.apache.cassandra.service.paxos.Commit.Agreed;
import static org.quicktheories.QuickTheory.qt;
/**
* Property-based tests for PaxosCommit response tracking logic.
*
* PaxosCommit uses a single-counter model: it counts accepts and failures in a
* flat manner (no committed vs pending distinction). DC-local filtering applies
* for datacenter-local consistency levels.
*
* Tests all write consistency levels: ANY, ONE, TWO, THREE, QUORUM, ALL,
* LOCAL_ONE, LOCAL_QUORUM, EACH_QUORUM.
*/
public class PaxosCommitPropertyTest extends ResponseHandlerPropertyTestBase
{
private static final ImmutableList<ConsistencyLevel> commitConsistencyLevels = ImmutableList.of(
ConsistencyLevel.ANY,
ConsistencyLevel.ONE,
ConsistencyLevel.TWO,
ConsistencyLevel.THREE,
ConsistencyLevel.QUORUM,
ConsistencyLevel.ALL,
ConsistencyLevel.LOCAL_ONE,
ConsistencyLevel.LOCAL_QUORUM,
ConsistencyLevel.EACH_QUORUM
);
@Override
protected List<ConsistencyLevel> consistencyLevels()
{
return commitConsistencyLevels;
}
// ========================================
// Independent Model
// ========================================
/**
* Independent model for PaxosCommit completion.
*
* PaxosCommit uses a single counter:
* - DC-local CLs filter out non-local DC responses entirely
* - SUCCESS: accepts == required (exact match for once-only signaling)
* - FAILURE: replicas.size() - failures == required - 1 (impossible to reach required)
*
* Note: replicas.size() is ALL replicas across ALL DCs, even for local CLs.
* This asymmetry (count only local, but use global size for failure) is
* production behavior.
*/
static ExpectedOutcome calculateCommitOutcome(TopologyConfig topology,
ConsistencyLevel cl,
List<ResponseMessage> responses,
int required,
int totalReplicaCount)
{
boolean dcLocalFilter = cl.isDatacenterLocal();
int accepts = 0;
int failures = 0;
for (ResponseMessage resp : responses)
{
if (dcLocalFilter)
{
String dc = getReplicaDatacenter(resp.replicaIdx, topology);
if (!"datacenter1".equals(dc))
continue;
}
if (resp.isSuccess())
accepts++;
else
failures++;
}
// Success: reached required accepts
if (accepts >= required)
return ExpectedOutcome.SUCCESS;
// Failure: impossible to reach required
// (uses totalReplicaCount which includes all DCs, even for local CLs)
if (totalReplicaCount - failures < required)
return ExpectedOutcome.FAILURE;
return null; // still in progress
}
// ========================================
// Handler creation and response application
// ========================================
/**
* 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>>
extends PaxosCommit<T>
{
private final Set<InetAddressAndPort> localEndpoints;
TestableCommit(Agreed commit, EndpointsForToken replicas, int required,
ConsistencyLevel consistencyForCommit, T onDone,
Set<InetAddressAndPort> localEndpoints)
{
super(commit, false, consistencyForCommit, consistencyForCommit, replicas, required, onDone);
this.localEndpoints = localEndpoints;
}
@Override
protected boolean isFromLocalDc(InetAddressAndPort endpoint)
{
return localEndpoints.contains(endpoint);
}
}
/**
* Build the set of datacenter1 endpoints from the replica sets, used to override
* DC filtering in TestableCommit.
*/
private static Set<InetAddressAndPort> localEndpointSet(TopologyConfig topology,
ReplicaSets replicaSets)
{
Set<InetAddressAndPort> local = new HashSet<>();
int dc1Full = topology.replicationFactors.getOrDefault("datacenter1", 0);
for (int i = 0; i < dc1Full; i++)
local.add(replicaSets.fullReplicas.get(i).endpoint());
// pending replicas for DC1 are at indices [dc1Full, dc1Full + dc1Pending)
// within replicaSets.pendingReplicas, but pendingReplicas is ordered per-DC too
int pendingOffset = 0;
for (Map.Entry<String, Integer> entry : topology.pendingReplicas.entrySet())
{
int count = entry.getValue();
if ("datacenter1".equals(entry.getKey()))
{
for (int i = 0; i < count; i++)
local.add(replicaSets.pendingReplicas.get(pendingOffset + i).endpoint());
break;
}
pendingOffset += count;
}
return local;
}
/**
* Compute blockFor for PaxosCommit, matching Participants.requiredFor() behavior.
*/
private static int computeBlockFor(Keyspace ks, ConsistencyLevel cl, EndpointsForToken pending)
{
return cl.blockForWrite(ks.getReplicationStrategy(), pending);
}
/**
* Creates a PaxosCommit handler for testing.
* Uses TestableCommit subclass to override DC filtering for LOCAL_* consistency levels.
* Builds a synthetic Agreed with an untracked keyspace so mutation tracking is bypassed.
*/
private static PaxosCommit<Consumer<PaxosCommit.Status>> createCommitHandler(
EndpointsForToken allReplicas, int required, ConsistencyLevel commitCl,
AtomicReference<PaxosCommit.Status> statusCapture,
TopologyConfig topology, ReplicaSets replicaSets, Keyspace ks)
{
Set<InetAddressAndPort> localEndpoints = localEndpointSet(topology, replicaSets);
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, allReplicas, required, commitCl, statusCapture::set, localEndpoints);
}
/**
* Applies a single response to the PaxosCommit handler.
*/
private static void applyCommitResponse(PaxosCommit<?> handler,
ResponseMessage response,
ReplicaSets replicaSets)
{
EndpointsForToken targets;
int adjustedIdx;
if (response.replicaIdx < replicaSets.fullReplicas.size())
{
targets = replicaSets.fullReplicas;
adjustedIdx = response.replicaIdx;
}
else
{
targets = replicaSets.pendingReplicas;
adjustedIdx = response.replicaIdx - replicaSets.fullReplicas.size();
}
InetAddressAndPort endpoint = targets.get(adjustedIdx).endpoint();
if (response.isSuccess())
{
Message<NoPayload> msg = Message.builder(Verb.ECHO_REQ, noPayload)
.from(endpoint)
.build();
handler.onResponse(msg);
}
else
{
handler.onFailure(endpoint, RequestFailure.forReason(response.failureReason));
}
}
/**
* Applies responses to the handler, stopping when onDone fires.
*/
private static List<ResponseMessage> applyCommitResponseSequence(
PaxosCommit<?> handler,
List<ResponseMessage> responses,
ReplicaSets replicaSets,
AtomicReference<PaxosCommit.Status> statusCapture)
{
List<ResponseMessage> appliedResponses = new ArrayList<>();
for (ResponseMessage response : responses)
{
if (statusCapture.get() != null)
break;
applyCommitResponse(handler, response, replicaSets);
appliedResponses.add(response);
}
return appliedResponses;
}
// ========================================
// Property Test
// ========================================
@Test
public void paxosCommitBehavior()
{
qt()
.withExamples(250)
.forAll(testCaseGen())
.assuming(testCase -> {
for (CLScenario scenario : testCase.scenarios)
{
if (testCase.topology.totalReplicas < minReplicasForCL(scenario.cl))
return false;
if (scenario.cl == ConsistencyLevel.EACH_QUORUM)
{
for (Map.Entry<String, Integer> entry : testCase.topology.replicationFactors.entrySet())
{
int dcRf = entry.getValue();
int dcPending = testCase.topology.pendingReplicas.get(entry.getKey());
int dcQuorum = (dcRf / 2 + 1) + dcPending;
int dcTotal = dcRf + dcPending;
if (dcTotal < dcQuorum)
return false;
}
}
}
return true;
})
.checkAssert(testCase -> {
try
{
Keyspace ks = getOrCreateKeyspace(testCase.topology);
ReplicaSets replicaSets = createReplicaSets(testCase.topology);
// Build combined replica list (full + pending) as PaxosCommit sees it
List<Replica> allReplicaList = new ArrayList<>();
replicaSets.fullReplicas.forEach(allReplicaList::add);
replicaSets.pendingReplicas.forEach(allReplicaList::add);
EndpointsForToken allReplicas = EndpointsForToken.of(
replicaSets.fullReplicas.token(),
allReplicaList.toArray(new Replica[0]));
for (CLScenario scenario : testCase.scenarios)
{
int blockFor = computeBlockFor(ks, scenario.cl, replicaSets.pendingReplicas);
AtomicReference<PaxosCommit.Status> statusCapture = new AtomicReference<>();
PaxosCommit<Consumer<PaxosCommit.Status>> handler =
createCommitHandler(allReplicas, blockFor, scenario.cl, statusCapture,
testCase.topology, replicaSets, ks);
List<ResponseMessage> appliedResponses =
applyCommitResponseSequence(handler, scenario.responses, replicaSets, statusCapture);
validateCommitOutcome(statusCapture.get(), appliedResponses,
testCase.topology, scenario.cl, blockFor,
allReplicas.size());
}
}
catch (Throwable e)
{
if (e instanceof AssertionError)
throw (AssertionError) e;
throw new AssertionError("Test setup failed: " + e.getMessage(), e);
}
});
}
/**
* Validates the PaxosCommit outcome against the independent model.
*/
private static void validateCommitOutcome(PaxosCommit.Status status,
List<ResponseMessage> responses,
TopologyConfig topology,
ConsistencyLevel cl,
int blockFor,
int totalReplicaCount)
{
ExpectedOutcome expected = calculateCommitOutcome(topology, cl, responses, blockFor, totalReplicaCount);
if (status == null)
{
if (expected != null)
{
throw new AssertionError(String.format(
"Model says %s but PaxosCommit handler didn't complete: " +
"topology=%s, CL=%s, blockFor=%d, responses=%d, totalReplicas=%d",
expected, topology, cl, blockFor, responses.size(), totalReplicaCount));
}
return;
}
if (expected == null)
{
throw new AssertionError(String.format(
"PaxosCommit handler completed but model says incomplete: " +
"topology=%s, CL=%s, blockFor=%d, responses=%d, totalReplicas=%d, status=%s",
topology, cl, blockFor, responses.size(), totalReplicaCount, status));
}
if (expected == ExpectedOutcome.SUCCESS && !status.isSuccess())
{
throw new AssertionError(String.format(
"Expected success but got failure: topology=%s, CL=%s, blockFor=%d, responses=%d",
topology, cl, blockFor, responses.size()));
}
if (expected == ExpectedOutcome.FAILURE && status.isSuccess())
{
throw new AssertionError(String.format(
"Expected failure but got success: topology=%s, CL=%s, blockFor=%d, responses=%d",
topology, cl, blockFor, responses.size()));
}
}
/**
* Minimum replicas needed for a consistency level.
*/
private static int minReplicasForCL(ConsistencyLevel cl)
{
switch (cl)
{
case ANY:
case ONE:
case LOCAL_ONE:
return 1;
case TWO:
return 2;
case THREE:
return 3;
case QUORUM:
case LOCAL_QUORUM:
case EACH_QUORUM:
return 3;
case ALL:
return 1;
default:
throw new IllegalArgumentException("Unsupported CL: " + cl);
}
}
}

352
test/unit/org/apache/cassandra/service/paxos/PaxosPreparePropertyTest.java Normal file
View File

@ -0,0 +1,352 @@
/*
* 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.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Predicate;
import com.google.common.collect.ImmutableList;
import org.junit.Test;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.exceptions.RequestFailure;
import org.apache.cassandra.exceptions.RequestFailureReason;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.Replica;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.ResponseHandlerPropertyTestBase;
import org.apache.cassandra.service.paxos.Commit.Committed;
import org.apache.cassandra.service.paxos.PaxosPrepare.Permitted;
import org.apache.cassandra.service.paxos.PaxosPrepare.Rejected;
import org.apache.cassandra.service.paxos.PaxosPrepare.Response;
import org.apache.cassandra.service.paxos.PaxosPrepare.Status;
import org.apache.cassandra.service.paxos.PaxosState.MaybePromise;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.quicktheories.core.Gen;
import static java.util.Collections.emptyMap;
import static org.quicktheories.QuickTheory.qt;
/**
* Property-based tests for PaxosPrepare response tracking logic.
* Tests the quorum counting aspects: when does a prepare achieve a quorum of
* permissions, and when is failure due to too many failures.
*
* Simplified model: all responses agree on the same "latest commit" (no divergence).
* Tests SERIAL and LOCAL_SERIAL only.
*/
public class PaxosPreparePropertyTest extends ResponseHandlerPropertyTestBase
{
private static final ImmutableList<ConsistencyLevel> CONSISTENCY_LEVELS = ImmutableList.of(
ConsistencyLevel.SERIAL,
ConsistencyLevel.LOCAL_SERIAL
);
@Override
protected List<ConsistencyLevel> consistencyLevels()
{
return CONSISTENCY_LEVELS;
}
// ========================================
// Data Structures
// ========================================
enum PrepareResponseType { PERMIT_WITH_LATEST, REJECT, FAIL }
static class PrepareResponseMessage
{
final int replicaIdx;
final PrepareResponseType type;
PrepareResponseMessage(int replicaIdx, PrepareResponseType type)
{
this.replicaIdx = replicaIdx;
this.type = type;
}
@Override
public String toString()
{
return String.format("replica=%d, %s", replicaIdx, type);
}
}
static class PrepareScenario
{
final ConsistencyLevel cl;
final Paxos.Participants participants;
final List<PrepareResponseMessage> responses;
PrepareScenario(ConsistencyLevel cl, Paxos.Participants participants,
List<PrepareResponseMessage> responses)
{
this.cl = cl;
this.participants = participants;
this.responses = responses;
}
}
static class PrepareTestCase
{
final TopologyConfig topology;
final List<PrepareScenario> scenarios;
PrepareTestCase(TopologyConfig topology, List<PrepareScenario> scenarios)
{
this.topology = topology;
this.scenarios = scenarios;
}
@Override
public String toString()
{
return String.format("topology=%s, scenarios=%d", topology, scenarios.size());
}
}
// ========================================
// Independent Model
// ========================================
/**
* Simplified independent model for PaxosPrepare completion.
*
* All permitted responses are treated as having the latest commit.
*
* SUPERSEDED: any rejection received (immediate termination)
* SUCCESS: withLatest >= sizeOfConsensusQuorum (and no rejection)
* FAILURE: failures + sizeOfConsensusQuorum > sizeOfPoll
* INCOMPLETE: otherwise
*/
static ExpectedOutcome calculatePrepareOutcome(int sizeOfPoll, int sizeOfConsensusQuorum,
List<PrepareResponseMessage> appliedResponses)
{
int withLatest = 0;
int failures = 0;
for (PrepareResponseMessage r : appliedResponses)
{
switch (r.type)
{
case PERMIT_WITH_LATEST: withLatest++; break;
case REJECT: return ExpectedOutcome.FAILURE;
case FAIL: failures++; break;
}
}
if (withLatest >= sizeOfConsensusQuorum)
return ExpectedOutcome.SUCCESS;
if (failures + sizeOfConsensusQuorum > sizeOfPoll)
return ExpectedOutcome.FAILURE;
return null;
}
// ========================================
// PaxosPrepare construction helpers
// ========================================
private static Paxos.Participants buildParticipants(ConsistencyLevel cl, Keyspace ks) throws Exception
{
var token = Murmur3Partitioner.instance.getToken(ByteBufferUtil.bytes(0));
TableMetadata table = ks.getColumnFamilyStores().iterator().next().metadata();
Predicate<Replica> allAlive = r -> true;
return Paxos.Participants.get(ClusterMetadata.current(), table, token, cl, allAlive);
}
private static Committed buildCommittedNone(Keyspace ks)
{
TableMetadata table = ks.getColumnFamilyStores().iterator().next().metadata();
DecoratedKey key = Murmur3Partitioner.instance.decorateKey(ByteBufferUtil.bytes(0));
return Committed.none(key, table);
}
private static Response buildPermitWithLatest(Committed latestCommitted)
{
return new Permitted(
MaybePromise.Outcome.PROMISE, 0L, null, latestCommitted,
null, true, emptyMap(), Epoch.EMPTY, null
);
}
private static Response buildRejected()
{
return new Rejected(Ballot.none());
}
// ========================================
// Test Case Generator
// ========================================
/**
* Generates test cases. Builds Participants from the actual replication strategy
* to ensure quorum parameters match reality, then generates responses sized
* to the actual electorate.
*/
Gen<PrepareTestCase> prepareTestCaseGen()
{
return random -> {
TopologyConfig topology = topologyGen().generate(random);
Keyspace ks;
try { ks = getOrCreateKeyspace(topology); }
catch (Exception e) { throw new RuntimeException(e); }
List<PrepareScenario> scenarios = new ArrayList<>();
for (ConsistencyLevel cl : CONSISTENCY_LEVELS)
{
Paxos.Participants participants;
try { participants = buildParticipants(cl, ks); }
catch (Exception e) { continue; }
int pollSize = participants.sizeOfPoll();
if (pollSize <= 0 || participants.sizeOfConsensusQuorum > pollSize)
continue;
Gen<List<PrepareResponseMessage>> responseGen =
typedResponseSequenceGen(pollSize, PrepareResponseType.values(), PrepareResponseMessage::new);
for (int i = 0; i < 25; i++)
scenarios.add(new PrepareScenario(cl, participants, responseGen.generate(random)));
for (PrepareResponseType type : PrepareResponseType.values())
scenarios.add(new PrepareScenario(cl, participants,
allSameTypeSequence(pollSize, type, PrepareResponseMessage::new)));
}
return new PrepareTestCase(topology, scenarios);
};
}
// ========================================
// Property Test
// ========================================
@Test
public void paxosPrepareQuorumCounting()
{
qt()
.withExamples(250)
.forAll(prepareTestCaseGen())
.assuming(testCase -> !testCase.scenarios.isEmpty())
.checkAssert(testCase -> {
try
{
Keyspace ks = getOrCreateKeyspace(testCase.topology);
Committed committedNone = buildCommittedNone(ks);
TableMetadata table = ks.getColumnFamilyStores().iterator().next().metadata();
DecoratedKey key = Murmur3Partitioner.instance.decorateKey(ByteBufferUtil.bytes(0));
for (PrepareScenario scenario : testCase.scenarios)
{
Paxos.Participants participants = scenario.participants;
int consensusQuorum = participants.sizeOfConsensusQuorum;
int pollSize = participants.sizeOfPoll();
AtomicReference<Status> statusCapture = new AtomicReference<>();
PaxosPrepare prepare = new PaxosPrepare(participants,
new PaxosPrepare.Request(Ballot.none(), participants.electorate, key, table, true, true),
false, statusCapture::set);
List<PrepareResponseMessage> applied = new ArrayList<>();
boolean completed = false;
for (PrepareResponseMessage msg : scenario.responses)
{
if (statusCapture.get() != null)
{
completed = true;
break;
}
InetAddressAndPort from = participants.voter(msg.replicaIdx);
switch (msg.type)
{
case PERMIT_WITH_LATEST:
prepare.onResponse(buildPermitWithLatest(committedNone), from);
break;
case REJECT:
prepare.onResponse(buildRejected(), from);
break;
case FAIL:
prepare.onFailure(from, RequestFailure.forReason(RequestFailureReason.TIMEOUT));
break;
}
applied.add(msg);
if (statusCapture.get() != null)
completed = true;
ExpectedOutcome expected = calculatePrepareOutcome(pollSize, consensusQuorum, applied);
if (expected != null && !completed)
{
throw new AssertionError(String.format(
"Model says %s but PaxosPrepare not complete: topology=%s, CL=%s, " +
"applied=%d, quorum=%d/%d, responses=%s",
expected, testCase.topology, scenario.cl,
applied.size(), consensusQuorum, pollSize, applied));
}
if (expected == null && completed)
{
throw new AssertionError(String.format(
"PaxosPrepare completed but model says incomplete: topology=%s, CL=%s, " +
"applied=%d, quorum=%d/%d, status=%s, responses=%s",
testCase.topology, scenario.cl,
applied.size(), consensusQuorum, pollSize,
statusCapture.get(), applied));
}
}
if (!completed)
{
ExpectedOutcome expected = calculatePrepareOutcome(pollSize, consensusQuorum, applied);
if (expected != null)
{
throw new AssertionError(String.format(
"After all %d responses, model says %s but PaxosPrepare not complete: " +
"topology=%s, CL=%s, quorum=%d/%d",
applied.size(), expected, testCase.topology, scenario.cl,
consensusQuorum, pollSize));
}
}
}
}
catch (Throwable e)
{
if (e instanceof AssertionError)
throw (AssertionError) e;
throw new AssertionError("Test setup failed: " + e.getMessage(), e);
}
});
}
}

341
test/unit/org/apache/cassandra/service/paxos/PaxosProposePropertyTest.java Normal file
View File

@ -0,0 +1,341 @@
/*
* 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.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.Predicate;
import com.google.common.collect.ImmutableList;
import org.junit.Test;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.exceptions.RequestFailure;
import org.apache.cassandra.exceptions.RequestFailureReason;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.Replica;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.ResponseHandlerPropertyTestBase;
import org.apache.cassandra.service.paxos.Commit.Proposal;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.quicktheories.core.Gen;
import static org.quicktheories.QuickTheory.qt;
/**
* Property-based tests for PaxosPropose response tracking logic.
* Tests through actual PaxosPropose instances, calling onResponse/onFailure
* and checking that the onDone callback fires at the correct point.
*
* Tests SERIAL and LOCAL_SERIAL only. Uses an independent model for the quorum math.
*/
public class PaxosProposePropertyTest extends ResponseHandlerPropertyTestBase
{
private static final ImmutableList<ConsistencyLevel> CONSISTENCY_LEVELS = ImmutableList.of(
ConsistencyLevel.SERIAL,
ConsistencyLevel.LOCAL_SERIAL
);
@Override
protected List<ConsistencyLevel> consistencyLevels()
{
return CONSISTENCY_LEVELS;
}
// ========================================
// Data Structures
// ========================================
enum ProposeResponseType { ACCEPT, REFUSE, FAIL }
static class ProposeResponseMessage
{
final int replicaIdx;
final ProposeResponseType type;
ProposeResponseMessage(int replicaIdx, ProposeResponseType type)
{
this.replicaIdx = replicaIdx;
this.type = type;
}
@Override
public String toString()
{
return String.format("replica=%d, %s", replicaIdx, type);
}
}
static class ProposeScenario
{
final ConsistencyLevel cl;
final Paxos.Participants participants;
final List<ProposeResponseMessage> responses;
ProposeScenario(ConsistencyLevel cl, Paxos.Participants participants,
List<ProposeResponseMessage> responses)
{
this.cl = cl;
this.participants = participants;
this.responses = responses;
}
}
static class ProposeTestCase
{
final TopologyConfig topology;
final List<ProposeScenario> scenarios;
ProposeTestCase(TopologyConfig topology, List<ProposeScenario> scenarios)
{
this.topology = topology;
this.scenarios = scenarios;
}
@Override
public String toString()
{
return String.format("topology=%s, scenarios=%d", topology, scenarios.size());
}
}
// ========================================
// Independent Model
// ========================================
/**
* Independent model for PaxosPropose completion.
*
* Rules:
* - SUCCESS: accepts >= required
* - Can still succeed: refusals == 0 AND required <= participants - failures
* - FAILURE: cannot succeed (any refusal, or too many failures)
*/
static ExpectedOutcome calculateProposeOutcome(int participants, int required,
List<ProposeResponseMessage> appliedResponses)
{
int accepts = 0, refusals = 0, failures = 0;
for (ProposeResponseMessage r : appliedResponses)
{
switch (r.type)
{
case ACCEPT: accepts++; break;
case REFUSE: refusals++; break;
case FAIL: failures++; break;
}
}
if (accepts >= required)
return ExpectedOutcome.SUCCESS;
boolean canSucceed = refusals == 0 && required <= participants - failures;
if (canSucceed)
return null; // still in progress
return ExpectedOutcome.FAILURE;
}
// ========================================
// PaxosPropose construction helpers
// ========================================
private static Paxos.Participants buildParticipants(ConsistencyLevel cl, Keyspace ks) throws Exception
{
var token = Murmur3Partitioner.instance.getToken(ByteBufferUtil.bytes(0));
TableMetadata table = ks.getColumnFamilyStores().iterator().next().metadata();
Predicate<Replica> allAlive = r -> true;
return Paxos.Participants.get(ClusterMetadata.current(), table, token, cl, allAlive);
}
private static Proposal buildEmptyProposal(Keyspace ks)
{
TableMetadata table = ks.getColumnFamilyStores().iterator().next().metadata();
DecoratedKey key = Murmur3Partitioner.instance.decorateKey(ByteBufferUtil.bytes(0));
return Proposal.empty(Ballot.none(), key, table);
}
// ========================================
// Test Case Generator
// ========================================
/**
* Generates test cases. Builds Participants from the actual replication strategy
* to ensure quorum parameters match reality, then generates responses sized
* to the actual electorate.
*/
Gen<ProposeTestCase> proposeTestCaseGen()
{
return random -> {
TopologyConfig topology = topologyGen().generate(random);
Keyspace ks;
try
{
ks = getOrCreateKeyspace(topology);
}
catch (Exception e)
{
throw new RuntimeException(e);
}
List<ProposeScenario> scenarios = new ArrayList<>();
for (ConsistencyLevel cl : CONSISTENCY_LEVELS)
{
Paxos.Participants participants;
try
{
participants = buildParticipants(cl, ks);
}
catch (Exception e)
{
continue;
}
int pollSize = participants.sizeOfPoll();
if (pollSize <= 0 || participants.sizeOfConsensusQuorum > pollSize)
continue;
Gen<List<ProposeResponseMessage>> responseGen =
typedResponseSequenceGen(pollSize, ProposeResponseType.values(), ProposeResponseMessage::new);
for (int i = 0; i < 25; i++)
scenarios.add(new ProposeScenario(cl, participants, responseGen.generate(random)));
for (ProposeResponseType type : ProposeResponseType.values())
scenarios.add(new ProposeScenario(cl, participants,
allSameTypeSequence(pollSize, type, ProposeResponseMessage::new)));
}
return new ProposeTestCase(topology, scenarios);
};
}
// ========================================
// Property Test
// ========================================
@Test
public void paxosProposeSignaling()
{
qt()
.withExamples(250)
.forAll(proposeTestCaseGen())
.assuming(testCase -> !testCase.scenarios.isEmpty())
.checkAssert(testCase -> {
try
{
Keyspace ks = getOrCreateKeyspace(testCase.topology);
Proposal proposal = buildEmptyProposal(ks);
for (ProposeScenario scenario : testCase.scenarios)
verifyScenario(testCase.topology, scenario, proposal);
}
catch (Throwable e)
{
if (e instanceof AssertionError)
throw (AssertionError) e;
throw new AssertionError("Test setup failed: " + e.getMessage(), e);
}
});
}
private void verifyScenario(TopologyConfig topology, ProposeScenario scenario, Proposal proposal)
{
Paxos.Participants participants = scenario.participants;
int consensusQuorum = participants.sizeOfConsensusQuorum;
int pollSize = participants.sizeOfPoll();
AtomicReference<PaxosPropose.Status> statusCapture = new AtomicReference<>();
PaxosPropose<Consumer<PaxosPropose.Status>> propose = new PaxosPropose<>(
proposal, pollSize, consensusQuorum, statusCapture::set);
List<ProposeResponseMessage> applied = new ArrayList<>();
boolean completed = false;
for (ProposeResponseMessage msg : scenario.responses)
{
if (statusCapture.get() != null)
{
completed = true;
break;
}
InetAddressAndPort from = participants.voter(msg.replicaIdx);
switch (msg.type)
{
case ACCEPT:
propose.onResponse(PaxosState.AcceptResult.SUCCESS, from);
break;
case REFUSE:
propose.onResponse(new PaxosState.AcceptResult(Ballot.none()), from);
break;
case FAIL:
propose.onFailure(from, RequestFailure.forReason(RequestFailureReason.TIMEOUT));
break;
}
applied.add(msg);
if (statusCapture.get() != null)
completed = true;
ExpectedOutcome expected = calculateProposeOutcome(pollSize, consensusQuorum, applied);
if (expected != null && !completed)
{
throw new AssertionError(String.format(
"Model says %s but PaxosPropose not complete: topology=%s, CL=%s, " +
"applied=%d, quorum=%d/%d, responses=%s",
expected, topology, scenario.cl,
applied.size(), consensusQuorum, pollSize, applied));
}
if (expected == null && completed)
{
throw new AssertionError(String.format(
"PaxosPropose completed but model says incomplete: topology=%s, CL=%s, " +
"applied=%d, quorum=%d/%d, status=%s, responses=%s",
topology, scenario.cl,
applied.size(), consensusQuorum, pollSize,
statusCapture.get(), applied));
}
}
if (!completed)
{
ExpectedOutcome expected = calculateProposeOutcome(pollSize, consensusQuorum, applied);
if (expected != null)
{
throw new AssertionError(String.format(
"After all %d responses, model says %s but PaxosPropose not complete: " +
"topology=%s, CL=%s, quorum=%d/%d",
applied.size(), expected, topology, scenario.cl,
consensusQuorum, pollSize));
}
}
}
}

537
test/unit/org/apache/cassandra/service/reads/tracked/ReadReconciliationsCoordinatorPropertyTest.java Normal file
View File

@ -0,0 +1,537 @@
/*
* 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.reads.tracked;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.quicktheories.core.Gen;
import static org.quicktheories.QuickTheory.qt;
import static org.quicktheories.generators.SourceDSL.booleans;
import static org.quicktheories.generators.SourceDSL.integers;
import static org.quicktheories.generators.SourceDSL.lists;
/**
* Property-based tests for ReadReconciliations.Coordinator completion logic.
* Tests that the three-counter system (mutations, summaries, syncAcks) correctly
* determines when reconciliation is complete.
*/
public class ReadReconciliationsCoordinatorPropertyTest
{
// LOCAL_NODE is initialized from ClusterMetadata to match production code
private static int LOCAL_NODE;
private static int REMOTE_NODE;
@BeforeClass
public static void setUpClass() throws Throwable
{
SchemaLoader.loadSchema();
// Get the actual local node ID that ReadReconciliations will use
LOCAL_NODE = ClusterMetadata.current().myNodeId().id();
REMOTE_NODE = LOCAL_NODE + 1; // Use a distinct remote node ID
}
/**
* Testable subclass that overrides complete() to skip messaging side effects,
* and exposes test methods to directly manipulate counters.
*/
static class TestableCoordinator extends ReadReconciliations.Coordinator
{
TestableCoordinator(int dataNode, int[] summaryNodes)
{
super(new TrackedRead.Id(LOCAL_NODE, 0), dataNode, summaryNodes);
}
@Override
protected boolean complete()
{
// Skip messaging - just return true
return true;
}
// Test methods that directly update counters without messaging side effects
boolean testAcceptLocalSummary()
{
return updateRemainingAndMaybeComplete(0, -1, 0);
}
boolean testAcceptRemoteSummary(int missingCount)
{
return updateRemainingAndMaybeComplete(missingCount, -1, 0);
}
}
/**
* Configuration for a coordinator test case.
*/
static class CoordinatorConfig
{
final int summaryNodeCount; // 0-5 summary nodes
final boolean isDataNode; // Whether local node is the data node
CoordinatorConfig(int summaryNodeCount, boolean isDataNode)
{
this.summaryNodeCount = summaryNodeCount;
this.isDataNode = isDataNode;
}
@Override
public String toString()
{
return String.format("summaryNodes=%d, isDataNode=%s", summaryNodeCount, isDataNode);
}
}
/**
* Response types for coordinator testing.
*/
enum ResponseType
{
LOCAL_SUMMARY,
REMOTE_SUMMARY,
SYNC_ACK,
MUTATION
}
/**
* A single response event.
*/
static class CoordinatorResponse
{
final ResponseType type;
final int missingCount; // Only used for REMOTE_SUMMARY
final int nodeId; // Used for SYNC_ACK
CoordinatorResponse(ResponseType type)
{
this(type, 0, 0);
}
CoordinatorResponse(ResponseType type, int missingCount)
{
this(type, missingCount, 0);
}
CoordinatorResponse(ResponseType type, int missingCount, int nodeId)
{
this.type = type;
this.missingCount = missingCount;
this.nodeId = nodeId;
}
@Override
public String toString()
{
if (type == ResponseType.REMOTE_SUMMARY)
return String.format("%s(missing=%d)", type, missingCount);
if (type == ResponseType.SYNC_ACK)
return String.format("%s(node=%d)", type, nodeId);
return type.toString();
}
}
/**
* A complete test case with configuration and response sequence.
*/
static class TestCase
{
final CoordinatorConfig config;
final List<CoordinatorResponse> responses;
TestCase(CoordinatorConfig config, List<CoordinatorResponse> responses)
{
this.config = config;
this.responses = responses;
}
@Override
public String toString()
{
return String.format("config=%s, responses=%d", config, responses.size());
}
}
/**
* Expected state tracker for validating coordinator behavior.
*/
static class ExpectedState
{
int remainingMutations;
int remainingSummaries;
int remainingSyncAcks;
ExpectedState(int summaryNodeCount, boolean isDataNode)
{
this.remainingMutations = 0;
this.remainingSummaries = 1 + summaryNodeCount;
this.remainingSyncAcks = isDataNode ? summaryNodeCount : 0;
}
boolean apply(CoordinatorResponse response)
{
switch (response.type)
{
case LOCAL_SUMMARY:
remainingSummaries--;
break;
case REMOTE_SUMMARY:
remainingMutations += response.missingCount;
remainingSummaries--;
break;
case SYNC_ACK:
remainingSyncAcks--;
break;
case MUTATION:
remainingMutations--;
break;
}
return isComplete();
}
boolean isComplete()
{
return remainingMutations == 0
&& remainingSummaries == 0
&& remainingSyncAcks == 0;
}
@Override
public String toString()
{
return String.format("mutations=%d, summaries=%d, syncAcks=%d",
remainingMutations, remainingSummaries, remainingSyncAcks);
}
}
// ========================================
// Generators
// ========================================
/**
* Generate coordinator configurations.
*/
Gen<CoordinatorConfig> configGen()
{
return integers().between(0, 5).flatMap(summaryNodes ->
booleans().all().map(isDataNode ->
new CoordinatorConfig(summaryNodes, isDataNode)
)
);
}
/**
* Generate test cases with valid response sequences.
* Responses are generated such that mutations always come after their parent remote summary.
*/
Gen<TestCase> testCaseGen()
{
return configGen().flatMap(config -> responseSequenceGen(config).map(responses ->
new TestCase(config, responses)
));
}
/**
* Generate a valid response sequence for a given configuration.
*/
Gen<List<CoordinatorResponse>> responseSequenceGen(CoordinatorConfig config)
{
// Generate missingCount for each remote summary (0-10)
return lists().of(integers().between(0, 10)).ofSize(config.summaryNodeCount).flatMap(missingCounts -> {
// Calculate total mutations needed
int totalMutations = missingCounts.stream().mapToInt(Integer::intValue).sum();
// Generate a random permutation using indices
int totalEvents = 1 + config.summaryNodeCount + totalMutations + (config.isDataNode ? config.summaryNodeCount : 0);
return permutationGen(totalEvents).map(permutation -> {
// Build the response sequence respecting ordering constraints
return buildResponseSequence(config, missingCounts, permutation);
});
});
}
/**
* Build response sequence respecting the constraint that mutations must come after their parent remote summary.
*/
private List<CoordinatorResponse> buildResponseSequence(CoordinatorConfig config,
List<Integer> missingCounts,
List<Integer> permutation)
{
// Create all events
List<CoordinatorResponse> events = new ArrayList<>();
// Add local summary
events.add(new CoordinatorResponse(ResponseType.LOCAL_SUMMARY));
// Add remote summaries with their mutations linked
List<List<CoordinatorResponse>> remoteSummaryGroups = new ArrayList<>();
for (int i = 0; i < config.summaryNodeCount; i++)
{
List<CoordinatorResponse> group = new ArrayList<>();
group.add(new CoordinatorResponse(ResponseType.REMOTE_SUMMARY, missingCounts.get(i)));
for (int j = 0; j < missingCounts.get(i); j++)
group.add(new CoordinatorResponse(ResponseType.MUTATION));
remoteSummaryGroups.add(group);
}
// Add sync acks (if data node)
List<CoordinatorResponse> syncAcks = new ArrayList<>();
if (config.isDataNode)
{
for (int i = 0; i < config.summaryNodeCount; i++)
syncAcks.add(new CoordinatorResponse(ResponseType.SYNC_ACK, 0, REMOTE_NODE + i + 1));
}
// Shuffle the independent groups using the permutation
// We shuffle: local summary, each remote summary group (as a unit), and sync acks
List<Object> shuffleable = new ArrayList<>();
shuffleable.add(events.get(0)); // local summary
shuffleable.addAll(remoteSummaryGroups);
shuffleable.addAll(syncAcks);
// Use permutation to reorder (just use first N elements of permutation as indices)
List<Object> shuffled = new ArrayList<>(shuffleable);
Collections.shuffle(shuffled, new java.util.Random(permutation.hashCode()));
// Flatten back to response list
List<CoordinatorResponse> result = new ArrayList<>();
for (Object item : shuffled)
{
if (item instanceof CoordinatorResponse)
result.add((CoordinatorResponse) item);
else if (item instanceof List)
{
@SuppressWarnings("unchecked")
List<CoordinatorResponse> group = (List<CoordinatorResponse>) item;
result.addAll(group);
}
}
return result;
}
/**
* Generate a random permutation of integers [0, n-1].
*/
private static Gen<List<Integer>> permutationGen(int n)
{
if (n == 0)
return lists().of(integers().all()).ofSize(0);
if (n == 1)
{
return lists().of(integers().all()).ofSize(1).map(list -> {
List<Integer> result = new ArrayList<>();
result.add(0);
return result;
});
}
// Generate n random integers for shuffling
return lists().of(integers().between(0, Integer.MAX_VALUE)).ofSize(n).map(randoms -> {
List<Integer> result = new ArrayList<>();
for (int i = 0; i < n; i++)
result.add(i);
// Fisher-Yates shuffle using generated random values
for (int i = n - 1; i > 0; i--)
{
int j = Math.abs(randoms.get(i)) % (i + 1);
Collections.swap(result, i, j);
}
return result;
});
}
// ========================================
// Test Methods
// ========================================
/**
* Apply a response to the coordinator and return whether it completed.
*/
private boolean applyResponse(TestableCoordinator coordinator, CoordinatorResponse response)
{
switch (response.type)
{
case LOCAL_SUMMARY:
return coordinator.testAcceptLocalSummary();
case REMOTE_SUMMARY:
return coordinator.testAcceptRemoteSummary(response.missingCount);
case SYNC_ACK:
return coordinator.acceptSyncAck(response.nodeId);
case MUTATION:
return coordinator.acceptMutation(null); // mutationId is ignored
default:
throw new IllegalArgumentException("Unknown response type: " + response.type);
}
}
/**
* Create a coordinator for testing.
*/
private TestableCoordinator createCoordinator(CoordinatorConfig config)
{
int dataNode = config.isDataNode ? LOCAL_NODE : REMOTE_NODE;
int[] summaryNodes = new int[config.summaryNodeCount];
for (int i = 0; i < config.summaryNodeCount; i++)
summaryNodes[i] = REMOTE_NODE + i + 1; // Use distinct node IDs
return new TestableCoordinator(dataNode, summaryNodes);
}
@Test
public void coordinatorCompletionBehavior()
{
qt()
.withExamples(500)
.forAll(testCaseGen())
.checkAssert(testCase -> {
TestableCoordinator coordinator = createCoordinator(testCase.config);
ExpectedState expected = new ExpectedState(
testCase.config.summaryNodeCount,
testCase.config.isDataNode
);
for (CoordinatorResponse response : testCase.responses)
{
boolean actualComplete = applyResponse(coordinator, response);
boolean expectedComplete = expected.apply(response);
Assert.assertEquals(
String.format("Completion mismatch after %s: expected=%s, actual=%s, state=%s, config=%s",
response, expectedComplete, actualComplete, expected, testCase.config),
expectedComplete, actualComplete
);
if (actualComplete)
break;
}
// Final state should be complete
Assert.assertTrue(
String.format("Should be complete after all responses: state=%s, config=%s",
expected, testCase.config),
expected.isComplete()
);
});
}
/**
* Test edge case: minimal coordinator with 0 summary nodes as data node.
* Should complete immediately after local summary.
*/
@Test
public void minimalDataNodeCompletes()
{
TestableCoordinator coordinator = new TestableCoordinator(LOCAL_NODE, new int[0]);
// Should complete after just the local summary
Assert.assertTrue("Should complete after local summary with 0 summary nodes",
coordinator.testAcceptLocalSummary());
}
/**
* Test edge case: minimal coordinator with 0 summary nodes as summary node.
* Should complete immediately after local summary.
*/
@Test
public void minimalSummaryNodeCompletes()
{
TestableCoordinator coordinator = new TestableCoordinator(REMOTE_NODE, new int[0]);
// Should complete after just the local summary
Assert.assertTrue("Should complete after local summary with 0 summary nodes",
coordinator.testAcceptLocalSummary());
}
/**
* Test that data node requires sync acks from all summary nodes.
*/
@Test
public void dataNodeRequiresSyncAcks()
{
TestableCoordinator coordinator = new TestableCoordinator(
LOCAL_NODE, new int[]{REMOTE_NODE, REMOTE_NODE + 1}
);
// Local summary
Assert.assertFalse(coordinator.testAcceptLocalSummary());
// Remote summaries with no missing mutations
Assert.assertFalse(coordinator.testAcceptRemoteSummary(0));
Assert.assertFalse(coordinator.testAcceptRemoteSummary(0));
// First sync ack
Assert.assertFalse(coordinator.acceptSyncAck(REMOTE_NODE));
// Second sync ack should complete
Assert.assertTrue(coordinator.acceptSyncAck(REMOTE_NODE + 1));
}
/**
* Test that summary node does NOT require sync acks.
*/
@Test
public void summaryNodeDoesNotRequireSyncAcks()
{
TestableCoordinator coordinator = new TestableCoordinator(
REMOTE_NODE, new int[]{REMOTE_NODE + 1, REMOTE_NODE + 2}
);
// Local summary
Assert.assertFalse(coordinator.testAcceptLocalSummary());
// Remote summaries with no missing mutations
Assert.assertFalse(coordinator.testAcceptRemoteSummary(0));
// Last remote summary should complete (no sync acks needed)
Assert.assertTrue(coordinator.testAcceptRemoteSummary(0));
}
/**
* Test that mutations must be received before completion.
*/
@Test
public void mutationsMustBeReceived()
{
TestableCoordinator coordinator = new TestableCoordinator(
REMOTE_NODE, new int[]{REMOTE_NODE + 1}
);
// Local summary
Assert.assertFalse(coordinator.testAcceptLocalSummary());
// Remote summary with 2 missing mutations
Assert.assertFalse(coordinator.testAcceptRemoteSummary(2));
// First mutation
Assert.assertFalse(coordinator.acceptMutation(null));
// Second mutation should complete
Assert.assertTrue(coordinator.acceptMutation(null));
}
}