From 02b67a54a581030b62f3d1ecb23f6d6dd7abf0bf Mon Sep 17 00:00:00 2001 From: Yannick Welsch Date: Wed, 23 May 2018 13:25:41 +0200 Subject: [PATCH] Use correct cluster state version for node fault detection (#30810) Since its introduction in ES 1.4, node fault detection has been using the wrong cluster state version to send as part of the ping request, by using always the constant -1 (ClusterState.UNKNOWN_VERSION). This can, in an unfortunate series of events, lead to a situation where a previous stale master can regain its authority and revert the cluster to an older state. This commit makes NodesFaultDetection use the correct current cluster state for sending ping requests, avoiding the situation where a stale master possibly forces a newer master to step down and rejoin the stale one. --- .../discovery/zen/NodesFaultDetection.java | 16 +++++++++++----- .../discovery/zen/ZenDiscovery.java | 2 +- .../discovery/ZenFaultDetectionTests.java | 8 +++++--- 3 files changed, 17 insertions(+), 9 deletions(-) diff --git a/core/src/main/java/org/elasticsearch/discovery/zen/NodesFaultDetection.java b/core/src/main/java/org/elasticsearch/discovery/zen/NodesFaultDetection.java index 5cd02a52504f5..c1824fdf99693 100644 --- a/core/src/main/java/org/elasticsearch/discovery/zen/NodesFaultDetection.java +++ b/core/src/main/java/org/elasticsearch/discovery/zen/NodesFaultDetection.java @@ -67,13 +67,16 @@ public void onPingReceived(PingRequest pingRequest) {} private final ConcurrentMap nodesFD = newConcurrentMap(); - private volatile long clusterStateVersion = ClusterState.UNKNOWN_VERSION; + private final java.util.function.Supplier clusterStateSupplier; private volatile DiscoveryNode localNode; - public NodesFaultDetection(Settings settings, ThreadPool threadPool, TransportService transportService, ClusterName clusterName) { + public NodesFaultDetection(Settings settings, ThreadPool threadPool, TransportService transportService, + java.util.function.Supplier clusterStateSupplier, ClusterName clusterName) { super(settings, threadPool, transportService, clusterName); + this.clusterStateSupplier = clusterStateSupplier; + logger.debug("[node ] uses ping_interval [{}], ping_timeout [{}], ping_retries [{}]", pingInterval, pingRetryTimeout, pingRetryCount); @@ -213,15 +216,18 @@ private boolean running() { return NodeFD.this.equals(nodesFD.get(node)); } + private PingRequest newPingRequest() { + return new PingRequest(node, clusterName, localNode, clusterStateSupplier.get().version()); + } + @Override public void run() { if (!running()) { return; } - final PingRequest pingRequest = new PingRequest(node, clusterName, localNode, clusterStateVersion); final TransportRequestOptions options = TransportRequestOptions.builder().withType(TransportRequestOptions.Type.PING) .withTimeout(pingRetryTimeout).build(); - transportService.sendRequest(node, PING_ACTION_NAME, pingRequest, options, new TransportResponseHandler() { + transportService.sendRequest(node, PING_ACTION_NAME, newPingRequest(), options, new TransportResponseHandler() { @Override public PingResponse newInstance() { return new PingResponse(); @@ -264,7 +270,7 @@ public void handleException(TransportException exp) { } } else { // resend the request, not reschedule, rely on send timeout - transportService.sendRequest(node, PING_ACTION_NAME, pingRequest, options, this); + transportService.sendRequest(node, PING_ACTION_NAME, newPingRequest(), options, this); } } diff --git a/core/src/main/java/org/elasticsearch/discovery/zen/ZenDiscovery.java b/core/src/main/java/org/elasticsearch/discovery/zen/ZenDiscovery.java index 8ea0c2e42c629..abf072b69cd0f 100644 --- a/core/src/main/java/org/elasticsearch/discovery/zen/ZenDiscovery.java +++ b/core/src/main/java/org/elasticsearch/discovery/zen/ZenDiscovery.java @@ -190,7 +190,7 @@ public ZenDiscovery(Settings settings, ThreadPool threadPool, TransportService t this.masterFD = new MasterFaultDetection(settings, threadPool, transportService, clusterService); this.masterFD.addListener(new MasterNodeFailureListener()); - this.nodesFD = new NodesFaultDetection(settings, threadPool, transportService, clusterService.getClusterName()); + this.nodesFD = new NodesFaultDetection(settings, threadPool, transportService, this::clusterState, clusterService.getClusterName()); this.nodesFD.addListener(new NodeFaultDetectionListener()); this.publishClusterState = diff --git a/core/src/test/java/org/elasticsearch/discovery/ZenFaultDetectionTests.java b/core/src/test/java/org/elasticsearch/discovery/ZenFaultDetectionTests.java index 11bd8e8ea6b4a..761f7f5a5febc 100644 --- a/core/src/test/java/org/elasticsearch/discovery/ZenFaultDetectionTests.java +++ b/core/src/test/java/org/elasticsearch/discovery/ZenFaultDetectionTests.java @@ -184,17 +184,19 @@ public void testNodesFaultDetectionConnectOnDisconnect() throws InterruptedExcep final Settings pingSettings = Settings.builder() .put(FaultDetection.CONNECT_ON_NETWORK_DISCONNECT_SETTING.getKey(), shouldRetry) .put(FaultDetection.PING_INTERVAL_SETTING.getKey(), "5m").build(); - ClusterState clusterState = ClusterState.builder(new ClusterName("test")).nodes(buildNodesForA(true)).build(); + ClusterState clusterState = ClusterState.builder(new ClusterName("test")).version(randomNonNegativeLong()) + .nodes(buildNodesForA(true)).build(); NodesFaultDetection nodesFDA = new NodesFaultDetection(Settings.builder().put(settingsA).put(pingSettings).build(), - threadPool, serviceA, clusterState.getClusterName()); + threadPool, serviceA, () -> clusterState, clusterState.getClusterName()); nodesFDA.setLocalNode(nodeA); NodesFaultDetection nodesFDB = new NodesFaultDetection(Settings.builder().put(settingsB).put(pingSettings).build(), - threadPool, serviceB, clusterState.getClusterName()); + threadPool, serviceB, () -> clusterState, clusterState.getClusterName()); nodesFDB.setLocalNode(nodeB); final CountDownLatch pingSent = new CountDownLatch(1); nodesFDB.addListener(new NodesFaultDetection.Listener() { @Override public void onPingReceived(NodesFaultDetection.PingRequest pingRequest) { + assertThat(pingRequest.clusterStateVersion(), equalTo(clusterState.version())); pingSent.countDown(); } });