<commit_before>/*
 * Licensed to Elasticsearch under one or more contributor
 * license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright
 * ownership. Elasticsearch licenses this file to you under
 * the Apache License, Version 2.0 (the "License"); you may
 * not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package org.elasticsearch.discovery;

import com.google.common.base.Predicate;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.block.ClusterBlock;
import org.elasticsearch.cluster.block.ClusterBlockException;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.cluster.node.DiscoveryNodes;
import org.elasticsearch.common.Nullable;
import org.elasticsearch.common.Priority;
import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.test.ElasticsearchIntegrationTest;
import org.elasticsearch.test.junit.annotations.TestLogging;
import org.elasticsearch.test.transport.MockTransportService;
import org.elasticsearch.transport.TransportModule;
import org.elasticsearch.transport.TransportService;
import org.junit.Test;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;

import static org.elasticsearch.test.ElasticsearchIntegrationTest.ClusterScope;
import static org.elasticsearch.test.ElasticsearchIntegrationTest.Scope;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
import static org.hamcrest.Matchers.*;

/**
 */
@ClusterScope(scope = Scope.TEST, numDataNodes = 0)
public class DiscoveryWithNetworkFailuresTests extends ElasticsearchIntegrationTest {

    private static final Settings nodeSettings = ImmutableSettings.settingsBuilder()
            .put("discovery.type", "zen") // <-- To override the local setting if set externally
            .put("discovery.zen.fd.ping_timeout", "1s") // <-- for hitting simulated network failures quickly
            .put("discovery.zen.fd.ping_retries", "1") // <-- for hitting simulated network failures quickly
            .put(DiscoverySettings.PUBLISH_TIMEOUT, "1s") // <-- for hitting simulated network failures quickly
            .put("discovery.zen.minimum_master_nodes", 2)
            .put(TransportModule.TRANSPORT_SERVICE_TYPE_KEY, MockTransportService.class.getName())
            .build();

    @Override
    protected int numberOfShards() {
        return 3;
    }

    @Override
    protected int numberOfReplicas() {
        return 1;
    }

    @Test
    @TestLogging("discovery.zen:TRACE")
    public void failWithMinimumMasterNodesConfigured() throws Exception {

        List<String> nodes = internalCluster().startNodesAsync(3, nodeSettings).get();

        // Wait until 3 nodes are part of the cluster
        ensureStableCluster(3);

        // Figure out what is the elected master node
        DiscoveryNode masterDiscoNode = findMasterNode(nodes);
        logger.info("---> legit elected master node=" + masterDiscoNode);
        final Client masterClient = internalCluster().masterClient();

        // Everything is stable now, it is now time to simulate evil...

        // Pick a node that isn't the elected master.
        String unluckyNode = null;
        for (String node : nodes) {
            if (!node.equals(masterDiscoNode.getName())) {
                unluckyNode = node;
            }
        }
        assert unluckyNode != null;

        // Simulate a network issue between the unlucky node and elected master node in both directions.
        addFailToSendNoConnectRule(masterDiscoNode.getName(), unluckyNode);
        addFailToSendNoConnectRule(unluckyNode, masterDiscoNode.getName());
        try {
            // Wait until elected master has removed that the unlucky node...
            boolean applied = awaitBusy(new Predicate<Object>() {
                @Override
                public boolean apply(Object input) {
                    return masterClient.admin().cluster().prepareState().setLocal(true).get().getState().nodes().size() == 2;
                }
            }, 1, TimeUnit.MINUTES);
            assertThat(applied, is(true));

            // The unlucky node must report *no* master node, since it can't connect to master and in fact it should
            // continuously ping until network failures have been resolved. However
            final Client isolatedNodeClient = internalCluster().client(unluckyNode);
            // It may a take a bit before the node detects it has been cut off from the elected master
            applied = awaitBusy(new Predicate<Object>() {
                @Override
                public boolean apply(Object input) {
                    ClusterState localClusterState = isolatedNodeClient.admin().cluster().prepareState().setLocal(true).get().getState();
                    DiscoveryNodes localDiscoveryNodes = localClusterState.nodes();
                    logger.info("localDiscoveryNodes=" + localDiscoveryNodes.prettyPrint());
                    return localDiscoveryNodes.masterNode() == null;
                }
            }, 10, TimeUnit.SECONDS);
            assertThat(applied, is(true));
        } finally {
            // stop simulating network failures, from this point on the unlucky node is able to rejoin
            // We also need to do this even if assertions fail, since otherwise the test framework can't work properly
            clearNoConnectRule(masterDiscoNode.getName(), unluckyNode);
            clearNoConnectRule(unluckyNode, masterDiscoNode.getName());
        }

        // Wait until the master node sees all 3 nodes again.
        ensureStableCluster(3);

        for (String node : nodes) {
            ClusterState state = internalCluster().client(node).admin().cluster().prepareState().setLocal(true).execute().actionGet().getState();
            assertThat(state.nodes().size(), equalTo(3));
            // The elected master shouldn't have changed, since the unlucky node never could have elected himself as
            // master since m_m_n of 2 could never be satisfied.
            assertThat(state.nodes().masterNode(), equalTo(masterDiscoNode));
        }
    }

