Zen2: Move most integration tests to Zen2 (#35678)

Zen2 is now feature-complete enough to run most ESIntegTestCase tests. The changes in this PR
are as follows:
- ClusterSettingsIT is adapted to not be Zen1 specific anymore (it was using Zen1 settings).
- Some of the integration tests require persistent storage of the cluster state, which is not fully
implemented yet (see #33958). These tests keep running with Zen1 for now but will be switched
over as soon as that is fully implemented.
- Some very few integration tests are not running yet with Zen2 for other reasons, depending on
some of the other open points in #32006.

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/IngestRestartIT.java

@@ -30,6 +30,7 @@ import org.elasticsearch.script.MockScriptEngine;
 import org.elasticsearch.script.MockScriptPlugin;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.InternalTestCluster;
+import org.elasticsearch.test.discovery.TestZenDiscovery;
 
 import java.util.Arrays;
 import java.util.Collection;
@@ -45,6 +46,13 @@ import static org.hamcrest.Matchers.equalTo;
 @ESIntegTestCase.ClusterScope(numDataNodes = 0, numClientNodes = 0, scope = ESIntegTestCase.Scope.TEST)
 public class IngestRestartIT extends ESIntegTestCase {
 
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return Settings.builder().put(super.nodeSettings(nodeOrdinal))
+            .put(TestZenDiscovery.USE_ZEN2.getKey(), false) // no state persistence yet
+            .build();
+    }
+
     @Override
     protected Collection<Class<? extends Plugin>> nodePlugins() {
         return Arrays.asList(IngestCommonPlugin.class, CustomScriptPlugin.class);

server/src/test/java/org/elasticsearch/action/admin/cluster/stats/ClusterStatsIT.java

@@ -34,6 +34,7 @@ import org.elasticsearch.node.Node;
 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.hamcrest.Matchers;
 
 import java.io.IOException;
@@ -210,14 +211,15 @@ public class ClusterStatsIT extends ESIntegTestCase {
     }
 
     public void testClusterStatusWhenStateNotRecovered() throws Exception {
-        internalCluster().startMasterOnlyNode(Settings.builder().put("gateway.recover_after_nodes", 2).build());
+        internalCluster().startMasterOnlyNode(Settings.builder().put("gateway.recover_after_nodes", 2)
+            .put(TestZenDiscovery.USE_ZEN2.getKey(), false).build());
         ClusterStatsResponse response = client().admin().cluster().prepareClusterStats().get();
         assertThat(response.getStatus(), equalTo(ClusterHealthStatus.RED));
 
         if (randomBoolean()) {
-            internalCluster().startMasterOnlyNode(Settings.EMPTY);
+            internalCluster().startMasterOnlyNode(Settings.builder().put(TestZenDiscovery.USE_ZEN2.getKey(), false).build());
         } else {
-            internalCluster().startDataOnlyNode(Settings.EMPTY);
+            internalCluster().startDataOnlyNode(Settings.builder().put(TestZenDiscovery.USE_ZEN2.getKey(), false).build());
         }
         // wait for the cluster status to settle
         ensureGreen();

server/src/test/java/org/elasticsearch/action/admin/indices/create/CreateIndexIT.java

@@ -42,6 +42,7 @@ import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.ESIntegTestCase.ClusterScope;
 import org.elasticsearch.test.ESIntegTestCase.Scope;
 import org.elasticsearch.test.InternalTestCluster;
+import org.elasticsearch.test.discovery.TestZenDiscovery;
 
 import java.util.HashMap;
 import java.util.HashSet;
@@ -66,6 +67,14 @@ import static org.hamcrest.core.IsNull.notNullValue;
 @ClusterScope(scope = Scope.TEST)
 public class CreateIndexIT extends ESIntegTestCase {
 
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return Settings.builder().put(super.nodeSettings(nodeOrdinal))
+            // testIndexWithUnknownSetting and testRestartIndexCreationAfterFullClusterRestart fail with Zen2
+            .put(TestZenDiscovery.USE_ZEN2.getKey(), false)
+            .build();
+    }
+
     public void testCreationDateGivenFails() {
         try {
             prepareCreate("test").setSettings(Settings.builder().put(IndexMetaData.SETTING_CREATION_DATE, 4L)).get();

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

@@ -21,6 +21,7 @@ package org.elasticsearch.client.transport;
 
 import org.elasticsearch.Version;
 import org.elasticsearch.client.Client;
+import org.elasticsearch.cluster.coordination.ClusterBootstrapService;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.network.NetworkModule;
 import org.elasticsearch.common.settings.Settings;
@@ -65,6 +66,8 @@ public class TransportClientIT extends ESIntegTestCase {
                 .put("transport.type", getTestTransportType())
                 .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)
                 .build(), Arrays.asList(getTestTransportPlugin(), TestZenDiscovery.TestPlugin.class,
                                         MockHttpTransport.TestPlugin.class)).start()) {
             TransportAddress transportAddress = node.injector().getInstance(TransportService.class).boundAddress().publishAddress();

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

@@ -74,6 +74,7 @@ public class MinimumMasterNodesIT extends ESIntegTestCase {
     @Override
     protected Settings nodeSettings(int nodeOrdinal) {
         return Settings.builder().put(super.nodeSettings(nodeOrdinal))
+            .put(TestZenDiscovery.USE_ZEN2.getKey(), false) // Zen2 does not have minimum_master_nodes
             .put(TestZenDiscovery.USE_MOCK_PINGS.getKey(), false).build();
     }
 

server/src/test/java/org/elasticsearch/cluster/NoMasterNodeIT.java

@@ -39,6 +39,7 @@ import org.elasticsearch.script.ScriptType;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.ESIntegTestCase.ClusterScope;
 import org.elasticsearch.test.ESIntegTestCase.Scope;
+import org.elasticsearch.test.discovery.TestZenDiscovery;
 
 import java.util.Collections;
 
@@ -51,6 +52,13 @@ import static org.hamcrest.Matchers.greaterThan;
 @ClusterScope(scope = Scope.TEST, numDataNodes = 0, autoMinMasterNodes = false)
 public class NoMasterNodeIT extends ESIntegTestCase {
 
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return Settings.builder().put(super.nodeSettings(nodeOrdinal))
+            .put(TestZenDiscovery.USE_ZEN2.getKey(), false) // tests here need adaption for Zen2
+            .build();
+    }
+
     public void testNoMasterActions() throws Exception {
         Settings settings = Settings.builder()
                 .put("action.auto_create_index", true)

server/src/test/java/org/elasticsearch/cluster/SpecificMasterNodesIT.java

@@ -28,6 +28,7 @@ import org.elasticsearch.node.Node;
 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.junit.annotations.TestLogging;
 
 import java.io.IOException;
@@ -44,6 +45,7 @@ public class SpecificMasterNodesIT extends ESIntegTestCase {
     @Override
     protected Settings nodeSettings(int nodeOrdinal) {
         return Settings.builder().put(super.nodeSettings(nodeOrdinal))
+            .put(TestZenDiscovery.USE_ZEN2.getKey(), false) // does unsafe things
             .put(DISCOVERY_ZEN_MINIMUM_MASTER_NODES_SETTING.getKey(), 1).build();
     }
 

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

@@ -326,7 +326,7 @@ public class PrimaryAllocationIT extends ESIntegTestCase {
 
     public void testNotWaitForQuorumCopies() throws Exception {
         logger.info("--> starting 3 nodes");
-        internalCluster().startNodes(3);
+        internalCluster().startNodes(3, Settings.builder().put(TestZenDiscovery.USE_ZEN2.getKey(), false).build()); // needs state recovery
         logger.info("--> creating index with 1 primary and 2 replicas");
         assertAcked(client().admin().indices().prepareCreate("test").setSettings(Settings.builder()
             .put("index.number_of_shards", randomIntBetween(1, 3)).put("index.number_of_replicas", 2)).get());

server/src/test/java/org/elasticsearch/cluster/settings/ClusterSettingsIT.java

@@ -26,17 +26,18 @@ import org.elasticsearch.action.admin.cluster.settings.ClusterUpdateSettingsResp
 import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
 import org.elasticsearch.cluster.metadata.MetaData;
 import org.elasticsearch.cluster.routing.allocation.decider.EnableAllocationDecider;
+import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeUnit;
-import org.elasticsearch.discovery.Discovery;
+import org.elasticsearch.common.unit.TimeValue;
-import org.elasticsearch.discovery.DiscoverySettings;
-import org.elasticsearch.discovery.zen.ZenDiscovery;
 import org.elasticsearch.indices.recovery.RecoverySettings;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.junit.After;
 
 import java.util.Arrays;
 
+import static org.elasticsearch.cluster.routing.allocation.DiskThresholdSettings.CLUSTER_ROUTING_ALLOCATION_REROUTE_INTERVAL_SETTING;
+import static org.elasticsearch.cluster.routing.allocation.decider.ThrottlingAllocationDecider.CLUSTER_ROUTING_ALLOCATION_NODE_INITIAL_PRIMARIES_RECOVERIES_SETTING;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertBlocked;
 import static org.hamcrest.Matchers.containsString;
@@ -68,123 +69,118 @@ public class ClusterSettingsIT extends ESIntegTestCase {
     }
 
     public void testDeleteIsAppliedFirst() {
-        DiscoverySettings discoverySettings = getDiscoverySettings();
+        final Setting<Integer> INITIAL_RECOVERIES = CLUSTER_ROUTING_ALLOCATION_NODE_INITIAL_PRIMARIES_RECOVERIES_SETTING;
-
+        final Setting<TimeValue> REROUTE_INTERVAL = CLUSTER_ROUTING_ALLOCATION_REROUTE_INTERVAL_SETTING;
-        assertEquals(discoverySettings.getPublishTimeout(), DiscoverySettings.PUBLISH_TIMEOUT_SETTING.get(Settings.EMPTY));
-        assertTrue(DiscoverySettings.PUBLISH_DIFF_ENABLE_SETTING.get(Settings.EMPTY));
 
         ClusterUpdateSettingsResponse response = client().admin().cluster()
             .prepareUpdateSettings()
             .setTransientSettings(Settings.builder()
-                .put(DiscoverySettings.PUBLISH_DIFF_ENABLE_SETTING.getKey(), false)
+                .put(INITIAL_RECOVERIES.getKey(), 7)
-                .put(DiscoverySettings.PUBLISH_TIMEOUT_SETTING.getKey(), "1s").build())
+                .put(REROUTE_INTERVAL.getKey(), "42s").build())
             .get();
 
         assertAcked(response);
-        assertEquals(response.getTransientSettings().get(DiscoverySettings.PUBLISH_TIMEOUT_SETTING.getKey()), "1s");
+        assertThat(INITIAL_RECOVERIES.get(response.getTransientSettings()), equalTo(7));
-        assertTrue(DiscoverySettings.PUBLISH_DIFF_ENABLE_SETTING.get(Settings.EMPTY));
+        assertThat(clusterService().getClusterSettings().get(INITIAL_RECOVERIES), equalTo(7));
-        assertFalse(response.getTransientSettings().getAsBoolean(DiscoverySettings.PUBLISH_DIFF_ENABLE_SETTING.getKey(), null));
+        assertThat(REROUTE_INTERVAL.get(response.getTransientSettings()), equalTo(TimeValue.timeValueSeconds(42)));
+        assertThat(clusterService().getClusterSettings().get(REROUTE_INTERVAL), equalTo(TimeValue.timeValueSeconds(42)));
 
         response = client().admin().cluster()
             .prepareUpdateSettings()
-            .setTransientSettings(Settings.builder().putNull((randomBoolean() ? "discovery.zen.*" : "*"))
+            .setTransientSettings(Settings.builder().putNull((randomBoolean() ? "cluster.routing.*" : "*"))
-                .put(DiscoverySettings.PUBLISH_TIMEOUT_SETTING.getKey(), "2s"))
+                .put(REROUTE_INTERVAL.getKey(), "43s"))
             .get();
-        assertEquals(response.getTransientSettings().get(DiscoverySettings.PUBLISH_TIMEOUT_SETTING.getKey()), "2s");
+        assertThat(INITIAL_RECOVERIES.get(response.getTransientSettings()), equalTo(INITIAL_RECOVERIES.get(Settings.EMPTY)));
-        assertNull(response.getTransientSettings().getAsBoolean(DiscoverySettings.PUBLISH_DIFF_ENABLE_SETTING.getKey(), null));
+        assertThat(clusterService().getClusterSettings().get(INITIAL_RECOVERIES), equalTo(INITIAL_RECOVERIES.get(Settings.EMPTY)));
+        assertThat(REROUTE_INTERVAL.get(response.getTransientSettings()), equalTo(TimeValue.timeValueSeconds(43)));
+        assertThat(clusterService().getClusterSettings().get(REROUTE_INTERVAL), equalTo(TimeValue.timeValueSeconds(43)));
     }
 
     public void testResetClusterSetting() {
-        DiscoverySettings discoverySettings = getDiscoverySettings();
+        final Setting<Integer> INITIAL_RECOVERIES = CLUSTER_ROUTING_ALLOCATION_NODE_INITIAL_PRIMARIES_RECOVERIES_SETTING;
-
+        final Setting<TimeValue> REROUTE_INTERVAL = CLUSTER_ROUTING_ALLOCATION_REROUTE_INTERVAL_SETTING;
-        assertThat(discoverySettings.getPublishTimeout(), equalTo(DiscoverySettings.PUBLISH_TIMEOUT_SETTING.get(Settings.EMPTY)));
-        assertThat(discoverySettings.getPublishDiff(), equalTo(DiscoverySettings.PUBLISH_DIFF_ENABLE_SETTING.get(Settings.EMPTY)));
 
         ClusterUpdateSettingsResponse response = client().admin().cluster()
                 .prepareUpdateSettings()
-                .setTransientSettings(Settings.builder().put(DiscoverySettings.PUBLISH_TIMEOUT_SETTING.getKey(), "1s").build())
+                .setTransientSettings(Settings.builder().put(INITIAL_RECOVERIES.getKey(), 7).build())
                 .get();
 
         assertAcked(response);
-        assertThat(response.getTransientSettings().get(DiscoverySettings.PUBLISH_TIMEOUT_SETTING.getKey()), equalTo("1s"));
+        assertThat(INITIAL_RECOVERIES.get(response.getTransientSettings()), equalTo(7));
-        assertThat(discoverySettings.getPublishTimeout().seconds(), equalTo(1L));
+        assertThat(clusterService().getClusterSettings().get(INITIAL_RECOVERIES), equalTo(7));
-        assertThat(discoverySettings.getPublishDiff(), equalTo(DiscoverySettings.PUBLISH_DIFF_ENABLE_SETTING.get(Settings.EMPTY)));
-
 
         response = client().admin().cluster()
                 .prepareUpdateSettings()
-                .setTransientSettings(Settings.builder().putNull(DiscoverySettings.PUBLISH_TIMEOUT_SETTING.getKey()))
+                .setTransientSettings(Settings.builder().putNull(INITIAL_RECOVERIES.getKey()))
                 .get();
 
         assertAcked(response);
-        assertNull(response.getTransientSettings().get(DiscoverySettings.PUBLISH_TIMEOUT_SETTING.getKey()));
+        assertNull(response.getTransientSettings().get(INITIAL_RECOVERIES.getKey()));
-        assertThat(discoverySettings.getPublishTimeout(), equalTo(DiscoverySettings.PUBLISH_TIMEOUT_SETTING.get(Settings.EMPTY)));
+        assertThat(clusterService().getClusterSettings().get(INITIAL_RECOVERIES),
-        assertThat(discoverySettings.getPublishDiff(), equalTo(DiscoverySettings.PUBLISH_DIFF_ENABLE_SETTING.get(Settings.EMPTY)));
+            equalTo(INITIAL_RECOVERIES.get(Settings.EMPTY)));
 
         response = client().admin().cluster()
                 .prepareUpdateSettings()
                 .setTransientSettings(Settings.builder()
-                        .put(DiscoverySettings.PUBLISH_TIMEOUT_SETTING.getKey(), "1s")
+                        .put(INITIAL_RECOVERIES.getKey(), 8)
-                        .put(DiscoverySettings.PUBLISH_DIFF_ENABLE_SETTING.getKey(), false).build())
+                        .put(REROUTE_INTERVAL.getKey(), "43s").build())
                 .get();
 
         assertAcked(response);
-        assertThat(response.getTransientSettings().get(DiscoverySettings.PUBLISH_TIMEOUT_SETTING.getKey()), equalTo("1s"));
+        assertThat(INITIAL_RECOVERIES.get(response.getTransientSettings()), equalTo(8));
-        assertThat(discoverySettings.getPublishTimeout().seconds(), equalTo(1L));
+        assertThat(clusterService().getClusterSettings().get(INITIAL_RECOVERIES), equalTo(8));
-        assertFalse(discoverySettings.getPublishDiff());
+        assertThat(REROUTE_INTERVAL.get(response.getTransientSettings()), equalTo(TimeValue.timeValueSeconds(43)));
+        assertThat(clusterService().getClusterSettings().get(REROUTE_INTERVAL), equalTo(TimeValue.timeValueSeconds(43)));
         response = client().admin().cluster()
                 .prepareUpdateSettings()
-                .setTransientSettings(Settings.builder().putNull((randomBoolean() ? "discovery.zen.*" : "*")))
+                .setTransientSettings(Settings.builder().putNull((randomBoolean() ? "cluster.routing.*" : "*")))
                 .get();
 
-        assertNull(response.getTransientSettings().get(DiscoverySettings.PUBLISH_TIMEOUT_SETTING.getKey()));
+        assertThat(INITIAL_RECOVERIES.get(response.getTransientSettings()), equalTo(INITIAL_RECOVERIES.get(Settings.EMPTY)));
-        assertNull(response.getTransientSettings().get(DiscoverySettings.PUBLISH_DIFF_ENABLE_SETTING.getKey()));
+        assertThat(clusterService().getClusterSettings().get(INITIAL_RECOVERIES), equalTo(INITIAL_RECOVERIES.get(Settings.EMPTY)));
-        assertThat(discoverySettings.getPublishTimeout(), equalTo(DiscoverySettings.PUBLISH_TIMEOUT_SETTING.get(Settings.EMPTY)));
+        assertThat(REROUTE_INTERVAL.get(response.getTransientSettings()), equalTo(REROUTE_INTERVAL.get(Settings.EMPTY)));
-        assertThat(discoverySettings.getPublishDiff(), equalTo(DiscoverySettings.PUBLISH_DIFF_ENABLE_SETTING.get(Settings.EMPTY)));
+        assertThat(clusterService().getClusterSettings().get(REROUTE_INTERVAL), equalTo(REROUTE_INTERVAL.get(Settings.EMPTY)));
 
         // now persistent
         response = client().admin().cluster()
                 .prepareUpdateSettings()
-                .setPersistentSettings(Settings.builder().put(DiscoverySettings.PUBLISH_TIMEOUT_SETTING.getKey(), "1s").build())
+                .setPersistentSettings(Settings.builder().put(INITIAL_RECOVERIES.getKey(), 9).build())
                 .get();
 
         assertAcked(response);
-        assertThat(response.getPersistentSettings().get(DiscoverySettings.PUBLISH_TIMEOUT_SETTING.getKey()), equalTo("1s"));
+        assertThat(INITIAL_RECOVERIES.get(response.getPersistentSettings()), equalTo(9));
-        assertThat(discoverySettings.getPublishTimeout().seconds(), equalTo(1L));
+        assertThat(clusterService().getClusterSettings().get(INITIAL_RECOVERIES), equalTo(9));
-        assertThat(discoverySettings.getPublishDiff(), equalTo(DiscoverySettings.PUBLISH_DIFF_ENABLE_SETTING.get(Settings.EMPTY)));
-
 
         response = client().admin().cluster()
                 .prepareUpdateSettings()
-                .setPersistentSettings(Settings.builder().putNull((DiscoverySettings.PUBLISH_TIMEOUT_SETTING.getKey())))
+                .setPersistentSettings(Settings.builder().putNull(INITIAL_RECOVERIES.getKey()))
                 .get();
 
         assertAcked(response);
-        assertNull(response.getPersistentSettings().get(DiscoverySettings.PUBLISH_TIMEOUT_SETTING.getKey()));
+        assertThat(INITIAL_RECOVERIES.get(response.getPersistentSettings()), equalTo(INITIAL_RECOVERIES.get(Settings.EMPTY)));
-        assertThat(discoverySettings.getPublishTimeout(), equalTo(DiscoverySettings.PUBLISH_TIMEOUT_SETTING.get(Settings.EMPTY)));
+        assertThat(clusterService().getClusterSettings().get(INITIAL_RECOVERIES), equalTo(INITIAL_RECOVERIES.get(Settings.EMPTY)));
-        assertThat(discoverySettings.getPublishDiff(), equalTo(DiscoverySettings.PUBLISH_DIFF_ENABLE_SETTING.get(Settings.EMPTY)));
-
 
         response = client().admin().cluster()
                 .prepareUpdateSettings()
                 .setPersistentSettings(Settings.builder()
-                        .put(DiscoverySettings.PUBLISH_TIMEOUT_SETTING.getKey(), "1s")
+                        .put(INITIAL_RECOVERIES.getKey(), 10)
-                        .put(DiscoverySettings.PUBLISH_DIFF_ENABLE_SETTING.getKey(), false).build())
+                        .put(REROUTE_INTERVAL.getKey(), "44s").build())
                 .get();
 
         assertAcked(response);
-        assertThat(response.getPersistentSettings().get(DiscoverySettings.PUBLISH_TIMEOUT_SETTING.getKey()), equalTo("1s"));
+        assertThat(INITIAL_RECOVERIES.get(response.getPersistentSettings()), equalTo(10));
-        assertThat(discoverySettings.getPublishTimeout().seconds(), equalTo(1L));
+        assertThat(clusterService().getClusterSettings().get(INITIAL_RECOVERIES), equalTo(10));
-        assertFalse(discoverySettings.getPublishDiff());
+        assertThat(REROUTE_INTERVAL.get(response.getPersistentSettings()), equalTo(TimeValue.timeValueSeconds(44)));
+        assertThat(clusterService().getClusterSettings().get(REROUTE_INTERVAL), equalTo(TimeValue.timeValueSeconds(44)));
         response = client().admin().cluster()
                 .prepareUpdateSettings()
-                .setPersistentSettings(Settings.builder().putNull((randomBoolean() ? "discovery.zen.*" : "*")))
+                .setPersistentSettings(Settings.builder().putNull((randomBoolean() ? "cluster.routing.*" : "*")))
                 .get();
 
-        assertNull(response.getPersistentSettings().get(DiscoverySettings.PUBLISH_TIMEOUT_SETTING.getKey()));
+        assertThat(INITIAL_RECOVERIES.get(response.getPersistentSettings()), equalTo(INITIAL_RECOVERIES.get(Settings.EMPTY)));
-        assertNull(response.getPersistentSettings().get(DiscoverySettings.PUBLISH_DIFF_ENABLE_SETTING.getKey()));
+        assertThat(clusterService().getClusterSettings().get(INITIAL_RECOVERIES), equalTo(INITIAL_RECOVERIES.get(Settings.EMPTY)));
-        assertThat(discoverySettings.getPublishTimeout(), equalTo(DiscoverySettings.PUBLISH_TIMEOUT_SETTING.get(Settings.EMPTY)));
+        assertThat(REROUTE_INTERVAL.get(response.getPersistentSettings()), equalTo(REROUTE_INTERVAL.get(Settings.EMPTY)));
-        assertThat(discoverySettings.getPublishDiff(), equalTo(DiscoverySettings.PUBLISH_DIFF_ENABLE_SETTING.get(Settings.EMPTY)));
+        assertThat(clusterService().getClusterSettings().get(REROUTE_INTERVAL), equalTo(REROUTE_INTERVAL.get(Settings.EMPTY)));
     }
 
     public void testClusterSettingsUpdateResponse() {
@@ -253,49 +249,42 @@ public class ClusterSettingsIT extends ESIntegTestCase {
             Arrays.asList("internal:index/shard/recovery/*", "internal:gateway/local*"));
     }
 
-    public void testUpdateDiscoveryPublishTimeout() {
+    public void testUpdateSettings() {
-
+        final Setting<Integer> INITIAL_RECOVERIES = CLUSTER_ROUTING_ALLOCATION_NODE_INITIAL_PRIMARIES_RECOVERIES_SETTING;
-        DiscoverySettings discoverySettings = getDiscoverySettings();
-
-        assertThat(discoverySettings.getPublishTimeout(), equalTo(DiscoverySettings.PUBLISH_TIMEOUT_SETTING.get(Settings.EMPTY)));
 
         ClusterUpdateSettingsResponse response = client().admin().cluster()
                 .prepareUpdateSettings()
-                .setTransientSettings(Settings.builder().put(DiscoverySettings.PUBLISH_TIMEOUT_SETTING.getKey(), "1s").build())
+                .setTransientSettings(Settings.builder().put(INITIAL_RECOVERIES.getKey(), 42).build())
                 .get();
 
         assertAcked(response);
-        assertThat(response.getTransientSettings().get(DiscoverySettings.PUBLISH_TIMEOUT_SETTING.getKey()), equalTo("1s"));
+        assertThat(INITIAL_RECOVERIES.get(response.getTransientSettings()), equalTo(42));
-        assertThat(discoverySettings.getPublishTimeout().seconds(), equalTo(1L));
+        assertThat(clusterService().getClusterSettings().get(INITIAL_RECOVERIES), equalTo(42));
 
         try {
             client().admin().cluster()
                     .prepareUpdateSettings()
-                    .setTransientSettings(Settings.builder().put(DiscoverySettings.PUBLISH_TIMEOUT_SETTING.getKey(), "whatever").build())
+                    .setTransientSettings(Settings.builder().put(INITIAL_RECOVERIES.getKey(), "whatever").build())
                     .get();
             fail("bogus value");
         } catch (IllegalArgumentException ex) {
-            assertEquals(ex.getMessage(), "failed to parse setting [discovery.zen.publish_timeout] with value [whatever]" +
+            assertEquals(ex.getMessage(), "Failed to parse value [whatever] for setting [" + INITIAL_RECOVERIES.getKey() + "]");
-                " as a time value: unit is missing or unrecognized");
         }
 
-        assertThat(discoverySettings.getPublishTimeout().seconds(), equalTo(1L));
+        assertThat(clusterService().getClusterSettings().get(INITIAL_RECOVERIES), equalTo(42));
 
         try {
             client().admin().cluster()
                     .prepareUpdateSettings()
-                    .setTransientSettings(Settings.builder().put(DiscoverySettings.PUBLISH_TIMEOUT_SETTING.getKey(), -1).build())
+                    .setTransientSettings(Settings.builder()
+                        .put(INITIAL_RECOVERIES.getKey(), -1).build())
                     .get();
             fail("bogus value");
         } catch (IllegalArgumentException ex) {
-            assertEquals(ex.getMessage(), "failed to parse value [-1] for setting [discovery.zen.publish_timeout], must be >= [0ms]");
+            assertEquals(ex.getMessage(), "Failed to parse value [-1] for setting [" + INITIAL_RECOVERIES.getKey() + "] must be >= 0");
         }
 
-        assertThat(discoverySettings.getPublishTimeout().seconds(), equalTo(1L));
+        assertThat(clusterService().getClusterSettings().get(INITIAL_RECOVERIES), equalTo(42));
-    }
-
-    private DiscoverySettings getDiscoverySettings() {
-        return ((ZenDiscovery) internalCluster().getInstance(Discovery.class)).getDiscoverySettings();
     }
 
     public void testClusterUpdateSettingsWithBlocks() {

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

@@ -63,6 +63,7 @@ public abstract class AbstractDisruptionTestCase extends ESIntegTestCase {
     @Override
     protected Settings nodeSettings(int nodeOrdinal) {
         return Settings.builder().put(super.nodeSettings(nodeOrdinal)).put(DEFAULT_SETTINGS)
+                .put(TestZenDiscovery.USE_ZEN2.getKey(), false) // requires more work
                 .put(TestZenDiscovery.USE_MOCK_PINGS.getKey(), false).build();
     }
 

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

@@ -68,6 +68,7 @@ public class SnapshotDisruptionIT extends ESIntegTestCase {
         return Settings.builder().put(super.nodeSettings(nodeOrdinal))
             .put(AbstractDisruptionTestCase.DEFAULT_SETTINGS)
             .put(TestZenDiscovery.USE_MOCK_PINGS.getKey(), false)
+            .put(TestZenDiscovery.USE_ZEN2.getKey(), false) // requires more work
             .put(DiscoverySettings.COMMIT_TIMEOUT_SETTING.getKey(), "30s")
             .build();
     }

server/src/test/java/org/elasticsearch/discovery/zen/ZenDiscoveryIT.java

@@ -41,6 +41,7 @@ import org.elasticsearch.discovery.DiscoveryStats;
 import org.elasticsearch.node.Node;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.TestCustomMetaData;
+import org.elasticsearch.test.discovery.TestZenDiscovery;
 import org.elasticsearch.test.junit.annotations.TestLogging;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.BytesTransportRequest;
@@ -72,6 +73,13 @@ import static org.hamcrest.Matchers.notNullValue;
 @TestLogging("_root:DEBUG")
 public class ZenDiscoveryIT extends ESIntegTestCase {
 
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return Settings.builder().put(super.nodeSettings(nodeOrdinal))
+            .put(TestZenDiscovery.USE_ZEN2.getKey(), false) // Zen1-specific stuff in some tests
+            .build();
+    }
+
     public void testNoShardRelocationsOccurWhenElectedMasterNodeFails() throws Exception {
         Settings defaultSettings = Settings.builder()
                 .put(FaultDetection.PING_TIMEOUT_SETTING.getKey(), "1s")

server/src/test/java/org/elasticsearch/gateway/GatewayIndexStateIT.java

@@ -49,6 +49,7 @@ import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.ESIntegTestCase.ClusterScope;
 import org.elasticsearch.test.ESIntegTestCase.Scope;
 import org.elasticsearch.test.InternalTestCluster.RestartCallback;
+import org.elasticsearch.test.discovery.TestZenDiscovery;
 
 import java.io.IOException;
 import java.util.List;
@@ -65,6 +66,14 @@ public class GatewayIndexStateIT extends ESIntegTestCase {
 
     private final Logger logger = LogManager.getLogger(GatewayIndexStateIT.class);
 
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return Settings.builder().put(super.nodeSettings(nodeOrdinal))
+            // testRecoverBrokenIndexMetadata, testRecoverMissingAnalyzer, testDanglingIndices and testArchiveBrokenClusterSettings fail
+            .put(TestZenDiscovery.USE_ZEN2.getKey(), false)
+            .build();
+    }
+
     public void testMappingMetaDataParsed() throws Exception {
         logger.info("--> starting 1 nodes");
         internalCluster().startNode();

server/src/test/java/org/elasticsearch/gateway/RecoverAfterNodesIT.java

@@ -29,6 +29,7 @@ import org.elasticsearch.node.Node;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.ESIntegTestCase.ClusterScope;
 import org.elasticsearch.test.ESIntegTestCase.Scope;
+import org.elasticsearch.test.discovery.TestZenDiscovery;
 
 import java.util.Set;
 
@@ -39,6 +40,13 @@ import static org.hamcrest.Matchers.hasItem;
 public class RecoverAfterNodesIT extends ESIntegTestCase {
     private static final TimeValue BLOCK_WAIT_TIMEOUT = TimeValue.timeValueSeconds(10);
 
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return Settings.builder().put(super.nodeSettings(nodeOrdinal))
+            .put(TestZenDiscovery.USE_ZEN2.getKey(), false) // recover_after no implemented in Zen2 yet
+            .build();
+    }
+
     public Set<ClusterBlock> waitForNoBlocksOnNode(TimeValue timeout, Client nodeClient) throws InterruptedException {
         long start = System.currentTimeMillis();
         Set<ClusterBlock> blocks;

server/src/test/java/org/elasticsearch/gateway/RecoveryFromGatewayIT.java

@@ -48,6 +48,7 @@ import org.elasticsearch.test.ESIntegTestCase.ClusterScope;
 import org.elasticsearch.test.ESIntegTestCase.Scope;
 import org.elasticsearch.test.InternalTestCluster;
 import org.elasticsearch.test.InternalTestCluster.RestartCallback;
+import org.elasticsearch.test.discovery.TestZenDiscovery;
 import org.elasticsearch.test.store.MockFSIndexStore;
 
 import java.nio.file.DirectoryStream;
@@ -84,6 +85,14 @@ public class RecoveryFromGatewayIT extends ESIntegTestCase {
         return Arrays.asList(MockFSIndexStore.TestPlugin.class);
     }
 
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return Settings.builder().put(super.nodeSettings(nodeOrdinal))
+            // testTwoNodeFirstNodeCleared does unsafe things, and testLatestVersionLoaded / testRecoveryDifferentNodeOrderStartup also fail
+            .put(TestZenDiscovery.USE_ZEN2.getKey(), false)
+            .build();
+    }
+
     public void testOneNodeRecoverFromGateway() throws Exception {
 
         internalCluster().startNode();

server/src/test/java/org/elasticsearch/index/shard/RemoveCorruptedShardDataCommandIT.java

@@ -70,6 +70,7 @@ import org.elasticsearch.test.CorruptionUtils;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.InternalSettingsPlugin;
 import org.elasticsearch.test.InternalTestCluster;
+import org.elasticsearch.test.discovery.TestZenDiscovery;
 import org.elasticsearch.test.engine.MockEngineSupport;
 import org.elasticsearch.test.transport.MockTransportService;
 
@@ -101,6 +102,13 @@ import static org.hamcrest.Matchers.startsWith;
 @ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.SUITE, numDataNodes = 0)
 public class RemoveCorruptedShardDataCommandIT extends ESIntegTestCase {
 
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return Settings.builder().put(super.nodeSettings(nodeOrdinal))
+            .put(TestZenDiscovery.USE_ZEN2.getKey(), false) // no state persistence yet
+            .build();
+    }
+
     @Override
     protected Collection<Class<? extends Plugin>> nodePlugins() {
         return Arrays.asList(MockTransportService.TestPlugin.class, MockEngineFactoryPlugin.class, InternalSettingsPlugin.class);

server/src/test/java/org/elasticsearch/index/store/CorruptedFileIT.java

@@ -68,6 +68,7 @@ import org.elasticsearch.test.CorruptionUtils;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.InternalSettingsPlugin;
 import org.elasticsearch.test.MockIndexEventListener;
+import org.elasticsearch.test.discovery.TestZenDiscovery;
 import org.elasticsearch.test.store.MockFSIndexStore;
 import org.elasticsearch.test.transport.MockTransportService;
 import org.elasticsearch.transport.TransportService;
@@ -116,6 +117,7 @@ public class CorruptedFileIT extends ESIntegTestCase {
             // speed up recoveries
             .put(ThrottlingAllocationDecider.CLUSTER_ROUTING_ALLOCATION_NODE_CONCURRENT_INCOMING_RECOVERIES_SETTING.getKey(), 5)
             .put(ThrottlingAllocationDecider.CLUSTER_ROUTING_ALLOCATION_NODE_CONCURRENT_OUTGOING_RECOVERIES_SETTING.getKey(), 5)
+            .put(TestZenDiscovery.USE_ZEN2.getKey(), false) // no state persistence yet
             .build();
     }
 

server/src/test/java/org/elasticsearch/indices/flush/FlushIT.java

@@ -56,6 +56,7 @@ import org.elasticsearch.index.shard.IndexShardTestCase;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.indices.IndicesService;
 import org.elasticsearch.test.ESIntegTestCase;
+import org.elasticsearch.test.discovery.TestZenDiscovery;
 
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.hamcrest.Matchers.emptyIterable;
@@ -64,6 +65,15 @@ import static org.hamcrest.Matchers.not;
 import static org.hamcrest.Matchers.nullValue;
 
 public class FlushIT extends ESIntegTestCase {
+
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return Settings.builder().put(super.nodeSettings(nodeOrdinal))
+            // uses fullClusterRestart
+            .put(TestZenDiscovery.USE_ZEN2.getKey(), false)
+            .build();
+    }
+
     public void testWaitIfOngoing() throws InterruptedException {
         createIndex("test");
         ensureGreen("test");

server/src/test/java/org/elasticsearch/indices/state/RareClusterStateIT.java

@@ -192,6 +192,7 @@ public class RareClusterStateIT extends ESIntegTestCase {
         Settings settings = Settings.builder()
             .put(DiscoverySettings.COMMIT_TIMEOUT_SETTING.getKey(), "30s") // explicitly set so it won't default to publish timeout
             .put(DiscoverySettings.PUBLISH_TIMEOUT_SETTING.getKey(), "0s") // don't wait post commit as we are blocking things by design
+            .put(TestZenDiscovery.USE_ZEN2.getKey(), false) // TODO: convert test to support Zen2
             .build();
         final List<String> nodeNames = internalCluster().startNodes(2, settings);
         assertFalse(client().admin().cluster().prepareHealth().setWaitForNodes("2").get().isTimedOut());
@@ -306,6 +307,7 @@ public class RareClusterStateIT extends ESIntegTestCase {
             Settings.builder()
                 .put(DiscoverySettings.COMMIT_TIMEOUT_SETTING.getKey(), "30s") // explicitly set so it won't default to publish timeout
                 .put(DiscoverySettings.PUBLISH_TIMEOUT_SETTING.getKey(), "0s") // don't wait post commit as we are blocking things by design
+                .put(TestZenDiscovery.USE_ZEN2.getKey(), false) // TODO: convert test to support Zen2
                 .build());
         assertFalse(client().admin().cluster().prepareHealth().setWaitForNodes("2").get().isTimedOut());
 

server/src/test/java/org/elasticsearch/persistent/PersistentTasksExecutorFullRestartIT.java

@@ -20,11 +20,13 @@ package org.elasticsearch.persistent;
 
 import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.common.UUIDs;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.persistent.PersistentTasksCustomMetaData.PersistentTask;
 import org.elasticsearch.persistent.TestPersistentTasksPlugin.TestParams;
 import org.elasticsearch.persistent.TestPersistentTasksPlugin.TestPersistentTasksExecutor;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.test.ESIntegTestCase;
+import org.elasticsearch.test.discovery.TestZenDiscovery;
 import org.elasticsearch.test.junit.annotations.TestLogging;
 
 import java.util.ArrayList;
@@ -38,6 +40,14 @@ import static org.hamcrest.Matchers.greaterThan;
 
 @ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST, minNumDataNodes = 1)
 public class PersistentTasksExecutorFullRestartIT extends ESIntegTestCase {
+
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return Settings.builder().put(super.nodeSettings(nodeOrdinal))
+            .put(TestZenDiscovery.USE_ZEN2.getKey(), false) // no state persistence yet
+            .build();
+    }
+    
     @Override
     protected Collection<Class<? extends Plugin>> nodePlugins() {
         return Collections.singletonList(TestPersistentTasksPlugin.class);

server/src/test/java/org/elasticsearch/persistent/decider/EnableAssignmentDeciderIT.java

@@ -32,6 +32,7 @@ import org.elasticsearch.persistent.TestPersistentTasksPlugin.TestParams;
 import org.elasticsearch.persistent.TestPersistentTasksPlugin.TestPersistentTasksExecutor;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.test.ESIntegTestCase;
+import org.elasticsearch.test.discovery.TestZenDiscovery;
 
 import java.util.Collection;
 import java.util.concurrent.CountDownLatch;
@@ -55,6 +56,13 @@ public class EnableAssignmentDeciderIT extends ESIntegTestCase {
         return nodePlugins();
     }
 
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return Settings.builder().put(super.nodeSettings(nodeOrdinal))
+            .put(TestZenDiscovery.USE_ZEN2.getKey(), false) // state recovery not completed in Zen2
+            .build();
+    }
+
     @Override
     protected boolean ignoreExternalCluster() {
         return true;

server/src/test/java/org/elasticsearch/snapshots/DedicatedClusterSnapshotRestoreIT.java

@@ -78,6 +78,7 @@ import org.elasticsearch.test.ESIntegTestCase.ClusterScope;
 import org.elasticsearch.test.ESIntegTestCase.Scope;
 import org.elasticsearch.test.InternalTestCluster;
 import org.elasticsearch.test.TestCustomMetaData;
+import org.elasticsearch.test.discovery.TestZenDiscovery;
 import org.elasticsearch.test.rest.FakeRestRequest;
 
 import java.io.IOException;
@@ -115,6 +116,13 @@ import static org.mockito.Mockito.mock;
 @ClusterScope(scope = Scope.TEST, numDataNodes = 0, transportClientRatio = 0)
 public class DedicatedClusterSnapshotRestoreIT extends AbstractSnapshotIntegTestCase {
 
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return Settings.builder().put(super.nodeSettings(nodeOrdinal))
+            .put(TestZenDiscovery.USE_ZEN2.getKey(), false) // requires more work
+            .build();
+    }
+
     public static class TestCustomMetaDataPlugin extends Plugin {
 
         private final List<NamedWriteableRegistry.Entry> namedWritables = new ArrayList<>();

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

@@ -1925,6 +1925,9 @@ public abstract class ESIntegTestCase extends ESTestCase {
             initialNodeSettings.put(NetworkModule.TRANSPORT_TYPE_KEY, getTestTransportType());
             initialTransportClientSettings.put(NetworkModule.TRANSPORT_TYPE_KEY, getTestTransportType());
         }
+        if (addTestZenDiscovery()) {
+            initialNodeSettings.put(TestZenDiscovery.USE_ZEN2.getKey(), true);
+        }
         return new NodeConfigurationSource() {
             @Override
             public Settings nodeSettings(int nodeOrdinal) {

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

@@ -73,6 +73,7 @@ import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.PageCacheRecycler;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.core.internal.io.IOUtils;
 import org.elasticsearch.discovery.DiscoveryModule;
 import org.elasticsearch.discovery.zen.ElectMasterService;
@@ -579,7 +580,7 @@ public final class InternalTestCluster extends TestCluster {
         final Stream<NodeAndClient> collection = n == 0
                 ? nodes.values().stream()
                 : nodes.values().stream()
-                        .filter(new DataNodePredicate().and(new MasterNodePredicate(getMasterName()).negate()));
+                        .filter(new DataNodePredicate().and(new NodeNamePredicate(getMasterName()).negate()));
         final Iterator<NodeAndClient> values = collection.iterator();
 
         logger.info("changing cluster size from {} data nodes to {}", size, n);
@@ -725,7 +726,7 @@ public final class InternalTestCluster extends TestCluster {
      */
     public synchronized Client masterClient() {
         ensureOpen();
-        NodeAndClient randomNodeAndClient = getRandomNodeAndClient(new MasterNodePredicate(getMasterName()));
+        NodeAndClient randomNodeAndClient = getRandomNodeAndClient(new NodeNamePredicate(getMasterName()));
         if (randomNodeAndClient != null) {
             return randomNodeAndClient.nodeClient(); // ensure node client master is requested
         }
@@ -738,7 +739,7 @@ public final class InternalTestCluster extends TestCluster {
      */
     public synchronized Client nonMasterClient() {
         ensureOpen();
-        NodeAndClient randomNodeAndClient = getRandomNodeAndClient(new MasterNodePredicate(getMasterName()).negate());
+        NodeAndClient randomNodeAndClient = getRandomNodeAndClient(new NodeNamePredicate(getMasterName()).negate());
         if (randomNodeAndClient != null) {
             return randomNodeAndClient.nodeClient(); // ensure node client non-master is requested
         }
@@ -1441,7 +1442,7 @@ public final class InternalTestCluster extends TestCluster {
      * in the cluster.
      */
     public synchronized <T> Iterable<T> getDataOrMasterNodeInstances(Class<T> clazz) {
-        return getInstances(clazz, new DataOrMasterNodePredicate());
+        return getInstances(clazz, new DataNodePredicate().or(new MasterNodePredicate()));
     }
 
     private synchronized <T> Iterable<T> getInstances(Class<T> clazz, Predicate<NodeAndClient> predicate) {
@@ -1464,6 +1465,10 @@ public final class InternalTestCluster extends TestCluster {
         return getInstance(clazz, new DataNodePredicate());
     }
 
+    public synchronized <T> T getMasterNodeInstance(Class<T> clazz) {
+        return getInstance(clazz, new MasterNodePredicate());
+    }
+
     private synchronized <T> T getInstance(Class<T> clazz, Predicate<NodeAndClient> predicate) {
         NodeAndClient randomNodeAndClient = getRandomNodeAndClient(predicate);
         assert randomNodeAndClient != null;
@@ -1538,7 +1543,7 @@ public final class InternalTestCluster extends TestCluster {
      * Stops any of the current nodes but not the master node.
      */
     public synchronized void stopRandomNonMasterNode() throws IOException {
-        NodeAndClient nodeAndClient = getRandomNodeAndClient(new MasterNodePredicate(getMasterName()).negate());
+        NodeAndClient nodeAndClient = getRandomNodeAndClient(new NodeNamePredicate(getMasterName()).negate());
         if (nodeAndClient != null) {
             logger.info("Closing random non master node [{}] current master [{}] ", nodeAndClient.name, getMasterName());
             stopNodesAndClient(nodeAndClient);
@@ -2091,7 +2096,7 @@ public final class InternalTestCluster extends TestCluster {
     }
 
     private synchronized Collection<NodeAndClient> dataAndMasterNodes() {
-        return filterNodes(nodes, new DataOrMasterNodePredicate());
+        return filterNodes(nodes, new DataNodePredicate().or(new MasterNodePredicate()));
     }
 
     private synchronized Collection<NodeAndClient> filterNodes(Map<String, InternalTestCluster.NodeAndClient> map,
@@ -2110,24 +2115,23 @@ public final class InternalTestCluster extends TestCluster {
         }
     }
 
-    private static final class DataOrMasterNodePredicate implements Predicate<NodeAndClient> {
-        @Override
-        public boolean test(NodeAndClient nodeAndClient) {
-            return DiscoveryNode.isDataNode(nodeAndClient.node.settings()) ||
-                DiscoveryNode.isMasterNode(nodeAndClient.node.settings());
-        }
-    }
-
     private static final class MasterNodePredicate implements Predicate<NodeAndClient> {
-        private final String masterNodeName;
+        @Override
+        public boolean test(NodeAndClient nodeAndClient) {
+            return DiscoveryNode.isMasterNode(nodeAndClient.node.settings());
+        }
+    }
 
-        MasterNodePredicate(String masterNodeName) {
+    private static final class NodeNamePredicate implements Predicate<NodeAndClient> {
-            this.masterNodeName = masterNodeName;
+        private final HashSet<String> nodeNames;
+
+        NodeNamePredicate(String... nodeNames) {
+            this.nodeNames = Sets.newHashSet(nodeNames);
         }
 
         @Override
         public boolean test(NodeAndClient nodeAndClient) {
-            return masterNodeName.equals(nodeAndClient.name);
+            return nodeNames.contains(nodeAndClient.getName());
         }
     }
 
@@ -2214,25 +2218,11 @@ public final class InternalTestCluster extends TestCluster {
     /**
      * Returns a predicate that only accepts settings of nodes with one of the given names.
      */
-    public static Predicate<Settings> nameFilter(String... nodeName) {
+    public static Predicate<Settings> nameFilter(String... nodeNames) {
-        return new NodeNamePredicate(new HashSet<>(Arrays.asList(nodeName)));
+        final Set<String> nodes = Sets.newHashSet(nodeNames);
+        return settings -> nodes.contains(settings.get("node.name"));
     }
 
-    private static final class NodeNamePredicate implements Predicate<Settings> {
-        private final HashSet<String> nodeNames;
-
-        NodeNamePredicate(HashSet<String> nodeNames) {
-            this.nodeNames = nodeNames;
-        }
-
-        @Override
-        public boolean test(Settings settings) {
-            return nodeNames.contains(settings.get("node.name"));
-
-        }
-    }
-
-
     /**
      * An abstract class that is called during {@link #rollingRestart(InternalTestCluster.RestartCallback)}
      * and / or {@link #fullRestart(InternalTestCluster.RestartCallback)} to execute actions at certain

x-pack/plugin/core/src/test/java/org/elasticsearch/license/AbstractLicensesIntegrationTestCase.java

@@ -26,7 +26,7 @@ public abstract class AbstractLicensesIntegrationTestCase extends ESIntegTestCas
 
     @Override
     protected Settings nodeSettings(int nodeOrdinal) {
-        return Settings.builder().put(XPackSettings.SECURITY_ENABLED.getKey(), false).build();
+        return Settings.builder().put(super.nodeSettings(nodeOrdinal)).put(XPackSettings.SECURITY_ENABLED.getKey(), false).build();
     }
 
     @Override

x-pack/plugin/core/src/test/java/org/elasticsearch/license/LicenseServiceClusterNotRecoveredTests.java

@@ -9,6 +9,7 @@ import org.elasticsearch.analysis.common.CommonAnalysisPlugin;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.test.ESIntegTestCase;
+import org.elasticsearch.test.discovery.TestZenDiscovery;
 import org.elasticsearch.transport.Netty4Plugin;
 import org.elasticsearch.xpack.core.LocalStateCompositeXPackPlugin;
 
@@ -35,6 +36,7 @@ public class LicenseServiceClusterNotRecoveredTests extends AbstractLicensesInte
         return Settings.builder()
                 .put(super.nodeSettings(nodeOrdinal))
                 .put("node.data", true)
+                .put(TestZenDiscovery.USE_ZEN2.getKey(), false) // this test is just weird
                 .put("resource.reload.interval.high", "500ms"); // for license mode file watcher
     }
 

x-pack/plugin/core/src/test/java/org/elasticsearch/license/LicenseServiceClusterTests.java

@@ -11,6 +11,7 @@ import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.test.ESIntegTestCase.ClusterScope;
+import org.elasticsearch.test.discovery.TestZenDiscovery;
 import org.elasticsearch.transport.Netty4Plugin;
 import org.elasticsearch.xpack.core.LocalStateCompositeXPackPlugin;
 import org.elasticsearch.xpack.core.XPackPlugin;
@@ -42,6 +43,7 @@ public class LicenseServiceClusterTests extends AbstractLicensesIntegrationTestC
         return Settings.builder()
                 .put(super.nodeSettings(nodeOrdinal))
                 .put("node.data", true)
+                .put(TestZenDiscovery.USE_ZEN2.getKey(), false) // no state persistence
                 .put("resource.reload.interval.high", "500ms"); // for license mode file watcher
     }
 

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/integration/MlDistributedFailureIT.java

@@ -20,6 +20,7 @@ import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.persistent.PersistentTasksCustomMetaData;
 import org.elasticsearch.persistent.PersistentTasksCustomMetaData.PersistentTask;
+import org.elasticsearch.test.discovery.TestZenDiscovery;
 import org.elasticsearch.test.junit.annotations.TestLogging;
 import org.elasticsearch.xpack.core.ml.action.CloseJobAction;
 import org.elasticsearch.xpack.core.ml.action.GetDatafeedsStatsAction;
@@ -46,6 +47,13 @@ import static org.elasticsearch.persistent.PersistentTasksClusterService.needsRe
 
 public class MlDistributedFailureIT extends BaseMlIntegTestCase {
 
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return Settings.builder().put(super.nodeSettings(nodeOrdinal))
+            .put(TestZenDiscovery.USE_ZEN2.getKey(), false) // no state persistence yet
+            .build();
+    }
+
     public void testFailOver() throws Exception {
         internalCluster().ensureAtLeastNumDataNodes(3);
         ensureStableClusterOnAllNodes(3);

x-pack/plugin/security/src/test/java/org/elasticsearch/license/LicensingTests.java

@@ -300,6 +300,7 @@ public class LicensingTests extends SecurityIntegTestCase {
             .put("path.home", home)
             .put(TestZenDiscovery.USE_MOCK_PINGS.getKey(), false)
             .put(DiscoveryModule.DISCOVERY_TYPE_SETTING.getKey(), "test-zen")
+            .put(TestZenDiscovery.USE_ZEN2.getKey(), true)
             .putList(DiscoveryModule.DISCOVERY_HOSTS_PROVIDER_SETTING.getKey())
             .putList(DISCOVERY_ZEN_PING_UNICAST_HOSTS_SETTING.getKey(), unicastHostsList)
             .build();

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/transport/ServerTransportFilterIntegrationTests.java

@@ -91,7 +91,7 @@ public class ServerTransportFilterIntegrationTests extends SecurityIntegTestCase
         Path xpackConf = home.resolve("config");
         Files.createDirectories(xpackConf);
 
-        Transport transport = internalCluster().getDataNodeInstance(Transport.class);
+        Transport transport = internalCluster().getMasterNodeInstance(Transport.class);
         TransportAddress transportAddress = transport.boundAddress().publishAddress();
         String unicastHost = NetworkAddress.format(transportAddress.address());
 
@@ -108,6 +108,7 @@ public class ServerTransportFilterIntegrationTests extends SecurityIntegTestCase
                 .put(XPackSettings.WATCHER_ENABLED.getKey(), false)
                 .put("path.home", home)
                 .put(Node.NODE_MASTER_SETTING.getKey(), false)
+                .put(TestZenDiscovery.USE_ZEN2.getKey(), true)
                 .put(TestZenDiscovery.USE_MOCK_PINGS.getKey(), false);
                 //.put("xpack.ml.autodetect_process", false);
         Collection<Class<? extends Plugin>> mockPlugins = Arrays.asList(
@@ -132,7 +133,7 @@ public class ServerTransportFilterIntegrationTests extends SecurityIntegTestCase
         writeFile(xpackConf, "users_roles", configUsersRoles());
         writeFile(xpackConf, "roles.yml", configRoles());
 
-        Transport transport = internalCluster().getDataNodeInstance(Transport.class);
+        Transport transport = internalCluster().getMasterNodeInstance(Transport.class);
         TransportAddress transportAddress = transport.profileBoundAddresses().get("client").publishAddress();
         String unicastHost = NetworkAddress.format(transportAddress.address());
 
@@ -151,6 +152,7 @@ public class ServerTransportFilterIntegrationTests extends SecurityIntegTestCase
                 .put("discovery.initial_state_timeout", "0s")
                 .put("path.home", home)
                 .put(Node.NODE_MASTER_SETTING.getKey(), false)
+                .put(TestZenDiscovery.USE_ZEN2.getKey(), true)
                 .put(TestZenDiscovery.USE_MOCK_PINGS.getKey(), false);
                 //.put("xpack.ml.autodetect_process", false);
         Collection<Class<? extends Plugin>> mockPlugins = Arrays.asList(