From 22f6f52b2711a28c632795db56cc24ca6c255606 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Szymon=20Mi=C4=99=C5=BCa=C5=82?= Date: Tue, 11 Mar 2025 16:11:01 +0100 Subject: [PATCH] Fix replacing node stuck in hibernation state When a node is replaced, it announces itself as hibernated (one of the silent shutdown states). If the replacement fails, other nodes continue to see the replacing node in this state. As a result, the replacing node does not receive gossip messages from the seed upon subsequent startup, leading to an exception. This patch adds an explicit shutdown announcement via gossip to let other nodes know that the node was explicitly shutdown - as it was due to the exception. This allows other nodes (seeds in particular) to contact the replacing node at its next startup, allowing it to retry the replacement. --- .../org/apache/cassandra/gms/Gossiper.java | 31 +++-- .../cassandra/service/StorageService.java | 4 + .../cassandra/distributed/impl/Instance.java | 1 + .../hostreplacement/HostReplacementTest.java | 106 ++++++++++++++++++ 4 files changed, 132 insertions(+), 10 deletions(-) diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java index 2e045296df..bdcbab95ac 100644 --- a/src/java/org/apache/cassandra/gms/Gossiper.java +++ b/src/java/org/apache/cassandra/gms/Gossiper.java @@ -1497,11 +1497,13 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean, /** * This method is called whenever there is a "big" change in ep state (a generation change for a known node). + * It is public as the state change simulation is needed in testing, otherwise should not be used directly. * * @param ep endpoint * @param epState EndpointState for the endpoint */ - private void handleMajorStateChange(InetAddressAndPort ep, EndpointState epState) + @VisibleForTesting + public void handleMajorStateChange(InetAddressAndPort ep, EndpointState epState) { checkProperThreadForStateMutation(); EndpointState localEpState = endpointStateMap.get(ep); @@ -2257,15 +2259,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean, EndpointState mystate = endpointStateMap.get(getBroadcastAddressAndPort()); if (mystate != null && !isSilentShutdownState(mystate) && StorageService.instance.isJoined()) { - logger.info("Announcing shutdown"); - addLocalApplicationState(ApplicationState.STATUS_WITH_PORT, StorageService.instance.valueFactory.shutdown(true)); - addLocalApplicationState(ApplicationState.STATUS, StorageService.instance.valueFactory.shutdown(true)); - // clone endpointstate to avoid it changing between serializedSize and serialize calls - EndpointState clone = new EndpointState(mystate); - Message message = Message.out(Verb.GOSSIP_SHUTDOWN, new GossipShutdown(clone)); - for (InetAddressAndPort ep : liveEndpoints) - MessagingService.instance().send(message, ep); - Uninterruptibles.sleepUninterruptibly(SHUTDOWN_ANNOUNCE_DELAY_IN_MS.getInt(), TimeUnit.MILLISECONDS); + announceShutdown(); } else logger.warn("No local state, state is in silent shutdown, or node hasn't joined, not announcing shutdown"); @@ -2273,6 +2267,23 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean, scheduledGossipTask.cancel(false); } + /** + * This method sends the node shutdown status to all live endpoints. + * It does not close the gossiper itself. + */ + public void announceShutdown() + { + logger.info("Announcing shutdown"); + addLocalApplicationState(ApplicationState.STATUS_WITH_PORT, StorageService.instance.valueFactory.shutdown(true)); + addLocalApplicationState(ApplicationState.STATUS, StorageService.instance.valueFactory.shutdown(true)); + // clone endpointstate to avoid it changing between serializedSize and serialize calls + EndpointState clone = new EndpointState(endpointStateMap.get(getBroadcastAddressAndPort())); + Message message = Message.out(Verb.GOSSIP_SHUTDOWN, new GossipShutdown(clone)); + for (InetAddressAndPort ep : liveEndpoints) + MessagingService.instance().send(message, ep); + Uninterruptibles.sleepUninterruptibly(SHUTDOWN_ANNOUNCE_DELAY_IN_MS.getInt(), TimeUnit.MILLISECONDS); + } + public boolean isEnabled() { ScheduledFuture scheduledGossipTask = this.scheduledGossipTask; diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 533ca08e6d..62c6f0cf5d 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -2188,7 +2188,11 @@ public class StorageService extends NotificationBroadcasterSupport implements IE SystemKeyspace.removeEndpoint(DatabaseDescriptor.getReplaceAddress()); } if (!Gossiper.instance.seenAnySeed()) + { + logger.info("Announcing shutdown to get out of the hibernation deadlock"); + Gossiper.instance.announceShutdown(); throw new IllegalStateException("Unable to contact any seeds: " + Gossiper.instance.getSeeds()); + } if (RESET_BOOTSTRAP_PROGRESS.getBoolean()) { diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java index e0eda7afce..3f63afae0f 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java @@ -765,6 +765,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance } catch (Throwable t) { + startedAt.set(0); if (t instanceof RuntimeException) throw (RuntimeException) t; throw new RuntimeException(t); diff --git a/test/distributed/org/apache/cassandra/distributed/test/hostreplacement/HostReplacementTest.java b/test/distributed/org/apache/cassandra/distributed/test/hostreplacement/HostReplacementTest.java index 8219d43ad1..a7918f9a4f 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/hostreplacement/HostReplacementTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/hostreplacement/HostReplacementTest.java @@ -19,9 +19,12 @@ package org.apache.cassandra.distributed.test.hostreplacement; import java.io.IOException; +import java.net.UnknownHostException; import java.util.Arrays; import java.util.List; +import java.util.UUID; +import org.apache.commons.io.FileUtils; import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -32,22 +35,37 @@ import org.apache.cassandra.distributed.api.ConsistencyLevel; import org.apache.cassandra.distributed.api.Feature; import org.apache.cassandra.distributed.api.ICoordinator; import org.apache.cassandra.distributed.api.IInvokableInstance; +import org.apache.cassandra.distributed.api.IIsolatedExecutor; import org.apache.cassandra.distributed.api.SimpleQueryResult; import org.apache.cassandra.distributed.api.TokenSupplier; +import org.apache.cassandra.distributed.impl.InstanceConfig; import org.apache.cassandra.distributed.shared.AssertUtils; import org.apache.cassandra.distributed.shared.ClusterUtils; +import org.apache.cassandra.distributed.shared.WithProperties; import org.apache.cassandra.distributed.test.TestBaseImpl; +import org.apache.cassandra.gms.ApplicationState; +import org.apache.cassandra.gms.EndpointState; +import org.apache.cassandra.gms.Gossiper; +import org.apache.cassandra.gms.VersionedValue; +import org.apache.cassandra.io.util.File; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.service.StorageService; import org.assertj.core.api.Assertions; import static org.apache.cassandra.config.CassandraRelevantProperties.BOOTSTRAP_SKIP_SCHEMA_CHECK; import static org.apache.cassandra.config.CassandraRelevantProperties.GOSSIPER_QUARANTINE_DELAY; +import static org.apache.cassandra.config.CassandraRelevantProperties.GOSSIP_DISABLE_THREAD_VALIDATION; +import static org.apache.cassandra.config.CassandraRelevantProperties.REPLACE_ADDRESS; import static org.apache.cassandra.distributed.shared.ClusterUtils.assertInRing; import static org.apache.cassandra.distributed.shared.ClusterUtils.assertRingIs; import static org.apache.cassandra.distributed.shared.ClusterUtils.awaitRingHealthy; import static org.apache.cassandra.distributed.shared.ClusterUtils.awaitRingJoin; +import static org.apache.cassandra.distributed.shared.ClusterUtils.getDirectories; import static org.apache.cassandra.distributed.shared.ClusterUtils.getTokenMetadataTokens; import static org.apache.cassandra.distributed.shared.ClusterUtils.replaceHostAndStart; import static org.apache.cassandra.distributed.shared.ClusterUtils.stopUnchecked; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; +import static org.junit.Assert.assertFalse; public class HostReplacementTest extends TestBaseImpl { @@ -206,6 +224,94 @@ public class HostReplacementTest extends TestBaseImpl } } + /** + * Make sure that a node stuck in hibernate state due to failed replacement can retry the replacement procedure and succeed. + */ + @Test + public void retryingFailedReplaceWithNodeInHibernateState() throws IOException + { + try (WithProperties properties = new WithProperties()) + { + properties.set(GOSSIP_DISABLE_THREAD_VALIDATION, "true"); + + // given a two node cluster with one need + TokenSupplier even = TokenSupplier.evenlyDistributedTokens(2); + try (Cluster cluster = Cluster.build(2) + .withConfig(c -> c.with(Feature.GOSSIP, Feature.NATIVE_PROTOCOL) + .set(Constants.KEY_DTEST_API_STARTUP_FAILURE_AS_SHUTDOWN, true)) + .withTokenSupplier(node -> even.token(node == 3 ? 2 : node)) + .start()) + { + IInvokableInstance seed = cluster.get(1); + IInvokableInstance nodeToReplace = cluster.get(2); + + setupCluster(cluster); + SimpleQueryResult expectedState = nodeToReplace.coordinator().executeWithResult("SELECT * FROM " + KEYSPACE + ".tbl", ConsistencyLevel.ALL); + + // when + // stop the node to replace + stopUnchecked(nodeToReplace); + // wipe the node to replace + for (File dir : getDirectories(nodeToReplace)) + { + FileUtils.deleteDirectory(dir.toJavaIOFile()); + } + + String toReplaceAddress = nodeToReplace.config().broadcastAddress().getAddress().getHostAddress(); + // set hibernate status for the node to replace on seed + seed.runOnInstance(putInHibernation(toReplaceAddress)); + + // we need to fake a new host id + ((InstanceConfig) nodeToReplace.config()).setHostId(UUID.randomUUID()); + // enable autoboostrap + nodeToReplace.config().set("auto_bootstrap", true); + + // first replacement will fail as the node was announced as hibernated and no-one can contact it as startup + assertThatExceptionOfType(IllegalStateException.class).isThrownBy(() -> { + ClusterUtils.start(nodeToReplace, props -> { + // set the replacement address + props.set(REPLACE_ADDRESS, toReplaceAddress); + }); + }).withMessageContaining("Unable to contact any seeds"); + + // then + // retrying replacement will succeed as the node announced itself as shutdown before killing itself + ClusterUtils.start(nodeToReplace, props -> { + // set the replacement address + props.set(REPLACE_ADDRESS, toReplaceAddress); + }); + assertFalse("replaces node should be up", nodeToReplace.isShutdown()); + + // the data after replacement should be consistent + awaitRingJoin(seed, nodeToReplace); + awaitRingJoin(nodeToReplace, seed); + + validateRows(seed.coordinator(), expectedState); + validateRows(nodeToReplace.coordinator(), expectedState); + } + } + } + + private static IIsolatedExecutor.SerializableRunnable putInHibernation(String address) + { + return () -> { + InetAddressAndPort endpoint; + try + { + endpoint = InetAddressAndPort.getByName(address); + } + catch (UnknownHostException e) + { + throw new RuntimeException(e); + } + EndpointState epState = Gossiper.instance.getEndpointStateForEndpoint(endpoint); + VersionedValue newStatus = StorageService.instance.valueFactory.hibernate(true); + epState.addApplicationState(ApplicationState.STATUS, newStatus); + epState.addApplicationState(ApplicationState.STATUS_WITH_PORT, newStatus); + Gossiper.instance.handleMajorStateChange(endpoint, epState); + }; + } + static void setupCluster(Cluster cluster) { fixDistributedSchemas(cluster);