    @Test
    @TestLogging("discovery.zen:TRACE,action:TRACE,cluster.service:TRACE,indices.recovery:TRACE,indices.cluster:TRACE")
    public void testDataConsistency() throws Exception {
        List<String> nodes = internalCluster().startNodesAsync(3, nodeSettings).get();

        // Wait until a 3 nodes are part of the cluster
        ensureStableCluster(3);

        assertAcked(prepareCreate("test")
                .addMapping("type", "field", "type=long")
                .get());

        IndexRequestBuilder[] indexRequests = new IndexRequestBuilder[scaledRandomIntBetween(1, 1000)];
        for (int i = 0; i < indexRequests.length; i++) {
            indexRequests[i] = client().prepareIndex("test", "type", String.valueOf(i)).setSource("field", i);
        }
        indexRandom(true, indexRequests);


        for (int i = 0; i < indexRequests.length; i++) {
            GetResponse getResponse = client().prepareGet("test", "type", String.valueOf(i)).get();
            assertThat(getResponse.isExists(), is(true));
            assertThat(getResponse.getVersion(), equalTo(1l));
            assertThat(getResponse.getId(), equalTo(String.valueOf(i)));
        }
        SearchResponse searchResponse = client().prepareSearch("test").setTypes("type")
                .addSort("field", SortOrder.ASC)
                .get();
        assertHitCount(searchResponse, indexRequests.length);
        for (int i = 0; i < searchResponse.getHits().getHits().length; i++) {
            SearchHit searchHit = searchResponse.getHits().getAt(i);
            assertThat(searchHit.id(), equalTo(String.valueOf(i)));
            assertThat((long) searchHit.sortValues()[0], equalTo((long) i));
        }

        // Everything is stable now, it is now time to simulate evil...
        // but first make sure we have no initializing shards and all is green
        // (waiting for green here, because indexing / search in a yellow index is fine as long as no other nodes go down)
        ensureGreen("test");

        // Pick a node that isn't the elected master.
        final String isolatedNode = nodes.get(0);
        final String nonIsolatedNode = nodes.get(1);

        // Simulate a network issue between the unlucky node and the rest of the cluster.
        randomIsolateNode(isolatedNode, nodes);
        try {
            logger.info("wait until elected master has removed [{}]", isolatedNode);
            boolean applied = awaitBusy(new Predicate<Object>() {
                @Override
                public boolean apply(Object input) {
                    return client(nonIsolatedNode).admin().cluster().prepareState().setLocal(true).get().getState().nodes().size() == 2;
                }
            }, 1, TimeUnit.MINUTES);
            assertThat(applied, is(true));

            // The unlucky node must report *no* master node, since it can't connect to master and in fact it should
            // continuously ping until network failures have been resolved. However
            // It may a take a bit before the node detects it has been cut off from the elected master
            logger.info("waiting for isolated node [{}] to have no master", isolatedNode);
            applied = awaitBusy(new Predicate<Object>() {
                @Override
                public boolean apply(Object input) {
                    ClusterState localClusterState = client(isolatedNode).admin().cluster().prepareState().setLocal(true).get().getState();
                    DiscoveryNodes localDiscoveryNodes = localClusterState.nodes();
                    logger.info("localDiscoveryNodes=" + localDiscoveryNodes.prettyPrint());
                    return localDiscoveryNodes.masterNode() == null;
                }
            }, 10, TimeUnit.SECONDS);
            assertThat(applied, is(true));
            ensureStableCluster(2, nonIsolatedNode);

            // Reads on the right side of the split must work
            logger.info("verifying healthy part of cluster returns data");
            searchResponse = client(nonIsolatedNode).prepareSearch("test").setTypes("type")
                    .addSort("field", SortOrder.ASC)
                    .get();
            assertHitCount(searchResponse, indexRequests.length);
            for (int i = 0; i < searchResponse.getHits().getHits().length; i++) {
                SearchHit searchHit = searchResponse.getHits().getAt(i);
                assertThat(searchHit.id(), equalTo(String.valueOf(i)));
                assertThat((long) searchHit.sortValues()[0], equalTo((long) i));
            }

            // Reads on the wrong side of the split are partial
            logger.info("verifying isolated node [{}] returns partial data", isolatedNode);
            searchResponse = client(isolatedNode).prepareSearch("test").setTypes("type")
                    .addSort("field", SortOrder.ASC).setPreference("_only_local")
                    .get();
            assertThat(searchResponse.getSuccessfulShards(), lessThan(searchResponse.getTotalShards()));
            assertThat(searchResponse.getHits().totalHits(), lessThan((long) indexRequests.length));

            logger.info("verifying writes on healthy cluster");
            UpdateResponse updateResponse = client(nonIsolatedNode).prepareUpdate("test", "type", "0").setDoc("field2", 2).get();
            assertThat(updateResponse.getVersion(), equalTo(2l));

            try {
                logger.info("verifying writes on isolated [{}] fail", isolatedNode);
                client(isolatedNode).prepareUpdate("test", "type", "0").setDoc("field2", 2)
                        .setTimeout("1s") // Fail quick, otherwise we wait 60 seconds.
                        .get();
                fail();
            } catch (ClusterBlockException exception) {
                assertThat(exception.status(), equalTo(RestStatus.SERVICE_UNAVAILABLE));
                assertThat(exception.blocks().size(), equalTo(1));
                ClusterBlock clusterBlock = exception.blocks().iterator().next();
                assertThat(clusterBlock.id(), equalTo(DiscoverySettings.NO_MASTER_BLOCK_ID));
            }
        } finally {
            // stop simulating network failures, from this point on the unlucky node is able to rejoin
            // We also need to do this even if assertions fail, since otherwise the test framework can't work properly
            restoreIsolation(isolatedNode, nodes);
        }

        // Wait until the master node sees all 3 nodes again.
        ensureStableCluster(3);

        logger.info("verifying all nodes return all data");
        for (Client client : clients()) {
            searchResponse = client.prepareSearch("test").setTypes("type")
                    .addSort("field", SortOrder.ASC)
                    .get();
            for (int i = 0; i < searchResponse.getHits().getHits().length; i++) {
                SearchHit searchHit = searchResponse.getHits().getAt(i);
                assertThat(searchHit.id(), equalTo(String.valueOf(i)));
                assertThat((long) searchHit.sortValues()[0], equalTo((long) i));
            }

            GetResponse getResponse = client.prepareGet("test", "type", "0").setPreference("_local").get();
            assertThat(getResponse.isExists(), is(true));
            assertThat(getResponse.getId(), equalTo("0"));
            assertThat(getResponse.getVersion(), equalTo(2l));
            for (int i = 1; i < indexRequests.length; i++) {
                getResponse = client.prepareGet("test", "type", String.valueOf(i)).setPreference("_local").get();
                assertThat(getResponse.isExists(), is(true));
                assertThat(getResponse.getVersion(), equalTo(1l));
                assertThat(getResponse.getId(), equalTo(String.valueOf(i)));
            }
        }
    }


