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

feedback

This commit is contained in:
Blake Eggleston 2026-06-17 15:14:49 -07:00
parent f4d9c4c3a3
commit 9bd69f2fec
2 changed files with 16 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -463,10 +463,10 @@ public abstract class AbstractReplicationStrategy
}
protected CoordinationPlan.ForWrite planForWriteInternal(ClusterMetadata metadata,
Keyspace keyspace,
ConsistencyLevel consistencyLevel,
Function<ClusterMetadata, ReplicaLayout.ForTokenWrite> liveAndDown,
ReplicaPlans.Selector selector)
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);
@ -651,7 +651,7 @@ public abstract class AbstractReplicationStrategy
/**
* Create ResponseTracker for read operation.
*/
public <E extends Endpoints<E>, P extends ReplicaPlan.ForRead<E, P>> ResponseTracker createTrackerForRead(P plan)
private <E extends Endpoints<E>, P extends ReplicaPlan.ForRead<E, P>> ResponseTracker createTrackerForRead(P plan)
{
int blockFor = plan.readQuorum();

24
src/java/org/apache/cassandra/service/paxos/PaxosCommit.java
View File

@ -461,19 +461,6 @@ public class PaxosCommit<OnDone extends Consumer<? super PaxosCommit.Status>> ex
boolean localExecutedSynchronously = false;
InetAddressAndPort localEndpoint = FBUtilities.getBroadcastAddressAndPort();
// Set up additional commit work from the replication strategy (e.g., satellite writes for SRS).
// This needs to happen before executeOnSelf() can trigger onPaxosDecision(), so that
// additionalCommitFuture is set before it's read. The base strategy returns an
// already-completed future, so this is a no-op for non-SRS keyspaces.
// For SRS, satellite messages are sent here (in parallel with local execution below).
// MutationTrackingService.retryFailedWrite for down satellite endpoints schedules async retries,
// which will find the mutation in the journal after executeOnSelf() completes below.
if (isTrackedKeyspace)
{
AbstractReplicationStrategy strategy = Keyspace.open(commit.metadata().keyspace).getReplicationStrategy();
setAugmentedCommitFuture(strategy.sendPaxosCommitMutations(commit, isUrgent));
}
if (isTrackedKeyspace)
{
// For tracked keyspaces, we MUST execute locally synchronously, regardless of USE_SELF_EXECUTION setting.
@ -504,11 +491,22 @@ public class PaxosCommit<OnDone extends Consumer<? super PaxosCommit.Status>> ex
}
}
}
if (localIsReplica)
{
executeOnSelf();
localExecutedSynchronously = true;
}
// Set up additional commit work from the replication strategy (e.g., satellite writes for SRS).
// This needs to happen before executeOnSelf() can trigger onPaxosDecision(), so that
// additionalCommitFuture is set before it's read. The base strategy returns an
// already-completed future, so this is a no-op for non-SRS keyspaces.
// For SRS, satellite messages are sent here (in parallel with local execution below).
// MutationTrackingService.retryFailedWrite for down satellite endpoints schedules async retries,
// which will find the mutation in the journal after executeOnSelf() completes below.
AbstractReplicationStrategy strategy = Keyspace.open(commit.metadata().keyspace).getReplicationStrategy();
setAugmentedCommitFuture(strategy.sendPaxosCommitMutations(commit, isUrgent));
}
// Now send to remote replicas in the electorate (and record local execution for non-tracked keyspaces)