diff --git a/CHANGES.txt b/CHANGES.txt index 570d7cfdd4..197b627b7c 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -14,6 +14,7 @@ Merged from 4.1: * Disk usage guardrail cannot be disabled when failure threshold is reached (CASSANDRA-21057) * ReadCommandController should close fast to avoid deadlock when building secondary index (CASSANDRA-19564) Merged from 4.0: + * Node does not send multiple inflight echos (CASSANDRA-18866) * Obsolete expired SSTables before compaction starts (CASSANDRA-19776) * Switch lz4-java to at.yawk.lz4 version due to CVE (CASSANDRA-21052) * Restrict BytesType compatibility to scalar types only (CASSANDRA-20982) diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java index 3d4bc95cac..647441ffba 100644 --- a/src/java/org/apache/cassandra/gms/Gossiper.java +++ b/src/java/org/apache/cassandra/gms/Gossiper.java @@ -165,6 +165,9 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean, @VisibleForTesting final Set liveEndpoints = new ConcurrentSkipListSet<>(); + /* Inflight echo requests. */ + private final Map inflightEcho = new ConcurrentHashMap<>(); + /* unreachable member set */ private final Map unreachableEndpoints = new ConcurrentHashMap<>(); @@ -206,6 +209,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean, { unreachableEndpoints.clear(); liveEndpoints.clear(); + inflightEcho.clear(); justRemovedEndpoints.clear(); expireTimeEndpointMap.clear(); endpointStateMap.clear(); @@ -725,6 +729,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean, logger.warn("Seeds list is now empty!"); } + inflightEcho.remove(endpoint); if (disableEndpointRemoval) return; @@ -1431,14 +1436,24 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean, { localState.markDead(); - Message echoMessage = Message.out(ECHO_REQ, noPayload); - logger.trace("Sending ECHO_REQ to {}", addr); - RequestCallback echoHandler = msg -> + EndpointState prevState = inflightEcho.put(addr, localState); + boolean sendEcho = !localState.equals(prevState); + if (sendEcho) { - runInGossipStageBlocking(() -> realMarkAlive(addr, localState)); - }; - - MessagingService.instance().sendWithCallback(echoMessage, addr, echoHandler); + Message echoMessage = Message.out(ECHO_REQ, noPayload); + logger.trace("Sending ECHO_REQ to {}", addr); + RequestCallback echoHandler = msg -> + { + runInGossipStageBlocking(() -> { + EndpointState epState = inflightEcho.remove(addr); + if (epState != null) + realMarkAlive(addr, epState); + }); + }; + MessagingService.instance().sendWithCallback(echoMessage, addr, echoHandler); + } + else + logger.trace("Skipping ECHO_REQ to {} since it is already inflight", addr); GossiperDiagnostics.markedAlive(this, addr, localState); } @@ -1488,6 +1503,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean, private void silentlyMarkDead(InetAddressAndPort addr, EndpointState localState) { localState.markDead(); + inflightEcho.remove(addr); if (!disableEndpointRemoval) { liveEndpoints.remove(addr);