You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by ma...@apache.org on 2026/08/13 22:36:06 UTC
(cassandra) branch cassandra-5.0 updated (797c73a247 -> 6e6077ebba)
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a change to branch cassandra-5.0
in repository https://gitbox.apache.org/repos/asf/cassandra.git
from 797c73a247 Merge branch 'cassandra-4.1' into cassandra-5.0
new 24dfcfe057 Ensure transferred_ranges reset on decommision re-attempt when pending ranges cannot be proven continous
new 8fa203de28 Merge branch 'cassandra-4.0' into cassandra-4.1
new 6e6077ebba Merge branch 'cassandra-4.1' into cassandra-5.0
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
CHANGES.txt | 1 +
.../org/apache/cassandra/db/SystemKeyspace.java | 31 ++++++-
.../apache/cassandra/service/StorageService.java | 32 ++++++-
.../distributed/test/DecommissionTest.java | 101 ++++++++++++++++++++-
.../distributed/test/ring/BootstrapTest.java | 14 ++-
5 files changed, 170 insertions(+), 9 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
(cassandra) 01/01: Merge branch 'cassandra-4.1' into cassandra-5.0
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch cassandra-5.0
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 6e6077ebbaabbe44ddad652dac92baa100f32115
Merge: 797c73a247 8fa203de28
Author: Caleb Rackliffe <ca...@gmail.com>
AuthorDate: Thu Aug 13 16:58:02 2026 -0500
Merge branch 'cassandra-4.1' into cassandra-5.0
* cassandra-4.1:
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 | 32 ++++++-
.../distributed/test/DecommissionTest.java | 101 ++++++++++++++++++++-
.../distributed/test/ring/BootstrapTest.java | 14 ++-
5 files changed, 170 insertions(+), 9 deletions(-)
diff --cc CHANGES.txt
index 3bae2041e9,96ebeb649d..8d2e5db0fb
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,9 -1,6 +1,10 @@@
-4.1.13
+5.0.10
+ * Avoid rebuilding per-SSTable SAI components unless missing or corrupted (CASSANDRA-21515)
+ * Propagate trickle_fsync settings to compressed SSTable writers (CASSANDRA-21487)
+ * Allow DatabaseDescriptor.setCompressedReadAheadBufferSizeInKb(0) to disable read-ahead buffer (CASSANDRA-21522)
+ * Return CorruptSSTableException if chunk metadata and file size are out of sync (CASSANDRA-21519)
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 a7318fb2de,bcd827e41a..9feb933948
--- a/src/java/org/apache/cassandra/db/SystemKeyspace.java
+++ b/src/java/org/apache/cassandra/db/SystemKeyspace.java
@@@ -1720,12 -1702,25 +1720,31 @@@ public final class SystemKeyspac
availableRanges.truncateBlockingWithoutSnapshot();
}
+ public static void resetAvailableStreamedRangesForKeyspace(String keyspace)
+ {
+ String cql = "DELETE FROM %s.%s WHERE keyspace_name = ?";
+ 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()
+ {
+ 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,
diff --cc src/java/org/apache/cassandra/service/StorageService.java
index 22ef542f2a,26e52ecc5f..a358841e2d
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@@ -272,12 -240,8 +272,13 @@@ import static org.apache.cassandra.inde
import static org.apache.cassandra.index.SecondaryIndexManager.isIndexColumnFamily;
import static org.apache.cassandra.net.NoPayload.noPayload;
import static org.apache.cassandra.net.Verb.REPLICATION_DONE_REQ;
+import static org.apache.cassandra.locator.InetAddressAndPort.stringify;
import static org.apache.cassandra.service.ActiveRepairService.ParentRepairStatus;
import static org.apache.cassandra.service.ActiveRepairService.repairCommandExecutor;
+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,10 -4989,35 +5332,36 @@@ public class StorageService extends Not
{
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;
++ && 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 (operationMode != Mode.LEAVING) // If we're already decommissioning there is no point checking RF/pending ranges
+ // If we're already decommissioning there is no point checking RF/pending ranges
+ if (operationMode != Mode.LEAVING)
{
int rf, numNodes;
for (String keyspaceName : Schema.instance.distributedKeyspaces().names())
diff --cc test/distributed/org/apache/cassandra/distributed/test/DecommissionTest.java
index 66091da3ef,0000000000..a81cf6475e
mode 100644,000000..100644
--- a/test/distributed/org/apache/cassandra/distributed/test/DecommissionTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/DecommissionTest.java
@@@ -1,220 -1,0 +1,319 @@@
+/*
+ * 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;
+
++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;
+import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
+import net.bytebuddy.implementation.MethodDelegation;
+import net.bytebuddy.implementation.bind.annotation.SuperCall;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.streaming.StreamState;
+import org.apache.cassandra.utils.concurrent.Future;
+
+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;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class DecommissionTest extends TestBaseImpl
+{
+ @Test
+ public void testDecommission() throws Throwable
+ {
+ try (Cluster cluster = init(Cluster.build(2)
+ .withConfig(config -> config.with(GOSSIP)
+ .with(NETWORK))
+ .withInstanceInitializer(DecommissionTest.BB::install)
+ .start()))
+ {
+ IInvokableInstance instance = cluster.get(1);
+
+ instance.runOnInstance(() -> {
+
+ assertEquals(COMPLETED.name(), StorageService.instance.getBootstrapState());
+
+ // pretend that decommissioning has failed in the middle
+
+ try
+ {
+ StorageService.instance.decommission(true);
+ fail("the first attempt to decommission should fail");
+ }
+ catch (Throwable t)
+ {
+ assertEquals("simulated error in prepareUnbootstrapStreaming", t.getMessage());
+ }
+
+ assertFalse(StorageService.instance.isDecommissioning());
+ assertTrue(StorageService.instance.isDecommissionFailed());
+
+ // still COMPLETED, nothing has changed
+ assertEquals(COMPLETED.name(), StorageService.instance.getBootstrapState());
+
+ String operationMode = StorageService.instance.getOperationMode();
+ assertEquals(DECOMMISSION_FAILED.name(), operationMode);
+
+ // try to decommission again, now successfully
+
+ try
+ {
+ StorageService.instance.decommission(true);
+
+ // decommission was successful, so we reset failed decommission mode
+ assertFalse(StorageService.instance.isDecommissionFailed());
+
+ assertEquals(DECOMMISSIONED.name(), StorageService.instance.getBootstrapState());
+ assertFalse(StorageService.instance.isDecommissioning());
+ }
+ catch (Throwable t)
+ {
+ fail("the second decommission attempt should pass but it failed on: " + t.getMessage());
+ }
+
+ // check that decommissioning of already decommissioned node has no effect
+
+ try
+ {
+ assertEquals(DECOMMISSIONED.name(), StorageService.instance.getBootstrapState());
+ assertFalse(StorageService.instance.isDecommissionFailed());
+
+ StorageService.instance.decommission(true);
+
+ assertEquals(DECOMMISSIONED.name(), StorageService.instance.getBootstrapState());
+ assertFalse(StorageService.instance.isDecommissionFailed());
+ assertFalse(StorageService.instance.isDecommissioning());
+ }
+ catch (Throwable t)
+ {
+ fail("Decommissioning already decommissioned node should be no-op operation.");
+ }
+ });
+ }
+ }
+
+ @Test
+ public void testDecommissionAfterNodeRestart() throws Throwable
+ {
+ try (Cluster cluster = init(Cluster.build(2)
+ .withConfig(config -> config.with(GOSSIP)
+ .with(NETWORK))
+ .withInstanceInitializer((classLoader, threadGroup, num, generation) -> {
+ // we do not want to install BB after restart of a node which
+ // failed to decommission which is the second generation, here
+ // as "1" as it is counted from 0.
+ if (num == 1 && generation != 1)
+ BB.install(classLoader, num);
+ })
+ .start()))
+ {
+ IInvokableInstance instance = cluster.get(1);
+
+ instance.runOnInstance(() -> {
+ assertEquals(COMPLETED.name(), StorageService.instance.getBootstrapState());
+
+ // pretend that decommissioning has failed in the middle
+
+ try
+ {
+ StorageService.instance.decommission(true);
+ fail("the first attempt to decommission should fail");
+ }
+ catch (Throwable t)
+ {
+ assertEquals("simulated error in prepareUnbootstrapStreaming", t.getMessage());
+ }
+
+ // node is in DECOMMISSION_FAILED mode
+ String operationMode = StorageService.instance.getOperationMode();
+ assertEquals(DECOMMISSION_FAILED.name(), operationMode);
+ });
+
+ // restart the node which we failed to decommission
+ stopUnchecked(instance);
+ instance.startup();
+
+ // it is back to normal so let's decommission again
+
+ String oprationMode = instance.callOnInstance(() -> StorageService.instance.getOperationMode());
+ assertEquals(NORMAL.name(), oprationMode);
+
+ instance.runOnInstance(() -> {
+ try
+ {
+ StorageService.instance.decommission(true);
+ }
+ catch (InterruptedException e)
+ {
+ fail("Should decommission the node");
+ }
+
+ assertEquals(DECOMMISSIONED.name(), StorageService.instance.getBootstrapState());
+ assertFalse(StorageService.instance.isDecommissionFailed());
+ assertFalse(StorageService.instance.isDecommissioning());
+ });
+ }
+ }
+
+
++ @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<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
+ {
++ private static int invocations = 0;
++
+ public static void install(ClassLoader classLoader, Integer num)
+ {
+ new ByteBuddy().rebase(StorageService.class)
+ .method(named("prepareUnbootstrapStreaming"))
+ .intercept(MethodDelegation.to(DecommissionTest.BB.class))
+ .make()
+ .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<Future<StreamState>> prepareUnbootstrapStreaming(@SuperCall Callable<Supplier<Future<StreamState>>> zuper)
+ {
+ ++invocations;
+
+ if (invocations == 1)
+ throw new RuntimeException("simulated error in prepareUnbootstrapStreaming");
+
+ try
+ {
+ return zuper.call();
+ }
+ catch (Exception e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
++
++ @SuppressWarnings("unused")
++ 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 possible moment
++ return ImmediateFuture.failure(new IOException("failing hints so that decomm fails at last moment possible"));
++ }
+ }
+}
diff --cc test/distributed/org/apache/cassandra/distributed/test/ring/BootstrapTest.java
index 46a179718e,1344a92e22..f70c10574b
--- a/test/distributed/org/apache/cassandra/distributed/test/ring/BootstrapTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/ring/BootstrapTest.java
@@@ -64,21 -35,10 +64,20 @@@ import org.apache.cassandra.distributed
import org.apache.cassandra.distributed.api.IInstanceConfig;
import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.distributed.api.TokenSupplier;
+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;
+import org.apache.cassandra.service.StorageServiceMBean;
import static java.util.Arrays.asList;
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
+import static org.apache.cassandra.config.CassandraRelevantProperties.JOIN_RING;
+import static org.apache.cassandra.config.CassandraRelevantProperties.MIGRATION_DELAY;
+import static org.apache.cassandra.config.CassandraRelevantProperties.RESET_BOOTSTRAP_PROGRESS;
import static org.apache.cassandra.distributed.action.GossipHelper.bootstrap;
import static org.apache.cassandra.distributed.action.GossipHelper.pullSchemaFrom;
import static org.apache.cassandra.distributed.action.GossipHelper.statusToBootstrap;
@@@ -347,118 -155,4 +351,122 @@@ public class BootstrapTest extends Test
.collect(Collectors.toMap(nodeId -> nodeId,
nodeId -> (Long) cluster.get(nodeId).executeInternal("SELECT count(*) FROM " + KEYSPACE + ".tbl")[0][0]));
}
+
+ public static class BB
+ {
+ public static void install(ClassLoader classLoader, Integer num)
+ {
+ if (num != 3)
+ {
+ 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);
+ }
+
+ private static int invocations = 0;
+
+ @SuppressWarnings("unused")
+ public static void bootstrapFinished(@SuperCall Callable<Void> zuper)
+ {
+ ++invocations;
+
+ if (invocations == 1)
+ throw new RuntimeException("simulated error in bootstrapFinished");
+
+ try
+ {
+ zuper.call();
+ }
+ catch (Exception e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ /**
+ * This regression test for CASSANDRA-19902 ensures {@link StorageServiceMBean} JMX
+ * interface is published before the node finishes bootstrapping
+ */
+ @Test
+ public void testStorageServiceMBeanIsPublishedOnJMXDuringBootstrap() throws Throwable
+ {
+ ExecutorService es = Executors.newFixedThreadPool(1);
+ try (Cluster cluster = builder().withNodes(2)
+ .withConfig(config -> config.with(GOSSIP)
+ .with(NETWORK)
+ .with(Feature.JMX)
+ .set("auto_bootstrap", true))
+ .withInstanceInitializer(BBBootstrapInterceptor::install)
+ .createWithoutStarting();
+ Closeable ignored = es::shutdown)
+ {
+ Runnable test = () ->
+ {
+ // Wait for bootstrap to start via countdown latch
+ IInvokableInstance joiningInstance = cluster.get(2);
+ joiningInstance.runOnInstance(() -> Uninterruptibles.awaitUninterruptibly(BBBootstrapInterceptor.bootstrapStart));
+ // At this point, it should be possible to check bootstrap status via JMX
+ IInstanceConfig config = joiningInstance.config();
+ try (JMXConnector jmxc = JMXUtil.getJmxConnector(config))
+ {
+ MBeanServerConnection mbsc = jmxc.getMBeanServerConnection();
+ StorageServiceMBean sp = JMX.newMBeanProxy(mbsc, new ObjectName("org.apache.cassandra.db:type=StorageService"), StorageServiceMBean.class);
+ assertEquals(sp.getOperationMode(), StorageService.Mode.JOINING.toString());
+ }
+ catch (IOException | MalformedObjectNameException e)
+ {
+ throw new RuntimeException(e);
+ }
+ finally
+ {
+ // Complete bootstrap via countdown latch so test will finish properly
+ joiningInstance.runOnInstance(() -> BBBootstrapInterceptor.bootstrapReady.countDown());
+ }
+ };
+
+ Future<?> testResult = es.submit(test);
+ try
+ {
+ cluster.startup();
+ }
+ catch (Exception ex) {
+ // ignore exceptions from startup process. More interested in the test result.
+ }
+ testResult.get();
+ }
+ es.awaitTermination(5, TimeUnit.SECONDS);
+ }
+
+ public static class BBBootstrapInterceptor
+ {
+ final static CountDownLatch bootstrapReady = new CountDownLatch(1);
+ final static CountDownLatch bootstrapStart = new CountDownLatch(1);
+ static void install(ClassLoader cl, int nodeNumber)
+ {
+ if (nodeNumber != 2)
+ return;
+ new ByteBuddy().rebase(StorageService.class)
+ .method(named("bootstrap").and(takesArguments(2)))
+ .intercept(MethodDelegation.to(BBBootstrapInterceptor.class))
+ .make()
+ .load(cl, ClassLoadingStrategy.Default.INJECTION);
+ }
+
+ public static boolean bootstrap(Collection<Token> tokens, long bootstrapTimeoutMillis)
+ {
+ bootstrapStart.countDown();
+ Uninterruptibles.awaitUninterruptibly(bootstrapReady);
+ return false; // bootstrap fails
+ }
+ }
+
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org