Skip to content

CASSANDRA-21585: Avoid "During operation execution,... at ONE the ring has changed fro… #5036

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
krummas wants to merge 2 commits into
base: trunk
Choose a base branch
from
Open

CASSANDRA-21585: Avoid "During operation execution,... at ONE the ring has changed fro… #5036

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3d431b9
Avoid "During operation execution,... at ONE the ring has changed fro…
krummas Aug 19, 2026
29dfa4c
test fix, avoid triggering snapshots
krummas Aug 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 23 additions & 11 deletions src/java/org/apache/cassandra/locator/ReplicaPlans.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,7 @@
import org.apache.cassandra.tcm.membership.Location;
import org.apache.cassandra.tcm.membership.NodeState;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;

import static com.google.common.collect.Iterables.any;
import static com.google.common.collect.Iterables.filter;
Expand Down Expand Up @@ -207,13 +208,14 @@ static void assureSufficientLiveReplicas(Locator locator, AbstractReplicationStr
/**
* Construct a ReplicaPlan for writing to exactly one node, with CL.ONE. This node is *assumed* to be alive.
*/
public static ReplicaPlan.ForWrite forSingleReplicaWrite(ClusterMetadata metadata, Keyspace keyspace, Token token, Function<ClusterMetadata, Replica> replicaSupplier)
public static ReplicaPlan.ForWrite forSingleReplicaWrite(ClusterMetadata metadata, Keyspace keyspace, Token token, Function<ClusterMetadata, Pair<Replica, EndpointsForToken>> replicasSupplier)
{
EndpointsForToken one = EndpointsForToken.of(token, replicaSupplier.apply(metadata));
Pair<Replica, EndpointsForToken> replicas = replicasSupplier.apply(metadata);
EndpointsForToken one = EndpointsForToken.of(token, replicas.left);
EndpointsForToken empty = EndpointsForToken.empty(token);

return new ReplicaPlan.ForWrite(keyspace, keyspace.getReplicationStrategy(), ConsistencyLevel.ONE, empty, one, one, one,
(newClusterMetadata) -> forSingleReplicaWrite(newClusterMetadata, keyspace, token, replicaSupplier),
return new ReplicaPlan.ForWrite(keyspace, keyspace.getReplicationStrategy(), ConsistencyLevel.ONE, empty, replicas.right, one, one,
(newClusterMetadata) -> forSingleReplicaWrite(newClusterMetadata, keyspace, token, replicasSupplier),
metadata.epoch);
}

Expand All @@ -228,16 +230,16 @@ public static ReplicaPlan.ForWrite forSingleReplicaWrite(ClusterMetadata metadat
* is unclear we want to mix those latencies with read latencies, so this
* may be a bit involved.
*/
public static Replica findCounterLeaderReplica(ClusterMetadata metadata, String keyspaceName, DecoratedKey key, String localDataCenter, ConsistencyLevel cl) throws UnavailableException
public static Pair<Replica, EndpointsForToken> findCounterLeaderReplica(ClusterMetadata metadata, String keyspaceName, DecoratedKey key, String localDataCenter, ConsistencyLevel cl) throws UnavailableException
{
Keyspace keyspace = Keyspace.open(keyspaceName);
NodeProximity proximity = DatabaseDescriptor.getNodeProximity();
AbstractReplicationStrategy replicationStrategy = keyspace.getReplicationStrategy();

EndpointsForToken replicas = metadata.placement(keyspace.getMetadata().params.replication).reads.forToken(key.getToken()).get();
EndpointsForToken allReplicas = metadata.placement(keyspace.getMetadata().params.replication).reads.forToken(key.getToken()).get();

// CASSANDRA-13043: filter out those endpoints not accepting clients yet, maybe because still bootstrapping
replicas = replicas.filter(replica -> StorageService.instance.isRpcReady(replica.endpoint()));
EndpointsForToken replicas = allReplicas.filter(replica -> StorageService.instance.isRpcReady(replica.endpoint()));

// CASSANDRA-17411: filter out endpoints that are not alive
replicas = replicas.filter(replica -> FailureDetector.instance.isAlive(replica.endpoint()));
Expand All @@ -260,19 +262,19 @@ public static Replica findCounterLeaderReplica(ClusterMetadata metadata, String

// No endpoint in local DC, pick the closest endpoint according to the configured proximity measures
replicas = proximity.sortedByProximity(FBUtilities.getBroadcastAddressAndPort(), replicas);
return replicas.get(0);
return Pair.create(replicas.get(0), allReplicas);
}

return localReplicas.get(ThreadLocalRandom.current().nextInt(localReplicas.size()));
return Pair.create(localReplicas.get(ThreadLocalRandom.current().nextInt(localReplicas.size())), allReplicas);
}

/**
* A forwarding counter write is always sent to a single owning coordinator for the range, by the original coordinator
* (if it is not itself an owner)
*/
public static ReplicaPlan.ForWrite forForwardingCounterWrite(ClusterMetadata metadata, Keyspace keyspace, Token token, Function<ClusterMetadata, Replica> replica)
public static ReplicaPlan.ForWrite forForwardingCounterWrite(ClusterMetadata metadata, Keyspace keyspace, Token token, Function<ClusterMetadata, Pair<Replica, EndpointsForToken>> replicas)
{
return forSingleReplicaWrite(metadata, keyspace, token, replica);
return forSingleReplicaWrite(metadata, keyspace, token, replicas);
}

public static ReplicaPlan.ForWrite forLocalBatchlogWrite()
Expand Down Expand Up @@ -840,6 +842,11 @@ public static ReplicaPlan.ForTokenRead forSingleReplicaRead(Keyspace keyspace, T
return forSingleReplicaRead(ClusterMetadata.current(), keyspace, token, replica);
}

/**
* Create a plan for reading from a specific single replica
*
* Recompute doesn't actually recompute the replica - this means that `AbstractForRead#stillAppliesTo` will always succeed
*/
private static ReplicaPlan.ForTokenRead forSingleReplicaRead(ClusterMetadata metadata, Keyspace keyspace, Token token, Replica replica)
{
EndpointsForToken one = EndpointsForToken.of(token, replica);
Expand All @@ -860,6 +867,11 @@ public static ReplicaPlan.ForRangeRead forSingleReplicaRead(Keyspace keyspace, A
return forSingleReplicaRead(ClusterMetadata.current(), keyspace, range, replica, vnodeCount);
}

/**
* Create a plan for reading from a specific single replica
*
* Recompute doesn't actually recompute the replica - this means that `AbstractForRead#stillAppliesTo` will always succeed
*/
private static ReplicaPlan.ForRangeRead forSingleReplicaRead(ClusterMetadata metadata, Keyspace keyspace, AbstractBounds<PartitionPosition> range, Replica replica, int vnodeCount)
{
// TODO: this is unsafe, as one.range() may be inconsistent with our supplied range; should refactor Range/AbstractBounds to single class
Expand Down
8 changes: 4 additions & 4 deletions src/java/org/apache/cassandra/service/StorageProxy.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2083,9 +2083,9 @@ protected Verb verb()
public static AbstractWriteResponseHandler<IMutation> mutateCounter(CounterMutation cm, String localDataCenter, Dispatcher.RequestTime requestTime) throws UnavailableException, OverloadedException
{
ClusterMetadata metadata = ClusterMetadata.current();
Replica replica = ReplicaPlans.findCounterLeaderReplica(metadata, cm.getKeyspaceName(), cm.key(), localDataCenter, cm.consistency());
Pair<Replica, EndpointsForToken> replicas = ReplicaPlans.findCounterLeaderReplica(metadata, cm.getKeyspaceName(), cm.key(), localDataCenter, cm.consistency());

if (replica.isSelf())
if (replicas.left.isSelf())
{
return applyCounterMutationOnCoordinator(cm, localDataCenter, requestTime);
}
Expand All @@ -2110,9 +2110,9 @@ public static AbstractWriteResponseHandler<IMutation> mutateCounter(CounterMutat
AbstractWriteResponseHandler<IMutation> responseHandler = new WriteResponseHandler<>(forWrite,
WriteType.COUNTER, null, requestTime);

Tracing.trace("Enqueuing counter update to {}", replica);
Tracing.trace("Enqueuing counter update to {}", replicas.left);
Message message = Message.outWithFlag(Verb.COUNTER_MUTATION_REQ, cm, MessageFlag.CALL_BACK_ON_FAILURE);
MessagingService.instance().sendWriteWithCallback(message, replica, responseHandler);
MessagingService.instance().sendWriteWithCallback(message, replicas.left, responseHandler);
return responseHandler;
}
}
Expand Down
18 changes: 18 additions & 0 deletions test/distributed/org/apache/cassandra/distributed/test/tcm/FailureDetectorRecomputeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@

import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.test.TestBaseImpl;
import org.apache.cassandra.gms.FailureDetector;
import org.apache.cassandra.locator.InetAddressAndPort;
Expand All @@ -49,6 +50,7 @@ public void readTest() throws IOException
{
try (Cluster cluster = init(Cluster.build(3)
.withInstanceInitializer(BB::install)
.withConfig(c -> c.set("metadata_snapshot_frequency", "10000"))
.start()))
{
cluster.schemaChange(withKeyspace("create table %s.tbl (id int primary key)"));
Expand All @@ -63,6 +65,7 @@ public void writeTest() throws IOException
{
try (Cluster cluster = init(Cluster.build(3)
.withInstanceInitializer(BB::install)
.withConfig(c -> c.set("metadata_snapshot_frequency", "10000"))
.start()))
{
cluster.schemaChange(withKeyspace("create table %s.tbl (id int primary key)"));
Expand All @@ -72,6 +75,21 @@ public void writeTest() throws IOException
}
}

@Test
public void counterWriteOneTest() throws IOException
{
try (Cluster cluster = init(Cluster.build(3)
.withInstanceInitializer(BB::install)
.withConfig(c -> c.with(Feature.values()).set("metadata_snapshot_frequency", "10000"))
.start()))
{
cluster.schemaChange(withKeyspace("create table %s.tbl (id int primary key, c counter)"));
cluster.get(1).runOnInstance(() -> BB.enabled.set(true));
for (int i = 0; i < 10; i++)
cluster.coordinator(1).execute(withKeyspace("update %s.tbl set c = c + 1 where id = ?"), ConsistencyLevel.ONE, i);
}
}

public static class BB
{
public static AtomicBoolean enabled = new AtomicBoolean();
Expand Down