This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch cassandra-4.0
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/cassandra-4.0 by this push:
new 24dfcfe057 Ensure transferred_ranges reset on decommision re-attempt
when pending ranges cannot be proven continous
24dfcfe057 is described below
commit 24dfcfe057412db2379bff48ff3bf8d54311c52b
Author: Matt Byrd <[email protected]>
AuthorDate: Mon Aug 3 16:25:49 2026 +0100
Ensure transferred_ranges reset on decommision re-attempt when pending
ranges cannot be proven continous
Patch by Matt Byrd; reviewed by Caleb Rackliffe and Sam Tunnicliffe for
CASSANDRA-16290
---
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 | 150 +++++++++++++++++++++
5 files changed, 214 insertions(+), 6 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index a9a45d9ec5..5bbb1e16e0 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
4.0.22
+ * 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 --git a/src/java/org/apache/cassandra/db/SystemKeyspace.java
b/src/java/org/apache/cassandra/db/SystemKeyspace.java
index 56dd03a3c4..808e0bbc73 100644
--- a/src/java/org/apache/cassandra/db/SystemKeyspace.java
+++ b/src/java/org/apache/cassandra/db/SystemKeyspace.java
@@ -1477,6 +1477,25 @@ public final class SystemKeyspace
availableRanges.truncateBlockingWithoutSnapshot();
}
+ /**
+ * Wipes the whole table rather than deleting per (operation, keyspace):
keyspace_name is part of the
+ * partition key, so an operation-scoped delete would have to iterate
keyspaces. Safe because the
+ * decommission path is the only reader - see getTransferredRanges.
Revisit if another topology
+ * change starts consulting this table.
+ */
+ public static synchronized void resetTransferredRanges()
+ {
+
Keyspace.open(SchemaConstants.SYSTEM_KEYSPACE_NAME).getColumnFamilyStore(TRANSFERRED_RANGES_V2).truncateBlockingWithoutSnapshot();
+ try
+ {
+
Keyspace.open(SchemaConstants.SYSTEM_KEYSPACE_NAME).getColumnFamilyStore(LEGACY_TRANSFERRED_RANGES).truncateBlockingWithoutSnapshot();
+ }
+ catch (Throwable t)
+ {
+ logger.warn("failed to truncate system table {} but it should no
longer be being consulted", LEGACY_TRANSFERRED_RANGES, t);
+ }
+ }
+
public static synchronized void updateTransferredRanges(StreamOperation
streamOperation,
InetAddressAndPort
peer,
String keyspace,
@@ -1493,11 +1512,19 @@ public final class SystemKeyspace
executeInternal(String.format(cql, TRANSFERRED_RANGES_V2),
rangesToUpdate, streamOperation.getDescription(), peer.address, peer.port,
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 = ?";
- UntypedResultSet rs = executeInternal(String.format(query,
TRANSFERRED_RANGES_V2), description, keyspace);
+ UntypedResultSet rs = executeInternal(String.format(query,
TRANSFERRED_RANGES_V2), streamOperation.getDescription(), keyspace);
for (UntypedResultSet.Row row : rs)
{
InetAddress peerAddress = row.getInetAddress("peer");
diff --git a/src/java/org/apache/cassandra/service/StorageService.java
b/src/java/org/apache/cassandra/service/StorageService.java
index d3dab125e4..8bcc67dd31 100644
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@ -4649,6 +4649,32 @@ public class StorageService extends
NotificationBroadcasterSupport implements IE
{
PendingRangeCalculatorService.instance.blockUntilFinished();
+ // In 5.0 we need to handle DECOMMISSION_FAILED which can be
legitimately resumed
+ // hence operationMode != NORMAL clause chosen, and kept same on
4.0/4.1/5.0 for simplicity
+ // Checking OperationMode alone is insufficient since consider the
following sequence:
+ //
+ // 1. Run a decommission, get far enough to persist transferred
ranges, then fail, setting operationMode to DECOMMISSION_FAILED
+ // 2. Restart instance go NORMAL again, lose pending endpoints
+ // 3. Write comes in and is now not written to pending endpoint
for one of the transferred ranges
+ // 4. Attempt decommission which fails before
resetTransferredRanges, leaving operationMode as DECOMMISSION_FAILED
+ // 5. Re-attempt decommission which now sees operationMode ==
DECOMMISSION_FAILED
+ // (!= NORMAL) so skips resetting transferred ranges
+ // 6. decommission completes and write from step 3 has not been
transferred via streaming or write path to the new owner.
+ //
+ // So checking tokenMetadata.isLeaving in addition means that on
step 5 we now fail the isLeaving check
+ // (cleared by the restart in step 2, and step 4 never reached
startLeaving) and truncate transferred ranges
+ boolean resumingInFlightDecommission =
tokenMetadata.isLeaving(FBUtilities.getBroadcastAddressAndPort())
+ && operationMode !=
Mode.NORMAL;
+
+ // We reset transferred ranges upon starting a new decommission so
that we fully stream
+ // anything written since a previous attempt, which may not have
been persisted to a pending endpoint.
+ // See CASSANDRA-16290.
+ if (!resumingInFlightDecommission)
+ {
+ logger.info("resetting transferred ranges to force
re-streaming");
+ SystemKeyspace.resetTransferredRanges();
+ }
+
String dc =
DatabaseDescriptor.getEndpointSnitch().getLocalDatacenter();
if (operationMode != Mode.LEAVING) // If we're already
decommissioning there is no point checking RF/pending ranges
@@ -5647,8 +5673,7 @@ public class StorageService extends
NotificationBroadcasterSupport implements IE
if (rangesWithEndpoints.isEmpty())
continue;
- //Description is always Unbootstrap? Is that right?
- Map<InetAddressAndPort, Set<Range<Token>>>
transferredRangePerKeyspace = SystemKeyspace.getTransferredRanges("Unbootstrap",
+ Map<InetAddressAndPort, Set<Range<Token>>>
transferredRangePerKeyspace =
SystemKeyspace.getTransferredRanges(StreamOperation.DECOMMISSION,
keyspace,
StorageService.instance.getTokenMetadata().partitioner);
RangesByEndpoint.Builder replicasPerEndpoint = new
RangesByEndpoint.Builder();
@@ -5659,7 +5684,7 @@ public class StorageService extends
NotificationBroadcasterSupport implements IE
Set<Range<Token>> transferredRanges =
transferredRangePerKeyspace.get(remote.endpoint());
if (transferredRanges != null &&
transferredRanges.contains(local.range()))
{
- logger.debug("Skipping transferred range {} of keyspace
{}, endpoint {}", local, keyspace, remote);
+ logger.info("Skipping transferred range {} of keyspace {},
endpoint {}", local, keyspace, remote);
continue;
}
diff --git
a/test/distributed/org/apache/cassandra/distributed/test/ring/BootstrapTest.java
b/test/distributed/org/apache/cassandra/distributed/test/ring/BootstrapTest.java
index d6e715a009..e0c6a7e8d0 100644
---
a/test/distributed/org/apache/cassandra/distributed/test/ring/BootstrapTest.java
+++
b/test/distributed/org/apache/cassandra/distributed/test/ring/BootstrapTest.java
@@ -107,9 +107,14 @@ public class BootstrapTest extends TestBaseImpl
}
public static void populate(ICluster cluster, int from, int to, int coord,
int rf, ConsistencyLevel cl)
+ {
+ populate(cluster, from, to, coord, rf, cl, "pk int, ck int, v int");
+ }
+
+ public static void populate(ICluster cluster, int from, int to, int coord,
int rf, ConsistencyLevel cl, String columnDefinitions)
{
cluster.schemaChange("CREATE KEYSPACE IF NOT EXISTS " + KEYSPACE + "
WITH replication = {'class': 'SimpleStrategy', 'replication_factor': " + rf +
"};");
- cluster.schemaChange("CREATE TABLE IF NOT EXISTS " + KEYSPACE + ".tbl
(pk int, ck int, v int, PRIMARY KEY (pk, ck))");
+ cluster.schemaChange("CREATE TABLE IF NOT EXISTS " + KEYSPACE + ".tbl
(" + columnDefinitions + ", PRIMARY KEY (pk, ck))");
for (int i = from; i < to; i++)
{
cluster.coordinator(coord).execute("INSERT INTO " + KEYSPACE +
".tbl (pk, ck, v) VALUES (?, ?, ?)",
diff --git
a/test/distributed/org/apache/cassandra/distributed/test/ring/DecommissionTest.java
b/test/distributed/org/apache/cassandra/distributed/test/ring/DecommissionTest.java
new file mode 100644
index 0000000000..1a913d1b74
--- /dev/null
+++
b/test/distributed/org/apache/cassandra/distributed/test/ring/DecommissionTest.java
@@ -0,0 +1,150 @@
+/*
+ * 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 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 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))
+ 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))));
+ 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"));
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
