[Tests] Added ServiceDisruptionScheme(s) and testAckedIndexing

This commit adds the notion of ServiceDisruptionScheme allowing for introducing disruptions in our test cluster. This
abstraction as used in a couple of wrappers around the functionality offered by MockTransportService to simulate various
network partions. There is also one implementation for causing a node to be slow in processing cluster state updates.

This new mechnaism is integrated into existing tests DiscoveryWithNetworkFailuresTests.

A new test called testAckedIndexing is added to verify retrieval of documents whose indexing was acked during various disruptions.

Closes #6505

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

@@ -342,7 +342,7 @@ public ClusterState execute(ClusterState currentState) {
 
                     @Override
                     public void onFailure(String source, Throwable t) {
-                            logger.error("unexpected failure during [{}]", t, source);
+                        logger.error("unexpected failure during [{}]", t, source);
                     }
 
                     @Override
@@ -408,8 +408,7 @@ public ClusterState execute(ClusterState currentState) {
                 public void onFailure(String source, Throwable t) {
                     if (t instanceof ClusterService.NoLongerMasterException) {
                         logger.debug("not processing {} leave request as we are no longer master", node);
-                    }
-                    else {
+                    } else {
                         logger.error("unexpected failure during [{}]", t, source);
                     }
                 }
@@ -448,8 +447,7 @@ public ClusterState execute(ClusterState currentState) {
             public void onFailure(String source, Throwable t) {
                 if (t instanceof ClusterService.NoLongerMasterException) {
                     logger.debug("not processing [{}] as we are no longer master", source);
-                }
-                else {
+                } else {
                     logger.error("unexpected failure during [{}]", t, source);
                 }
             }
@@ -486,8 +484,7 @@ public ClusterState execute(ClusterState currentState) {
             public void onFailure(String source, Throwable t) {
                 if (t instanceof ClusterService.NoLongerMasterException) {
                     logger.debug("not processing [{}] as we are no longer master", source);
-                }
-                else {
+                } else {
                     logger.error("unexpected failure during [{}]", t, source);
                 }
             }

src/main/java/org/elasticsearch/transport/TransportService.java

@@ -245,6 +245,10 @@ public void removeHandler(String action) {
         }
     }
 
+    protected TransportRequestHandler getHandler(String action) {
+        return serverHandlers.get(action);
+    }
+
     class Adapter implements TransportServiceAdapter {
 
         final MeanMetric rxMetric = new MeanMetric();

src/test/java/org/elasticsearch/cluster/ClusterServiceTests.java

@@ -263,12 +263,12 @@ public void testMasterAwareExecution() throws Exception {
                 .put("discovery.type", "local")
                 .build();
 
-        ListenableFuture<String> master = cluster().startNodeAsync(settings);
-        ListenableFuture<String> nonMaster = cluster().startNodeAsync(settingsBuilder().put(settings).put("node.master", false).build());
+        ListenableFuture<String> master = internalCluster().startNodeAsync(settings);
+        ListenableFuture<String> nonMaster = internalCluster().startNodeAsync(settingsBuilder().put(settings).put("node.master", false).build());
         master.get();
         ensureGreen(); // make sure we have a cluster
 
-        ClusterService clusterService = cluster().getInstance(ClusterService.class, nonMaster.get());
+        ClusterService clusterService = internalCluster().getInstance(ClusterService.class, nonMaster.get());
 
         final boolean[] taskFailed = {false};
         final CountDownLatch latch1 = new CountDownLatch(1);

src/test/java/org/elasticsearch/cluster/NoMasterNodeTests.java

@@ -21,9 +21,9 @@
 
 import org.elasticsearch.action.ActionRequestBuilder;
 import com.google.common.base.Predicate;
+import org.elasticsearch.action.bulk.BulkRequestBuilder;
 import org.elasticsearch.action.count.CountResponse;
 import org.elasticsearch.action.get.GetResponse;
-import org.elasticsearch.action.bulk.BulkRequestBuilder;
 import org.elasticsearch.action.percolate.PercolateSourceBuilder;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.cluster.block.ClusterBlockException;
@@ -48,11 +48,6 @@
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertThrows;
 import static org.hamcrest.Matchers.*;
-import static org.elasticsearch.test.ElasticsearchIntegrationTest.*;
-import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertExists;
-import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
-import static org.hamcrest.Matchers.equalTo;
-import static org.hamcrest.Matchers.greaterThan;
 
 /**
  */

src/test/java/org/elasticsearch/recovery/RecoveryWhileUnderLoadTests.java

@@ -43,7 +43,6 @@
 import static org.elasticsearch.common.settings.ImmutableSettings.settingsBuilder;
 import static org.elasticsearch.index.query.QueryBuilders.matchAllQuery;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.*;
-import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoTimeout;
 import static org.hamcrest.Matchers.equalTo;
 
 public class RecoveryWhileUnderLoadTests extends ElasticsearchIntegrationTest {

src/test/java/org/elasticsearch/test/BackgroundIndexer.java

@@ -217,7 +217,7 @@ public void continueIndexing(int numOfDocs) {
         setBudget(numOfDocs);
     }
 
-    /** Stop all background threads **/
+    /** Stop all background threads * */
     public void stop() throws InterruptedException {
         if (stop.get()) {
             return;

src/test/java/org/elasticsearch/test/ElasticsearchIntegrationTest.java

@@ -97,6 +97,7 @@
 import org.elasticsearch.search.SearchService;
 import org.elasticsearch.test.client.RandomizingClient;
 import org.hamcrest.Matchers;
+import org.elasticsearch.test.disruption.ServiceDisruptionScheme;
 import org.junit.*;
 
 import java.io.IOException;
@@ -583,6 +584,7 @@ protected final void afterInternal() throws IOException {
         boolean success = false;
         try {
             logger.info("[{}#{}]: cleaning up after test", getTestClass().getSimpleName(), getTestName());
+            clearDisruptionScheme();
             final Scope currentClusterScope = getCurrentClusterScope();
             try {
                 if (currentClusterScope != Scope.TEST) {
@@ -696,6 +698,15 @@ protected int numberOfReplicas() {
         return between(minimumNumberOfReplicas(), maximumNumberOfReplicas());
     }
 
+
+    public void setDisruptionScheme(ServiceDisruptionScheme scheme) {
+        internalCluster().setDisruptionScheme(scheme);
+    }
+
+    public void clearDisruptionScheme() {
+        internalCluster().clearDisruptionScheme();
+    }
+
     /**
      * Returns a settings object used in {@link #createIndex(String...)} and {@link #prepareCreate(String)} and friends.
      * This method can be overwritten by subclasses to set defaults for the indices that are created by the test.

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

@@ -76,6 +76,7 @@
 import org.elasticsearch.search.SearchService;
 import org.elasticsearch.test.cache.recycler.MockBigArraysModule;
 import org.elasticsearch.test.cache.recycler.MockPageCacheRecyclerModule;
+import org.elasticsearch.test.disruption.ServiceDisruptionScheme;
 import org.elasticsearch.test.engine.MockEngineModule;
 import org.elasticsearch.test.store.MockFSIndexStoreModule;
 import org.elasticsearch.test.transport.AssertingLocalTransport;
@@ -185,6 +186,8 @@ public final class InternalTestCluster extends TestCluster {
 
     private final boolean hasFilterCache;
 
+    private ServiceDisruptionScheme activeDisruptionScheme;
+
     public InternalTestCluster(long clusterSeed, String clusterName) {
         this(clusterSeed, DEFAULT_MIN_NUM_DATA_NODES, DEFAULT_MAX_NUM_DATA_NODES, clusterName, SettingsSource.EMPTY, DEFAULT_NUM_CLIENT_NODES, DEFAULT_ENABLE_RANDOM_BENCH_NODES);
     }
@@ -288,6 +291,10 @@ public String getClusterName() {
         return clusterName;
     }
 
+    public String[] getNodeNames() {
+        return nodes.keySet().toArray(Strings.EMPTY_ARRAY);
+    }
+
     private static boolean isLocalTransportConfigured() {
         if ("local".equals(System.getProperty("es.node.mode", "network"))) {
             return true;
@@ -486,6 +493,7 @@ public synchronized void ensureAtMostNumDataNodes(int n) throws IOException {
         while (limit.hasNext()) {
             NodeAndClient next = limit.next();
             nodesToRemove.add(next);
+            removeDistruptionSchemeFromNode(next);
             next.close();
         }
         for (NodeAndClient toRemove : nodesToRemove) {
@@ -660,6 +668,10 @@ public boolean apply(NodeAndClient nodeAndClient) {
     @Override
     public void close() {
         if (this.open.compareAndSet(true, false)) {
+            if (activeDisruptionScheme != null) {
+                activeDisruptionScheme.testClusterClosed();
+                activeDisruptionScheme = null;
+            }
             IOUtils.closeWhileHandlingException(nodes.values());
             nodes.clear();
             executor.shutdownNow();
@@ -824,6 +836,7 @@ public synchronized void beforeTest(Random random, double transportClientRatio)
     }
 
     private synchronized void reset(boolean wipeData) throws IOException {
+        clearDisruptionScheme();
         randomlyResetClients();
         if (wipeData) {
             wipeDataDirectories();
@@ -1023,6 +1036,7 @@ public synchronized void stopRandomDataNode() throws IOException {
         NodeAndClient nodeAndClient = getRandomNodeAndClient(new DataNodePredicate());
         if (nodeAndClient != null) {
             logger.info("Closing random node [{}] ", nodeAndClient.name);
+            removeDistruptionSchemeFromNode(nodeAndClient);
             nodes.remove(nodeAndClient.name);
             nodeAndClient.close();
         }
@@ -1042,6 +1056,7 @@ public boolean apply(NodeAndClient nodeAndClient) {
         });
         if (nodeAndClient != null) {
             logger.info("Closing filtered random node [{}] ", nodeAndClient.name);
+            removeDistruptionSchemeFromNode(nodeAndClient);
             nodes.remove(nodeAndClient.name);
             nodeAndClient.close();
         }
@@ -1056,6 +1071,7 @@ public synchronized void stopCurrentMasterNode() throws IOException {
         String masterNodeName = getMasterName();
         assert nodes.containsKey(masterNodeName);
         logger.info("Closing master node [{}] ", masterNodeName);
+        removeDistruptionSchemeFromNode(nodes.get(masterNodeName));
         NodeAndClient remove = nodes.remove(masterNodeName);
         remove.close();
     }
@@ -1067,6 +1083,7 @@ public void stopRandomNonMasterNode() throws IOException {
         NodeAndClient nodeAndClient = getRandomNodeAndClient(Predicates.not(new MasterNodePredicate(getMasterName())));
         if (nodeAndClient != null) {
             logger.info("Closing random non master node [{}] current master [{}] ", nodeAndClient.name, getMasterName());
+            removeDistruptionSchemeFromNode(nodeAndClient);
             nodes.remove(nodeAndClient.name);
             nodeAndClient.close();
         }
@@ -1120,6 +1137,9 @@ private void restartAllNodes(boolean rollingRestart, RestartCallback callback) t
                 if (!callback.doRestart(nodeAndClient.name)) {
                     logger.info("Closing node [{}] during restart", nodeAndClient.name);
                     toRemove.add(nodeAndClient);
+                    if (activeDisruptionScheme != null) {
+                        activeDisruptionScheme.removeFromNode(nodeAndClient.name, this);
+                    }
                     nodeAndClient.close();
                 }
             }
@@ -1134,18 +1154,33 @@ private void restartAllNodes(boolean rollingRestart, RestartCallback callback) t
             for (NodeAndClient nodeAndClient : nodes.values()) {
                 callback.doAfterNodes(numNodesRestarted++, nodeAndClient.nodeClient());
                 logger.info("Restarting node [{}] ", nodeAndClient.name);
+                if (activeDisruptionScheme != null) {
+                    activeDisruptionScheme.removeFromNode(nodeAndClient.name, this);
+                }
                 nodeAndClient.restart(callback);
+                if (activeDisruptionScheme != null) {
+                    activeDisruptionScheme.applyToNode(nodeAndClient.name, this);
+                }
             }
         } else {
             int numNodesRestarted = 0;
             for (NodeAndClient nodeAndClient : nodes.values()) {
                 callback.doAfterNodes(numNodesRestarted++, nodeAndClient.nodeClient());
                 logger.info("Stopping node [{}] ", nodeAndClient.name);
+                if (activeDisruptionScheme != null) {
+                    activeDisruptionScheme.removeFromNode(nodeAndClient.name, this);
+                }
                 nodeAndClient.node.close();
             }
             for (NodeAndClient nodeAndClient : nodes.values()) {
                 logger.info("Starting node [{}] ", nodeAndClient.name);
+                if (activeDisruptionScheme != null) {
+                    activeDisruptionScheme.removeFromNode(nodeAndClient.name, this);
+                }
                 nodeAndClient.restart(callback);
+                if (activeDisruptionScheme != null) {
+                    activeDisruptionScheme.applyToNode(nodeAndClient.name, this);
+                }
             }
         }
     }
@@ -1343,6 +1378,7 @@ private synchronized void publishNode(NodeAndClient nodeAndClient) {
             dataDirToClean.addAll(Arrays.asList(nodeEnv.nodeDataLocations()));
         }
         nodes.put(nodeAndClient.name, nodeAndClient);
+        applyDisruptionSchemeToNode(nodeAndClient);
     }
 
     public void closeNonSharedNodes(boolean wipeData) throws IOException {
@@ -1364,6 +1400,33 @@ public boolean hasFilterCache() {
         return hasFilterCache;
     }
 
+    public void setDisruptionScheme(ServiceDisruptionScheme scheme) {
+        clearDisruptionScheme();
+        scheme.applyToCluster(this);
+        activeDisruptionScheme = scheme;
+    }
+
+    public void clearDisruptionScheme() {
+        if (activeDisruptionScheme != null) {
+            activeDisruptionScheme.removeFromCluster(this);
+        }
+        activeDisruptionScheme = null;
+    }
+
+    private void applyDisruptionSchemeToNode(NodeAndClient nodeAndClient) {
+        if (activeDisruptionScheme != null) {
+            assert nodes.containsKey(nodeAndClient.name);
+            activeDisruptionScheme.applyToNode(nodeAndClient.name, this);
+        }
+    }
+
+    private void removeDistruptionSchemeFromNode(NodeAndClient nodeAndClient) {
+        if (activeDisruptionScheme != null) {
+            assert nodes.containsKey(nodeAndClient.name);
+            activeDisruptionScheme.removeFromNode(nodeAndClient.name, this);
+        }
+    }
+
     private synchronized Collection<NodeAndClient> dataNodeAndClients() {
         return Collections2.filter(nodes.values(), new DataNodePredicate());
     }

src/test/java/org/elasticsearch/test/TestCluster.java

@@ -24,6 +24,7 @@
 import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.logging.ESLogger;
 import org.elasticsearch.common.logging.Loggers;
 import org.elasticsearch.indices.IndexMissingException;