From 64d953fdcac91715898454a3cb04d932910a70b7 Mon Sep 17 00:00:00 2001 From: Cameron Zemek Date: Mon, 12 Jun 2023 00:17:02 +0200 Subject: [PATCH] Wait for live endpoints in gossip waiting to settle patch by Cameron Zemek; reviewed by Stefan Miklosovic and Brandon Williams for CASSANDRA-18543 --- CHANGES.txt | 1 + .../org/apache/cassandra/gms/Gossiper.java | 44 ++++++++++++++++--- 2 files changed, 38 insertions(+), 7 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 942651a37e..abbd7181f7 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 3.11.16 + * Wait for live endpoints in gossip waiting to settle (CASSANDRA-18543) * Fix error message handling when trying to use CLUSTERING ORDER with non-clustering column (CASSANDRA-17818 * Add keyspace and table name to exception message during ColumnSubselection deserialization (CASSANDRA-18346) * Remove unnecessary String.format invocation in QueryProcessor when getting a prepared statement from cache (CASSANDRA-17202) diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java index 613b566f9a..ac35aa8642 100644 --- a/src/java/org/apache/cassandra/gms/Gossiper.java +++ b/src/java/org/apache/cassandra/gms/Gossiper.java @@ -39,6 +39,8 @@ import com.google.common.util.concurrent.ListenableFutureTask; import com.google.common.util.concurrent.Uninterruptibles; import io.netty.util.concurrent.FastThreadLocal; +import org.apache.cassandra.exceptions.RequestFailureReason; +import org.apache.cassandra.net.IAsyncCallbackWithFailure; import org.apache.cassandra.utils.ExecutorUtils; import org.apache.cassandra.utils.MBeanWrapper; import org.apache.cassandra.utils.NoSpamLogger; @@ -135,6 +137,9 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean /* live member set */ private final Set liveEndpoints = new ConcurrentSkipListSet(inetcomparator); + /* Inflight echo requests. */ + private final Set inflightEcho = new ConcurrentSkipListSet<>(inetcomparator); + /* unreachable member set */ private final Map unreachableEndpoints = new ConcurrentHashMap(); @@ -529,6 +534,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean } liveEndpoints.remove(endpoint); + inflightEcho.remove(endpoint); unreachableEndpoints.remove(endpoint); MessagingService.instance().resetVersion(endpoint); quarantineEndpoint(endpoint); @@ -1126,12 +1132,23 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean return; } + if (inflightEcho.contains(addr)) + { + return; + } + inflightEcho.add(addr); + localState.markDead(); MessageOut echoMessage = new MessageOut(MessagingService.Verb.ECHO, EchoMessage.instance, EchoMessage.serializer); logger.trace("Sending a EchoMessage to {}", addr); - IAsyncCallback echoHandler = new IAsyncCallback() + IAsyncCallbackWithFailure echoHandler = new IAsyncCallbackWithFailure() { + public void onFailure(InetAddress from, RequestFailureReason failureReason) + { + inflightEcho.remove(addr); + } + public boolean isLatencyForSnitch() { return false; @@ -1139,11 +1156,20 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean public void response(MessageIn msg) { - runInGossipStageBlocking(() -> realMarkAlive(addr, localState)); + runInGossipStageBlocking(() -> { + try + { + realMarkAlive(addr, localState); + } + finally + { + inflightEcho.remove(addr); + } + }); } }; - MessagingService.instance().sendRR(echoMessage, addr, echoHandler); + MessagingService.instance().sendRRWithFailure(echoMessage, addr, echoHandler); } @VisibleForTesting @@ -1173,6 +1199,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean logger.trace("marking as down {}", addr); localState.markDead(); liveEndpoints.remove(addr); + inflightEcho.remove(addr); unreachableEndpoints.put(addr, System.nanoTime()); logger.info("InetAddress {} is now DOWN", addr); for (IEndpointStateChangeSubscriber subscriber : subscribers) @@ -1862,21 +1889,23 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean { return; } - final int GOSSIP_SETTLE_MIN_WAIT_MS = 5000; - final int GOSSIP_SETTLE_POLL_INTERVAL_MS = 1000; - final int GOSSIP_SETTLE_POLL_SUCCESSES_REQUIRED = 3; + final int GOSSIP_SETTLE_MIN_WAIT_MS = Integer.getInteger("cassandra.gossip_settle_min_wait_ms", 5000); + final int GOSSIP_SETTLE_POLL_INTERVAL_MS = Integer.getInteger("cassandra.gossip_settle_interval_ms", 1000); + final int GOSSIP_SETTLE_POLL_SUCCESSES_REQUIRED = Integer.getInteger("cassandra.gossip_settle_poll_success_required", 3); logger.info("Waiting for gossip to settle..."); Uninterruptibles.sleepUninterruptibly(GOSSIP_SETTLE_MIN_WAIT_MS, TimeUnit.MILLISECONDS); int totalPolls = 0; int numOkay = 0; int epSize = Gossiper.instance.getEndpointStates().size(); + int liveSize = Gossiper.instance.getLiveMembers().size(); while (numOkay < GOSSIP_SETTLE_POLL_SUCCESSES_REQUIRED) { Uninterruptibles.sleepUninterruptibly(GOSSIP_SETTLE_POLL_INTERVAL_MS, TimeUnit.MILLISECONDS); int currentSize = Gossiper.instance.getEndpointStates().size(); + int currentLive = Gossiper.instance.getLiveMembers().size(); totalPolls++; - if (currentSize == epSize) + if (currentSize == epSize && currentLive == liveSize) { logger.debug("Gossip looks settled."); numOkay++; @@ -1887,6 +1916,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean numOkay = 0; } epSize = currentSize; + liveSize = currentLive; if (forceAfter > 0 && totalPolls > forceAfter) { logger.warn("Gossip not settled but startup forced by cassandra.skip_wait_for_gossip_to_settle. Gossip total polls: {}",