This is an automated email from the ASF dual-hosted git repository. asf-gitbox-commits pushed a commit to branch cassandra-6.0 in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit a082093c788c51e20f17b328d68b7b22d848b094 Author: Marcus Eriksson <[email protected]> AuthorDate: Fri Sep 4 15:29:04 2026 +0200 Don't replicate transformations to LEFT nodes Patch by marcuse; reviewed by Sam Tunnicliffe for CASSANDRA-21652 --- CHANGES.txt | 1 + src/java/org/apache/cassandra/tcm/Commit.java | 12 ++++++++++-- .../distributed/test/ring/DecommissionTest.java | 16 ++++++++++++++++ .../tcm/listeners/PlacementsChangeListenerTest.java | 4 ++-- 4 files changed, 29 insertions(+), 4 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 2ea5846d1d..d07a5d2158 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 6.0-alpha3 + * Don't replicate transformations to LEFT nodes (CASSANDRA-21652) * Don't fail host specific repairs against system_cluster_metadata (CASSANDRA-21651) * During major upgrade, if a node has replacement arguments it attempts to bootstrap even if already bootstrapped (CASSANDRA-21634) * Node with empty data directory is allowed to rejoin the ring on the second start after the first start is correctly rejected (CASSANDRA-21633) diff --git a/src/java/org/apache/cassandra/tcm/Commit.java b/src/java/org/apache/cassandra/tcm/Commit.java index 847cd22a7c..dab4f05cde 100644 --- a/src/java/org/apache/cassandra/tcm/Commit.java +++ b/src/java/org/apache/cassandra/tcm/Commit.java @@ -48,6 +48,7 @@ import org.apache.cassandra.tcm.log.LogState; import org.apache.cassandra.tcm.membership.Directory; import org.apache.cassandra.tcm.membership.EndpointLookup; import org.apache.cassandra.tcm.membership.NodeId; +import org.apache.cassandra.tcm.membership.NodeState; import org.apache.cassandra.tcm.membership.NodeVersion; import org.apache.cassandra.tcm.serialization.Version; import org.apache.cassandra.utils.FBUtilities; @@ -461,6 +462,11 @@ public class Commit { return endpoints.endpoint(id); }; + + boolean hasLeft(NodeId id) + { + return directory.peerState(id) == NodeState.LEFT; + } } private final Supplier<RoutingHelper> routingSupplier; @@ -490,10 +496,12 @@ public class Commit { InetAddressAndPort endpoint = routing.endpoint(peerId); boolean upgraded = routing.isUpgraded(peerId); - // Do not replicate to self and to the peer that has requested to commit this message + boolean hasLeft = routing.hasLeft(peerId); + // Do not replicate to self, to any left nodes and to the peer that has requested to commit this message if (endpoint.equals(FBUtilities.getBroadcastAddressAndPort()) || (source != null && source.equals(endpoint)) || - !upgraded) + !upgraded || + hasLeft) { continue; } diff --git a/test/distributed/org/apache/cassandra/distributed/test/ring/DecommissionTest.java b/test/distributed/org/apache/cassandra/distributed/test/ring/DecommissionTest.java index d4277173dc..06bd4a1881 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/ring/DecommissionTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/ring/DecommissionTest.java @@ -27,6 +27,7 @@ import java.util.Set; import java.util.concurrent.Callable; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.stream.Collectors; @@ -438,5 +439,20 @@ public class DecommissionTest extends TestBaseImpl } } + @Test + public void testDontReplicateToLeftNodes() throws IOException, TimeoutException + { + try (Cluster cluster = init(builder().withNodes(3) + .withConfig(config -> config.with(NETWORK, GOSSIP)) + .start())) + { + cluster.get(2).nodetoolResult("decommission", "--force").asserts().success(); + long mark = cluster.get(1).logs().mark(); + cluster.coordinator(1).execute(withKeyspace("create table %s.tbl (id int primary key)"), ConsistencyLevel.ONE); + cluster.get(3).logs().watchFor("Enacted.*CreateTableStatement.*tbl"); + assertTrue(cluster.get(1).logs().grep(mark, "Replicating newly committed transformations up to.*127.0.0.2.*").getResult().isEmpty()); + } + } + } diff --git a/test/unit/org/apache/cassandra/tcm/listeners/PlacementsChangeListenerTest.java b/test/unit/org/apache/cassandra/tcm/listeners/PlacementsChangeListenerTest.java index e8429e5bb7..f2c38d619d 100644 --- a/test/unit/org/apache/cassandra/tcm/listeners/PlacementsChangeListenerTest.java +++ b/test/unit/org/apache/cassandra/tcm/listeners/PlacementsChangeListenerTest.java @@ -39,7 +39,6 @@ import org.apache.cassandra.tcm.CMSMembership; import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.tcm.Epoch; import org.apache.cassandra.tcm.membership.Directory; -import org.apache.cassandra.tcm.membership.MembershipUtils; import org.apache.cassandra.tcm.membership.NodeAddresses; import org.apache.cassandra.tcm.membership.NodeId; import org.apache.cassandra.tcm.ownership.DataPlacement; @@ -81,7 +80,8 @@ public class PlacementsChangeListenerTest DataPlacements.Builder builder = before.unbuild(); before.forEach((params, placement) -> { Replica remove = placement.writes.byEndpoint().flattenValues().iterator().next(); - Replica add = Replica.fullReplica(MembershipUtils.endpoint(99), remove.range()); + // randomPlacements gives 127.0.0.{1-255} ip addresses, need to add a non-conflicting one here: + Replica add = Replica.fullReplica(InetAddressAndPort.getByNameUnchecked("127.0.1.99"), remove.range()); DataPlacement newPlacement = placement.unbuild() .withoutWriteReplica(e.nextEpoch(), remove) .withWriteReplica(e.nextEpoch(), add).build(); --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
