diff --git a/src/java/org/apache/cassandra/db/SystemKeyspace.java b/src/java/org/apache/cassandra/db/SystemKeyspace.java index a7318fb2dee2..d2ed7b5aad95 100644 --- a/src/java/org/apache/cassandra/db/SystemKeyspace.java +++ b/src/java/org/apache/cassandra/db/SystemKeyspace.java @@ -1726,6 +1726,18 @@ public static void resetAvailableStreamedRangesForKeyspace(String keyspace) executeInternal(format(cql, SchemaConstants.SYSTEM_KEYSPACE_NAME, AVAILABLE_RANGES_V2), keyspace); } + /** + * 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() + { + for (String table : Arrays.asList(LEGACY_TRANSFERRED_RANGES, TRANSFERRED_RANGES_V2)) + Keyspace.open(SchemaConstants.SYSTEM_KEYSPACE_NAME).getColumnFamilyStore(table).truncateBlockingWithoutSnapshot(); + } + public static synchronized void updateTransferredRanges(StreamOperation streamOperation, InetAddressAndPort peer, String keyspace, @@ -1742,11 +1754,19 @@ public static synchronized void updateTransferredRanges(StreamOperation streamOp executeInternal(String.format(cql, TRANSFERRED_RANGES_V2), rangesToUpdate, streamOperation.getDescription(), peer.getAddress(), peer.getPort(), keyspace); } - public static synchronized Map>> 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>> getTransferredRanges(StreamOperation streamOperation, String keyspace, IPartitioner partitioner) { Map>> 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 22ef542f2ae0..9d8a13f12573 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -278,6 +278,7 @@ import static org.apache.cassandra.service.StorageService.Mode.DECOMMISSIONED; import static org.apache.cassandra.service.StorageService.Mode.DECOMMISSION_FAILED; import static org.apache.cassandra.service.StorageService.Mode.JOINING_FAILED; +import static org.apache.cassandra.service.StorageService.Mode.NORMAL; import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis; import static org.apache.cassandra.utils.Clock.Global.nanoTime; import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort; @@ -5331,6 +5332,19 @@ public void decommission(boolean force) throws InterruptedException { PendingRangeCalculatorService.instance.blockUntilFinished(); + // This check needs to happen above startLeaving + boolean resumingInFlightDecommission = tokenMetadata.isLeaving(FBUtilities.getBroadcastAddressAndPort()) + && operationMode != 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 we're already decommissioning there is no point checking RF/pending ranges @@ -6385,8 +6399,7 @@ private Future streamRanges(Map rangesT if (rangesWithEndpoints.isEmpty()) continue; - //Description is always Unbootstrap? Is that right? - Map>> transferredRangePerKeyspace = SystemKeyspace.getTransferredRanges("Unbootstrap", + Map>> transferredRangePerKeyspace = SystemKeyspace.getTransferredRanges(StreamOperation.DECOMMISSION, keyspace, StorageService.instance.getTokenMetadata().partitioner); RangesByEndpoint.Builder replicasPerEndpoint = new RangesByEndpoint.Builder(); @@ -6397,7 +6410,7 @@ private Future streamRanges(Map rangesT Set> 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/DecommissionTest.java b/test/distributed/org/apache/cassandra/distributed/test/DecommissionTest.java index 66091da3ef9e..a81cf6475e9b 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/DecommissionTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/DecommissionTest.java @@ -18,9 +18,18 @@ package org.apache.cassandra.distributed.test; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Arrays; +import java.util.List; import java.util.concurrent.Callable; import java.util.function.Supplier; +import java.util.stream.Collectors; +import org.apache.cassandra.dht.Murmur3Partitioner; +import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.distributed.shared.ClusterUtils; +import org.apache.cassandra.utils.concurrent.ImmediateFuture; import org.junit.Test; import net.bytebuddy.ByteBuddy; @@ -36,9 +45,11 @@ import static net.bytebuddy.matcher.ElementMatchers.named; import static org.apache.cassandra.db.SystemKeyspace.BootstrapState.COMPLETED; import static org.apache.cassandra.db.SystemKeyspace.BootstrapState.DECOMMISSIONED; +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.shared.ClusterUtils.stopUnchecked; +import static org.apache.cassandra.distributed.test.ring.BootstrapTest.populate; import static org.apache.cassandra.service.StorageService.Mode.DECOMMISSION_FAILED; import static org.apache.cassandra.service.StorageService.Mode.NORMAL; import static org.junit.Assert.assertEquals; @@ -186,8 +197,80 @@ public void testDecommissionAfterNodeRestart() throws Throwable } + @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 + try (Cluster cluster = builder().withNodes(4) + .withConfig(config -> config.with(NETWORK, GOSSIP) + // disable hints to simplify test + .set("hinted_handoff_enabled", false) + ) + // 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(); + leavingNode.runOnInstance(() -> assertEquals(DECOMMISSION_FAILED, StorageService.Mode.valueOf(StorageService.instance.getOperationMode()))); + + // 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)); + + List 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 { + private static int invocations = 0; + public static void install(ClassLoader classLoader, Integer num) { new ByteBuddy().rebase(StorageService.class) @@ -197,7 +280,15 @@ public static void install(ClassLoader classLoader, Integer num) .load(classLoader, ClassLoadingStrategy.Default.INJECTION); } - private static int invocations = 0; + 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") public static Supplier> prepareUnbootstrapStreaming(@SuperCall Callable>> zuper) @@ -216,5 +307,13 @@ public static Supplier> prepareUnbootstrapStreaming(@SuperCa throw new RuntimeException(e); } } + + @SuppressWarnings("unused") + public static Future streamHints(@SuperCall Callable> 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 possible moment + return ImmediateFuture.failure(new IOException("failing hints so that decomm fails at last moment possible")); + } } } 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 46a179718e2d..f70c10574b4a 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/ring/BootstrapTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/ring/BootstrapTest.java @@ -67,7 +67,6 @@ import org.apache.cassandra.distributed.shared.JMXUtil; import org.apache.cassandra.distributed.shared.NetworkTopology; import org.apache.cassandra.distributed.shared.WithProperties; -import org.apache.cassandra.distributed.test.DecommissionTest; import org.apache.cassandra.distributed.test.TestBaseImpl; import org.apache.cassandra.schema.SchemaConstants; import org.apache.cassandra.service.StorageService; @@ -329,9 +328,14 @@ public static void populate(ICluster cluster, int from, int to) } 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 (?, ?, ?)", @@ -356,9 +360,13 @@ public static void install(ClassLoader classLoader, Integer num) { return; } + // bootstrapFinished is implemented by this class, so delegate here. The previous + // target (DecommissionTest.BB) was wrong but bound anyway while that class had + // exactly one @SuperCall interceptor method; CASSANDRA-16290 adds a second one, + // which makes the binding ambiguous. new ByteBuddy().rebase(StorageService.class) .method(named("bootstrapFinished")) - .intercept(MethodDelegation.to(DecommissionTest.BB.class)) + .intercept(MethodDelegation.to(BB.class)) .make() .load(classLoader, ClassLoadingStrategy.Default.INJECTION); }