    @Test
    @TestLogging("discovery.zen:TRACE,action:TRACE,cluster.service:TRACE,indices.recovery:TRACE,indices.cluster:TRACE")
    public void voidIsolateMasterAndVerifyClusterStateConsensus() throws Exception {
        final List<String> nodes = internalCluster().startNodesAsync(3, nodeSettings).get();
        ensureStableCluster(3);

        assertAcked(prepareCreate("test")
                .setSettings(ImmutableSettings.builder()
                                .put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, 1 + randomInt(2))
                                .put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, randomInt(2))
                ));

        ensureGreen();
        String isolatedNode = findMasterNode(nodes).name();
        String nonIsolatedNode = null;
        for (String node : nodes) {
            if (!node.equals(isolatedNode)) {
                nonIsolatedNode = node;
                break;
            }
        }
        randomIsolateNode(isolatedNode, nodes);

        // make sure cluster reforms
        ensureStableCluster(2, nonIsolatedNode);

        // restore isolation
        restoreIsolation(isolatedNode, nodes);

        ensureStableCluster(3);

        logger.info("issue a reroute");
        // trigger a reroute now, instead of waiting for the background reroute of RerouteService
        assertAcked(client().admin().cluster().prepareReroute());
        // and wait for it to finish.
        assertFalse(client().admin().cluster().prepareHealth().setWaitForRelocatingShards(0).get().isTimedOut());


        // verify all cluster states are the same
        ClusterState state = null;
        for (String node : nodes) {
            ClusterState nodeState = client(node).admin().cluster().prepareState().setLocal(true).get().getState();
            if (state == null) {
                state = nodeState;
                continue;
            }
            // assert nodes are identical
            try {
                assertEquals("unequal versions", state.version(), nodeState.version());
                assertEquals("unequal node count", state.nodes().size(), nodeState.nodes().size());
                assertEquals("different masters ", state.nodes().masterNodeId(), nodeState.nodes().masterNodeId());
                assertEquals("different meta data version", state.metaData().version(), nodeState.metaData().version());
                if (!state.routingTable().prettyPrint().equals(nodeState.routingTable().prettyPrint())) {
                    fail("different routing");
                }
            } catch (AssertionError t) {
                fail("failed comparing cluster state: " + t.getMessage() + "\n" +
                        "--- cluster state of node [" + nodes.get(0) + "]: ---\n" + state.prettyPrint() +
                        "\n--- cluster state [" + node + "]: ---\n" + nodeState.prettyPrint());
            }

        }

    }


    @Test<commit_msg>[Test]  testIsolateMasterAndVerifyClusterStateConsensus didn't wait on initializing shards before comparing cluster states<commit_after>/*
 * Licensed to Elasticsearch under one or more contributor
 * license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright
 * ownership. Elasticsearch licenses this file to you under
 * the Apache License, Version 2.0 (the "License"); you may
 * not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package org.elasticsearch.discovery;

