Disc: Move AbstractDisruptionTC to filebased D. (#34461)

* Discovery: Move AbstractDisruptionTestCase to file-based discovery.
* Relates #33675
* Simplify away ClusterDiscoveryConfiguration

Author: original-brownbear

server/src/test/java/org/elasticsearch/discovery/AbstractDisruptionTestCase.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.discovery;
 
+import java.nio.file.Path;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.block.ClusterBlock;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
@@ -33,7 +34,8 @@
 import org.elasticsearch.env.NodeEnvironment;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.test.ESIntegTestCase;
-import org.elasticsearch.test.discovery.ClusterDiscoveryConfiguration;
+import org.elasticsearch.test.InternalTestCluster;
+import org.elasticsearch.test.NodeConfigurationSource;
 import org.elasticsearch.test.discovery.TestZenDiscovery;
 import org.elasticsearch.test.disruption.NetworkDisruption;
 import org.elasticsearch.test.disruption.NetworkDisruption.Bridge;
@@ -52,17 +54,17 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
-import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 
+import static org.elasticsearch.discovery.DiscoveryModule.DISCOVERY_HOSTS_PROVIDER_SETTING;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.not;
 
 public abstract class AbstractDisruptionTestCase extends ESIntegTestCase {
 
     static final TimeValue DISRUPTION_HEALING_OVERHEAD = TimeValue.timeValueSeconds(40); // we use 30s as timeout in many places.
 
-    private ClusterDiscoveryConfiguration discoveryConfig;
+    private NodeConfigurationSource discoveryConfig;
 
     @Override
     protected Settings nodeSettings(int nodeOrdinal) {
@@ -116,18 +118,14 @@ protected void beforeIndexDeletion() throws Exception {
         }
     }
 
-    List<String> startCluster(int numberOfNodes) throws ExecutionException, InterruptedException {
+    List<String> startCluster(int numberOfNodes) {
         return startCluster(numberOfNodes, -1);
     }
 
-    List<String> startCluster(int numberOfNodes, int minimumMasterNode) throws ExecutionException, InterruptedException {
-        return startCluster(numberOfNodes, minimumMasterNode, null);
-    }
-
-    List<String> startCluster(int numberOfNodes, int minimumMasterNode, @Nullable int[] unicastHostsOrdinals) throws
-            ExecutionException, InterruptedException {
-        configureCluster(numberOfNodes, unicastHostsOrdinals, minimumMasterNode);
-        List<String> nodes = internalCluster().startNodes(numberOfNodes);
+    List<String> startCluster(int numberOfNodes, int minimumMasterNode) {
+        configureCluster(numberOfNodes, minimumMasterNode);
+        InternalTestCluster internalCluster = internalCluster();
+        List<String> nodes = internalCluster.startNodes(numberOfNodes);
         ensureStableCluster(numberOfNodes);
 
         // TODO: this is a temporary solution so that nodes will not base their reaction to a partition based on previous successful results
@@ -154,20 +152,11 @@ protected Collection<Class<? extends Plugin>> nodePlugins() {
         return Arrays.asList(MockTransportService.TestPlugin.class);
     }
 
-    void configureCluster(
-            int numberOfNodes,
-            @Nullable int[] unicastHostsOrdinals,
-            int minimumMasterNode
-    ) throws ExecutionException, InterruptedException {
-        configureCluster(DEFAULT_SETTINGS, numberOfNodes, unicastHostsOrdinals, minimumMasterNode);
+    void configureCluster(int numberOfNodes, int minimumMasterNode) {
+        configureCluster(DEFAULT_SETTINGS, numberOfNodes, minimumMasterNode);
     }
 
-    void configureCluster(
-            Settings settings,
-            int numberOfNodes,
-            @Nullable int[] unicastHostsOrdinals,
-            int minimumMasterNode
-    ) throws ExecutionException, InterruptedException {
+    void configureCluster(Settings settings, int numberOfNodes, int minimumMasterNode) {
         if (minimumMasterNode < 0) {
             minimumMasterNode = numberOfNodes / 2 + 1;
         }
@@ -177,14 +166,21 @@ void configureCluster(
                 .put(settings)
                 .put(NodeEnvironment.MAX_LOCAL_STORAGE_NODES_SETTING.getKey(), numberOfNodes)
                 .put(ElectMasterService.DISCOVERY_ZEN_MINIMUM_MASTER_NODES_SETTING.getKey(), minimumMasterNode)
+                .putList(DISCOVERY_HOSTS_PROVIDER_SETTING.getKey(), "file")
                 .build();
 
         if (discoveryConfig == null) {
-            if (unicastHostsOrdinals == null) {
-                discoveryConfig = new ClusterDiscoveryConfiguration.UnicastZen(numberOfNodes, nodeSettings);
-            } else {
-                discoveryConfig = new ClusterDiscoveryConfiguration.UnicastZen(numberOfNodes, nodeSettings, unicastHostsOrdinals);
-            }
+            discoveryConfig = new NodeConfigurationSource() {
+                @Override
+                public Settings nodeSettings(final int nodeOrdinal) {
+                    return nodeSettings;
+                }
+
+                @Override
+                public Path nodeConfigPath(final int nodeOrdinal) {
+                    return null;
+                }
+            };
         }
     }
 

server/src/test/java/org/elasticsearch/discovery/ClusterDisruptionIT.java

@@ -363,7 +363,7 @@ public void onFailure(Exception e) {
      */
     public void testSearchWithRelocationAndSlowClusterStateProcessing() throws Exception {
         // don't use DEFAULT settings (which can cause node disconnects on a slow CI machine)
-        configureCluster(Settings.EMPTY, 3, null, 1);
+        configureCluster(Settings.EMPTY, 3, 1);
         internalCluster().startMasterOnlyNode();
         final String node_1 = internalCluster().startDataOnlyNode();
 
@@ -390,7 +390,7 @@ public void testSearchWithRelocationAndSlowClusterStateProcessing() throws Excep
 
     public void testIndexImportedFromDataOnlyNodesIfMasterLostDataFolder() throws Exception {
         // test for https://github.com/elastic/elasticsearch/issues/8823
-        configureCluster(2, null, 1);
+        configureCluster(2, 1);
         String masterNode = internalCluster().startMasterOnlyNode(Settings.EMPTY);
         internalCluster().startDataOnlyNode(Settings.EMPTY);
 
@@ -421,7 +421,7 @@ public void testIndicesDeleted() throws Exception {
             .put(DiscoverySettings.COMMIT_TIMEOUT_SETTING.getKey(), "30s") // wait till cluster state is committed
             .build();
         final String idxName = "test";
-        configureCluster(settings, 3, null, 2);
+        configureCluster(settings, 3, 2);
         final List<String> allMasterEligibleNodes = internalCluster().startMasterOnlyNodes(2);
         final String dataNode = internalCluster().startDataOnlyNode();
         ensureStableCluster(3);

server/src/test/java/org/elasticsearch/discovery/DiscoveryDisruptionIT.java

@@ -59,7 +59,8 @@
 public class DiscoveryDisruptionIT extends AbstractDisruptionTestCase {
 
     public void testIsolatedUnicastNodes() throws Exception {
-        List<String> nodes = startCluster(4, -1, new int[]{0});
+        internalCluster().setHostsListContainsOnlyFirstNode(true);
+        List<String> nodes = startCluster(4, -1);
         // Figure out what is the elected master node
         final String unicastTarget = nodes.get(0);
 
@@ -98,7 +99,8 @@ public void testIsolatedUnicastNodes() throws Exception {
      * The rejoining node should take this master node and connect.
      */
     public void testUnicastSinglePingResponseContainsMaster() throws Exception {
-        List<String> nodes = startCluster(4, -1, new int[]{0});
+        internalCluster().setHostsListContainsOnlyFirstNode(true);
+        List<String> nodes = startCluster(4, -1);
         // Figure out what is the elected master node
         final String masterNode = internalCluster().getMasterName();
         logger.info("---> legit elected master node={}", masterNode);
@@ -194,7 +196,7 @@ public void testClusterJoinDespiteOfPublishingIssues() throws Exception {
     }
 
     public void testClusterFormingWithASlowNode() throws Exception {
-        configureCluster(3, null, 2);
+        configureCluster(3, 2);
 
         SlowClusterStateProcessing disruption = new SlowClusterStateProcessing(random(), 0, 0, 1000, 2000);
 
@@ -210,7 +212,7 @@ public void testClusterFormingWithASlowNode() throws Exception {
     }
 
     public void testElectMasterWithLatestVersion() throws Exception {
-        configureCluster(3, null, 2);
+        configureCluster(3, 2);
         final Set<String> nodes = new HashSet<>(internalCluster().startNodes(3));
         ensureStableCluster(3);
         ServiceDisruptionScheme isolateAllNodes =

server/src/test/java/org/elasticsearch/discovery/SnapshotDisruptionIT.java

@@ -59,7 +59,7 @@ public void testDisruptionOnSnapshotInitialization() throws Exception {
             .put(DiscoverySettings.COMMIT_TIMEOUT_SETTING.getKey(), "30s") // wait till cluster state is committed
             .build();
         final String idxName = "test";
-        configureCluster(settings, 4, null, 2);
+        configureCluster(settings, 4, 2);
         final List<String> allMasterEligibleNodes = internalCluster().startMasterOnlyNodes(3);
         final String dataNode = internalCluster().startDataOnlyNode();
         ensureStableCluster(4);

test/framework/src/main/java/org/elasticsearch/test/InternalTestCluster.java

@@ -231,6 +231,9 @@ public final class InternalTestCluster extends TestCluster {
     private ServiceDisruptionScheme activeDisruptionScheme;
     private Function<Client, Client> clientWrapper;
 
+    // If set to true only the first node in the cluster will be made a unicast node
+    private boolean hostsListContainsOnlyFirstNode;
+
     public InternalTestCluster(
             final long clusterSeed,
             final Path baseDir,
@@ -1605,12 +1608,17 @@ private synchronized void startAndPublishNodesAndClients(List<NodeAndClient> nod
 
     private final Object discoveryFileMutex = new Object();
 
-    private void rebuildUnicastHostFiles(Collection<NodeAndClient> newNodes) {
+    private void rebuildUnicastHostFiles(List<NodeAndClient> newNodes) {
         // cannot be a synchronized method since it's called on other threads from within synchronized startAndPublishNodesAndClients()
         synchronized (discoveryFileMutex) {
             try {
-                List<String> discoveryFileContents = Stream.concat(nodes.values().stream(), newNodes.stream())
-                    .map(nac -> nac.node.injector().getInstance(TransportService.class)).filter(Objects::nonNull)
+                Stream<NodeAndClient> unicastHosts = Stream.concat(nodes.values().stream(), newNodes.stream());
+                if (hostsListContainsOnlyFirstNode) {
+                    unicastHosts = unicastHosts.limit(1L);
+                }
+                List<String> discoveryFileContents = unicastHosts.map(
+                        nac -> nac.node.injector().getInstance(TransportService.class)
+                    ).filter(Objects::nonNull)
                     .map(TransportService::getLocalNode).filter(Objects::nonNull).filter(DiscoveryNode::isMasterNode)
                     .map(n -> n.getAddress().toString())
                     .distinct().collect(Collectors.toList());
@@ -2038,6 +2046,9 @@ public synchronized int numMasterNodes() {
       return filterNodes(nodes, NodeAndClient::isMasterEligible).size();
     }
 
+    public void setHostsListContainsOnlyFirstNode(boolean hostsListContainsOnlyFirstNode) {
+        this.hostsListContainsOnlyFirstNode = hostsListContainsOnlyFirstNode;
+    }
 
     public void setDisruptionScheme(ServiceDisruptionScheme scheme) {
         assert activeDisruptionScheme == null :