frankgh commented on code in PR #4708: URL: https://github.com/apache/cassandra/pull/4708#discussion_r3500914682
########## src/java/org/apache/cassandra/locator/CoordinationPlan.java: ########## @@ -0,0 +1,321 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.locator; + +import java.util.Set; +import java.util.function.Function; +import java.util.function.Supplier; + +import javax.annotation.Nullable; + +import org.apache.cassandra.db.ConsistencyLevel; +import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.db.PartitionPosition; +import org.apache.cassandra.dht.AbstractBounds; +import org.apache.cassandra.dht.Token; +import org.apache.cassandra.exceptions.UnavailableException; +import org.apache.cassandra.index.Index; +import org.apache.cassandra.schema.SchemaConstants; +import org.apache.cassandra.schema.TableId; +import org.apache.cassandra.service.reads.ReadCoordinator; +import org.apache.cassandra.service.reads.SpeculativeRetryPolicy; +import org.apache.cassandra.tcm.ClusterMetadata; + +/** + * Ties together replica selection and response tracking for a single operation. + * + * This immutable container ensures that the replica plan (who to contact) and + * the response tracker (how to determine success) are created atomically by the + * replication strategy, consulting the same state. This is particularly important + * for strategies that have per-range state (e.g., failover state in SRS) where + * the replica selection and quorum requirements must be consistent. + * + * The separation between ReplicaPlan and ResponseTracker allows: + * - ReplicaPlan to focus on replica topology and selection + * - ResponseTracker to encapsulate completion logic + * - Replication strategies to customize both consistently Review Comment: NIT, can we have proper javadoc syntax here? ``` * <p>The separation between ReplicaPlan and ResponseTracker allows: * <ul> * <li>ReplicaPlan to focus on replica topology and selection * <li>ResponseTracker to encapsulate completion logic * <li>Replication strategies to customize both consistently * </ul> ``` ########## src/java/org/apache/cassandra/locator/CoordinationPlan.java: ########## @@ -0,0 +1,321 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.locator; + +import java.util.Set; +import java.util.function.Function; +import java.util.function.Supplier; + +import javax.annotation.Nullable; + +import org.apache.cassandra.db.ConsistencyLevel; +import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.db.PartitionPosition; +import org.apache.cassandra.dht.AbstractBounds; +import org.apache.cassandra.dht.Token; +import org.apache.cassandra.exceptions.UnavailableException; +import org.apache.cassandra.index.Index; +import org.apache.cassandra.schema.SchemaConstants; +import org.apache.cassandra.schema.TableId; +import org.apache.cassandra.service.reads.ReadCoordinator; +import org.apache.cassandra.service.reads.SpeculativeRetryPolicy; +import org.apache.cassandra.tcm.ClusterMetadata; + +/** + * Ties together replica selection and response tracking for a single operation. + * + * This immutable container ensures that the replica plan (who to contact) and + * the response tracker (how to determine success) are created atomically by the + * replication strategy, consulting the same state. This is particularly important + * for strategies that have per-range state (e.g., failover state in SRS) where + * the replica selection and quorum requirements must be consistent. + * + * The separation between ReplicaPlan and ResponseTracker allows: + * - ReplicaPlan to focus on replica topology and selection + * - ResponseTracker to encapsulate completion logic + * - Replication strategies to customize both consistently + * + * No polymorphism is needed at this level - the variation is captured in the + * ResponseTracker implementations. This is just a typed tuple ensuring the + * two pieces travel together. + * + * @param <P> the type of ReplicaPlan (ForRead, ForWrite, ForPaxosWrite, etc.) + */ +public abstract class CoordinationPlan<E extends Endpoints<E>, P extends ReplicaPlan<E, P>> +{ + // TODO (now): consolidate callback Condition instances into this. Replace all condition await calls with calls to this. + private final ResponseTracker responses; + + /** + * Create a coordination plan. + * + * @param responses the response tracker for determining completion/success + */ + public CoordinationPlan(ResponseTracker responses) + { + if (responses == null) + throw new IllegalArgumentException("tracker cannot be null"); + + this.responses = responses; Review Comment: NIT, should this be a NPE instead? ``` this.responses = Objects.requireNonNull(responses, "tracker cannot be null"); ``` ########## src/java/org/apache/cassandra/locator/CoordinationPlan.java: ########## @@ -0,0 +1,321 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.locator; + +import java.util.Set; +import java.util.function.Function; +import java.util.function.Supplier; + +import javax.annotation.Nullable; + +import org.apache.cassandra.db.ConsistencyLevel; +import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.db.PartitionPosition; +import org.apache.cassandra.dht.AbstractBounds; +import org.apache.cassandra.dht.Token; +import org.apache.cassandra.exceptions.UnavailableException; +import org.apache.cassandra.index.Index; +import org.apache.cassandra.schema.SchemaConstants; +import org.apache.cassandra.schema.TableId; +import org.apache.cassandra.service.reads.ReadCoordinator; +import org.apache.cassandra.service.reads.SpeculativeRetryPolicy; +import org.apache.cassandra.tcm.ClusterMetadata; + +/** + * Ties together replica selection and response tracking for a single operation. + * + * This immutable container ensures that the replica plan (who to contact) and + * the response tracker (how to determine success) are created atomically by the + * replication strategy, consulting the same state. This is particularly important + * for strategies that have per-range state (e.g., failover state in SRS) where + * the replica selection and quorum requirements must be consistent. + * + * The separation between ReplicaPlan and ResponseTracker allows: + * - ReplicaPlan to focus on replica topology and selection + * - ResponseTracker to encapsulate completion logic + * - Replication strategies to customize both consistently + * + * No polymorphism is needed at this level - the variation is captured in the + * ResponseTracker implementations. This is just a typed tuple ensuring the + * two pieces travel together. + * + * @param <P> the type of ReplicaPlan (ForRead, ForWrite, ForPaxosWrite, etc.) + */ +public abstract class CoordinationPlan<E extends Endpoints<E>, P extends ReplicaPlan<E, P>> +{ + // TODO (now): consolidate callback Condition instances into this. Replace all condition await calls with calls to this. + private final ResponseTracker responses; + + /** + * Create a coordination plan. + * + * @param responses the response tracker for determining completion/success + */ + public CoordinationPlan(ResponseTracker responses) + { + if (responses == null) + throw new IllegalArgumentException("tracker cannot be null"); + + this.responses = responses; + } + + public ConsistencyLevel consistencyLevel() + { + return replicas().consistencyLevel(); + } + + public AbstractReplicationStrategy replicationStrategy() + { + return replicas().replicationStrategy(); + } + + public abstract P replicas(); + + /** + * The response tracker for determining completion/success. + * + * The tracker encapsulates the logic for: + * - Recording responses and failures + * - Determining when the operation is complete + * - Checking if the operation succeeded + * + * @return the response tracker + */ + public ResponseTracker responses() + { + return responses; + } + + @Override + public String toString() + { + return String.format("CoordinationPlan[replicaPlan=%s, tracker=%s]", replicas(), responses.getClass().getSimpleName()); + } + + /** + * Extended plan including source cluster metadata and ideal coordination plan. + */ + public static class ForWrite extends CoordinationPlan<EndpointsForToken, ReplicaPlan.ForWrite> + { + private final ReplicaPlan.ForWrite replicas; + + public ForWrite(ReplicaPlan.ForWrite replicas, ResponseTracker responses) + { + super(responses); + this.replicas = replicas; + } + + @Override + public ReplicaPlan.ForWrite replicas() + { + return replicas; + } + } + + public static class ForWriteWithIdeal extends CoordinationPlan.ForWrite + { + public final ClusterMetadata metadata; Review Comment: unused and can be removed unless we have plans to use `ClusterMetadata` in the future ########## src/java/org/apache/cassandra/batchlog/BatchlogManager.java: ########## @@ -635,26 +630,13 @@ private static ReplayWriteResponseHandler<Mutation> sendSingleReplayMutation(fin } } - ReplayWriteResponseHandler<Mutation> handler = new ReplayWriteResponseHandler<>(replicaPlan, mutation, Dispatcher.RequestTime.forImmediateExecution()); + ReplayWriteResponseHandler<Mutation> handler = new ReplayWriteResponseHandler<>(replayPlan, () -> mutation, Dispatcher.RequestTime.forImmediateExecution()); Review Comment: `Mutation` already implements `Supplier<Mutation>`. Do we need the change here? or should we preserve the original? ``` ReplayWriteResponseHandler<Mutation> handler = new ReplayWriteResponseHandler<>(replayPlan, mutation, Dispatcher.RequestTime.forImmediateExecution()); ``` ########## src/java/org/apache/cassandra/locator/ReplicaPlans.java: ########## @@ -287,6 +278,35 @@ public static ReplicaPlan.ForWrite forLocalBatchlogWrite() return forWrite(systemKeyspace, ConsistencyLevel.ONE, (cm) -> liveAndDown, (cm) -> true, writeAll); } + /** + * Create a replica plan for replaying a mutation from the batchlog. + * + * When recovering failed batches, mutations are replayed to live remote replicas only + * (local replica is handled separately by the caller). + * + * @param metadata the cluster metadata + * @param keyspace the keyspace + * @param token the token for the mutation + * @return replica plan targeting live remote replicas with CL.ONE + */ + public static ReplicaPlan.ForWrite forReplayMutation(ClusterMetadata metadata, Keyspace keyspace, Token token) + { + ReplicaLayout.ForTokenWrite liveAndDown = ReplicaLayout.forTokenWriteLiveAndDown(metadata, keyspace.getMetadata(), token); + Replicas.temporaryAssertFull(liveAndDown.all()); // TODO in CASSANDRA-14549 + + Replica selfReplica = liveAndDown.all().selfIfPresent(); + ReplicaLayout.ForTokenWrite liveRemoteOnly = liveAndDown.filter(r -> FailureDetector.isReplicaAlive.test(r) && r != selfReplica); + + return new ReplicaPlan.ForWrite(keyspace, keyspace.getReplicationStrategy(), + ConsistencyLevel.ONE, + liveRemoteOnly.pending(), + liveRemoteOnly.all(), + liveRemoteOnly.all(), + liveRemoteOnly.all(), Review Comment: NIT: extract `liveRemoteOnly.all()` into a local variable: ``` EndpointsForToken allLiveRemoteOnly = liveRemoteOnly.all(); return new ReplicaPlan.ForWrite(keyspace, keyspace.getReplicationStrategy(), ConsistencyLevel.ONE, liveRemoteOnly.pending(), allLiveRemoteOnly, allLiveRemoteOnly, allLiveRemoteOnly, ``` ########## src/java/org/apache/cassandra/locator/WriteResponseTracker.java: ########## @@ -0,0 +1,237 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.locator; + +import java.util.function.Predicate; + +import org.apache.cassandra.exceptions.RequestFailureReason; + +/** + * Response tracker for writes with pending replicas using the double count model. + * <p> + * Two requirements must be satisfied for success: + * <ol> + * <li>Committed replicas must satisfy base CL: {@code committedSuccesses >= baseBlockFor}</li> + * <li>Total replicas (committed + pending) must satisfy: {@code totalSuccesses >= totalBlockFor}</li> + * </ol> + * <p> + * This ensures both consistency (committed replicas have the data) and bootstrap safety + * (pending replicas also receive the write). + * <p> + * Implemented by composing two {@link SimpleResponseTracker}s: + * <ul> + * <li>committedTracker: tracks responses from committed replicas only</li> + * <li>totalTracker: tracks responses from all replicas (committed + pending)</li> + * </ul> + * <p> + * Thread-safe through delegation to thread-safe SimpleResponseTrackers. + */ +public class WriteResponseTracker implements ResponseTracker +{ + private final SimpleResponseTracker natural; + private final SimpleResponseTracker total; + + /** + * Create a write response tracker with the double count model. + * + * @param baseBlockFor number of committed replica responses required for CL + * @param totalBlockFor total responses required (baseBlockFor + pending count) + * @param committedReplicas number of committed replicas available + * @param pendingReplicas number of pending replicas available + * @param isPending predicate to determine if an endpoint is pending + */ + public WriteResponseTracker(int baseBlockFor, + int totalBlockFor, + int committedReplicas, + int pendingReplicas, + Predicate<InetAddressAndPort> isPending) + { + this(baseBlockFor, totalBlockFor, committedReplicas, pendingReplicas, isPending, null); + } + + /** + * Create a write response tracker with the double count model and a filter. + * + * @param naturalBlockFor number of committed replica responses required for CL + * @param totalBlockFor total responses required (naturalBlockFor + pending count) + * @param naturalReplicas number of committed replicas available + * @param pendingReplicas number of pending replicas available + * @param isPending predicate to determine if an endpoint is pending + * @param filter predicate to filter which responses count (e.g., InOurDc for LOCAL_QUORUM) + */ + public WriteResponseTracker(int naturalBlockFor, + int totalBlockFor, + int naturalReplicas, + int pendingReplicas, + Predicate<InetAddressAndPort> isPending, + Predicate<InetAddressAndPort> filter) + { + if (naturalBlockFor < 0) + throw new IllegalArgumentException("naturalBlockFor must be non-negative: " + naturalBlockFor); + if (totalBlockFor < naturalBlockFor) + throw new IllegalArgumentException("totalBlockFor (" + totalBlockFor + ") must be >= naturalBlockFor (" + naturalBlockFor + ")"); + if (naturalReplicas < 0) + throw new IllegalArgumentException("naturalReplicas must be non-negative: " + naturalReplicas); + if (pendingReplicas < 0) + throw new IllegalArgumentException("pendingReplicas must be non-negative: " + pendingReplicas); + if (naturalBlockFor > naturalReplicas) + throw new IllegalArgumentException("naturalBlockFor (" + naturalBlockFor + ") cannot exceed naturalReplicas (" + naturalReplicas + ")"); + if (totalBlockFor > naturalReplicas + pendingReplicas) + throw new IllegalArgumentException("totalBlockFor (" + totalBlockFor + ") cannot exceed total replicas (" + (naturalReplicas + pendingReplicas) + ")"); + if (isPending == null) + throw new IllegalArgumentException("isPending predicate cannot be null"); + + Predicate<InetAddressAndPort> naturalFilter = isPending.negate(); + if (filter != null) + naturalFilter = naturalFilter.and(filter); + + this.natural = new SimpleResponseTracker(naturalBlockFor, naturalReplicas, naturalFilter); + this.total = new SimpleResponseTracker(totalBlockFor, naturalReplicas + pendingReplicas, filter); + } + + private WriteResponseTracker(SimpleResponseTracker natural, SimpleResponseTracker total) + { + this.natural = natural; + this.total = total; + } + + @Override + public void onResponse(InetAddressAndPort from) + { + natural.onResponse(from); + total.onResponse(from); + } + + @Override + public void onFailure(InetAddressAndPort from, RequestFailureReason reason) + { + natural.onFailure(from, reason); + total.onFailure(from, reason); + } + + @Override + public boolean isComplete() + { + // Early failure if either tracker fails + if (natural.isComplete() && !natural.isSuccessful()) + return true; + if (total.isComplete() && !total.isSuccessful()) + return true; + + // Success requires both to succeed + return natural.isSuccessful() && total.isSuccessful(); + } + + @Override + public boolean isSuccessful() + { + return natural.isSuccessful() && total.isSuccessful(); + } + + @Override + public int required() + { + // Return total requirement for error messages + return total.required(); + } + + @Override + public int received() + { + // Return total successes for error messages + return total.received(); + } + + @Override + public int failures() + { + // Return total failures for error messages + return total.failures(); + } + + @Override + public boolean countsTowardQuorum(InetAddressAndPort from) + { + return total.countsTowardQuorum(from); + } + + public int committedReceived() + { + return natural.received(); + } + + public int pendingReceived() Review Comment: This is not really atomic. Thread 1 calls onResponse, and natural updates `responses` , then thread 2 calls `pendingReceived` (or `pendingFailures`) and reads updated value from both `total` and `natural`. thread 1 has not yet updated `responses` of total. Since we are only using this for test, I think we should at least call it out, also maybe not make it public unless we need it ########## src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java: ########## @@ -437,4 +459,362 @@ public RangesAtEndpoint getLocalRanges(ClusterMetadata cm) return newRanges; } } + + protected CoordinationPlan.ForWrite planForWriteInternal(ClusterMetadata metadata, + Keyspace keyspace, + ConsistencyLevel consistencyLevel, + Function<ClusterMetadata, ReplicaLayout.ForTokenWrite> liveAndDown, + ReplicaPlans.Selector selector) + { + ReplicaPlan.ForWrite plan = ReplicaPlans.forWrite(metadata, keyspace, consistencyLevel, liveAndDown, selector); + ResponseTracker tracker = createTrackerForWrite(consistencyLevel, plan, plan.pending, metadata); + return new CoordinationPlan.ForWrite(plan, tracker); + } + + protected ReplicaPlan.ForWrite createReplicaPlanForWrite(ClusterMetadata metadata, + Keyspace keyspace, + ConsistencyLevel consistencyLevel, + Function<ClusterMetadata, ReplicaLayout.ForTokenWrite> liveAndDown, + ReplicaPlans.Selector selector) + { + return ReplicaPlans.forWrite(metadata, keyspace, consistencyLevel, liveAndDown, selector); + } + + public CoordinationPlan.ForWriteWithIdeal planForWrite(ClusterMetadata metadata, + Keyspace keyspace, + ConsistencyLevel consistencyLevel, + Function<ClusterMetadata, ReplicaLayout.ForTokenWrite> liveAndDown, + ReplicaPlans.Selector selector) + { + ReplicaPlan.ForWrite plan = createReplicaPlanForWrite(metadata, keyspace, consistencyLevel, liveAndDown, selector); + ResponseTracker tracker = createTrackerForWrite(consistencyLevel, plan, plan.pending, metadata); + + CoordinationPlan.ForWrite ideal = null; + ConsistencyLevel idealCL = DatabaseDescriptor.getIdealConsistencyLevel(); + if (idealCL != null) + { + if (idealCL == consistencyLevel) + { + ideal = new CoordinationPlan.ForWrite(plan, tracker); + } + else + { + ideal = new CoordinationPlan.ForWrite(createReplicaPlanForWrite(metadata, keyspace, idealCL, liveAndDown, selector), + createTrackerForWrite(idealCL, plan, plan.pending, metadata)); + } + } + + return new CoordinationPlan.ForWriteWithIdeal(metadata, plan, tracker, ideal); + } + + public CoordinationPlan.ForWriteWithIdeal planForWrite(ClusterMetadata metadata, + Keyspace keyspace, + ConsistencyLevel consistencyLevel, + Token token, + ReplicaPlans.Selector selector) + { + return planForWrite(metadata, keyspace, consistencyLevel, + (newClusterMetadata) -> ReplicaLayout.forTokenWriteLiveAndDown(newClusterMetadata, keyspace, token), selector); + } + + /** + * Create coordination plan for forwarding a counter write to the leader replica. + * + * In cases where the original coordinator is not a replica of the counter key, the counter + * mutation is forwarded to a leader replica that will coordinate the actual counter update. + */ + public CoordinationPlan.ForWrite planForForwardingCounterWrite(ClusterMetadata metadata, + Keyspace keyspace, + Token token, + Function<ClusterMetadata, Replica> replicaSupplier) + { + ReplicaPlan.ForWrite plan = ReplicaPlans.forSingleReplicaWrite(metadata, keyspace, token, replicaSupplier); + ResponseTracker tracker = createTrackerForWrite(plan.consistencyLevel(), plan, plan.pending, metadata); + + return new CoordinationPlan.ForWriteWithIdeal(metadata, plan, tracker, null); + } + + /** + * Create coordination plan for replaying a mutation from the batchlog. + * + * When recovering failed batches, mutations are replayed to remote replicas only + * (local replica is handled separately). This method creates a coordination plan + * targeting live remote replicas with CL.ONE. + */ + public CoordinationPlan.ForWriteWithIdeal planForReplayMutation(ClusterMetadata metadata, + Keyspace keyspace, + Token token) + { + Preconditions.checkState(!replicationType.isTracked(), "Batch replay not supported with tracked keyspaces"); + + ReplicaPlan.ForWrite plan = ReplicaPlans.forReplayMutation(metadata, keyspace, token); + ResponseTracker tracker = createTrackerForWrite(plan.consistencyLevel(), plan, plan.pending, metadata); + + return new CoordinationPlan.ForWriteWithIdeal(metadata, plan, tracker, null); + } + + /** + * Create coordination plan for a single-partition token read. + */ + public CoordinationPlan.ForTokenRead planForTokenRead(ClusterMetadata metadata, + Keyspace keyspace, + TableId tableId, + Token token, + @Nullable Index.QueryPlan indexQueryPlan, + ConsistencyLevel consistencyLevel, + SpeculativeRetryPolicy retry, + ReadCoordinator coordinator) + { + ReplicaPlan.ForTokenRead plan = ReplicaPlans.forRead(metadata, keyspace, tableId, token, indexQueryPlan, consistencyLevel, retry, coordinator); + ReplicaPlan.SharedForTokenRead shared = ReplicaPlan.shared(plan); + ResponseTracker tracker = createTrackerForRead(plan); + return new CoordinationPlan.ForTokenRead(shared, tracker); + } + + /** + * Create coordination plan for a range read. + */ + public CoordinationPlan.ForRangeRead planForRangeRead(ClusterMetadata metadata, + Keyspace keyspace, + TableId tableId, + @Nullable Index.QueryPlan indexQueryPlan, + ConsistencyLevel consistencyLevel, + AbstractBounds<PartitionPosition> range, + int vnodeCount) + { + ReplicaPlan.ForRangeRead plan = ReplicaPlans.forRangeRead(metadata, keyspace, tableId, indexQueryPlan, consistencyLevel, range, vnodeCount, true); + ReplicaPlan.SharedForRangeRead shared = ReplicaPlan.shared(plan); + ResponseTracker tracker = createTrackerForRead(plan); + return new CoordinationPlan.ForRangeRead(shared, tracker); + } + + /** + * Attempt to merge two adjacent range read coordination plans into one. + * + * If the two plans share enough live endpoints to satisfy the consistency level + * and the merge is worthwhile returns a merged plan otherwise returns null. + */ + public CoordinationPlan.ForRangeRead maybeMergeRangeReads(ClusterMetadata metadata, + Keyspace keyspace, + TableId tableId, + ConsistencyLevel consistencyLevel, + ReplicaPlan.ForRangeRead left, + ReplicaPlan.ForRangeRead right) + { + ReplicaPlan.ForRangeRead merged = ReplicaPlans.maybeMerge(metadata, keyspace, tableId, consistencyLevel, left, right); + if (merged == null) + return null; + + ReplicaPlan.SharedForRangeRead shared = ReplicaPlan.shared(merged); + ResponseTracker tracker = createTrackerForRead(merged); + return new CoordinationPlan.ForRangeRead(shared, tracker); + } + + /** + * Create coordination plan for a full range read + */ + public CoordinationPlan.ForRangeRead planForFullRangeRead(Keyspace keyspace, + ConsistencyLevel consistencyLevel, + AbstractBounds<PartitionPosition> range, + Set<InetAddressAndPort> endpointsToContact, + int vnodeCount) + { + ReplicaPlan.ForRangeRead plan = ReplicaPlans.forFullRangeRead(keyspace, consistencyLevel, range, endpointsToContact, vnodeCount); + ReplicaPlan.SharedForRangeRead shared = ReplicaPlan.shared(plan); + ResponseTracker tracker = createTrackerForRead(plan); + return new CoordinationPlan.ForRangeRead(shared, tracker); + } + + /** + * Create coordination plan for a single-replica token read. + */ + public CoordinationPlan.ForTokenRead planForSingleReplicaTokenRead(Keyspace keyspace, Token token, Replica replica) + { + ReplicaPlan.ForTokenRead plan = ReplicaPlans.forSingleReplicaRead(keyspace, token, replica); + ReplicaPlan.SharedForTokenRead shared = ReplicaPlan.shared(plan); + ResponseTracker tracker = createTrackerForRead(plan); + return new CoordinationPlan.ForTokenRead(shared, tracker); + } + + /** + * Create coordination plan for a single-replica range read. + * + * Used by short read protection to fetch additional partitions from a + * specific replica. blockFor=1, totalReplicas=1. + */ + public CoordinationPlan.ForRangeRead planForSingleReplicaRangeRead(Keyspace keyspace, + AbstractBounds<PartitionPosition> range, + Replica replica, + int vnodeCount) + { + ReplicaPlan.ForRangeRead plan = ReplicaPlans.forSingleReplicaRead(keyspace, range, replica, vnodeCount); + ReplicaPlan.SharedForRangeRead shared = ReplicaPlan.shared(plan); + ResponseTracker tracker = createTrackerForRead(plan); + return new CoordinationPlan.ForRangeRead(shared, tracker); + } + + /** + * Create ResponseTracker for read operation. + */ + public <E extends Endpoints<E>, P extends ReplicaPlan.ForRead<E, P>> ResponseTracker createTrackerForRead(P plan) + { + int blockFor = plan.readQuorum(); + + // Use candidates.size() for totalReplicas to allow for speculation + // (speculation can contact additional candidates beyond initial contacts) + int totalReplicas = plan.readCandidates().size(); + + return new SimpleResponseTracker(blockFor, totalReplicas); + } + + public Paxos.Participants paxosParticipants(ClusterMetadata metadata, + TableMetadata table, + Token token, + ConsistencyLevel consistencyForConsensus, + Predicate<Replica> isReplicaAlive) + { + + KeyspaceMetadata keyspaceMetadata = metadata.schema.getKeyspaceMetadata(table.keyspace); + // MetaStrategy distributes the entire keyspace to all replicas. In addition, its tables (currently only + // the dist log table) don't use the globally configured partitioner. For these reasons we don't lookup the + // replicas using the supplied token as this can actually be of the incorrect type (for example when + // performing Paxos repair). + final Token actualToken = table.partitioner == MetaStrategy.partitioner ? MetaStrategy.entireRange.right : token; + ReplicaLayout.ForTokenWrite all = forTokenWriteLiveAndDown(keyspaceMetadata, actualToken); + ReplicaLayout.ForTokenWrite electorate = consistencyForConsensus.isDatacenterLocal() + ? all.filter(InOurDc.replicas()) : all; + + EndpointsForToken live = all.all().filter(isReplicaAlive); + return new Paxos.Participants(metadata.epoch, Keyspace.open(table.keyspace), consistencyForConsensus, all, electorate, live, + (cm) -> Paxos.Participants.get(cm, table, actualToken, consistencyForConsensus)); + } + + /** + * Create ResponseTracker for write operation based on consistency level. + */ + @VisibleForTesting + public ResponseTracker createTrackerForWrite(ConsistencyLevel cl, ReplicaPlan.ForWrite plan, Endpoints<?> pending, ClusterMetadata metadata) + { + switch (cl) + { + case ANY: + case ONE: + case TWO: + case THREE: + case QUORUM: + case ALL: + int totalContacts = plan.contacts().size(); + if (pending.isEmpty()) + { + int blockFor = cl.blockFor(this); + return new SimpleResponseTracker(blockFor, totalContacts); + } + else + { + // Check if double count model applies (some CLs like ANY don't add pending) + int baseBlockFor = cl.blockFor(this); + int totalBlockFor = cl.blockForWrite(this, pending); + + // If totalBlockFor == baseBlockFor, no double-count needed (e.g., ANY) + if (totalBlockFor == baseBlockFor) + return new SimpleResponseTracker(baseBlockFor, totalContacts); + + // Double count model: committed must satisfy base CL, total must include pending + int pendingReplicas = pending.size(); + // contacts() includes both committed and pending replicas + int committedReplicas = totalContacts - pendingReplicas; + return new WriteResponseTracker(baseBlockFor, totalBlockFor, + committedReplicas, pendingReplicas, + endpoint -> pending.endpoints().contains(endpoint)); + } Review Comment: Can be simplified to : ``` int totalContacts = plan.contacts().size(); int baseBlockFor = cl.blockFor(this); int totalBlockFor = pending.isEmpty() ? baseBlockFor : cl.blockForWrite(this, pending); // Check if double count model applies (some CLs like ANY don't add pending) // If totalBlockFor == baseBlockFor, no double-count needed (e.g., ANY) if (totalBlockFor == baseBlockFor) return new SimpleResponseTracker(baseBlockFor, totalContacts); // Double count model: natural must satisfy base CL, total must include pending int pendingReplicas = pending.size(); // contacts() includes both natural and pending replicas int naturalReplicas = totalContacts - pendingReplicas; return new WriteResponseTracker(baseBlockFor, totalBlockFor, naturalReplicas, pendingReplicas, endpoint -> pending.endpoints().contains(endpoint)); ``` ########## src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java: ########## @@ -437,4 +459,362 @@ public RangesAtEndpoint getLocalRanges(ClusterMetadata cm) return newRanges; } } + + protected CoordinationPlan.ForWrite planForWriteInternal(ClusterMetadata metadata, + Keyspace keyspace, + ConsistencyLevel consistencyLevel, + Function<ClusterMetadata, ReplicaLayout.ForTokenWrite> liveAndDown, + ReplicaPlans.Selector selector) + { + ReplicaPlan.ForWrite plan = ReplicaPlans.forWrite(metadata, keyspace, consistencyLevel, liveAndDown, selector); + ResponseTracker tracker = createTrackerForWrite(consistencyLevel, plan, plan.pending, metadata); + return new CoordinationPlan.ForWrite(plan, tracker); + } + + protected ReplicaPlan.ForWrite createReplicaPlanForWrite(ClusterMetadata metadata, + Keyspace keyspace, + ConsistencyLevel consistencyLevel, + Function<ClusterMetadata, ReplicaLayout.ForTokenWrite> liveAndDown, + ReplicaPlans.Selector selector) + { + return ReplicaPlans.forWrite(metadata, keyspace, consistencyLevel, liveAndDown, selector); + } + + public CoordinationPlan.ForWriteWithIdeal planForWrite(ClusterMetadata metadata, + Keyspace keyspace, + ConsistencyLevel consistencyLevel, + Function<ClusterMetadata, ReplicaLayout.ForTokenWrite> liveAndDown, + ReplicaPlans.Selector selector) + { + ReplicaPlan.ForWrite plan = createReplicaPlanForWrite(metadata, keyspace, consistencyLevel, liveAndDown, selector); + ResponseTracker tracker = createTrackerForWrite(consistencyLevel, plan, plan.pending, metadata); + + CoordinationPlan.ForWrite ideal = null; + ConsistencyLevel idealCL = DatabaseDescriptor.getIdealConsistencyLevel(); + if (idealCL != null) + { + if (idealCL == consistencyLevel) + { + ideal = new CoordinationPlan.ForWrite(plan, tracker); + } + else + { + ideal = new CoordinationPlan.ForWrite(createReplicaPlanForWrite(metadata, keyspace, idealCL, liveAndDown, selector), + createTrackerForWrite(idealCL, plan, plan.pending, metadata)); + } + } + + return new CoordinationPlan.ForWriteWithIdeal(metadata, plan, tracker, ideal); + } + + public CoordinationPlan.ForWriteWithIdeal planForWrite(ClusterMetadata metadata, + Keyspace keyspace, + ConsistencyLevel consistencyLevel, + Token token, + ReplicaPlans.Selector selector) + { + return planForWrite(metadata, keyspace, consistencyLevel, + (newClusterMetadata) -> ReplicaLayout.forTokenWriteLiveAndDown(newClusterMetadata, keyspace, token), selector); + } + + /** + * Create coordination plan for forwarding a counter write to the leader replica. + * + * In cases where the original coordinator is not a replica of the counter key, the counter + * mutation is forwarded to a leader replica that will coordinate the actual counter update. + */ + public CoordinationPlan.ForWrite planForForwardingCounterWrite(ClusterMetadata metadata, + Keyspace keyspace, + Token token, + Function<ClusterMetadata, Replica> replicaSupplier) + { + ReplicaPlan.ForWrite plan = ReplicaPlans.forSingleReplicaWrite(metadata, keyspace, token, replicaSupplier); + ResponseTracker tracker = createTrackerForWrite(plan.consistencyLevel(), plan, plan.pending, metadata); + + return new CoordinationPlan.ForWriteWithIdeal(metadata, plan, tracker, null); + } + + /** + * Create coordination plan for replaying a mutation from the batchlog. + * + * When recovering failed batches, mutations are replayed to remote replicas only + * (local replica is handled separately). This method creates a coordination plan + * targeting live remote replicas with CL.ONE. + */ + public CoordinationPlan.ForWriteWithIdeal planForReplayMutation(ClusterMetadata metadata, + Keyspace keyspace, + Token token) + { + Preconditions.checkState(!replicationType.isTracked(), "Batch replay not supported with tracked keyspaces"); + + ReplicaPlan.ForWrite plan = ReplicaPlans.forReplayMutation(metadata, keyspace, token); + ResponseTracker tracker = createTrackerForWrite(plan.consistencyLevel(), plan, plan.pending, metadata); + + return new CoordinationPlan.ForWriteWithIdeal(metadata, plan, tracker, null); + } + + /** + * Create coordination plan for a single-partition token read. + */ + public CoordinationPlan.ForTokenRead planForTokenRead(ClusterMetadata metadata, + Keyspace keyspace, + TableId tableId, + Token token, + @Nullable Index.QueryPlan indexQueryPlan, + ConsistencyLevel consistencyLevel, + SpeculativeRetryPolicy retry, + ReadCoordinator coordinator) + { + ReplicaPlan.ForTokenRead plan = ReplicaPlans.forRead(metadata, keyspace, tableId, token, indexQueryPlan, consistencyLevel, retry, coordinator); + ReplicaPlan.SharedForTokenRead shared = ReplicaPlan.shared(plan); + ResponseTracker tracker = createTrackerForRead(plan); + return new CoordinationPlan.ForTokenRead(shared, tracker); + } + + /** + * Create coordination plan for a range read. + */ + public CoordinationPlan.ForRangeRead planForRangeRead(ClusterMetadata metadata, + Keyspace keyspace, + TableId tableId, + @Nullable Index.QueryPlan indexQueryPlan, + ConsistencyLevel consistencyLevel, + AbstractBounds<PartitionPosition> range, + int vnodeCount) + { + ReplicaPlan.ForRangeRead plan = ReplicaPlans.forRangeRead(metadata, keyspace, tableId, indexQueryPlan, consistencyLevel, range, vnodeCount, true); + ReplicaPlan.SharedForRangeRead shared = ReplicaPlan.shared(plan); + ResponseTracker tracker = createTrackerForRead(plan); + return new CoordinationPlan.ForRangeRead(shared, tracker); + } + + /** + * Attempt to merge two adjacent range read coordination plans into one. + * + * If the two plans share enough live endpoints to satisfy the consistency level + * and the merge is worthwhile returns a merged plan otherwise returns null. + */ + public CoordinationPlan.ForRangeRead maybeMergeRangeReads(ClusterMetadata metadata, + Keyspace keyspace, + TableId tableId, + ConsistencyLevel consistencyLevel, + ReplicaPlan.ForRangeRead left, + ReplicaPlan.ForRangeRead right) + { + ReplicaPlan.ForRangeRead merged = ReplicaPlans.maybeMerge(metadata, keyspace, tableId, consistencyLevel, left, right); + if (merged == null) + return null; + + ReplicaPlan.SharedForRangeRead shared = ReplicaPlan.shared(merged); + ResponseTracker tracker = createTrackerForRead(merged); + return new CoordinationPlan.ForRangeRead(shared, tracker); + } + + /** + * Create coordination plan for a full range read + */ + public CoordinationPlan.ForRangeRead planForFullRangeRead(Keyspace keyspace, + ConsistencyLevel consistencyLevel, + AbstractBounds<PartitionPosition> range, + Set<InetAddressAndPort> endpointsToContact, + int vnodeCount) + { + ReplicaPlan.ForRangeRead plan = ReplicaPlans.forFullRangeRead(keyspace, consistencyLevel, range, endpointsToContact, vnodeCount); + ReplicaPlan.SharedForRangeRead shared = ReplicaPlan.shared(plan); + ResponseTracker tracker = createTrackerForRead(plan); + return new CoordinationPlan.ForRangeRead(shared, tracker); + } + + /** + * Create coordination plan for a single-replica token read. + */ + public CoordinationPlan.ForTokenRead planForSingleReplicaTokenRead(Keyspace keyspace, Token token, Replica replica) + { + ReplicaPlan.ForTokenRead plan = ReplicaPlans.forSingleReplicaRead(keyspace, token, replica); + ReplicaPlan.SharedForTokenRead shared = ReplicaPlan.shared(plan); + ResponseTracker tracker = createTrackerForRead(plan); + return new CoordinationPlan.ForTokenRead(shared, tracker); + } + + /** + * Create coordination plan for a single-replica range read. + * + * Used by short read protection to fetch additional partitions from a + * specific replica. blockFor=1, totalReplicas=1. + */ + public CoordinationPlan.ForRangeRead planForSingleReplicaRangeRead(Keyspace keyspace, + AbstractBounds<PartitionPosition> range, + Replica replica, + int vnodeCount) + { + ReplicaPlan.ForRangeRead plan = ReplicaPlans.forSingleReplicaRead(keyspace, range, replica, vnodeCount); + ReplicaPlan.SharedForRangeRead shared = ReplicaPlan.shared(plan); + ResponseTracker tracker = createTrackerForRead(plan); + return new CoordinationPlan.ForRangeRead(shared, tracker); + } + + /** + * Create ResponseTracker for read operation. + */ + public <E extends Endpoints<E>, P extends ReplicaPlan.ForRead<E, P>> ResponseTracker createTrackerForRead(P plan) + { + int blockFor = plan.readQuorum(); + + // Use candidates.size() for totalReplicas to allow for speculation + // (speculation can contact additional candidates beyond initial contacts) + int totalReplicas = plan.readCandidates().size(); + + return new SimpleResponseTracker(blockFor, totalReplicas); + } + + public Paxos.Participants paxosParticipants(ClusterMetadata metadata, + TableMetadata table, + Token token, + ConsistencyLevel consistencyForConsensus, + Predicate<Replica> isReplicaAlive) + { + + KeyspaceMetadata keyspaceMetadata = metadata.schema.getKeyspaceMetadata(table.keyspace); + // MetaStrategy distributes the entire keyspace to all replicas. In addition, its tables (currently only + // the dist log table) don't use the globally configured partitioner. For these reasons we don't lookup the + // replicas using the supplied token as this can actually be of the incorrect type (for example when + // performing Paxos repair). + final Token actualToken = table.partitioner == MetaStrategy.partitioner ? MetaStrategy.entireRange.right : token; + ReplicaLayout.ForTokenWrite all = forTokenWriteLiveAndDown(keyspaceMetadata, actualToken); + ReplicaLayout.ForTokenWrite electorate = consistencyForConsensus.isDatacenterLocal() + ? all.filter(InOurDc.replicas()) : all; + + EndpointsForToken live = all.all().filter(isReplicaAlive); + return new Paxos.Participants(metadata.epoch, Keyspace.open(table.keyspace), consistencyForConsensus, all, electorate, live, + (cm) -> Paxos.Participants.get(cm, table, actualToken, consistencyForConsensus)); + } + + /** + * Create ResponseTracker for write operation based on consistency level. + */ + @VisibleForTesting + public ResponseTracker createTrackerForWrite(ConsistencyLevel cl, ReplicaPlan.ForWrite plan, Endpoints<?> pending, ClusterMetadata metadata) + { + switch (cl) + { + case ANY: + case ONE: + case TWO: + case THREE: + case QUORUM: + case ALL: + int totalContacts = plan.contacts().size(); + if (pending.isEmpty()) + { + int blockFor = cl.blockFor(this); + return new SimpleResponseTracker(blockFor, totalContacts); + } + else + { + // Check if double count model applies (some CLs like ANY don't add pending) + int baseBlockFor = cl.blockFor(this); + int totalBlockFor = cl.blockForWrite(this, pending); + + // If totalBlockFor == baseBlockFor, no double-count needed (e.g., ANY) + if (totalBlockFor == baseBlockFor) + return new SimpleResponseTracker(baseBlockFor, totalContacts); + + // Double count model: committed must satisfy base CL, total must include pending + int pendingReplicas = pending.size(); + // contacts() includes both committed and pending replicas + int committedReplicas = totalContacts - pendingReplicas; + return new WriteResponseTracker(baseBlockFor, totalBlockFor, + committedReplicas, pendingReplicas, + endpoint -> pending.endpoints().contains(endpoint)); + } + + case LOCAL_ONE: + case LOCAL_QUORUM: + int localContacts = plan.contacts().filter(InOurDc.replicas()).size(); + if (pending.isEmpty()) + { + int localBlockFor = cl.blockFor(this); + return new SimpleResponseTracker(localBlockFor, localContacts, InOurDc.endpoints()); + } + else + { + // Check if double count model applies (depends on local pending) + int baseBlockFor = cl.blockFor(this); + int totalBlockFor = cl.blockForWrite(this, pending); + + // If totalBlockFor == baseBlockFor, no local pending so no double-count needed + if (totalBlockFor == baseBlockFor) + return new SimpleResponseTracker(baseBlockFor, localContacts, InOurDc.endpoints()); + + // Double count model for local DC + int localPending = pending.count(InOurDc.replicas()); + // localContacts includes both committed and pending in local DC + int localCommitted = localContacts - localPending; + return new WriteResponseTracker(baseBlockFor, totalBlockFor, + localCommitted, localPending, + endpoint -> pending.endpoints().contains(endpoint), + InOurDc.endpoints()); + } Review Comment: Can be simplified to : ``` { int localContacts = plan.contacts().filter(InOurDc.replicas()).size(); int baseBlockFor = cl.blockFor(this); int totalBlockFor = pending.isEmpty() ? baseBlockFor : cl.blockForWrite(this, pending); // Check if double count model applies (depends on local pending) // If totalBlockFor == baseBlockFor, no local pending so no double-count needed if (totalBlockFor == baseBlockFor) return new SimpleResponseTracker(baseBlockFor, localContacts, InOurDc.endpoints()); // Double count model for local DC int localPending = pending.count(InOurDc.replicas()); // localContacts includes both natural and pending in local DC int localNatural = localContacts - localPending; return new WriteResponseTracker(baseBlockFor, totalBlockFor, localNatural, localPending, endpoint -> pending.endpoints().contains(endpoint), InOurDc.endpoints()); } ``` ########## src/java/org/apache/cassandra/service/reads/tracked/TrackedRead.java: ########## @@ -285,7 +285,7 @@ private void start(Dispatcher.RequestTime requestTime, Consumer<PartialTrackedRe { // TODO: skip local coordination if this node knows its recovering from an outage // TODO: read speculation - Replica localReplica = replicaPlan.lookup(FBUtilities.getBroadcastAddressAndPort()); + Replica localReplica = plan.replicas().lookup(FBUtilities.getBroadcastAddressAndPort()); Review Comment: NIT extract `plan.replicas()` into a local variable and reuse it below: ``` P replicaPlan = plan.replicas(); Replica localReplica = replicaPlan.lookup(FBUtilities.getBroadcastAddressAndPort()); ``` ########## src/java/org/apache/cassandra/service/reads/ReplicaFilteringProtection.java: ########## @@ -636,20 +636,20 @@ private UnfilteredPartitionIterator fetchFromSource() key, filter); - ReplicaPlan.ForTokenRead replicaPlan = ReplicaPlans.forSingleReplicaRead(keyspace, key.getToken(), source); + CoordinationPlan.ForTokenRead plan = CoordinationPlan.forSingleReplicaTokenRead(keyspace, key.getToken(), source); try { - return executeReadCommand(cmd, source, ReplicaPlan.shared(replicaPlan)); + return executeReadCommand(cmd, source, plan); Review Comment: Do we no longer need to make it shared here? ########## src/java/org/apache/cassandra/service/reads/ReadCallback.java: ########## @@ -141,8 +137,8 @@ public void awaitResults() throws ReadFailureException, ReadTimeoutException * See {@link DigestResolver#preprocess(Message)} * CASSANDRA-16097 */ - int received = resolver.responses.size(); - boolean failed = failures > 0 && (replicaPlan().readQuorum() > received || !resolver.isDataPresent()); + int received = plan.responses().received(); Review Comment: NIT: assign `plan.responses()` to a local variable and use it below ``` ResponseTracker replicaPlan = plan.responses(); int received = replicaPlan.received(); ``` ########## src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java: ########## @@ -345,14 +357,17 @@ public Dispatcher.RequestTime getRequestTime() protected void signal() { + if (condition.isSignalled()) + return; + //The ideal CL should only count as a strike if the requested CL was achieved. //If the requested CL is not achieved it's fine for the ideal CL to also not be achieved. - if (idealCLDelegate != null && blockFor() + failures <= candidateReplicaCount()) + if (idealCLDelegate != null && plan.responses().isSuccessful()) Review Comment: do we also need to check for completeness here? ``` if (plan.responses().isComplete() && !plan.responses().isSuccessful()) ``` ########## src/java/org/apache/cassandra/service/paxos/Paxos.java: ########## @@ -464,26 +462,15 @@ public void collectFailure(InetAddressAndPort inetAddressAndPort, RequestFailure } - static Participants get(ClusterMetadata metadata, TableMetadata table, Token token, ConsistencyLevel consistencyForConsensus) + public static Participants get(ClusterMetadata metadata, TableMetadata table, Token token, ConsistencyLevel consistencyForConsensus) { return get(metadata, table, token, consistencyForConsensus, FailureDetector.isReplicaAlive); } static Participants get(ClusterMetadata metadata, TableMetadata table, Token token, ConsistencyLevel consistencyForConsensus, Predicate<Replica> isReplicaAlive) { - KeyspaceMetadata keyspaceMetadata = metadata.schema.getKeyspaceMetadata(table.keyspace); - // MetaStrategy distributes the entire keyspace to all replicas. In addition, its tables (currently only - // the dist log table) don't use the globally configured partitioner. For these reasons we don't lookup the - // replicas using the supplied token as this can actually be of the incorrect type (for example when - // performing Paxos repair). - final Token actualToken = table.partitioner == MetaStrategy.partitioner ? MetaStrategy.entireRange.right : token; - ReplicaLayout.ForTokenWrite all = forTokenWriteLiveAndDown(keyspaceMetadata, actualToken); - ReplicaLayout.ForTokenWrite electorate = consistencyForConsensus.isDatacenterLocal() - ? all.filter(InOurDc.replicas()) : all; - - EndpointsForToken live = all.all().filter(isReplicaAlive); - return new Participants(metadata.epoch, Keyspace.open(table.keyspace), consistencyForConsensus, all, electorate, live, - (cm) -> get(cm, table, actualToken, consistencyForConsensus)); + AbstractReplicationStrategy strategy = Keyspace.open(table.keyspace).getReplicationStrategy(); Review Comment: can the RF coming from `metadata.schema.getKeyspaceMetadata(table.keyspace)` diverge from `Keyspace.open(table.keyspace).getReplicationStrategy()` ? should we use the RF from `metadata` here instead? ########## src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java: ########## @@ -145,7 +154,7 @@ public void get() throws WriteTimeoutException, WriteFailureException, RetryOnDi throwTimeout(); int candidateReplicaCount = candidateReplicaCount(); - if (blockFor() + failures > candidateReplicaCount) + if (!plan.responses().isSuccessful()) Review Comment: do we need to check for completeness here too? ``` if (plan.responses().isComplete() && !plan.responses().isSuccessful()) ``` ########## src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java: ########## @@ -437,4 +461,376 @@ public RangesAtEndpoint getLocalRanges(ClusterMetadata cm) return newRanges; } } + + protected CoordinationPlan.ForWrite planForWriteInternal(ClusterMetadata metadata, + Keyspace keyspace, + ConsistencyLevel consistencyLevel, + Function<ClusterMetadata, ReplicaLayout.ForTokenWrite> liveAndDown, + ReplicaPlans.Selector selector) + { + ReplicaPlan.ForWrite plan = ReplicaPlans.forWrite(metadata, keyspace, consistencyLevel, liveAndDown, selector); + ResponseTracker tracker = createTrackerForWrite(consistencyLevel, plan, plan.pending, metadata); + return new CoordinationPlan.ForWrite(plan, tracker); + } + + public CoordinationPlan.ForWriteWithIdeal planForWrite(ClusterMetadata metadata, + Keyspace keyspace, + ConsistencyLevel consistencyLevel, + Function<ClusterMetadata, ReplicaLayout.ForTokenWrite> liveAndDown, + ReplicaPlans.Selector selector) + { + CoordinationPlan.ForWrite actual = planForWriteInternal(metadata, keyspace, consistencyLevel, liveAndDown, selector); + + CoordinationPlan.ForWrite ideal = null; + ConsistencyLevel idealCL = DatabaseDescriptor.getIdealConsistencyLevel(); + if (idealCL != null) + { + if (idealCL == consistencyLevel) + { + ideal = actual; + } + else + { + ideal = planForWriteInternal(metadata, keyspace, idealCL, liveAndDown, selector); + } + } + + return new CoordinationPlan.ForWriteWithIdeal(metadata, actual.replicas(), actual.responses(), ideal); + } + + public CoordinationPlan.ForWriteWithIdeal planForWrite(ClusterMetadata metadata, + Keyspace keyspace, + ConsistencyLevel consistencyLevel, + Token token, + ReplicaPlans.Selector selector) + { + return planForWrite(metadata, keyspace, consistencyLevel, + (newClusterMetadata) -> ReplicaLayout.forTokenWriteLiveAndDown(newClusterMetadata, keyspace, token), selector); + } + + /** + * Create coordination plan for forwarding a counter write to the leader replica. + * + * In cases where the original coordinator is not a replica of the counter key, the counter + * mutation is forwarded to a leader replica that will coordinate the actual counter update. + */ + public CoordinationPlan.ForWrite planForForwardingCounterWrite(ClusterMetadata metadata, + Keyspace keyspace, + Token token, + Function<ClusterMetadata, Replica> replicaSupplier) + { + ReplicaPlan.ForWrite plan = ReplicaPlans.forSingleReplicaWrite(metadata, keyspace, token, replicaSupplier); + ResponseTracker tracker = createTrackerForWrite(plan.consistencyLevel(), plan, plan.pending, metadata); + + return new CoordinationPlan.ForWriteWithIdeal(metadata, plan, tracker, null); + } + + /** + * Create coordination plan for replaying a mutation from the batchlog. + * + * When recovering failed batches, mutations are replayed to remote replicas only + * (local replica is handled separately). This method creates a replica plan + * targeting live remote replicas with CL.ONE, and a response tracker that waits on + * all contacts + */ + public CoordinationPlan.ForWriteWithIdeal planForReplayMutation(ClusterMetadata metadata, + Keyspace keyspace, + Token token) + { + Preconditions.checkState(!replicationType.isTracked(), "Batch replay not supported with tracked keyspaces"); + + ReplicaPlan.ForWrite plan = ReplicaPlans.forReplayMutation(metadata, keyspace, token); + + // wait until all contacts respond + int blockFor = plan.contacts().size(); + ResponseTracker tracker = new SimpleResponseTracker(blockFor, blockFor); + + return new CoordinationPlan.ForWriteWithIdeal(metadata, plan, tracker, null); + } + + /** + * Create coordination plan for a single-partition token read. + */ + public CoordinationPlan.ForTokenRead planForTokenRead(ClusterMetadata metadata, + Keyspace keyspace, + TableId tableId, + Token token, + @Nullable Index.QueryPlan indexQueryPlan, + ConsistencyLevel consistencyLevel, + SpeculativeRetryPolicy retry, + ReadCoordinator coordinator) + { + ReplicaPlan.ForTokenRead plan = ReplicaPlans.forRead(metadata, keyspace, tableId, token, indexQueryPlan, consistencyLevel, retry, coordinator); + ReplicaPlan.SharedForTokenRead shared = ReplicaPlan.shared(plan); + ResponseTracker tracker = createTrackerForRead(plan); + return new CoordinationPlan.ForTokenRead(shared, tracker); + } + + /** + * Create coordination plan for a range read. + */ + public CoordinationPlan.ForRangeRead planForRangeRead(ClusterMetadata metadata, + Keyspace keyspace, + TableId tableId, + @Nullable Index.QueryPlan indexQueryPlan, + ConsistencyLevel consistencyLevel, + AbstractBounds<PartitionPosition> range, + int vnodeCount) + { + ReplicaPlan.ForRangeRead plan = ReplicaPlans.forRangeRead(metadata, keyspace, tableId, indexQueryPlan, consistencyLevel, range, vnodeCount, true); + ReplicaPlan.SharedForRangeRead shared = ReplicaPlan.shared(plan); + ResponseTracker tracker = createTrackerForRead(plan); + return new CoordinationPlan.ForRangeRead(shared, tracker); + } + + /** + * Attempt to merge two adjacent range read coordination plans into one. + * + * If the two plans share enough live endpoints to satisfy the consistency level + * and the merge is worthwhile returns a merged plan otherwise returns null. + */ + public CoordinationPlan.ForRangeRead maybeMergeRangeReads(ClusterMetadata metadata, + Keyspace keyspace, + TableId tableId, + ConsistencyLevel consistencyLevel, + ReplicaPlan.ForRangeRead left, + ReplicaPlan.ForRangeRead right) + { + ReplicaPlan.ForRangeRead merged = ReplicaPlans.maybeMerge(metadata, keyspace, tableId, consistencyLevel, left, right); + if (merged == null) + return null; + + ReplicaPlan.SharedForRangeRead shared = ReplicaPlan.shared(merged); + ResponseTracker tracker = createTrackerForRead(merged); + return new CoordinationPlan.ForRangeRead(shared, tracker); + } + + /** + * Create coordination plan for a full range read + */ + public CoordinationPlan.ForRangeRead planForFullRangeRead(Keyspace keyspace, + ConsistencyLevel consistencyLevel, + AbstractBounds<PartitionPosition> range, + Set<InetAddressAndPort> endpointsToContact, + int vnodeCount) + { + ReplicaPlan.ForRangeRead plan = ReplicaPlans.forFullRangeRead(keyspace, consistencyLevel, range, endpointsToContact, vnodeCount); + ReplicaPlan.SharedForRangeRead shared = ReplicaPlan.shared(plan); + ResponseTracker tracker = createTrackerForRead(plan); + return new CoordinationPlan.ForRangeRead(shared, tracker); + } + + /** + * Create coordination plan for a single-replica token read. + */ + public CoordinationPlan.ForTokenRead planForSingleReplicaTokenRead(Keyspace keyspace, Token token, Replica replica) + { + ReplicaPlan.ForTokenRead plan = ReplicaPlans.forSingleReplicaRead(keyspace, token, replica); + ReplicaPlan.SharedForTokenRead shared = ReplicaPlan.shared(plan); + ResponseTracker tracker = createTrackerForRead(plan); + return new CoordinationPlan.ForTokenRead(shared, tracker); + } + + /** + * Create coordination plan for a single-replica range read. + * + * Used by short read protection to fetch additional partitions from a + * specific replica. blockFor=1, totalReplicas=1. + */ + public CoordinationPlan.ForRangeRead planForSingleReplicaRangeRead(Keyspace keyspace, + AbstractBounds<PartitionPosition> range, + Replica replica, + int vnodeCount) + { + ReplicaPlan.ForRangeRead plan = ReplicaPlans.forSingleReplicaRead(keyspace, range, replica, vnodeCount); + ReplicaPlan.SharedForRangeRead shared = ReplicaPlan.shared(plan); + ResponseTracker tracker = createTrackerForRead(plan); + return new CoordinationPlan.ForRangeRead(shared, tracker); + } + + /** + * Create ResponseTracker for read operation. + */ + public <E extends Endpoints<E>, P extends ReplicaPlan.ForRead<E, P>> ResponseTracker createTrackerForRead(P plan) + { + int blockFor = plan.readQuorum(); + + // Use candidates.size() for totalReplicas to allow for speculation + // (speculation can contact additional candidates beyond initial contacts) + int totalReplicas = plan.readCandidates().size(); + + return new SimpleResponseTracker(blockFor, totalReplicas); + } + + public Paxos.Participants paxosParticipants(ClusterMetadata metadata, + TableMetadata table, + Token token, + ConsistencyLevel consistencyForConsensus, + Predicate<Replica> isReplicaAlive) + { + + KeyspaceMetadata keyspaceMetadata = metadata.schema.getKeyspaceMetadata(table.keyspace); + // MetaStrategy distributes the entire keyspace to all replicas. In addition, its tables (currently only + // the dist log table) don't use the globally configured partitioner. For these reasons we don't lookup the + // replicas using the supplied token as this can actually be of the incorrect type (for example when + // performing Paxos repair). + final Token actualToken = table.partitioner == MetaStrategy.partitioner ? MetaStrategy.entireRange.right : token; + ReplicaLayout.ForTokenWrite all = forTokenWriteLiveAndDown(keyspaceMetadata, actualToken); Review Comment: `forTokenWriteLiveAndDown` uses ClusterMetadata.current(), which might diverge from the `ClusterMetadata` metadata object here. I am also wondering why we are not doing this instead: ``` ReplicaLayout.ForTokenWrite all = forTokenWriteLiveAndDown(metadata, keyspaceMetadata, actualToken); ``` ########## src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java: ########## @@ -437,4 +461,376 @@ public RangesAtEndpoint getLocalRanges(ClusterMetadata cm) return newRanges; } } + + protected CoordinationPlan.ForWrite planForWriteInternal(ClusterMetadata metadata, + Keyspace keyspace, + ConsistencyLevel consistencyLevel, + Function<ClusterMetadata, ReplicaLayout.ForTokenWrite> liveAndDown, + ReplicaPlans.Selector selector) + { + ReplicaPlan.ForWrite plan = ReplicaPlans.forWrite(metadata, keyspace, consistencyLevel, liveAndDown, selector); + ResponseTracker tracker = createTrackerForWrite(consistencyLevel, plan, plan.pending, metadata); + return new CoordinationPlan.ForWrite(plan, tracker); + } + + public CoordinationPlan.ForWriteWithIdeal planForWrite(ClusterMetadata metadata, + Keyspace keyspace, + ConsistencyLevel consistencyLevel, + Function<ClusterMetadata, ReplicaLayout.ForTokenWrite> liveAndDown, + ReplicaPlans.Selector selector) + { + CoordinationPlan.ForWrite actual = planForWriteInternal(metadata, keyspace, consistencyLevel, liveAndDown, selector); + + CoordinationPlan.ForWrite ideal = null; + ConsistencyLevel idealCL = DatabaseDescriptor.getIdealConsistencyLevel(); + if (idealCL != null) + { + if (idealCL == consistencyLevel) + { + ideal = actual; + } + else + { + ideal = planForWriteInternal(metadata, keyspace, idealCL, liveAndDown, selector); + } + } + + return new CoordinationPlan.ForWriteWithIdeal(metadata, actual.replicas(), actual.responses(), ideal); + } + + public CoordinationPlan.ForWriteWithIdeal planForWrite(ClusterMetadata metadata, + Keyspace keyspace, + ConsistencyLevel consistencyLevel, + Token token, + ReplicaPlans.Selector selector) + { + return planForWrite(metadata, keyspace, consistencyLevel, + (newClusterMetadata) -> ReplicaLayout.forTokenWriteLiveAndDown(newClusterMetadata, keyspace, token), selector); + } + + /** + * Create coordination plan for forwarding a counter write to the leader replica. + * + * In cases where the original coordinator is not a replica of the counter key, the counter + * mutation is forwarded to a leader replica that will coordinate the actual counter update. + */ + public CoordinationPlan.ForWrite planForForwardingCounterWrite(ClusterMetadata metadata, + Keyspace keyspace, + Token token, + Function<ClusterMetadata, Replica> replicaSupplier) + { + ReplicaPlan.ForWrite plan = ReplicaPlans.forSingleReplicaWrite(metadata, keyspace, token, replicaSupplier); + ResponseTracker tracker = createTrackerForWrite(plan.consistencyLevel(), plan, plan.pending, metadata); + + return new CoordinationPlan.ForWriteWithIdeal(metadata, plan, tracker, null); + } + + /** + * Create coordination plan for replaying a mutation from the batchlog. + * + * When recovering failed batches, mutations are replayed to remote replicas only + * (local replica is handled separately). This method creates a replica plan + * targeting live remote replicas with CL.ONE, and a response tracker that waits on + * all contacts + */ + public CoordinationPlan.ForWriteWithIdeal planForReplayMutation(ClusterMetadata metadata, + Keyspace keyspace, + Token token) + { + Preconditions.checkState(!replicationType.isTracked(), "Batch replay not supported with tracked keyspaces"); + + ReplicaPlan.ForWrite plan = ReplicaPlans.forReplayMutation(metadata, keyspace, token); + + // wait until all contacts respond + int blockFor = plan.contacts().size(); + ResponseTracker tracker = new SimpleResponseTracker(blockFor, blockFor); + + return new CoordinationPlan.ForWriteWithIdeal(metadata, plan, tracker, null); + } + + /** + * Create coordination plan for a single-partition token read. + */ + public CoordinationPlan.ForTokenRead planForTokenRead(ClusterMetadata metadata, + Keyspace keyspace, + TableId tableId, + Token token, + @Nullable Index.QueryPlan indexQueryPlan, + ConsistencyLevel consistencyLevel, + SpeculativeRetryPolicy retry, + ReadCoordinator coordinator) + { + ReplicaPlan.ForTokenRead plan = ReplicaPlans.forRead(metadata, keyspace, tableId, token, indexQueryPlan, consistencyLevel, retry, coordinator); + ReplicaPlan.SharedForTokenRead shared = ReplicaPlan.shared(plan); + ResponseTracker tracker = createTrackerForRead(plan); + return new CoordinationPlan.ForTokenRead(shared, tracker); + } + + /** + * Create coordination plan for a range read. + */ + public CoordinationPlan.ForRangeRead planForRangeRead(ClusterMetadata metadata, + Keyspace keyspace, + TableId tableId, + @Nullable Index.QueryPlan indexQueryPlan, + ConsistencyLevel consistencyLevel, + AbstractBounds<PartitionPosition> range, + int vnodeCount) + { + ReplicaPlan.ForRangeRead plan = ReplicaPlans.forRangeRead(metadata, keyspace, tableId, indexQueryPlan, consistencyLevel, range, vnodeCount, true); + ReplicaPlan.SharedForRangeRead shared = ReplicaPlan.shared(plan); + ResponseTracker tracker = createTrackerForRead(plan); + return new CoordinationPlan.ForRangeRead(shared, tracker); + } + + /** + * Attempt to merge two adjacent range read coordination plans into one. + * + * If the two plans share enough live endpoints to satisfy the consistency level + * and the merge is worthwhile returns a merged plan otherwise returns null. + */ + public CoordinationPlan.ForRangeRead maybeMergeRangeReads(ClusterMetadata metadata, + Keyspace keyspace, + TableId tableId, + ConsistencyLevel consistencyLevel, + ReplicaPlan.ForRangeRead left, + ReplicaPlan.ForRangeRead right) + { + ReplicaPlan.ForRangeRead merged = ReplicaPlans.maybeMerge(metadata, keyspace, tableId, consistencyLevel, left, right); + if (merged == null) + return null; + + ReplicaPlan.SharedForRangeRead shared = ReplicaPlan.shared(merged); + ResponseTracker tracker = createTrackerForRead(merged); + return new CoordinationPlan.ForRangeRead(shared, tracker); + } + + /** + * Create coordination plan for a full range read + */ + public CoordinationPlan.ForRangeRead planForFullRangeRead(Keyspace keyspace, + ConsistencyLevel consistencyLevel, + AbstractBounds<PartitionPosition> range, + Set<InetAddressAndPort> endpointsToContact, + int vnodeCount) + { + ReplicaPlan.ForRangeRead plan = ReplicaPlans.forFullRangeRead(keyspace, consistencyLevel, range, endpointsToContact, vnodeCount); + ReplicaPlan.SharedForRangeRead shared = ReplicaPlan.shared(plan); + ResponseTracker tracker = createTrackerForRead(plan); + return new CoordinationPlan.ForRangeRead(shared, tracker); + } + + /** + * Create coordination plan for a single-replica token read. + */ + public CoordinationPlan.ForTokenRead planForSingleReplicaTokenRead(Keyspace keyspace, Token token, Replica replica) + { + ReplicaPlan.ForTokenRead plan = ReplicaPlans.forSingleReplicaRead(keyspace, token, replica); + ReplicaPlan.SharedForTokenRead shared = ReplicaPlan.shared(plan); + ResponseTracker tracker = createTrackerForRead(plan); + return new CoordinationPlan.ForTokenRead(shared, tracker); + } + + /** + * Create coordination plan for a single-replica range read. + * + * Used by short read protection to fetch additional partitions from a + * specific replica. blockFor=1, totalReplicas=1. + */ + public CoordinationPlan.ForRangeRead planForSingleReplicaRangeRead(Keyspace keyspace, + AbstractBounds<PartitionPosition> range, + Replica replica, + int vnodeCount) + { + ReplicaPlan.ForRangeRead plan = ReplicaPlans.forSingleReplicaRead(keyspace, range, replica, vnodeCount); + ReplicaPlan.SharedForRangeRead shared = ReplicaPlan.shared(plan); + ResponseTracker tracker = createTrackerForRead(plan); + return new CoordinationPlan.ForRangeRead(shared, tracker); + } + + /** + * Create ResponseTracker for read operation. + */ + public <E extends Endpoints<E>, P extends ReplicaPlan.ForRead<E, P>> ResponseTracker createTrackerForRead(P plan) + { + int blockFor = plan.readQuorum(); + + // Use candidates.size() for totalReplicas to allow for speculation + // (speculation can contact additional candidates beyond initial contacts) + int totalReplicas = plan.readCandidates().size(); + + return new SimpleResponseTracker(blockFor, totalReplicas); + } + + public Paxos.Participants paxosParticipants(ClusterMetadata metadata, + TableMetadata table, + Token token, + ConsistencyLevel consistencyForConsensus, + Predicate<Replica> isReplicaAlive) + { + + KeyspaceMetadata keyspaceMetadata = metadata.schema.getKeyspaceMetadata(table.keyspace); + // MetaStrategy distributes the entire keyspace to all replicas. In addition, its tables (currently only + // the dist log table) don't use the globally configured partitioner. For these reasons we don't lookup the + // replicas using the supplied token as this can actually be of the incorrect type (for example when + // performing Paxos repair). + final Token actualToken = table.partitioner == MetaStrategy.partitioner ? MetaStrategy.entireRange.right : token; + ReplicaLayout.ForTokenWrite all = forTokenWriteLiveAndDown(keyspaceMetadata, actualToken); + ReplicaLayout.ForTokenWrite electorate = consistencyForConsensus.isDatacenterLocal() + ? all.filter(InOurDc.replicas()) : all; + + EndpointsForToken live = all.all().filter(isReplicaAlive); + return new Paxos.Participants(metadata.epoch, Keyspace.open(table.keyspace), consistencyForConsensus, all, electorate, live, + (cm) -> Paxos.Participants.get(cm, table, actualToken, consistencyForConsensus)); + } + + /** + * Hook for replication strategies to send additional mutations alongside a paxos commit. + * Called from PaxosCommit.start() after local synchronous execution for tracked keyspaces. + * + * If the method doesn't return null, the returned future is composed with the paxos consensus: + * onDone fires only after both the paxos quorum decision AND this future complete, or after + * one of them fails. + */ + public Future<Void> sendPaxosCommitMutations(Agreed commit, boolean isUrgent) + { + return null; + } + + /** + * Check whether paxos operations should be rejected for the given token. + */ + public boolean shouldRejectPaxos(Token token) + { + return false; + } + + /** + * Create ResponseTracker for write operation based on consistency level. + */ + @VisibleForTesting + public ResponseTracker createTrackerForWrite(ConsistencyLevel cl, ReplicaPlan.ForWrite plan, Endpoints<?> pending, ClusterMetadata metadata) + { + switch (cl) + { + case ANY: + case ONE: + case TWO: + case THREE: + case QUORUM: + case ALL: + int totalContacts = plan.contacts().size(); + if (pending.isEmpty()) + { + int blockFor = cl.blockFor(this); + return new SimpleResponseTracker(blockFor, totalContacts); + } + else + { + // Check if double count model applies (some CLs like ANY don't add pending) + int baseBlockFor = cl.blockFor(this); + int totalBlockFor = cl.blockForWrite(this, pending); + + // If totalBlockFor == baseBlockFor, no double-count needed (e.g., ANY) + if (totalBlockFor == baseBlockFor) + return new SimpleResponseTracker(baseBlockFor, totalContacts); + + // Double count model: committed must satisfy base CL, total must include pending + int pendingReplicas = pending.size(); + // contacts() includes both committed and pending replicas + int committedReplicas = totalContacts - pendingReplicas; + return new WriteResponseTracker(baseBlockFor, totalBlockFor, + committedReplicas, pendingReplicas, + endpoint -> pending.endpoints().contains(endpoint)); + } + + case LOCAL_ONE: + case LOCAL_QUORUM: + int localContacts = plan.contacts().filter(InOurDc.replicas()).size(); + if (pending.isEmpty()) + { + int localBlockFor = cl.blockFor(this); + return new SimpleResponseTracker(localBlockFor, localContacts, InOurDc.endpoints()); + } + else + { + // Check if double count model applies (depends on local pending) + int baseBlockFor = cl.blockFor(this); + int totalBlockFor = cl.blockForWrite(this, pending); + + // If totalBlockFor == baseBlockFor, no local pending so no double-count needed + if (totalBlockFor == baseBlockFor) + return new SimpleResponseTracker(baseBlockFor, localContacts, InOurDc.endpoints()); + + // Double count model for local DC + int localPending = pending.count(InOurDc.replicas()); + // localContacts includes both committed and pending in local DC + int localCommitted = localContacts - localPending; + return new WriteResponseTracker(baseBlockFor, totalBlockFor, + localCommitted, localPending, + endpoint -> pending.endpoints().contains(endpoint), + InOurDc.endpoints()); + } + + case EACH_QUORUM: + return createPerDcTracker(plan, pending, metadata); + + default: + throw new UnsupportedOperationException("Unsupported consistency level for writes: " + cl); + } + } + + /** + * Create per-datacenter tracker for EACH_QUORUM. + */ + private ResponseTracker createPerDcTracker(ReplicaPlan.ForWrite plan, Endpoints<?> pending, ClusterMetadata metadata) + { + Map<String, ResponseTracker> trackerPerDc = new HashMap<>(); + Locator locator = metadata.locator; + + // Group replicas by datacenter + Map<String, List<Replica>> replicasByDc = new HashMap<>(); + for (Replica replica : plan.contacts()) + { + String dc = locator.location(replica.endpoint()).datacenter; + replicasByDc.computeIfAbsent(dc, k -> new ArrayList<>()).add(replica); + } + + // Group pending replicas by datacenter + Map<String, List<Replica>> pendingByDc = new HashMap<>(); + for (Replica replica : pending) + { + String dc = locator.location(replica.endpoint()).datacenter; + pendingByDc.computeIfAbsent(dc, k -> new ArrayList<>()).add(replica); + } + + // Create tracker for each DC + for (Map.Entry<String, List<Replica>> entry : replicasByDc.entrySet()) + { + String dc = entry.getKey(); + int dcContacts = entry.getValue().size(); + List<Replica> dcPending = pendingByDc.getOrDefault(dc, Collections.emptyList()); + int dcPendingCount = dcPending.size(); + int dcCommitted = dcContacts - dcPendingCount; + int dcBlockFor = dcCommitted / 2 + 1; Review Comment: I had the same question, but I think it would be handled with `int totalBlockFor = dcBlockFor + dcPendingCount;` ########## src/java/org/apache/cassandra/locator/WriteResponseTracker.java: ########## @@ -0,0 +1,237 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.locator; + +import java.util.function.Predicate; + +import org.apache.cassandra.exceptions.RequestFailureReason; + +/** + * Response tracker for writes with pending replicas using the double count model. + * <p> + * Two requirements must be satisfied for success: + * <ol> + * <li>Committed replicas must satisfy base CL: {@code committedSuccesses >= baseBlockFor}</li> + * <li>Total replicas (committed + pending) must satisfy: {@code totalSuccesses >= totalBlockFor}</li> + * </ol> + * <p> + * This ensures both consistency (committed replicas have the data) and bootstrap safety + * (pending replicas also receive the write). + * <p> + * Implemented by composing two {@link SimpleResponseTracker}s: + * <ul> + * <li>committedTracker: tracks responses from committed replicas only</li> + * <li>totalTracker: tracks responses from all replicas (committed + pending)</li> + * </ul> + * <p> + * Thread-safe through delegation to thread-safe SimpleResponseTrackers. + */ +public class WriteResponseTracker implements ResponseTracker +{ + private final SimpleResponseTracker natural; + private final SimpleResponseTracker total; + + /** + * Create a write response tracker with the double count model. + * + * @param baseBlockFor number of committed replica responses required for CL + * @param totalBlockFor total responses required (baseBlockFor + pending count) + * @param committedReplicas number of committed replicas available + * @param pendingReplicas number of pending replicas available + * @param isPending predicate to determine if an endpoint is pending + */ + public WriteResponseTracker(int baseBlockFor, + int totalBlockFor, + int committedReplicas, + int pendingReplicas, + Predicate<InetAddressAndPort> isPending) + { + this(baseBlockFor, totalBlockFor, committedReplicas, pendingReplicas, isPending, null); + } + + /** + * Create a write response tracker with the double count model and a filter. + * + * @param naturalBlockFor number of committed replica responses required for CL + * @param totalBlockFor total responses required (naturalBlockFor + pending count) + * @param naturalReplicas number of committed replicas available + * @param pendingReplicas number of pending replicas available + * @param isPending predicate to determine if an endpoint is pending + * @param filter predicate to filter which responses count (e.g., InOurDc for LOCAL_QUORUM) + */ + public WriteResponseTracker(int naturalBlockFor, + int totalBlockFor, + int naturalReplicas, + int pendingReplicas, + Predicate<InetAddressAndPort> isPending, + Predicate<InetAddressAndPort> filter) + { + if (naturalBlockFor < 0) + throw new IllegalArgumentException("naturalBlockFor must be non-negative: " + naturalBlockFor); + if (totalBlockFor < naturalBlockFor) + throw new IllegalArgumentException("totalBlockFor (" + totalBlockFor + ") must be >= naturalBlockFor (" + naturalBlockFor + ")"); + if (naturalReplicas < 0) + throw new IllegalArgumentException("naturalReplicas must be non-negative: " + naturalReplicas); + if (pendingReplicas < 0) + throw new IllegalArgumentException("pendingReplicas must be non-negative: " + pendingReplicas); + if (naturalBlockFor > naturalReplicas) + throw new IllegalArgumentException("naturalBlockFor (" + naturalBlockFor + ") cannot exceed naturalReplicas (" + naturalReplicas + ")"); + if (totalBlockFor > naturalReplicas + pendingReplicas) + throw new IllegalArgumentException("totalBlockFor (" + totalBlockFor + ") cannot exceed total replicas (" + (naturalReplicas + pendingReplicas) + ")"); + if (isPending == null) + throw new IllegalArgumentException("isPending predicate cannot be null"); + + Predicate<InetAddressAndPort> naturalFilter = isPending.negate(); + if (filter != null) + naturalFilter = naturalFilter.and(filter); + + this.natural = new SimpleResponseTracker(naturalBlockFor, naturalReplicas, naturalFilter); + this.total = new SimpleResponseTracker(totalBlockFor, naturalReplicas + pendingReplicas, filter); + } + + private WriteResponseTracker(SimpleResponseTracker natural, SimpleResponseTracker total) + { + this.natural = natural; + this.total = total; + } + + @Override + public void onResponse(InetAddressAndPort from) + { + natural.onResponse(from); + total.onResponse(from); + } + + @Override + public void onFailure(InetAddressAndPort from, RequestFailureReason reason) + { + natural.onFailure(from, reason); + total.onFailure(from, reason); + } + + @Override + public boolean isComplete() + { + // Early failure if either tracker fails + if (natural.isComplete() && !natural.isSuccessful()) + return true; + if (total.isComplete() && !total.isSuccessful()) + return true; + + // Success requires both to succeed + return natural.isSuccessful() && total.isSuccessful(); + } + + @Override + public boolean isSuccessful() + { + return natural.isSuccessful() && total.isSuccessful(); + } + + @Override + public int required() + { + // Return total requirement for error messages + return total.required(); + } + + @Override + public int received() + { + // Return total successes for error messages + return total.received(); + } + + @Override + public int failures() + { + // Return total failures for error messages + return total.failures(); + } + + @Override + public boolean countsTowardQuorum(InetAddressAndPort from) + { + return total.countsTowardQuorum(from); + } + + public int committedReceived() + { + return natural.received(); + } + + public int pendingReceived() + { + return total.received() - natural.received(); + } + + public int committedFailures() + { + return natural.failures(); + } + + public int pendingFailures() Review Comment: also not atomic ########## src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java: ########## @@ -17,109 +17,69 @@ */ package org.apache.cassandra.service; -import java.util.HashMap; import java.util.Map; -import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Supplier; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.ConsistencyLevel; import org.apache.cassandra.db.MessageParams; import org.apache.cassandra.db.Mutation; import org.apache.cassandra.db.WriteType; +import org.apache.cassandra.locator.CoordinationPlan; +import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.locator.Locator; -import org.apache.cassandra.locator.NetworkTopologyStrategy; -import org.apache.cassandra.locator.Replica; -import org.apache.cassandra.locator.ReplicaPlan; import org.apache.cassandra.net.Message; import org.apache.cassandra.net.ParamType; import org.apache.cassandra.service.writes.thresholds.WriteWarningContext; import org.apache.cassandra.transport.Dispatcher; +import org.apache.cassandra.utils.FBUtilities; /** * This class blocks for a quorum of responses _in all datacenters_ (CL.EACH_QUORUM). + * + * The PerDcResponseTracker handles per-datacenter counting. */ public class DatacenterSyncWriteResponseHandler<T> extends AbstractWriteResponseHandler<T> { private static final Locator locator = DatabaseDescriptor.getLocator(); - private final Map<String, AtomicInteger> responses = new HashMap<String, AtomicInteger>(); - private final AtomicInteger acks = new AtomicInteger(0); - - public DatacenterSyncWriteResponseHandler(ReplicaPlan.ForWrite replicaPlan, + public DatacenterSyncWriteResponseHandler(CoordinationPlan.ForWrite coordinationPlan, Review Comment: `locator` is no longer used, we can remove it as well -- 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]