import com.google.common.base.Predicate;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.block.ClusterBlock;
import org.elasticsearch.cluster.block.ClusterBlockException;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.cluster.node.DiscoveryNodes;
import org.elasticsearch.common.Nullable;
import org.elasticsearch.common.Priority;
import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.test.ElasticsearchIntegrationTest;
import org.elasticsearch.test.junit.annotations.TestLogging;
import org.elasticsearch.test.transport.MockTransportService;
import org.elasticsearch.transport.TransportModule;
import org.elasticsearch.transport.TransportService;
import org.junit.Test;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;

import static org.elasticsearch.test.ElasticsearchIntegrationTest.ClusterScope;
import static org.elasticsearch.test.ElasticsearchIntegrationTest.Scope;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
import static org.hamcrest.Matchers.*;

/**
 */
@ClusterScope(scope = Scope.TEST, numDataNodes = 0)
public class DiscoveryWithNetworkFailuresTests extends ElasticsearchIntegrationTest {

    private static final Settings nodeSettings = ImmutableSettings.settingsBuilder()
            .put("discovery.type", "zen") // <-- To override the local setting if set externally
            .put("discovery.zen.fd.ping_timeout", "1s") // <-- for hitting simulated network failures quickly
            .put("discovery.zen.fd.ping_retries", "1") // <-- for hitting simulated network failures quickly
            .put(DiscoverySettings.PUBLISH_TIMEOUT, "1s") // <-- for hitting simulated network failures quickly
            .put("discovery.zen.minimum_master_nodes", 2)
            .put(TransportModule.TRANSPORT_SERVICE_TYPE_KEY, MockTransportService.class.getName())
            .build();

    @Override
    protected int numberOfShards() {
        return 3;
    }

    @Override
    protected int numberOfReplicas() {
        return 1;
    }

