[Zen2] Allow Setting a List of Bootstrap Nodes to Wait for

server/src/main/java/org/elasticsearch/action/admin/cluster/bootstrap/GetDiscoveredNodesRequest.java

@@ -26,6 +26,8 @@
 import org.elasticsearch.common.unit.TimeValue;
 
 import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
 
 /**
  * Request the set of master-eligible nodes discovered by this node. Most useful in a brand-new cluster as a precursor to setting the
@@ -38,6 +40,8 @@ public class GetDiscoveredNodesRequest extends ActionRequest {
     @Nullable // if the request should wait indefinitely
     private TimeValue timeout = TimeValue.timeValueSeconds(30);
 
+    private List<String> requiredNodes = Collections.emptyList();
+
     public GetDiscoveredNodesRequest() {
     }
 
@@ -95,6 +99,14 @@ public TimeValue getTimeout() {
         return timeout;
     }
 
+    public List<String> getRequiredNodes() {
+        return requiredNodes;
+    }
+
+    public void setRequiredNodes(final List<String> requiredNodes) {
+        this.requiredNodes = requiredNodes;
+    }
+
     @Override
     public ActionRequestValidationException validate() {
         return null;
@@ -110,13 +122,14 @@ public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         out.writeInt(waitForNodes);
         out.writeOptionalTimeValue(timeout);
+        out.writeStringList(requiredNodes);
     }
 
     @Override
     public String toString() {
         return "GetDiscoveredNodesRequest{" +
             "waitForNodes=" + waitForNodes +
             ", timeout=" + timeout +
-            '}';
+            ", requiredNodes=[" + String.join(",", requiredNodes) + "]}";
     }
 }

server/src/main/java/org/elasticsearch/action/admin/cluster/bootstrap/TransportGetDiscoveredNodesAction.java

@@ -37,6 +37,7 @@
 import org.elasticsearch.threadpool.ThreadPool.Names;
 import org.elasticsearch.transport.TransportService;
 
+import java.util.HashSet;
 import java.util.LinkedHashSet;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
@@ -93,7 +94,7 @@ public void accept(Iterable<DiscoveryNode> nodes) {
                 nodesSet.add(localNode);
                 nodes.forEach(nodesSet::add);
                 logger.trace("discovered {}", nodesSet);
-                if (nodesSet.size() >= request.getWaitForNodes() && listenerNotified.compareAndSet(false, true)) {
+                if (checkWaitRequirements(request, nodesSet) && listenerNotified.compareAndSet(false, true)) {
                     listenableFuture.onResponse(new GetDiscoveredNodesResponse(nodesSet));
                 }
             }
@@ -124,4 +125,16 @@ public String toString() {
             });
         }
     }
+
+    private static boolean checkWaitRequirements(GetDiscoveredNodesRequest request, Set<DiscoveryNode> nodes) {
+        Set<String> requiredNodes = new HashSet<>(request.getRequiredNodes());
+        for (final DiscoveryNode node : nodes) {
+            requiredNodes.remove(node.getAddress().toString());
+            requiredNodes.remove(node.getName());
+            if (requiredNodes.isEmpty()) {
+                break;
+            }
+        }
+        return requiredNodes.isEmpty() && nodes.size() >= request.getWaitForNodes();
+    }
 }

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

@@ -41,6 +41,9 @@
 import org.elasticsearch.transport.TransportService;
 
 import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.function.Function;
 
 public class ClusterBootstrapService {
 
@@ -51,12 +54,17 @@ public class ClusterBootstrapService {
     public static final Setting<Integer> INITIAL_MASTER_NODE_COUNT_SETTING =
         Setting.intSetting("cluster.unsafe_initial_master_node_count", 0, 0, Property.NodeScope);
 
+    public static final Setting<List<String>> INITIAL_MASTER_NODES_SETTING =
+        Setting.listSetting("cluster.unsafe_initial_master_nodes", Collections.emptyList(), Function.identity(), Property.NodeScope);
+
     private final int initialMasterNodeCount;
+    private final List<String> initialMasterNodes;
     private final TransportService transportService;
     private volatile boolean running;
 
     public ClusterBootstrapService(Settings settings, TransportService transportService) {
         initialMasterNodeCount = INITIAL_MASTER_NODE_COUNT_SETTING.get(settings);
+        initialMasterNodes = INITIAL_MASTER_NODES_SETTING.get(settings);
         this.transportService = transportService;
     }
 
@@ -73,6 +81,7 @@ public void start() {
 
                 final GetDiscoveredNodesRequest request = new GetDiscoveredNodesRequest();
                 request.setWaitForNodes(initialMasterNodeCount);
+                request.setRequiredNodes(initialMasterNodes);
                 request.setTimeout(null);
                 logger.trace("sending {}", request);
                 transportService.sendRequest(transportService.getLocalNode(), GetDiscoveredNodesAction.NAME, request,

server/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java

@@ -470,6 +470,7 @@ public void apply(Settings value, Settings current, Settings previous) {
                     LeaderChecker.LEADER_CHECK_RETRY_COUNT_SETTING,
                     Reconfigurator.CLUSTER_AUTO_SHRINK_VOTING_CONFIGURATION,
                     TransportAddVotingTombstonesAction.MAXIMUM_VOTING_TOMBSTONES_SETTING,
+                    ClusterBootstrapService.INITIAL_MASTER_NODES_SETTING,
                     ClusterBootstrapService.INITIAL_MASTER_NODE_COUNT_SETTING,
                     LagDetector.CLUSTER_FOLLOWER_LAG_TIMEOUT_SETTING
             )));

server/src/test/java/org/elasticsearch/client/transport/TransportClientIT.java

@@ -40,6 +40,7 @@
 
 import java.io.IOException;
 import java.util.Arrays;
+import java.util.Collections;
 
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
@@ -67,7 +68,9 @@ public void testNodeVersionIsUpdated() throws IOException, NodeValidationExcepti
                 .put(Node.NODE_DATA_SETTING.getKey(), false)
                 .put("cluster.name", "foobar")
                 .put(TestZenDiscovery.USE_ZEN2.getKey(), true)
-                .put(ClusterBootstrapService.INITIAL_MASTER_NODE_COUNT_SETTING.getKey(), 1)
+                .putList(
+                    ClusterBootstrapService.INITIAL_MASTER_NODES_SETTING.getKey(), Collections.singletonList("testNodeVersionIsUpdated")
+                ).put(ClusterBootstrapService.INITIAL_MASTER_NODE_COUNT_SETTING.getKey(), 1)
                 .build(), Arrays.asList(getTestTransportPlugin(), TestZenDiscovery.TestPlugin.class,
                                         MockHttpTransport.TestPlugin.class)).start()) {
             TransportAddress transportAddress = node.injector().getInstance(TransportService.class).boundAddress().publishAddress();