diff --git a/CHANGES.txt b/CHANGES.txt index 94270a0b46..9fa2bafff2 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -154,6 +154,7 @@ Merged from 4.0: * Fix Down nodes counter in nodetool describecluster (CASSANDRA-18512) * Remove unnecessary shuffling of GossipDigests in Gossiper#makeRandomGossipDigest (CASSANDRA-18546) Merged from 3.11: + * 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) Merged from 3.0: diff --git a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java index 8eaaa7a9e0..fbb20766fa 100644 --- a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java +++ b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java @@ -234,6 +234,22 @@ public enum CassandraRelevantProperties GOSSIPER_QUARANTINE_DELAY("cassandra.gossip_quarantine_delay_ms"), GOSSIPER_SKIP_WAITING_TO_SETTLE("cassandra.skip_wait_for_gossip_to_settle", "-1"), GOSSIP_DISABLE_THREAD_VALIDATION("cassandra.gossip.disable_thread_validation"), + + /** + * Delay before checking if gossip is settled. + */ + GOSSIP_SETTLE_MIN_WAIT_MS("cassandra.gossip_settle_min_wait_ms", "5000"), + + /** + * Interval delay between checking gossip is settled. + */ + GOSSIP_SETTLE_POLL_INTERVAL_MS("cassandra.gossip_settle_interval_ms", "1000"), + + /** + * Number of polls without gossip state change to consider gossip as settled. + */ + GOSSIP_SETTLE_POLL_SUCCESSES_REQUIRED("cassandra.gossip_settle_poll_success_required", "3"), + IGNORED_SCHEMA_CHECK_ENDPOINTS("cassandra.skip_schema_check_for_endpoints"), IGNORED_SCHEMA_CHECK_VERSIONS("cassandra.skip_schema_check_for_versions"), IGNORE_CORRUPTED_SCHEMA_TABLES("cassandra.ignore_corrupted_schema_tables"), diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java index adff044420..dd6605cca7 100644 --- a/src/java/org/apache/cassandra/gms/Gossiper.java +++ b/src/java/org/apache/cassandra/gms/Gossiper.java @@ -61,9 +61,11 @@ import org.slf4j.LoggerFactory; import org.apache.cassandra.concurrent.FutureTask; import org.apache.cassandra.concurrent.ScheduledExecutorPlus; import org.apache.cassandra.concurrent.Stage; +import org.apache.cassandra.config.CassandraRelevantProperties; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.SystemKeyspace; import org.apache.cassandra.dht.Token; +import org.apache.cassandra.exceptions.RequestFailureReason; import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.net.Message; import org.apache.cassandra.net.MessagingService; @@ -160,6 +162,9 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean @VisibleForTesting final Set liveEndpoints = new ConcurrentSkipListSet<>(); + /* Inflight echo requests. */ + private final Set inflightEcho = new ConcurrentSkipListSet<>(); + /* unreachable member set */ private final Map unreachableEndpoints = new ConcurrentHashMap<>(); @@ -671,6 +676,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean return; liveEndpoints.remove(endpoint); + inflightEcho.remove(endpoint); unreachableEndpoints.remove(endpoint); MessagingService.instance().versions.reset(endpoint); quarantineEndpoint(endpoint); @@ -1361,14 +1367,45 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean private void markAlive(final InetAddressAndPort addr, final EndpointState localState) { + if (inflightEcho.contains(addr)) + { + return; + } + inflightEcho.add(addr); + localState.markDead(); Message echoMessage = Message.out(ECHO_REQ, noPayload); logger.trace("Sending ECHO_REQ to {}", addr); - RequestCallback echoHandler = msg -> + RequestCallback echoHandler = new RequestCallback() { - // force processing of the echo response onto the gossip stage, as it comes in on the REQUEST_RESPONSE stage - runInGossipStageBlocking(() -> realMarkAlive(addr, localState)); + @Override + public void onResponse(Message msg) + { + // force processing of the echo response onto the gossip stage, as it comes in on the REQUEST_RESPONSE stage + runInGossipStageBlocking(() -> { + try + { + realMarkAlive(addr, localState); + } + finally + { + inflightEcho.remove(addr); + } + }); + } + + @Override + public boolean invokeOnFailure() + { + return true; + } + + @Override + public void onFailure(InetAddressAndPort from, RequestFailureReason failureReason) + { + inflightEcho.remove(addr); + } }; MessagingService.instance().sendWithCallback(echoMessage, addr, echoHandler); @@ -1424,6 +1461,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean if (!disableEndpointRemoval) { liveEndpoints.remove(addr); + inflightEcho.remove(addr); unreachableEndpoints.put(addr, nanoTime()); } } @@ -2378,21 +2416,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 = CassandraRelevantProperties.GOSSIP_SETTLE_MIN_WAIT_MS.getInt(); + final int GOSSIP_SETTLE_POLL_INTERVAL_MS = CassandraRelevantProperties.GOSSIP_SETTLE_POLL_INTERVAL_MS.getInt(); + final int GOSSIP_SETTLE_POLL_SUCCESSES_REQUIRED = CassandraRelevantProperties.GOSSIP_SETTLE_POLL_SUCCESSES_REQUIRED.getInt(); 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.getEndpointCount(); + 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.getEndpointCount(); + int currentLive = Gossiper.instance.getLiveMembers().size(); totalPolls++; - if (currentSize == epSize) + if (currentSize == epSize && currentLive == liveSize) { logger.debug("Gossip looks settled."); numOkay++; @@ -2403,6 +2443,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: {}",