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 <[email protected]> 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: [email protected] For additional commands, e-mail: [email protected]
