natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge 22f6f52b27 into 209ef2b6c4

This commit is contained in:
Szymon Miężał 2026-07-29 13:35:25 +08:00 committed by GitHub
parent 209ef2b6c4 22f6f52b27
commit fa2fb913b3
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 132 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
src/java/org/apache/cassandra/gms/Gossiper.java
View File

@ -1532,11 +1532,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);
@ -2286,15 +2288,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<GossipShutdown> 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");
@ -2302,6 +2296,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<GossipShutdown> 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;

4
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -2232,7 +2232,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())
{

1
test/distributed/org/apache/cassandra/distributed/impl/Instance.java
View File

@ -764,6 +764,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);

106
test/distributed/org/apache/cassandra/distributed/test/hostreplacement/HostReplacementTest.java
View File

@ -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);