bdeggleston commented on code in PR #4708: URL: https://github.com/apache/cassandra/pull/4708#discussion_r3560728902
########## test/unit/org/apache/cassandra/service/WriteResponseHandlerPropertyTest.java: ########## @@ -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; Review Comment: Yes, this just for determining the minimum cluster size we'd need to test a given CL. For instance testing CL_THREE or CL_QUORUM with a 2 node cluster isn't very useful. We can test CL_ALL with a single node cluster though -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
