MasterNodeChangePredicate should use the node instance to detect master change (#25877)

This predicate is used to deal with the intricacies of detecting when a master is reelected/nodes rejoins an existing master. The current implementation is based on nodeIds, which is fine if the master really change. If the nodeId is equal the code falls back to detecting an increment in the cluster state version which happens when a node is re-elected or when the node rejoins. Sadly this doesn't cover the case where the same node is elected after a full restart of all master nodes. In that case we recover the cluster state from disk but the version is reset back to 0. To fix this, the check should be done based on ephemeral IDs which are reset on restart.

Fixes #25471

Author: bleskes

core/src/main/java/org/elasticsearch/cluster/MasterNodeChangePredicate.java

@@ -19,6 +19,8 @@
 
 package org.elasticsearch.cluster;
 
+import org.elasticsearch.cluster.node.DiscoveryNode;
+
 import java.util.function.Predicate;
 
 public final class MasterNodeChangePredicate {
@@ -33,13 +35,14 @@ private MasterNodeChangePredicate() {
      */
     public static Predicate<ClusterState> build(ClusterState currentState) {
         final long currentVersion = currentState.version();
-        final String currentMaster = currentState.nodes().getMasterNodeId();
+        final DiscoveryNode masterNode = currentState.nodes().getMasterNode();
+        final String currentMasterId = masterNode == null ? null : masterNode.getEphemeralId();
         return newState -> {
-            final String newMaster = newState.nodes().getMasterNodeId();
+            final DiscoveryNode newMaster = newState.nodes().getMasterNode();
             final boolean accept;
             if (newMaster == null) {
                 accept = false;
-            } else if (newMaster.equals(currentMaster) == false){
+            } else if (newMaster.getEphemeralId().equals(currentMasterId) == false) {
                 accept = true;
             } else {
                 accept = newState.version() > currentVersion;

core/src/test/java/org/elasticsearch/action/support/master/TransportMasterNodeActionTests.java

@@ -36,6 +36,7 @@
 import org.elasticsearch.cluster.block.ClusterBlocks;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.transport.LocalTransportAddress;
@@ -299,21 +300,57 @@ public void testDelegateToMaster() throws ExecutionException, InterruptedExcepti
     }
 
     public void testDelegateToFailingMaster() throws ExecutionException, InterruptedException {
-        boolean failsWithConnectTransportException = randomBoolean();
+        final boolean failsWithConnectTransportException = randomBoolean();
+        final boolean rejoinSameMaster = failsWithConnectTransportException && randomBoolean();
         Request request = new Request().masterNodeTimeout(TimeValue.timeValueSeconds(failsWithConnectTransportException ? 60 : 0));
-        setState(clusterService, ClusterStateCreationUtils.state(localNode, remoteNode, allNodes));
+        DiscoveryNode masterNode = this.remoteNode;
+        setState(clusterService, ClusterState.builder(ClusterStateCreationUtils.state(localNode, masterNode, allNodes))
+            .version(randomIntBetween(0, 10))); // use a random base version so it can go down when simulating a restart.
 
         PlainActionFuture<Response> listener = new PlainActionFuture<>();
         new Action(Settings.EMPTY, "testAction", transportService, clusterService, threadPool).execute(request, listener);
 
-        assertThat(transport.capturedRequests().length, equalTo(1));
-        CapturingTransport.CapturedRequest capturedRequest = transport.capturedRequests()[0];
+        CapturingTransport.CapturedRequest[] capturedRequests = transport.getCapturedRequestsAndClear();
+        assertThat(capturedRequests.length, equalTo(1));
+        CapturingTransport.CapturedRequest capturedRequest = capturedRequests[0];
         assertTrue(capturedRequest.node.isMasterNode());
         assertThat(capturedRequest.request, equalTo(request));
         assertThat(capturedRequest.action, equalTo("testAction"));
 
-        if (failsWithConnectTransportException) {
-            transport.handleRemoteError(capturedRequest.requestId, new ConnectTransportException(remoteNode, "Fake error"));
+        if (rejoinSameMaster) {
+            transport.handleRemoteError(capturedRequest.requestId, new ConnectTransportException(masterNode, "Fake error"));
+            assertFalse(listener.isDone());
+            if (randomBoolean()) {
+                // simulate master node removal
+                final DiscoveryNodes.Builder nodesBuilder = DiscoveryNodes.builder(clusterService.state().nodes());
+                nodesBuilder.masterNodeId(null);
+                setState(clusterService, ClusterState.builder(clusterService.state()).nodes(nodesBuilder));
+            }
+            if (randomBoolean()) {
+                // reset the same state to increment a version simulating a join of an existing node
+                // simulating use being disconnected
+                final DiscoveryNodes.Builder nodesBuilder = DiscoveryNodes.builder(clusterService.state().nodes());
+                nodesBuilder.masterNodeId(masterNode.getId());
+                setState(clusterService, ClusterState.builder(clusterService.state()).nodes(nodesBuilder));
+            } else {
+                // simulate master restart followed by a state recovery - this will reset the cluster state version
+                final DiscoveryNodes.Builder nodesBuilder = DiscoveryNodes.builder(clusterService.state().nodes());
+                nodesBuilder.remove(masterNode);
+                masterNode = new DiscoveryNode(masterNode.getId(), masterNode.getAddress(), masterNode.getVersion());
+                nodesBuilder.add(masterNode);
+                nodesBuilder.masterNodeId(masterNode.getId());
+                final ClusterState.Builder builder = ClusterState.builder(clusterService.state()).nodes(nodesBuilder);
+                setState(clusterService, builder.version(0));
+            }
+            assertFalse(listener.isDone());
+            capturedRequests = transport.getCapturedRequestsAndClear();
+            assertThat(capturedRequests.length, equalTo(1));
+            capturedRequest = capturedRequests[0];
+            assertTrue(capturedRequest.node.isMasterNode());
+            assertThat(capturedRequest.request, equalTo(request));
+            assertThat(capturedRequest.action, equalTo("testAction"));
+        } else if (failsWithConnectTransportException) {
+            transport.handleRemoteError(capturedRequest.requestId, new ConnectTransportException(masterNode, "Fake error"));
             assertFalse(listener.isDone());
             setState(clusterService, ClusterStateCreationUtils.state(localNode, localNode, allNodes));
             assertTrue(listener.isDone());