    @Test
    @TestLogging("discovery.zen:TRACE")
    public void failWithMinimumMasterNodesConfigured() throws Exception {

        List<String> nodes = internalCluster().startNodesAsync(3, nodeSettings).get();

        // Wait until 3 nodes are part of the cluster
        ensureStableCluster(3);

        // Figure out what is the elected master node
        DiscoveryNode masterDiscoNode = findMasterNode(nodes);
        logger.info("---> legit elected master node=" + masterDiscoNode);
        final Client masterClient = internalCluster().masterClient();

        // Everything is stable now, it is now time to simulate evil...

        // Pick a node that isn't the elected master.
        String unluckyNode = null;
        for (String node : nodes) {
            if (!node.equals(masterDiscoNode.getName())) {
                unluckyNode = node;
            }
        }
        assert unluckyNode != null;

        // Simulate a network issue between the unlucky node and elected master node in both directions.
        addFailToSendNoConnectRule(masterDiscoNode.getName(), unluckyNode);
        addFailToSendNoConnectRule(unluckyNode, masterDiscoNode.getName());
        try {
            // Wait until elected master has removed that the unlucky node...
            boolean applied = awaitBusy(new Predicate<Object>() {
                @Override
                public boolean apply(Object input) {
                    return masterClient.admin().cluster().prepareState().setLocal(true).get().getState().nodes().size() == 2;
                }
            }, 1, TimeUnit.MINUTES);
            assertThat(applied, is(true));

            // The unlucky node must report *no* master node, since it can't connect to master and in fact it should
            // continuously ping until network failures have been resolved. However
            final Client isolatedNodeClient = internalCluster().client(unluckyNode);
            // It may a take a bit before the node detects it has been cut off from the elected master
            applied = awaitBusy(new Predicate<Object>() {
                @Override
                public boolean apply(Object input) {
                    ClusterState localClusterState = isolatedNodeClient.admin().cluster().prepareState().setLocal(true).get().getState();
                    DiscoveryNodes localDiscoveryNodes = localClusterState.nodes();
                    logger.info("localDiscoveryNodes=" + localDiscoveryNodes.prettyPrint());
                    return localDiscoveryNodes.masterNode() == null;
                }
            }, 10, TimeUnit.SECONDS);
            assertThat(applied, is(true));
        } finally {
            // stop simulating network failures, from this point on the unlucky node is able to rejoin
            // We also need to do this even if assertions fail, since otherwise the test framework can't work properly
            clearNoConnectRule(masterDiscoNode.getName(), unluckyNode);
            clearNoConnectRule(unluckyNode, masterDiscoNode.getName());
        }

        // Wait until the master node sees all 3 nodes again.
        ensureStableCluster(3);

        for (String node : nodes) {
            ClusterState state = internalCluster().client(node).admin().cluster().prepareState().setLocal(true).execute().actionGet().getState();
            assertThat(state.nodes().size(), equalTo(3));
            // The elected master shouldn't have changed, since the unlucky node never could have elected himself as
            // master since m_m_n of 2 could never be satisfied.
            assertThat(state.nodes().masterNode(), equalTo(masterDiscoNode));
        }
    }

