[Discovery] when master is gone, flush all pending cluster states

If the master FD flags master as gone while there are still pending cluster states, the processing of those cluster states we re-instate that node a master again. Closes #6526
elastic · Jun 17, 2014 · 797b4b5 · 797b4b5
1 parent 84ba369
commit 797b4b5
Showing 1 changed file with 11 additions and 0 deletions.
diff --git a/src/main/java/org/elasticsearch/discovery/zen/ZenDiscovery.java b/src/main/java/org/elasticsearch/discovery/zen/ZenDiscovery.java
@@ -61,6 +61,7 @@
 import org.elasticsearch.transport.*;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -522,6 +523,11 @@ public ClusterState execute(ClusterState currentState) {
                         .masterNodeId(null).build();
                 latestDiscoNodes = discoveryNodes;
 
+                // flush any pending cluster states from old master, so it will not be set as master again
+                ArrayList<ProcessClusterState> pendingNewClusterStates = new ArrayList<>();
+                processNewClusterStates.drainTo(pendingNewClusterStates);
+                logger.trace("removed [{}] pending cluster states", pendingNewClusterStates.size());
+
                 if (rejoinOnMasterGone) {
                     return rejoin(ClusterState.builder(currentState).nodes(discoveryNodes).build(), "master left (reason = " + reason + ")");
                 }
@@ -668,6 +674,11 @@ public ClusterState execute(ClusterState currentState) {
 
                             // we are going to use it for sure, poll (remove) it
                             potentialState = processNewClusterStates.poll();
+                            if (potentialState == null) {
+                                // might happen if the queue is drained
+                                break;
+                            }
+
                             potentialState.processed = true;
 
                             if (potentialState.clusterState.version() > stateToProcess.clusterState.version()) {