bdeggleston commented on code in PR #4708: URL: https://github.com/apache/cassandra/pull/4708#discussion_r3560716534
########## test/unit/org/apache/cassandra/service/ResponseHandlerPropertyTestBase.java: ########## @@ -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; Review Comment: Right, they're handled identically here because we just need them registered for internal topology checks (rack/DC membership). The tests themselves create response handlers with synthetic replica plans (see WriteResponseHandlerPropertyTest#createHandler, for instance), so we don't actually need to register them as pending, etc -- 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]
