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

ResponseTracker feedback

This commit is contained in:
Blake Eggleston 2026-07-10 14:45:15 -07:00
parent ab05667ba8
commit 58e96cec11
7 changed files with 52 additions and 67 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/java/org/apache/cassandra/batchlog/BatchlogManager.java
View File

@ -630,7 +630,7 @@ public class BatchlogManager implements BatchlogManagerMBean
}
}
ReplayWriteResponseHandler<Mutation> handler = new ReplayWriteResponseHandler<>(replayPlan, () -> mutation, Dispatcher.RequestTime.forImmediateExecution());
ReplayWriteResponseHandler<Mutation> handler = new ReplayWriteResponseHandler<>(replayPlan, mutation, Dispatcher.RequestTime.forImmediateExecution());
Message<Mutation> message = Message.outWithFlag(MUTATION_REQ, mutation, MessageFlag.CALL_BACK_ON_FAILURE);
for (Replica replica : replayPlan.replicas().liveAndDown())
MessagingService.instance().sendWriteWithCallback(message, replica, handler);

82
src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java
View File

@ -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);

20
src/java/org/apache/cassandra/locator/CoordinationPlan.java
View File

@ -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
* <ul>
* <li>ReplicaPlan to focus on replica topology and selection</li>
* <li>ResponseTracker to encapsulate completion logic</li>
* <li>Replication strategies to customize both consistently</li>
* </ul>
*
* 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<E extends Endpoints<E>, 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<E extends Endpoints<E>, 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<E extends Endpoints<E>, 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);
}
}

7
src/java/org/apache/cassandra/locator/ReplicaPlans.java
View File

@ -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);
}

4
src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java
View File

@ -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<T> extends AbstractWriteResponseHandler<T>
{
private static final Locator locator = DatabaseDescriptor.getLocator();
public DatacenterSyncWriteResponseHandler(CoordinationPlan.ForWrite coordinationPlan,
Runnable callback,
WriteType writeType,

2
src/java/org/apache/cassandra/service/paxos/Paxos.java
View File

@ -472,7 +472,7 @@ public class Paxos
static Participants get(ClusterMetadata metadata, TableMetadata table, Token token, ConsistencyLevel consistencyForConsensus, Predicate<Replica> isReplicaAlive)
{
AbstractReplicationStrategy strategy = Keyspace.open(table.keyspace).getReplicationStrategy();
AbstractReplicationStrategy strategy = metadata.schema.getKeyspaceMetadata(table.keyspace).replicationStrategy;
return strategy.paxosParticipants(metadata, table, token, consistencyForConsensus, isReplicaAlive);
}

2
test/unit/org/apache/cassandra/locator/CoordinationPlans.java
View File

@ -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)