|
69 | 69 | import org.apache.cassandra.db.SystemKeyspace; |
70 | 70 | import org.apache.cassandra.dht.IPartitioner; |
71 | 71 | import org.apache.cassandra.dht.Token; |
| 72 | +import org.apache.cassandra.exceptions.RequestFailure; |
72 | 73 | import org.apache.cassandra.locator.InetAddressAndPort; |
73 | 74 | import org.apache.cassandra.net.Message; |
74 | 75 | import org.apache.cassandra.net.MessagingService; |
@@ -659,6 +660,7 @@ static int getMaxEndpointStateVersion(EndpointState epState) |
659 | 660 | public void evictFromMembership(InetAddressAndPort endpoint) |
660 | 661 | { |
661 | 662 | checkProperThreadForStateMutation(); |
| 663 | + inflightEcho.remove(endpoint); |
662 | 664 | unreachableEndpoints.remove(endpoint); |
663 | 665 | endpointStateMap.remove(endpoint); |
664 | 666 | expireTimeEndpointMap.remove(endpoint); |
@@ -1201,13 +1203,30 @@ private void markAlive(final InetAddressAndPort addr, final EndpointState localS |
1201 | 1203 | { |
1202 | 1204 | Message<NoPayload> echoMessage = Message.out(ECHO_REQ, noPayload); |
1203 | 1205 | logger.trace("Sending ECHO_REQ to {}", addr); |
1204 | | - RequestCallback echoHandler = msg -> |
| 1206 | + RequestCallback echoHandler = new RequestCallback() |
1205 | 1207 | { |
1206 | | - runInGossipStageBlocking(() -> { |
1207 | | - EndpointState epState = inflightEcho.remove(addr); |
1208 | | - if (epState != null) |
1209 | | - realMarkAlive(addr, epState); |
1210 | | - }); |
| 1208 | + @Override |
| 1209 | + public void onResponse(Message msg) |
| 1210 | + { |
| 1211 | + runInGossipStageBlocking(() -> { |
| 1212 | + EndpointState epState = inflightEcho.remove(addr); |
| 1213 | + if (epState != null) |
| 1214 | + realMarkAlive(addr, epState); |
| 1215 | + }); |
| 1216 | + } |
| 1217 | + |
| 1218 | + @Override |
| 1219 | + public boolean invokeOnFailure() |
| 1220 | + { |
| 1221 | + return true; |
| 1222 | + } |
| 1223 | + |
| 1224 | + @Override |
| 1225 | + public void onFailure(InetAddressAndPort from, RequestFailure requestFailure) |
| 1226 | + { |
| 1227 | + logger.trace("ECHO_REQ to {} failed ({})", addr, requestFailure); |
| 1228 | + inflightEcho.remove(addr); |
| 1229 | + } |
1211 | 1230 | }; |
1212 | 1231 | MessagingService.instance().sendWithCallback(echoMessage, addr, echoHandler); |
1213 | 1232 | } |
|
0 commit comments