diff --git a/src/java/org/apache/cassandra/batchlog/BatchlogManager.java b/src/java/org/apache/cassandra/batchlog/BatchlogManager.java index ce834a8114..76bcd86e71 100644 --- a/src/java/org/apache/cassandra/batchlog/BatchlogManager.java +++ b/src/java/org/apache/cassandra/batchlog/BatchlogManager.java @@ -630,7 +630,7 @@ public class BatchlogManager implements BatchlogManagerMBean } } - ReplayWriteResponseHandler handler = new ReplayWriteResponseHandler<>(replayPlan, () -> mutation, Dispatcher.RequestTime.forImmediateExecution()); + ReplayWriteResponseHandler handler = new ReplayWriteResponseHandler<>(replayPlan, mutation, Dispatcher.RequestTime.forImmediateExecution()); Message message = Message.outWithFlag(MUTATION_REQ, mutation, MessageFlag.CALL_BACK_ON_FAILURE); for (Replica replica : replayPlan.replicas().liveAndDown()) MessagingService.instance().sendWriteWithCallback(message, replica, handler); diff --git a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java index cb61e44baa..58aa2446aa 100644 --- a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java +++ b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java @@ -495,7 +495,7 @@ public abstract class AbstractReplicationStrategy } } - return new CoordinationPlan.ForWriteWithIdeal(metadata, actual.replicas(), actual.responses(), ideal); + return new CoordinationPlan.ForWriteWithIdeal(actual.replicas(), actual.responses(), ideal); } public CoordinationPlan.ForWriteWithIdeal planForWrite(ClusterMetadata metadata, @@ -522,7 +522,7 @@ public abstract class AbstractReplicationStrategy 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); + return new CoordinationPlan.ForWriteWithIdeal(plan, tracker, null); } /** @@ -545,7 +545,7 @@ public abstract class AbstractReplicationStrategy int blockFor = plan.contacts().size(); ResponseTracker tracker = new SimpleResponseTracker(blockFor, blockFor); - return new CoordinationPlan.ForWriteWithIdeal(metadata, plan, tracker, null); + return new CoordinationPlan.ForWriteWithIdeal(plan, tracker, null); } /** @@ -675,7 +675,7 @@ public abstract class AbstractReplicationStrategy // 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 all = forTokenWriteLiveAndDown(metadata, keyspaceMetadata, actualToken); ReplicaLayout.ForTokenWrite electorate = consistencyForConsensus.isDatacenterLocal() ? all.filter(InOurDc.replicas()) : all; @@ -719,58 +719,46 @@ public abstract class AbstractReplicationStrategy 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); + 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); + // 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)); - } + // 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)); + } 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); + // 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()); + // 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()); - } + // 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()); + } case EACH_QUORUM: return createPerDcTracker(plan, pending, metadata); diff --git a/src/java/org/apache/cassandra/locator/CoordinationPlan.java b/src/java/org/apache/cassandra/locator/CoordinationPlan.java index 38d4d1e642..7260b65a24 100644 --- a/src/java/org/apache/cassandra/locator/CoordinationPlan.java +++ b/src/java/org/apache/cassandra/locator/CoordinationPlan.java @@ -24,6 +24,8 @@ import java.util.function.Supplier; import javax.annotation.Nullable; +import com.google.common.base.Preconditions; + import org.apache.cassandra.db.ConsistencyLevel; import org.apache.cassandra.db.Keyspace; import org.apache.cassandra.db.PartitionPosition; @@ -47,9 +49,11 @@ import org.apache.cassandra.tcm.ClusterMetadata; * 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 + *
    + *
  • 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 @@ -69,9 +73,7 @@ public abstract class CoordinationPlan, P extends Replica */ public CoordinationPlan(ResponseTracker responses) { - if (responses == null) - throw new IllegalArgumentException("tracker cannot be null"); - + Preconditions.checkNotNull(responses); this.responses = responses; } @@ -130,13 +132,11 @@ public abstract class CoordinationPlan, P extends Replica public static class ForWriteWithIdeal extends CoordinationPlan.ForWrite { - public final ClusterMetadata metadata; public final CoordinationPlan.ForWrite ideal; - public ForWriteWithIdeal(ClusterMetadata metadata, ReplicaPlan.ForWrite replicas, ResponseTracker responses, CoordinationPlan.ForWrite ideal) + public ForWriteWithIdeal(ReplicaPlan.ForWrite replicas, ResponseTracker responses, CoordinationPlan.ForWrite ideal) { super(replicas, responses); - this.metadata = metadata; this.ideal = ideal; } @@ -159,7 +159,7 @@ public abstract class CoordinationPlan, P extends Replica ReplicaPlan.ForWrite plan = ReplicaPlans.forBatchlogWrite(metadata, isAny); int blockFor = plan.consistencyLevel().blockFor(plan.replicationStrategy()); ResponseTracker tracker = new SimpleResponseTracker(blockFor, plan.contacts().size()); - return new ForWriteWithIdeal(metadata, plan, tracker, null); + return new ForWriteWithIdeal(plan, tracker, null); } } diff --git a/src/java/org/apache/cassandra/locator/ReplicaPlans.java b/src/java/org/apache/cassandra/locator/ReplicaPlans.java index 3c1ac78c0c..c2f7623554 100644 --- a/src/java/org/apache/cassandra/locator/ReplicaPlans.java +++ b/src/java/org/apache/cassandra/locator/ReplicaPlans.java @@ -297,12 +297,13 @@ public class ReplicaPlans Replica selfReplica = liveAndDown.all().selfIfPresent(); ReplicaLayout.ForTokenWrite liveRemoteOnly = liveAndDown.filter(r -> FailureDetector.isReplicaAlive.test(r) && r != selfReplica); + EndpointsForToken allLiveRemoteOnly = liveRemoteOnly.all(); return new ReplicaPlan.ForWrite(keyspace, keyspace.getReplicationStrategy(), ConsistencyLevel.ONE, liveRemoteOnly.pending(), - liveRemoteOnly.all(), - liveRemoteOnly.all(), - liveRemoteOnly.all(), + allLiveRemoteOnly, + allLiveRemoteOnly, + allLiveRemoteOnly, (cm) -> forReplayMutation(cm, keyspace, token), metadata.epoch); } diff --git a/src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java b/src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java index dc822c2994..53a68cde1e 100644 --- a/src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java +++ b/src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java @@ -20,14 +20,12 @@ package org.apache.cassandra.service; import java.util.Map; 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.net.Message; import org.apache.cassandra.net.ParamType; import org.apache.cassandra.service.writes.thresholds.WriteWarningContext; @@ -41,8 +39,6 @@ import org.apache.cassandra.utils.FBUtilities; */ public class DatacenterSyncWriteResponseHandler extends AbstractWriteResponseHandler { - private static final Locator locator = DatabaseDescriptor.getLocator(); - public DatacenterSyncWriteResponseHandler(CoordinationPlan.ForWrite coordinationPlan, Runnable callback, WriteType writeType, diff --git a/src/java/org/apache/cassandra/service/paxos/Paxos.java b/src/java/org/apache/cassandra/service/paxos/Paxos.java index 32f8cd0fd6..3791db6cc6 100644 --- a/src/java/org/apache/cassandra/service/paxos/Paxos.java +++ b/src/java/org/apache/cassandra/service/paxos/Paxos.java @@ -472,7 +472,7 @@ public class Paxos static Participants get(ClusterMetadata metadata, TableMetadata table, Token token, ConsistencyLevel consistencyForConsensus, Predicate isReplicaAlive) { - AbstractReplicationStrategy strategy = Keyspace.open(table.keyspace).getReplicationStrategy(); + AbstractReplicationStrategy strategy = metadata.schema.getKeyspaceMetadata(table.keyspace).replicationStrategy; return strategy.paxosParticipants(metadata, table, token, consistencyForConsensus, isReplicaAlive); } diff --git a/test/unit/org/apache/cassandra/locator/CoordinationPlans.java b/test/unit/org/apache/cassandra/locator/CoordinationPlans.java index e0b9b97741..ecaa84e6dd 100644 --- a/test/unit/org/apache/cassandra/locator/CoordinationPlans.java +++ b/test/unit/org/apache/cassandra/locator/CoordinationPlans.java @@ -43,7 +43,7 @@ public class CoordinationPlans idealPlan = new CoordinationPlan.ForWrite(idealReplicaPlan, idealTracker); } - return new CoordinationPlan.ForWriteWithIdeal(ClusterMetadata.current(), plan, tracker, idealPlan); + return new CoordinationPlan.ForWriteWithIdeal(plan, tracker, idealPlan); } public static CoordinationPlan.ForTokenRead create(ReplicaPlan.ForTokenRead plan)