[Zen2] Only elect master-eligible nodes

server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java

@@ -344,6 +344,7 @@ private Join joinLeaderInTerm(StartJoinRequest startJoinRequest) {
 
     private void handleJoinRequest(JoinRequest joinRequest, JoinHelper.JoinCallback joinCallback) {
         assert Thread.holdsLock(mutex) == false;
+        assert getLocalNode().isMasterNode() : getLocalNode() + " received a join but is not master-eligible";
         logger.trace("handleJoinRequest: as {}, handling {}", mode, joinRequest);
         transportService.connectToNode(joinRequest.getSourceNode());
 
@@ -392,6 +393,8 @@ void becomeCandidate(String method) {
     void becomeLeader(String method) {
         assert Thread.holdsLock(mutex) : "Coordinator mutex not held";
         assert mode == Mode.CANDIDATE : "expected candidate but was " + mode;
+        assert getLocalNode().isMasterNode() : getLocalNode() + " became a leader but is not master-eligible";
+
         logger.debug("{}: becoming LEADER (was {}, lastKnownLeader was [{}])", method, mode, lastKnownLeader);
 
         mode = Mode.LEADER;
@@ -409,6 +412,8 @@ void becomeLeader(String method) {
 
     void becomeFollower(String method, DiscoveryNode leaderNode) {
         assert Thread.holdsLock(mutex) : "Coordinator mutex not held";
+        assert leaderNode.isMasterNode() : leaderNode + " became a leader but is not master-eligible";
+
         logger.debug("{}: becoming FOLLOWER of [{}] (was {}, lastKnownLeader was [{}])", method, leaderNode, mode, lastKnownLeader);
 
         final boolean restartLeaderChecker = (mode == Mode.FOLLOWER && Optional.of(leaderNode).equals(lastKnownLeader)) == false;
@@ -642,9 +647,9 @@ public boolean setInitialConfiguration(final VotingConfiguration votingConfigura
             logger.info("setting initial configuration to {}", votingConfiguration);
             final Builder builder = masterService.incrementVersion(currentState);
             final CoordinationMetaData coordinationMetaData = CoordinationMetaData.builder(currentState.coordinationMetaData())
-                    .lastAcceptedConfiguration(votingConfiguration)
-                    .lastCommittedConfiguration(votingConfiguration)
-                    .build();
+                .lastAcceptedConfiguration(votingConfiguration)
+                .lastCommittedConfiguration(votingConfiguration)
+                .build();
 
             MetaData.Builder metaDataBuilder = MetaData.builder();
             // automatically generate a UID for the metadata if we need to
@@ -927,6 +932,11 @@ protected void onFoundPeersUpdated() {
 
     private void startElectionScheduler() {
         assert electionScheduler == null : electionScheduler;
+
+        if (getLocalNode().isMasterNode() == false) {
+            return;
+        }
+
         final TimeValue gracePeriod = TimeValue.ZERO; // TODO variable grace period
         electionScheduler = electionSchedulerFactory.startElectionScheduler(gracePeriod, new Runnable() {
             @Override

server/src/main/java/org/elasticsearch/cluster/coordination/JoinHelper.java

@@ -166,6 +166,7 @@ public String toString() {
     }
 
     public void sendJoinRequest(DiscoveryNode destination, Optional<Join> optionalJoin) {
+        assert destination.isMasterNode() : "trying to join master-ineligible " + destination;
         final JoinRequest joinRequest = new JoinRequest(transportService.getLocalNode(), optionalJoin);
         final Tuple<DiscoveryNode, JoinRequest> dedupKey = Tuple.tuple(destination, joinRequest);
         if (pendingOutgoingJoins.add(dedupKey)) {
@@ -210,6 +211,8 @@ public String executor() {
     }
 
     public void sendStartJoinRequest(final StartJoinRequest startJoinRequest, final DiscoveryNode destination) {
+        assert startJoinRequest.getSourceNode().isMasterNode()
+            : "sending start-join request for master-ineligible " + startJoinRequest.getSourceNode();
         transportService.sendRequest(destination, START_JOIN_ACTION_NAME,
             startJoinRequest, new TransportResponseHandler<Empty>() {
                 @Override

server/src/test/java/org/elasticsearch/cluster/coordination/Zen2IT.java

@@ -0,0 +1,37 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.elasticsearch.cluster.coordination;
+
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.test.ESIntegTestCase;
+
+import java.io.IOException;
+
+@ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST, numDataNodes = 0)
+public class Zen2IT extends ESIntegTestCase {
+    public void testNonMasterNodeDoesNotDisruptElections() throws IOException {
+        // Non-master nodes cannot publish cluster states (they have no STATE channels) but can theoretically win elections, which disrupts
+        // the cluster and prevents master-eligible nodes from winning their elections, repeatedly.
+        internalCluster().startNodes(2);
+        internalCluster().startDataOnlyNode();
+        internalCluster().stopCurrentMasterNode();
+        // There's only one master-eligible node left so it should elect itself pretty quickly.
+        assertFalse(client().admin().cluster().prepareHealth().setTimeout(TimeValue.timeValueSeconds(10)).get().isTimedOut());
+    }
+}