    @Test
    @TestLogging("discovery.zen:TRACE,action:TRACE,cluster.service:TRACE,indices.recovery:TRACE,indices.cluster:TRACE")
    public void testDataConsistency() throws Exception {
        List<String> nodes = internalCluster().startNodesAsync(3, nodeSettings).get();

        // Wait until a 3 nodes are part of the cluster
        ensureStableCluster(3);

        assertAcked(prepareCreate("test")
                .addMapping("type", "field", "type=long")
                .get());

        IndexRequestBuilder[] indexRequests = new IndexRequestBuilder[scaledRandomIntBetween(1, 1000)];
        for (int i = 0; i < indexRequests.length; i++) {
            indexRequests[i] = client().prepareIndex("test", "type", String.valueOf(i)).setSource("field", i);
        }
        indexRandom(true, indexRequests);


        for (int i = 0; i < indexRequests.length; i++) {
            GetResponse getResponse = client().prepareGet("test", "type", String.valueOf(i)).get();
            assertThat(getResponse.isExists(), is(true));
            assertThat(getResponse.getVersion(), equalTo(1l));
            assertThat(getResponse.getId(), equalTo(String.valueOf(i)));
        }
        SearchResponse searchResponse = client().prepareSearch("test").setTypes("type")
                .addSort("field", SortOrder.ASC)
                .get();
        assertHitCount(searchResponse, indexRequests.length);
        for (int i = 0; i < searchResponse.getHits().getHits().length; i++) {
            SearchHit searchHit = searchResponse.getHits().getAt(i);
            assertThat(searchHit.id(), equalTo(String.valueOf(i)));
            assertThat((long) searchHit.sortValues()[0], equalTo((long) i));
        }

        // Everything is stable now, it is now time to simulate evil...
        // but first make sure we have no initializing shards and all is green
        // (waiting for green here, because indexing / search in a yellow index is fine as long as no other nodes go down)
        ensureGreen("test");

        // Pick a node that isn't the elected master.
        final String isolatedNode = nodes.get(0);
        final String nonIsolatedNode = nodes.get(1);

        // Simulate a network issue between the unlucky node and the rest of the cluster.
        randomIsolateNode(isolatedNode, nodes);
        try {
            logger.info("wait until elected master has removed [{}]", isolatedNode);
            boolean applied = awaitBusy(new Predicate<Object>() {
                @Override
                public boolean apply(Object input) {
                    return client(nonIsolatedNode).admin().cluster().prepareState().setLocal(true).get().getState().nodes().size() == 2;
                }
            }, 1, TimeUnit.MINUTES);
            assertThat(applied, is(true));

            // The unlucky node must report *no* master node, since it can't connect to master and in fact it should
            // continuously ping until network failures have been resolved. However
            // It may a take a bit before the node detects it has been cut off from the elected master
            logger.info("waiting for isolated node [{}] to have no master", isolatedNode);
            applied = awaitBusy(new Predicate<Object>() {
                @Override
                public boolean apply(Object input) {
                    ClusterState localClusterState = client(isolatedNode).admin().cluster().prepareState().setLocal(true).get().getState();
                    DiscoveryNodes localDiscoveryNodes = localClusterState.nodes();
                    logger.info("localDiscoveryNodes=" + localDiscoveryNodes.prettyPrint());
                    return localDiscoveryNodes.masterNode() == null;
                }
            }, 10, TimeUnit.SECONDS);
            assertThat(applied, is(true));
            ensureStableCluster(2, nonIsolatedNode);

            // Reads on the right side of the split must work
            logger.info("verifying healthy part of cluster returns data");
            searchResponse = client(nonIsolatedNode).prepareSearch("test").setTypes("type")
                    .addSort("field", SortOrder.ASC)
                    .get();
            assertHitCount(searchResponse, indexRequests.length);
            for (int i = 0; i < searchResponse.getHits().getHits().length; i++) {
                SearchHit searchHit = searchResponse.getHits().getAt(i);
                assertThat(searchHit.id(), equalTo(String.valueOf(i)));
                assertThat((long) searchHit.sortValues()[0], equalTo((long) i));
            }

            // Reads on the wrong side of the split are partial
            logger.info("verifying isolated node [{}] returns partial data", isolatedNode);
            searchResponse = client(isolatedNode).prepareSearch("test").setTypes("type")
                    .addSort("field", SortOrder.ASC).setPreference("_only_local")
                    .get();
            assertThat(searchResponse.getSuccessfulShards(), lessThan(searchResponse.getTotalShards()));
            assertThat(searchResponse.getHits().totalHits(), lessThan((long) indexRequests.length));

            logger.info("verifying writes on healthy cluster");
            UpdateResponse updateResponse = client(nonIsolatedNode).prepareUpdate("test", "type", "0").setDoc("field2", 2).get();
            assertThat(updateResponse.getVersion(), equalTo(2l));

            try {
                logger.info("verifying writes on isolated [{}] fail", isolatedNode);
                client(isolatedNode).prepareUpdate("test", "type", "0").setDoc("field2", 2)
                        .setTimeout("1s") // Fail quick, otherwise we wait 60 seconds.
                        .get();
                fail();
            } catch (ClusterBlockException exception) {
                assertThat(exception.status(), equalTo(RestStatus.SERVICE_UNAVAILABLE));
                assertThat(exception.blocks().size(), equalTo(1));
                ClusterBlock clusterBlock = exception.blocks().iterator().next();
                assertThat(clusterBlock.id(), equalTo(DiscoverySettings.NO_MASTER_BLOCK_ID));
            }
        } finally {
            // stop simulating network failures, from this point on the unlucky node is able to rejoin
            // We also need to do this even if assertions fail, since otherwise the test framework can't work properly
            restoreIsolation(isolatedNode, nodes);
        }

        // Wait until the master node sees all 3 nodes again.
        ensureStableCluster(3);

        logger.info("verifying all nodes return all data");
        for (Client client : clients()) {
            searchResponse = client.prepareSearch("test").setTypes("type")
                    .addSort("field", SortOrder.ASC)
                    .get();
            for (int i = 0; i < searchResponse.getHits().getHits().length; i++) {
                SearchHit searchHit = searchResponse.getHits().getAt(i);
                assertThat(searchHit.id(), equalTo(String.valueOf(i)));
                assertThat((long) searchHit.sortValues()[0], equalTo((long) i));
            }

            GetResponse getResponse = client.prepareGet("test", "type", "0").setPreference("_local").get();
            assertThat(getResponse.isExists(), is(true));
            assertThat(getResponse.getId(), equalTo("0"));
            assertThat(getResponse.getVersion(), equalTo(2l));
            for (int i = 1; i < indexRequests.length; i++) {
                getResponse = client.prepareGet("test", "type", String.valueOf(i)).setPreference("_local").get();
                assertThat(getResponse.isExists(), is(true));
                assertThat(getResponse.getVersion(), equalTo(1l));
                assertThat(getResponse.getId(), equalTo(String.valueOf(i)));
            }
        }
    }


