This is an automated email from the ASF dual-hosted git repository. asf-gitbox-commits pushed a commit to branch cassandra-4.1 in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 8fa203de28340d0c1870946ea5eddca3157127a8 Merge: 91b968eddc 24dfcfe057 Author: Caleb Rackliffe <[email protected]> AuthorDate: Thu Aug 13 16:44:12 2026 -0500 Merge branch 'cassandra-4.0' into cassandra-4.1 * cassandra-4.0: Ensure transferred_ranges reset on decommision re-attempt when pending ranges cannot be proven continous CHANGES.txt | 1 + .../org/apache/cassandra/db/SystemKeyspace.java | 31 ++++- .../apache/cassandra/service/StorageService.java | 31 ++++- .../distributed/test/ring/BootstrapTest.java | 7 +- .../distributed/test/ring/DecommissionTest.java | 143 +++++++++++++++++++++ 5 files changed, 207 insertions(+), 6 deletions(-) diff --cc CHANGES.txt index 93201a663f,5bbb1e16e0..96ebeb649d --- a/CHANGES.txt +++ b/CHANGES.txt @@@ -1,5 -1,5 +1,6 @@@ -4.0.22 +4.1.13 +Merged from 4.0: + * Ensure transferred_ranges reset on decommision re-attempt when pending ranges cannot be proven continous (CASSANDRA-16290) * Add validation to uncompressed length during decompression (CASSANDRA-21567) * Fix regression in PasswordObfuscator for dollar-quoted passwords (CASSANDRA-21559) * Do not make DNS lookup when querying system_views.clients for hostname column by removing it (CASSANDRA-21539) diff --cc src/java/org/apache/cassandra/db/SystemKeyspace.java index 4b8cd29f0f,808e0bbc73..bcd827e41a --- a/src/java/org/apache/cassandra/db/SystemKeyspace.java +++ b/src/java/org/apache/cassandra/db/SystemKeyspace.java @@@ -1713,12 -1507,20 +1732,20 @@@ public final class SystemKeyspac { rangesToUpdate.add(rangeToBytes(range)); } - executeInternal(format(cql, LEGACY_TRANSFERRED_RANGES), rangesToUpdate, streamOperation.getDescription(), peer.address, keyspace); + executeInternal(format(cql, LEGACY_TRANSFERRED_RANGES), rangesToUpdate, streamOperation.getDescription(), peer.getAddress(), keyspace); cql = "UPDATE system.%s SET ranges = ranges + ? WHERE operation = ? AND peer = ? AND peer_port = ? AND keyspace_name = ?"; - executeInternal(String.format(cql, TRANSFERRED_RANGES_V2), rangesToUpdate, streamOperation.getDescription(), peer.address, peer.port, keyspace); + executeInternal(String.format(cql, TRANSFERRED_RANGES_V2), rangesToUpdate, streamOperation.getDescription(), peer.getAddress(), peer.getPort(), keyspace); } - public static synchronized Map<InetAddressAndPort, Set<Range<Token>>> getTransferredRanges(String description, String keyspace, IPartitioner partitioner) + // Only consulted on the decommission path, where the leaving node is the only streamer and its + // local transferred_ranges_v2 is therefore a complete record of what has already moved. + // + // Being node-local rules out the other topology changes. Under removenode the surviving replicas + // stream, and the node running it need not be one of them, so its local table may record nothing; + // even when it is a replica it can only account for its own streams. Move could use it for the + // ranges the moving node gives up, but the move path registers no listener so nothing is recorded, + // and it would still miss the ranges moving the other way. + public static synchronized Map<InetAddressAndPort, Set<Range<Token>>> getTransferredRanges(StreamOperation streamOperation, String keyspace, IPartitioner partitioner) { Map<InetAddressAndPort, Set<Range<Token>>> result = new HashMap<>(); String query = "SELECT * FROM system.%s WHERE operation = ? AND keyspace_name = ?"; diff --cc test/distributed/org/apache/cassandra/distributed/test/ring/DecommissionTest.java index 0000000000,1a913d1b74..318e9c6ba0 mode 000000,100644..100644 --- a/test/distributed/org/apache/cassandra/distributed/test/ring/DecommissionTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/ring/DecommissionTest.java @@@ -1,0 -1,150 +1,143 @@@ + /* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + + package org.apache.cassandra.distributed.test.ring; + + import java.io.IOException; + import java.nio.ByteBuffer; + import java.util.Arrays; -import java.util.Collections; + import java.util.List; + import java.util.concurrent.Callable; -import java.util.concurrent.Future; -import java.util.concurrent.atomic.AtomicBoolean; - -import com.google.common.util.concurrent.Futures; ++import java.util.stream.Collectors; + + import net.bytebuddy.ByteBuddy; + import net.bytebuddy.dynamic.loading.ClassLoadingStrategy; + import net.bytebuddy.implementation.MethodDelegation; + import net.bytebuddy.implementation.bind.annotation.SuperCall; + + import org.junit.Test; + + import org.apache.cassandra.dht.Murmur3Partitioner; + import org.apache.cassandra.distributed.Cluster; + import org.apache.cassandra.distributed.api.ConsistencyLevel; + import org.apache.cassandra.distributed.api.IInvokableInstance; + import org.apache.cassandra.distributed.shared.ClusterUtils; + import org.apache.cassandra.distributed.test.TestBaseImpl; + import org.apache.cassandra.service.StorageService; ++import org.apache.cassandra.utils.concurrent.Future; ++import org.apache.cassandra.utils.concurrent.ImmediateFuture; + + import static net.bytebuddy.matcher.ElementMatchers.named; + import static org.apache.cassandra.db.SystemKeyspace.TRANSFERRED_RANGES_V2; + import static org.apache.cassandra.distributed.api.Feature.GOSSIP; + import static org.apache.cassandra.distributed.api.Feature.NETWORK; + import static org.apache.cassandra.distributed.test.ring.BootstrapTest.populate; + import static org.junit.Assert.assertEquals; + import static org.junit.Assert.assertTrue; + + public class DecommissionTest extends TestBaseImpl + { + @Test + public void testAbortingDecommissionRestreams() throws Exception + { + // https://issues.apache.org/jira/browse/CASSANDRA-16290 + // We demonstrate here that decommissioning and then aborting decommission is unsafe + // if we've persisted transferred ranges and then skip them for something which was delivered after we aborted the decommission but before we resumed - - // Only install on the first classloader handed out for node2: a restarted instance gets a - // fresh classloader, so any static "already failed once" state would be reset and the - // resumed decommission would fail again. On this branch the initializer is a - // BiConsumer<ClassLoader, Integer>, so there is no generation argument to key off. - AtomicBoolean streamHintsInstalled = new AtomicBoolean(); - + try (Cluster cluster = builder().withNodes(4) + .withConfig(config -> config.with(NETWORK, GOSSIP) + // disable hints to simplify test + .set("hinted_handoff_enabled", false) + ) - .withInstanceInitializer((cl, num) -> { - if (num == 2 && streamHintsInstalled.compareAndSet(false, true)) ++ // only install on the first generation of node2: a restarted instance gets a ++ // fresh classloader, so any static "already failed once" state would be reset ++ // and the resumed decommission would fail again ++ .withInstanceInitializer((cl, threadGroup, num, generation) -> { ++ if (num == 2 && generation == 0) + BB.streamHintsInstall(cl); + }) + .start()) + { + // We need blob columns here so later we can do Murmur3Partitioner.LongToken.keyForToken(token); + populate(cluster, 0, 100, 1, 2, ConsistencyLevel.QUORUM, "pk blob, ck blob, v blob"); + + IInvokableInstance leavingNode = cluster.get(2); + + leavingNode.nodetoolResult("decommission").asserts().failure(); + + // abort the decommission + ClusterUtils.stopUnchecked(leavingNode); + ClusterUtils.start(leavingNode, props -> {}); + ClusterUtils.awaitRingHealthy(leavingNode); + + // Stop the non leaving nodes so we can write at ONE and fail to stream that datum + ClusterUtils.stopUnchecked(cluster.get(1)); + ClusterUtils.stopUnchecked(cluster.get(3)); + ClusterUtils.stopUnchecked(cluster.get(4)); + - // Mirroring the upstream getLocalTokens which we don't have here - List<Murmur3Partitioner.LongToken> tokens = Collections.singletonList(new Murmur3Partitioner.LongToken(Long.parseLong(ClusterUtils.getLocalToken(leavingNode)))); ++ List<Murmur3Partitioner.LongToken> tokens = ClusterUtils.getLocalTokens(leavingNode).stream().map(t -> new Murmur3Partitioner.LongToken(Long.parseLong(t))).collect(Collectors.toList()); + for (Murmur3Partitioner.LongToken token : tokens) + { + ByteBuffer key = Murmur3Partitioner.LongToken.keyForToken(token); + leavingNode.coordinator().execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (?, ?, ?)", ConsistencyLevel.ONE, key, key, key); + } + + ClusterUtils.start(cluster.get(1), props -> {}); + ClusterUtils.start(cluster.get(3), props -> {}); + ClusterUtils.start(cluster.get(4), props -> {}); + + ClusterUtils.awaitRingHealthy(leavingNode); + + Object[][] ranges = leavingNode.executeInternal("SELECT keyspace_name from system." + TRANSFERRED_RANGES_V2); + + assertTrue("transferred ranges missing entirely", ranges.length > 0); + assertTrue("transferred ranges present for keyspace", Arrays.stream(ranges).anyMatch(x -> x[0].equals(KEYSPACE))); + + // Resume decomm + leavingNode.nodetoolResult("decommission").asserts().success(); + + // Try and read data we wrote at ONE at ALL + for (Murmur3Partitioner.LongToken token : tokens) + { + ByteBuffer key = Murmur3Partitioner.LongToken.keyForToken(token); + Object[][] resp = cluster.get(1).coordinator().execute("SELECT pk from " + KEYSPACE + ".tbl where pk=?", ConsistencyLevel.ALL, key); + assertTrue("We should get a response for this key we wrote it at ONE", resp.length > 0); + assertEquals(key, resp[0][0]); + } + } + } + + public static class BB + { + static void streamHintsInstall(ClassLoader cl) + { + new ByteBuddy().rebase(StorageService.class) + .method(named("streamHints")) + .intercept(MethodDelegation.to(BB.class)) + .make() + .load(cl, ClassLoadingStrategy.Default.INJECTION); + } + + @SuppressWarnings({ "unused", "rawtypes" }) + public static Future streamHints(@SuperCall Callable<Future> zuper) + { + // this is only installed on the first startup of the leaving node, so every invocation + // here belongs to the decommission attempt we want to fail at the last moment possible - return Futures.immediateFailedFuture(new IOException("failing hints so that decomm fails at last moment possible")); ++ return ImmediateFuture.failure(new IOException("failing hints so that decomm fails at last moment possible")); + } + } + } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
