Zen2 node joining (elastic#35)

Adds node joining based on a simplified MasterService abstraction.

Author: ywelsch

build.gradle

@@ -145,8 +145,8 @@ task verifyVersions {
  * the enabled state of every bwc task. It should be set back to true
  * after the backport of the backcompat code is complete.
  */
-final boolean bwc_tests_enabled = true
-final String bwc_tests_disabled_issue = "" /* place a PR link here when commiting bwc changes */
+final boolean bwc_tests_enabled = false
+final String bwc_tests_disabled_issue = "Zen2 prototype development" /* place a PR link here when commiting bwc changes */
 if (bwc_tests_enabled == false) {
   if (bwc_tests_disabled_issue.isEmpty()) {
     throw new GradleException("bwc_tests_disabled_issue must be set when bwc_tests_enabled == false")

server/src/main/java/org/elasticsearch/cluster/ClusterState.java

@@ -720,6 +720,9 @@ public static ClusterState readFrom(StreamInput in, DiscoveryNode localNode) thr
         ClusterName clusterName = new ClusterName(in);
         Builder builder = new Builder(clusterName);
         builder.version = in.readLong();
+        if (in.getVersion().onOrAfter(Version.V_7_0_0_alpha1)) {
+            builder.term = in.readLong();
+        }
         builder.uuid = in.readString();
         if (in.getVersion().onOrAfter(Version.V_7_0_0_alpha1)) {
             builder.lastCommittedConfiguration(new VotingConfiguration(in));
@@ -741,6 +744,9 @@ public static ClusterState readFrom(StreamInput in, DiscoveryNode localNode) thr
     public void writeTo(StreamOutput out) throws IOException {
         clusterName.writeTo(out);
         out.writeLong(version);
+        if (out.getVersion().onOrAfter(Version.V_7_0_0_alpha1)) {
+            out.writeLong(term);
+        }
         out.writeString(stateUUID);
         if (out.getVersion().onOrAfter(Version.V_7_0_0_alpha1)) {
             lastCommittedConfiguration.writeTo(out);
@@ -900,7 +906,7 @@ public void writeTo(StreamOutput out) throws IOException {
 
         public boolean hasQuorum(Collection<String> votes) {
             if (nodeIds.isEmpty()) {
-                throw new IllegalStateException("cannot check quorum on an empty configuration");
+                return false; // TODO: should we even allow this check on an empty configuration?
             }
             final HashSet<String> intersection = new HashSet<>(nodeIds);
             intersection.retainAll(votes);

server/src/main/java/org/elasticsearch/discovery/zen/MasterFaultDetection.java

@@ -311,7 +311,7 @@ public Throwable fillInStackTrace() {
         }
     }
 
-    static class NodeDoesNotExistOnMasterException extends IllegalStateException {
+    public static class NodeDoesNotExistOnMasterException extends IllegalStateException {
         @Override
         public Throwable fillInStackTrace() {
             return null;

server/src/main/java/org/elasticsearch/discovery/zen/MembershipAction.java

@@ -205,7 +205,7 @@ public void messageReceived(ValidateJoinRequest request, TransportChannel channe
      * @see Version#minimumIndexCompatibilityVersion()
      * @throws IllegalStateException if any index is incompatible with the given version
      */
-    static void ensureIndexCompatibility(final Version nodeVersion, MetaData metaData) {
+    public static void ensureIndexCompatibility(final Version nodeVersion, MetaData metaData) {
         Version supportedIndexVersion = nodeVersion.minimumIndexCompatibilityVersion();
         // we ensure that all indices in the cluster we join are compatible with us no matter if they are
         // closed or not we can't read mappings of these indices so we need to reject the join...
@@ -229,7 +229,7 @@ static void ensureNodesCompatibility(final Version joiningNodeVersion, Discovery
     }
 
     /** ensures that the joining node has a version that's compatible with a given version range */
-    static void ensureNodesCompatibility(Version joiningNodeVersion, Version minClusterNodeVersion, Version maxClusterNodeVersion) {
+    public static void ensureNodesCompatibility(Version joiningNodeVersion, Version minClusterNodeVersion, Version maxClusterNodeVersion) {
         assert minClusterNodeVersion.onOrBefore(maxClusterNodeVersion) : minClusterNodeVersion + " > " + maxClusterNodeVersion;
         if (joiningNodeVersion.isCompatible(maxClusterNodeVersion) == false) {
             throw new IllegalStateException("node version [" + joiningNodeVersion + "] is not supported. " +
@@ -246,7 +246,7 @@ static void ensureNodesCompatibility(Version joiningNodeVersion, Version minClus
      * to ensure that if the master is already fully operating under the new major version, it doesn't go back to mixed
      * version mode
      **/
-    static void ensureMajorVersionBarrier(Version joiningNodeVersion, Version minClusterNodeVersion) {
+    public static void ensureMajorVersionBarrier(Version joiningNodeVersion, Version minClusterNodeVersion) {
         final byte clusterMajor = minClusterNodeVersion.major;
         if (joiningNodeVersion.major < clusterMajor) {
             throw new IllegalStateException("node version [" + joiningNodeVersion + "] is not supported. " +

server/src/main/java/org/elasticsearch/discovery/zen/NodeJoinController.java

@@ -344,11 +344,11 @@ public void onFailure(String source, Exception e) {
 
     }
 
-    static class JoinTaskListener implements ClusterStateTaskListener {
+    public static class JoinTaskListener implements ClusterStateTaskListener {
         final List<MembershipAction.JoinCallback> callbacks;
         private final Logger logger;
 
-        JoinTaskListener(MembershipAction.JoinCallback callback, Logger logger) {
+        public JoinTaskListener(MembershipAction.JoinCallback callback, Logger logger) {
             this(Collections.singletonList(callback), logger);
         }
 

server/src/main/java/org/elasticsearch/discovery/zen2/ConsensusState.java

@@ -103,6 +103,10 @@ public long getLastPublishedVersion() {
         return lastPublishedVersion;
     }
 
+    public boolean hasElectionQuorum(VotingConfiguration votingConfiguration) {
+        return joinVotes.isQuorum(votingConfiguration);
+    }
+
     /**
      * May be safely called at any time to move this instance to a new term. It is vitally important for safety that
      * the resulting Join is sent to no more than one node.
@@ -214,7 +218,7 @@ && getLastCommittedConfiguration().equals(getLastAcceptedConfiguration()) == fal
             logger.debug("handleClientValue: only allow reconfiguration while not already reconfiguring");
             throw new ConsensusMessageRejectedException("only allow reconfiguration while not already reconfiguring");
         }
-        if (clusterState.getLastAcceptedConfiguration().hasQuorum(joinVotes.nodes.keySet()) == false) {
+        if (hasElectionQuorum(clusterState.getLastAcceptedConfiguration()) == false) {
             logger.debug("handleClientValue: only allow reconfiguration if join quorum available for new config");
             throw new ConsensusMessageRejectedException("only allow reconfiguration if join quorum available for new config");
         }