    @Test
    @TestLogging("discovery.zen:TRACE,action:TRACE,cluster.service:TRACE,indices.recovery:TRACE,indices.cluster:TRACE")
    public void testIsolateMasterAndVerifyClusterStateConsensus() throws Exception {
        final List<String> nodes = internalCluster().startNodesAsync(3, nodeSettings).get();
        ensureStableCluster(3);

        assertAcked(prepareCreate("test")
                .setSettings(ImmutableSettings.builder()
                                .put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, 1 + randomInt(2))
                                .put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, randomInt(2))
                ));

        ensureGreen();
        String isolatedNode = findMasterNode(nodes).name();
        String nonIsolatedNode = null;
        for (String node : nodes) {
            if (!node.equals(isolatedNode)) {
                nonIsolatedNode = node;
                break;
            }
        }
        randomIsolateNode(isolatedNode, nodes);

        // make sure cluster reforms
        ensureStableCluster(2, nonIsolatedNode);

        // restore isolation
        restoreIsolation(isolatedNode, nodes);

        ensureStableCluster(3);

        logger.info("issue a reroute");
        // trigger a reroute now, instead of waiting for the background reroute of RerouteService
        assertAcked(client().admin().cluster().prepareReroute());
        // and wait for it to finish and for the cluster to stabilize
        ensureGreen("test");

        // verify all cluster states are the same
        ClusterState state = null;
        for (String node : nodes) {
            ClusterState nodeState = client(node).admin().cluster().prepareState().setLocal(true).get().getState();
            if (state == null) {
                state = nodeState;
                continue;
            }
            // assert nodes are identical
            try {
                assertEquals("unequal versions", state.version(), nodeState.version());
                assertEquals("unequal node count", state.nodes().size(), nodeState.nodes().size());
                assertEquals("different masters ", state.nodes().masterNodeId(), nodeState.nodes().masterNodeId());
                assertEquals("different meta data version", state.metaData().version(), nodeState.metaData().version());
                if (!state.routingTable().prettyPrint().equals(nodeState.routingTable().prettyPrint())) {
                    fail("different routing");
                }
            } catch (AssertionError t) {
                fail("failed comparing cluster state: " + t.getMessage() + "\n" +
                        "--- cluster state of node [" + nodes.get(0) + "]: ---\n" + state.prettyPrint() +
                        "\n--- cluster state [" + node + "]: ---\n" + nodeState.prettyPrint());
            }

        }

    }


    @Test
{'fraction_non_alphanumeric': 0.10979547900968784, 'fraction_numerical': 0.004002097651182689, 'mean_word_length': 2.61741214057508, 'pattern_counts': {'":': 0, '<': 37, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '379438', 'n_tokens_mistral': 9161, 'n_tokens_neox': 8869, 'n_words': 2355}