Tests: Add TestZenDiscovery and replace uses of MockZenPing with it (elastic#21488)

This changes adds a test discovery (which internally uses the existing
mock zenping by default). Having the mock the test framework selects be a discovery
greatly simplifies discovery setup (no more weird callback to a Node
method).

Author: rjernst

core/src/main/java/org/elasticsearch/discovery/DiscoveryModule.java

@@ -53,11 +53,9 @@ public class DiscoveryModule {
         new Setting<>("discovery.zen.hosts_provider", (String)null, Optional::ofNullable, Property.NodeScope);
 
     private final Discovery discovery;
-    private final ZenPing zenPing;
 
     public DiscoveryModule(Settings settings, ThreadPool threadPool, TransportService transportService, NetworkService networkService,
-                           ClusterService clusterService, Function<UnicastHostsProvider, ZenPing> createZenPing,
-                           List<DiscoveryPlugin> plugins) {
+                           ClusterService clusterService, List<DiscoveryPlugin> plugins) {
         final UnicastHostsProvider hostsProvider;
 
         Map<String, Supplier<UnicastHostsProvider>> hostProviders = new HashMap<>();
@@ -79,14 +77,11 @@ public DiscoveryModule(Settings settings, ThreadPool threadPool, TransportServic
             hostsProvider = Collections::emptyList;
         }
 
-        zenPing = createZenPing.apply(hostsProvider);
-
         Map<String, Supplier<Discovery>> discoveryTypes = new HashMap<>();
-        discoveryTypes.put("zen",
-            () -> new ZenDiscovery(settings, threadPool, transportService, clusterService, clusterService.getClusterSettings(), zenPing));
+        discoveryTypes.put("zen", () -> new ZenDiscovery(settings, threadPool, transportService, clusterService, hostsProvider));
         discoveryTypes.put("none", () -> new NoneDiscovery(settings, clusterService, clusterService.getClusterSettings()));
         for (DiscoveryPlugin plugin : plugins) {
-            plugin.getDiscoveryTypes(threadPool, transportService, clusterService, zenPing).entrySet().forEach(entry -> {
+            plugin.getDiscoveryTypes(threadPool, transportService, clusterService, hostsProvider).entrySet().forEach(entry -> {
                 if (discoveryTypes.put(entry.getKey(), entry.getValue()) != null) {
                     throw new IllegalArgumentException("Cannot register discovery type [" + entry.getKey() + "] twice");
                 }
@@ -103,9 +98,4 @@ public DiscoveryModule(Settings settings, ThreadPool threadPool, TransportServic
     public Discovery getDiscovery() {
         return discovery;
     }
-
-    // TODO: remove this, it should be completely local to discovery, but service disruption tests want to mess with it
-    public ZenPing getZenPing() {
-        return zenPing;
-    }
 }

core/src/main/java/org/elasticsearch/discovery/zen/ZenDiscovery.java

@@ -107,7 +107,7 @@ public class ZenDiscovery extends AbstractLifecycleComponent implements Discover
     private AllocationService allocationService;
     private final ClusterName clusterName;
     private final DiscoverySettings discoverySettings;
-    private final ZenPing zenPing;
+    protected final ZenPing zenPing; // protected to allow tests access
     private final MasterFaultDetection masterFD;
     private final NodesFaultDetection nodesFD;
     private final PublishClusterStateAction publishClusterState;
@@ -139,13 +139,13 @@ public class ZenDiscovery extends AbstractLifecycleComponent implements Discover
     private volatile NodeRemovalClusterStateTaskExecutor nodeRemovalExecutor;
 
     public ZenDiscovery(Settings settings, ThreadPool threadPool, TransportService transportService,
-                        ClusterService clusterService, ClusterSettings clusterSettings, ZenPing zenPing) {
+                        ClusterService clusterService, UnicastHostsProvider hostsProvider) {
         super(settings);
         this.clusterService = clusterService;
         this.clusterName = clusterService.getClusterName();
         this.transportService = transportService;
-        this.discoverySettings = new DiscoverySettings(settings, clusterSettings);
-        this.zenPing = zenPing;
+        this.discoverySettings = new DiscoverySettings(settings, clusterService.getClusterSettings());
+        this.zenPing = newZenPing(settings, threadPool, transportService, hostsProvider);
         this.electMaster = new ElectMasterService(settings);
         this.pingTimeout = PING_TIMEOUT_SETTING.get(settings);
         this.joinTimeout = JOIN_TIMEOUT_SETTING.get(settings);
@@ -160,12 +160,15 @@ public ZenDiscovery(Settings settings, ThreadPool threadPool, TransportService t
         logger.debug("using ping_timeout [{}], join.timeout [{}], master_election.ignore_non_master [{}]",
                 this.pingTimeout, joinTimeout, masterElectionIgnoreNonMasters);
 
-        clusterSettings.addSettingsUpdateConsumer(ElectMasterService.DISCOVERY_ZEN_MINIMUM_MASTER_NODES_SETTING, this::handleMinimumMasterNodesChanged, (value) -> {
-            final ClusterState clusterState = clusterService.state();
-            int masterNodes = clusterState.nodes().getMasterNodes().size();
-            if (value > masterNodes) {
-                throw new IllegalArgumentException("cannot set " + ElectMasterService.DISCOVERY_ZEN_MINIMUM_MASTER_NODES_SETTING.getKey() + " to more than the current master nodes count [" + masterNodes + "]");
-            }
+        clusterService.getClusterSettings().addSettingsUpdateConsumer(ElectMasterService.DISCOVERY_ZEN_MINIMUM_MASTER_NODES_SETTING,
+            this::handleMinimumMasterNodesChanged, (value) -> {
+                final ClusterState clusterState = clusterService.state();
+                int masterNodes = clusterState.nodes().getMasterNodes().size();
+                if (value > masterNodes) {
+                    throw new IllegalArgumentException("cannot set "
+                        + ElectMasterService.DISCOVERY_ZEN_MINIMUM_MASTER_NODES_SETTING.getKey() + " to more than the current" +
+                        " master nodes count [" + masterNodes + "]");
+                }
         });
 
         this.masterFD = new MasterFaultDetection(settings, threadPool, transportService, clusterService);
@@ -188,6 +191,12 @@ public ZenDiscovery(Settings settings, ThreadPool threadPool, TransportService t
             DISCOVERY_REJOIN_ACTION_NAME, RejoinClusterRequest::new, ThreadPool.Names.SAME, new RejoinClusterRequestHandler());
     }
 
+    // protected to allow overriding in tests
+    protected ZenPing newZenPing(Settings settings, ThreadPool threadPool, TransportService transportService,
+                                 UnicastHostsProvider hostsProvider) {
+        return new UnicastZenPing(settings, threadPool, transportService, hostsProvider);
+    }
+
     @Override
     public void setAllocationService(AllocationService allocationService) {
         this.allocationService = allocationService;

core/src/main/java/org/elasticsearch/node/Node.java

@@ -406,8 +406,7 @@ protected Node(final Environment environment, Collection<Class<? extends Plugin>
             }
 
             final DiscoveryModule discoveryModule = new DiscoveryModule(this.settings, threadPool, transportService,
-                networkService, clusterService, hostsProvider -> newZenPing(settings, threadPool, transportService, hostsProvider),
-                pluginsService.filterPlugins(DiscoveryPlugin.class));
+                networkService, clusterService, pluginsService.filterPlugins(DiscoveryPlugin.class));
             pluginsService.processModules(modules);
             modules.add(b -> {
                     b.bind(IndicesQueriesRegistry.class).toInstance(searchModule.getQueryParserRegistry());
@@ -441,7 +440,6 @@ protected Node(final Environment environment, Collection<Class<? extends Plugin>
                         indicesModule.getMapperRegistry(), settingsModule.getIndexScopedSettings()));
                     b.bind(ClusterInfoService.class).toInstance(clusterInfoService);
                     b.bind(Discovery.class).toInstance(discoveryModule.getDiscovery());
-                    b.bind(ZenPing.class).toInstance(discoveryModule.getZenPing());
                     {
                         RecoverySettings recoverySettings = new RecoverySettings(settings, settingsModule.getClusterSettings());
                         processRecoverySettings(settingsModule.getClusterSettings(), recoverySettings);
@@ -880,12 +878,6 @@ private List<NetworkService.CustomNameResolver> getCustomNameResolvers(List<Disc
         return customNameResolvers;
     }
 
-    /** Create a new ZenPing instance for use in zen discovery. */
-    protected ZenPing newZenPing(Settings settings, ThreadPool threadPool, TransportService transportService,
-                                 UnicastHostsProvider hostsProvider) {
-        return new UnicastZenPing(settings, threadPool, transportService, hostsProvider);
-    }
-
     /** Constructs an internal node used as a client into a cluster fronted by this tribe node. */
     protected Node newTribeClientNode(Settings settings, Collection<Class<? extends Plugin>> classpathPlugins) {
         return new Node(new Environment(settings), classpathPlugins);

core/src/main/java/org/elasticsearch/plugins/DiscoveryPlugin.java

@@ -57,10 +57,10 @@ public interface DiscoveryPlugin {
      * @param threadPool Use to schedule ping actions
      * @param transportService Use to communicate with other nodes
      * @param clusterService Use to find current nodes in the cluster
-     * @param zenPing Use to ping other nodes with zen unicast host list
+     * @param hostsProvider Use to find configured hosts which should be pinged for initial discovery
      */
     default Map<String, Supplier<Discovery>> getDiscoveryTypes(ThreadPool threadPool, TransportService transportService,
-                                                               ClusterService clusterService, ZenPing zenPing) {
+                                                               ClusterService clusterService, UnicastHostsProvider hostsProvider) {
         return Collections.emptyMap();
     }
 

core/src/test/java/org/elasticsearch/action/support/master/IndexingMasterFailoverIT.java

@@ -27,6 +27,7 @@
 import org.elasticsearch.discovery.zen.FaultDetection;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.test.ESIntegTestCase;
+import org.elasticsearch.test.discovery.TestZenDiscovery;
 import org.elasticsearch.test.disruption.NetworkDisruption;
 import org.elasticsearch.test.disruption.NetworkDisruption.NetworkDisconnect;
 import org.elasticsearch.test.disruption.NetworkDisruption.TwoPartitions;
@@ -46,18 +47,19 @@
 @ESIntegTestCase.SuppressLocalMode
 public class IndexingMasterFailoverIT extends ESIntegTestCase {
 
-    @Override
-    protected boolean addMockZenPings() {
-        return false;
-    }
-
     @Override
     protected Collection<Class<? extends Plugin>> nodePlugins() {
         final HashSet<Class<? extends Plugin>> classes = new HashSet<>(super.nodePlugins());
         classes.add(MockTransportService.TestPlugin.class);
         return classes;
     }
 
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return Settings.builder().put(super.nodeSettings(nodeOrdinal))
+            .put(TestZenDiscovery.USE_MOCK_PINGS.getKey(), false).build();
+    }
+
     /**
      * Indexing operations which entail mapping changes require a blocking request to the master node to update the mapping.
      * If the master node is being disrupted or if it cannot commit cluster state changes, it needs to retry within timeout limits.

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

@@ -19,6 +19,9 @@
 
 package org.elasticsearch.client.transport;
 
+import java.io.IOException;
+import java.util.Collections;
+
 import org.elasticsearch.Version;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.node.DiscoveryNode;
@@ -32,13 +35,10 @@
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.ESIntegTestCase.ClusterScope;
 import org.elasticsearch.test.ESIntegTestCase.Scope;
-import org.elasticsearch.test.discovery.MockZenPing;
+import org.elasticsearch.test.discovery.TestZenDiscovery;
 import org.elasticsearch.transport.MockTransportClient;
 import org.elasticsearch.transport.TransportService;
 
-import java.io.IOException;
-import java.util.Collections;
-
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.is;
@@ -65,7 +65,7 @@ public void testNodeVersionIsUpdated() throws IOException, NodeValidationExcepti
                 .put(NetworkModule.HTTP_ENABLED.getKey(), false)
                 .put(Node.NODE_DATA_SETTING.getKey(), false)
                 .put("cluster.name", "foobar")
-                .build(), Collections.singletonList(MockZenPing.TestPlugin.class)).start()) {
+                .build(), Collections.singletonList(TestZenDiscovery.TestPlugin.class)).start()) {
             TransportAddress transportAddress = node.injector().getInstance(TransportService.class).boundAddress().publishAddress();
             client.addTransportAddress(transportAddress);
             // since we force transport clients there has to be one node started that we connect to.

core/src/test/java/org/elasticsearch/cluster/MinimumMasterNodesIT.java

@@ -36,6 +36,7 @@
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.ESIntegTestCase.ClusterScope;
 import org.elasticsearch.test.ESIntegTestCase.Scope;
+import org.elasticsearch.test.discovery.TestZenDiscovery;
 import org.elasticsearch.test.disruption.NetworkDisruption;
 import org.elasticsearch.test.disruption.NetworkDisruption.NetworkDelay;
 import org.elasticsearch.test.disruption.NetworkDisruption.TwoPartitions;
@@ -75,8 +76,9 @@ protected Collection<Class<? extends Plugin>> nodePlugins() {
     }
 
     @Override
-    protected boolean addMockZenPings() {
-        return false;
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return Settings.builder().put(super.nodeSettings(nodeOrdinal))
+            .put(TestZenDiscovery.USE_MOCK_PINGS.getKey(), false).build();
     }
 
     public void testSimpleMinimumMasterNodes() throws Exception {

core/src/test/java/org/elasticsearch/cluster/routing/PrimaryAllocationIT.java

@@ -34,6 +34,7 @@
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.InternalTestCluster;
+import org.elasticsearch.test.discovery.TestZenDiscovery;
 import org.elasticsearch.test.disruption.NetworkDisruption;
 import org.elasticsearch.test.disruption.NetworkDisruption.NetworkDisconnect;
 import org.elasticsearch.test.disruption.NetworkDisruption.TwoPartitions;
@@ -63,8 +64,9 @@ protected Collection<Class<? extends Plugin>> nodePlugins() {
     }
 
     @Override
-    protected boolean addMockZenPings() {
-        return false;
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return Settings.builder().put(super.nodeSettings(nodeOrdinal))
+            .put(TestZenDiscovery.USE_MOCK_PINGS.getKey(), false).build();
     }
 
     private void createStaleReplicaScenario() throws Exception {