solr/core/src/test/org/apache/solr/cloud/SharedFSAutoReplicaFailoverTest.java - lucene-solr - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF 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.apache.solr.cloud;

 import java.io.IOException;
 import java.lang.invoke.MethodHandles;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CompletionService;
 import java.util.concurrent.ExecutorCompletionService;
 import java.util.concurrent.Future;
 import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;

 import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.apache.lucene.util.LuceneTestCase;
 import org.apache.lucene.util.QuickPatchThreadsFilter;
 import org.apache.lucene.util.LuceneTestCase.Slow;
 import com.carrotsearch.randomizedtesting.annotations.Nightly;

 import org.apache.solr.SolrIgnoredThreadsFilter;
 import org.apache.solr.SolrTestCaseJ4.SuppressSSL;
 import org.apache.solr.client.solrj.SolrQuery;
 import org.apache.solr.client.solrj.SolrServerException;
 import org.apache.solr.client.solrj.embedded.JettySolrRunner;
 import org.apache.solr.client.solrj.request.CollectionAdminRequest;
 import org.apache.solr.client.solrj.request.CollectionAdminRequest.Create;
 import org.apache.solr.client.solrj.response.CollectionAdminResponse;
 import org.apache.solr.client.solrj.response.QueryResponse;
 import org.apache.solr.cloud.hdfs.HdfsTestUtil;
 import org.apache.solr.common.SolrException;
 import org.apache.solr.common.SolrInputDocument;
 import org.apache.solr.common.cloud.ClusterState;
 import org.apache.solr.common.cloud.ClusterStateUtil;
 import org.apache.solr.common.cloud.DocCollection;
 import org.apache.solr.common.cloud.Replica;
 import org.apache.solr.common.cloud.Slice;
 import org.apache.solr.common.cloud.ZkStateReader;
 import org.apache.solr.common.params.SolrParams;
 import org.apache.solr.common.util.ExecutorUtil;
 import org.apache.solr.common.util.TimeSource;
 import org.apache.solr.core.CoreDescriptor;
 import org.apache.solr.common.util.SolrNamedThreadFactory;
 import org.apache.solr.util.BadHdfsThreadsFilter;
 import org.apache.solr.util.LogLevel;
 import org.apache.solr.util.TestInjection;
 import org.apache.solr.util.TimeOut;
 import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.Test;

 import com.carrotsearch.randomizedtesting.annotations.ThreadLeakFilters;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 @Nightly
 @Slow
 @SuppressSSL
 @ThreadLeakFilters(defaultFilters = true, filters = {
     SolrIgnoredThreadsFilter.class,
     QuickPatchThreadsFilter.class,
     BadHdfsThreadsFilter.class // hdfs currently leaks thread(s)
 })
 @LogLevel("org.apache.solr.cloud.autoscaling=DEBUG;org.apache.solr.cloud.*=DEBUG")
 @LuceneTestCase.BadApple(bugUrl="https://issues.apache.org/jira/browse/SOLR-12028") // added 20-Jul-2018
 public class SharedFSAutoReplicaFailoverTest extends AbstractFullDistribZkTestBase {
   private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());

   private static final boolean DEBUG = true;
   private static MiniDFSCluster dfsCluster;

   ThreadPoolExecutor executor = new ExecutorUtil.MDCAwareThreadPoolExecutor(0,
       Integer.MAX_VALUE, 5, TimeUnit.SECONDS, new SynchronousQueue<Runnable>(),
       new SolrNamedThreadFactory("testExecutor"));

   CompletionService<Object> completionService;
   Set<Future<Object>> pending;
   private final Map<String, String> collectionUlogDirMap = new HashMap<>();

   @BeforeClass
   public static void hdfsFailoverBeforeClass() throws Exception {
     System.setProperty("solr.hdfs.blockcache.blocksperbank", "512");
     dfsCluster = HdfsTestUtil.setupClass(createTempDir().toFile().getAbsolutePath());
     System.setProperty("solr.hdfs.blockcache.global", "true"); // always use global cache, this test can create a lot of directories
     schemaString = "schema15.xml";
   }

   @AfterClass
   public static void hdfsFailoverAfterClass() throws Exception {
     try {
       HdfsTestUtil.teardownClass(dfsCluster);
     } finally {
       System.clearProperty("solr.hdfs.blockcache.blocksperbank");
       dfsCluster = null;
     }
   }

   @Before
   @Override
   public void setUp() throws Exception {
     super.setUp();
     collectionUlogDirMap.clear();
     if (random().nextBoolean()) {
       CollectionAdminRequest.setClusterProperty("legacyCloud", "false").process(cloudClient);
     } else {
       CollectionAdminRequest.setClusterProperty("legacyCloud", "true").process(cloudClient);
     }
   }

   @Override
   public void distribSetUp() throws Exception {
     super.distribSetUp();
     useJettyDataDir = false;
   }

   protected String getSolrXml() {
     return "solr.xml";
   }


   public SharedFSAutoReplicaFailoverTest() {
     completionService = new ExecutorCompletionService<>(executor);
     pending = new HashSet<>();
   }

   @Test
   @ShardsFixed(num = 4)
   // 12-Jun-2018 @BadApple(bugUrl="https://issues.apache.org/jira/browse/SOLR-12028")
   //commented 23-AUG-2018  @BadApple(bugUrl="https://issues.apache.org/jira/browse/SOLR-12028") // added 20-Jul-2018
   public void test() throws Exception {
     try {
       // to keep uncommitted docs during failover
       TestInjection.skipIndexWriterCommitOnClose = true;
       testBasics();
     } finally {
       TestInjection.reset();
       if (DEBUG) {
         super.printLayout();
       }
     }
   }

   // very slow tests, especially since jetty is started and stopped
   // serially
   private void testBasics() throws Exception {
     String collection1 = "solrj_collection";
     Create createCollectionRequest = CollectionAdminRequest.createCollection(collection1,"conf1",2,2)
             .setMaxShardsPerNode(2)
             .setRouterField("myOwnField")
             .setAutoAddReplicas(true);
     CollectionAdminResponse response = createCollectionRequest.process(cloudClient);

     assertEquals(0, response.getStatus());
     assertTrue(response.isSuccess());
     waitForRecoveriesToFinish(collection1, false);

     String collection2 = "solrj_collection2";
     createCollectionRequest = CollectionAdminRequest.createCollection(collection2,"conf1",2,2)
             .setMaxShardsPerNode(2)
             .setRouterField("myOwnField")
             .setAutoAddReplicas(false);
     CollectionAdminResponse response2 = createCollectionRequest.process(getCommonCloudSolrClient());

     assertEquals(0, response2.getStatus());
     assertTrue(response2.isSuccess());

     waitForRecoveriesToFinish(collection2, false);

     String collection3 = "solrj_collection3";
     createCollectionRequest = CollectionAdminRequest.createCollection(collection3,"conf1",5,1)
             .setMaxShardsPerNode(1)
             .setRouterField("myOwnField")
             .setAutoAddReplicas(true);
     CollectionAdminResponse response3 = createCollectionRequest.process(getCommonCloudSolrClient());

     assertEquals(0, response3.getStatus());
     assertTrue(response3.isSuccess());

     waitForRecoveriesToFinish(collection3, false);

     // a collection has only 1 replica per a shard
     String collection4 = "solrj_collection4";
     createCollectionRequest = CollectionAdminRequest.createCollection(collection4,"conf1",5,1)
         .setMaxShardsPerNode(5)
         .setRouterField("text")
         .setAutoAddReplicas(true);
     CollectionAdminResponse response4 = createCollectionRequest.process(getCommonCloudSolrClient());

     assertEquals(0, response4.getStatus());
     assertTrue(response4.isSuccess());

     waitForRecoveriesToFinish(collection4, false);

     // all collections
     String[] collections = {collection1, collection2, collection3, collection4};

     // add some documents to collection4
     final int numDocs = 100;
     addDocs(collection4, numDocs, false);  // indexed but not committed

     // no result because not committed yet
     queryAndAssertResultSize(collection4, 0, 10000);

     assertUlogDir(collections);

     jettys.get(1).stop();
     jettys.get(2).stop();

     Thread.sleep(5000);

     assertSliceAndReplicaCount(collection1, 2, 2, 120000);

     assertEquals(4, ClusterStateUtil.getLiveAndActiveReplicaCount(cloudClient.getZkStateReader(), collection1));
     assertTrue(ClusterStateUtil.getLiveAndActiveReplicaCount(cloudClient.getZkStateReader(), collection2) < 4);

     // collection3 has maxShardsPerNode=1, there are 4 standard jetties and one control jetty and 2 nodes stopped
     ClusterStateUtil.waitForLiveAndActiveReplicaCount(cloudClient.getZkStateReader(), collection3, 3, 30000);

     // collection4 has maxShardsPerNode=5 and setMaxShardsPerNode=5
     ClusterStateUtil.waitForLiveAndActiveReplicaCount(cloudClient.getZkStateReader(), collection4, 5, 30000);

     // all docs should be queried after failover
     cloudClient.commit(); // to query all docs
     assertSliceAndReplicaCount(collection4, 5, 1, 120000);
     queryAndAssertResultSize(collection4, numDocs, 10000);

     // collection1 should still be at 4
     assertEquals(4, ClusterStateUtil.getLiveAndActiveReplicaCount(cloudClient.getZkStateReader(), collection1));
     // and collection2 less than 4
     assertTrue(ClusterStateUtil.getLiveAndActiveReplicaCount(cloudClient.getZkStateReader(), collection2) < 4);

     assertUlogDir(collections);

     boolean allowOverseerRestart = random().nextBoolean();
     List<JettySolrRunner> stoppedJetties = allowOverseerRestart
         ? jettys.stream().filter(jettySolrRunner -> random().nextBoolean()).collect(Collectors.toList()) : notOverseerJetties();
     ChaosMonkey.stop(stoppedJetties);
     controlJetty.stop();

     assertTrue("Timeout waiting for all not live", waitingForReplicasNotLive(cloudClient.getZkStateReader(), 45000, stoppedJetties));

     ChaosMonkey.start(stoppedJetties);
     controlJetty.start();

     assertSliceAndReplicaCount(collection1, 2, 2, 120000);
     assertSliceAndReplicaCount(collection3, 5, 1, 120000);

     // all docs should be queried
     assertSliceAndReplicaCount(collection4, 5, 1, 120000);
     queryAndAssertResultSize(collection4, numDocs, 10000);

     assertUlogDir(collections);

     int jettyIndex = random().nextInt(jettys.size());
     jettys.get(jettyIndex).stop();
     jettys.get(jettyIndex).start();

     assertSliceAndReplicaCount(collection1, 2, 2, 120000);

     assertUlogDir(collections);

     assertSliceAndReplicaCount(collection3, 5, 1, 120000);
     assertSliceAndReplicaCount(collection4, 5, 1, 120000);
   }

   private void queryAndAssertResultSize(String collection, int expectedResultSize, int timeoutMS)
       throws SolrServerException, IOException, InterruptedException {
     long startTimestamp = System.nanoTime();

     long actualResultSize = 0;
     while(true) {
       if (System.nanoTime() - startTimestamp > TimeUnit.MILLISECONDS.toNanos(timeoutMS) || actualResultSize > expectedResultSize) {
         fail("expected: " + expectedResultSize + ", actual: " + actualResultSize);
       }
       SolrParams queryAll = new SolrQuery("*:*");
       cloudClient.setDefaultCollection(collection);
       try {
         QueryResponse queryResponse = cloudClient.query(queryAll);
         actualResultSize = queryResponse.getResults().getNumFound();
         if(expectedResultSize == actualResultSize) {
           return;
         }
       } catch (SolrServerException | IOException e) {
         log.warn("Querying solr threw an exception. This can be expected to happen during restarts.", e);
       }

       Thread.sleep(1000);
     }
   }

   private void addDocs(String collection, int numDocs, boolean commit) throws SolrServerException, IOException {
     for (int docId = 1; docId <= numDocs; docId++) {
       SolrInputDocument doc = new SolrInputDocument();
       doc.addField("id", docId);
       doc.addField("text", "shard" + docId % 5);
       cloudClient.setDefaultCollection(collection);
       cloudClient.add(doc);
     }
     if (commit) {
       cloudClient.commit();
     }
   }

   /**
    * After failover, ulogDir should not be changed.
    */
   private void assertUlogDir(String... collections) {
     for (String collection : collections) {
       Collection<Slice> slices = cloudClient.getZkStateReader().getClusterState().getCollection(collection).getSlices();
       for (Slice slice : slices) {
         for (Replica replica : slice.getReplicas()) {
           Map<String, Object> properties = replica.getProperties();
           String coreName = replica.getCoreName();
           String curUlogDir = (String) properties.get(CoreDescriptor.CORE_ULOGDIR);
           String prevUlogDir = collectionUlogDirMap.get(coreName);
           if (curUlogDir != null) {
             if (prevUlogDir == null) {
               collectionUlogDirMap.put(coreName, curUlogDir);
             } else {
               assertEquals(prevUlogDir, curUlogDir);
             }
           }
         }
       }
     }
   }

   // Overseer failover is not currently guaranteed with MoveReplica or Policy Framework
   private List<JettySolrRunner> notOverseerJetties() throws IOException, SolrServerException {
     CollectionAdminResponse response = CollectionAdminRequest.getOverseerStatus().process(cloudClient);
     String overseerNode = (String) response.getResponse().get("leader");
     return jettys.stream().filter(jetty -> !(jetty.getCoreContainer() != null && overseerNode.equals(jetty.getNodeName())))
     .collect(Collectors.toList());
   }

   private boolean waitingForReplicasNotLive(ZkStateReader zkStateReader, int timeoutInMs, List<JettySolrRunner> jetties) {
     Set<String> nodeNames = jetties.stream()
         .filter(jetty -> jetty.getCoreContainer() != null)
         .map(JettySolrRunner::getNodeName)
         .collect(Collectors.toSet());
     long timeout = System.nanoTime()
         + TimeUnit.NANOSECONDS.convert(timeoutInMs, TimeUnit.MILLISECONDS);
     boolean success = false;
     while (!success && System.nanoTime() < timeout) {
       success = true;
       ClusterState clusterState = zkStateReader.getClusterState();
       if (clusterState != null) {
         Map<String, DocCollection> collections = clusterState.getCollectionsMap();
         for (Map.Entry<String, DocCollection> entry : collections.entrySet()) {
           DocCollection docCollection = entry.getValue();
           Collection<Slice> slices = docCollection.getSlices();
           for (Slice slice : slices) {
             // only look at active shards
             if (slice.getState() == Slice.State.ACTIVE) {
               Collection<Replica> replicas = slice.getReplicas();
               for (Replica replica : replicas) {
                 if (nodeNames.contains(replica.getNodeName())) {
                   boolean live = clusterState.liveNodesContain(replica
                       .getNodeName());
                   if (live) {
                     success = false;
                   }
                 }
               }
             }
           }
         }
         if (!success) {
           try {
             Thread.sleep(500);
           } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
             throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "Interrupted");
           }
         }
       }
     }

     return success;
   }

   private void assertSliceAndReplicaCount(String collection, int numSlices, int numReplicas, int timeOutInMs) throws InterruptedException {
     TimeOut timeOut = new TimeOut(timeOutInMs, TimeUnit.MILLISECONDS, TimeSource.NANO_TIME);
     while (!timeOut.hasTimedOut()) {
       ClusterState clusterState = cloudClient.getZkStateReader().getClusterState();
       Collection<Slice> slices = clusterState.getCollection(collection).getActiveSlices();
       if (slices.size() == numSlices) {
         boolean isMatch = true;
         for (Slice slice : slices) {
           int count = 0;
           for (Replica replica : slice.getReplicas()) {
             if (replica.getState() == Replica.State.ACTIVE && clusterState.liveNodesContain(replica.getNodeName())) {
               count++;
             }
           }
           if (count < numReplicas) {
             isMatch = false;
           }
         }
         if (isMatch) return;
       }
       Thread.sleep(200);
     }
     fail("Expected numSlices=" + numSlices + " numReplicas=" + numReplicas + " but found " + cloudClient.getZkStateReader().getClusterState().getCollection(collection) + " with /live_nodes: " + cloudClient.getZkStateReader().getClusterState().getLiveNodes());
   }

 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF 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.apache.solr.cloud;

	import java.io.IOException;
	import java.lang.invoke.MethodHandles;
	import java.util.Collection;
	import java.util.HashMap;
	import java.util.HashSet;
	import java.util.List;
	import java.util.Map;
	import java.util.Set;
	import java.util.concurrent.CompletionService;
	import java.util.concurrent.ExecutorCompletionService;
	import java.util.concurrent.Future;
	import java.util.concurrent.SynchronousQueue;
	import java.util.concurrent.ThreadPoolExecutor;
	import java.util.concurrent.TimeUnit;
	import java.util.stream.Collectors;

	import org.apache.hadoop.hdfs.MiniDFSCluster;
	import org.apache.lucene.util.LuceneTestCase;
	import org.apache.lucene.util.QuickPatchThreadsFilter;
	import org.apache.lucene.util.LuceneTestCase.Slow;
	import com.carrotsearch.randomizedtesting.annotations.Nightly;

	import org.apache.solr.SolrIgnoredThreadsFilter;
	import org.apache.solr.SolrTestCaseJ4.SuppressSSL;
	import org.apache.solr.client.solrj.SolrQuery;
	import org.apache.solr.client.solrj.SolrServerException;
	import org.apache.solr.client.solrj.embedded.JettySolrRunner;
	import org.apache.solr.client.solrj.request.CollectionAdminRequest;
	import org.apache.solr.client.solrj.request.CollectionAdminRequest.Create;
	import org.apache.solr.client.solrj.response.CollectionAdminResponse;
	import org.apache.solr.client.solrj.response.QueryResponse;
	import org.apache.solr.cloud.hdfs.HdfsTestUtil;
	import org.apache.solr.common.SolrException;
	import org.apache.solr.common.SolrInputDocument;
	import org.apache.solr.common.cloud.ClusterState;
	import org.apache.solr.common.cloud.ClusterStateUtil;
	import org.apache.solr.common.cloud.DocCollection;
	import org.apache.solr.common.cloud.Replica;
	import org.apache.solr.common.cloud.Slice;
	import org.apache.solr.common.cloud.ZkStateReader;
	import org.apache.solr.common.params.SolrParams;
	import org.apache.solr.common.util.ExecutorUtil;
	import org.apache.solr.common.util.TimeSource;
	import org.apache.solr.core.CoreDescriptor;
	import org.apache.solr.common.util.SolrNamedThreadFactory;
	import org.apache.solr.util.BadHdfsThreadsFilter;
	import org.apache.solr.util.LogLevel;
	import org.apache.solr.util.TestInjection;
	import org.apache.solr.util.TimeOut;
	import org.junit.AfterClass;
	import org.junit.Before;
	import org.junit.BeforeClass;
	import org.junit.Test;

	import com.carrotsearch.randomizedtesting.annotations.ThreadLeakFilters;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	@Nightly
	@Slow
	@SuppressSSL
	@ThreadLeakFilters(defaultFilters = true, filters = {
	SolrIgnoredThreadsFilter.class,
	QuickPatchThreadsFilter.class,
	BadHdfsThreadsFilter.class // hdfs currently leaks thread(s)
	})
	@LogLevel("org.apache.solr.cloud.autoscaling=DEBUG;org.apache.solr.cloud.*=DEBUG")
	@LuceneTestCase.BadApple(bugUrl="https://issues.apache.org/jira/browse/SOLR-12028") // added 20-Jul-2018
	public class SharedFSAutoReplicaFailoverTest extends AbstractFullDistribZkTestBase {
	private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());

	private static final boolean DEBUG = true;
	private static MiniDFSCluster dfsCluster;

	ThreadPoolExecutor executor = new ExecutorUtil.MDCAwareThreadPoolExecutor(0,
	Integer.MAX_VALUE, 5, TimeUnit.SECONDS, new SynchronousQueue<Runnable>(),
	new SolrNamedThreadFactory("testExecutor"));

	CompletionService<Object> completionService;
	Set<Future<Object>> pending;
	private final Map<String, String> collectionUlogDirMap = new HashMap<>();

	@BeforeClass
	public static void hdfsFailoverBeforeClass() throws Exception {
	System.setProperty("solr.hdfs.blockcache.blocksperbank", "512");
	dfsCluster = HdfsTestUtil.setupClass(createTempDir().toFile().getAbsolutePath());
	System.setProperty("solr.hdfs.blockcache.global", "true"); // always use global cache, this test can create a lot of directories
	schemaString = "schema15.xml";
	}

	@AfterClass
	public static void hdfsFailoverAfterClass() throws Exception {
	try {
	HdfsTestUtil.teardownClass(dfsCluster);
	} finally {
	System.clearProperty("solr.hdfs.blockcache.blocksperbank");
	dfsCluster = null;
	}
	}

	@Before
	@Override
	public void setUp() throws Exception {
	super.setUp();
	collectionUlogDirMap.clear();
	if (random().nextBoolean()) {
	CollectionAdminRequest.setClusterProperty("legacyCloud", "false").process(cloudClient);
	} else {
	CollectionAdminRequest.setClusterProperty("legacyCloud", "true").process(cloudClient);
	}
	}

	@Override
	public void distribSetUp() throws Exception {
	super.distribSetUp();
	useJettyDataDir = false;
	}

	protected String getSolrXml() {
	return "solr.xml";
	}


	public SharedFSAutoReplicaFailoverTest() {
	completionService = new ExecutorCompletionService<>(executor);
	pending = new HashSet<>();
	}

	@Test
	@ShardsFixed(num = 4)
	// 12-Jun-2018 @BadApple(bugUrl="https://issues.apache.org/jira/browse/SOLR-12028")
	//commented 23-AUG-2018 @BadApple(bugUrl="https://issues.apache.org/jira/browse/SOLR-12028") // added 20-Jul-2018
	public void test() throws Exception {
	try {
	// to keep uncommitted docs during failover
	TestInjection.skipIndexWriterCommitOnClose = true;
	testBasics();
	} finally {
	TestInjection.reset();
	if (DEBUG) {
	super.printLayout();
	}
	}
	}

	// very slow tests, especially since jetty is started and stopped
	// serially
	private void testBasics() throws Exception {
	String collection1 = "solrj_collection";
	Create createCollectionRequest = CollectionAdminRequest.createCollection(collection1,"conf1",2,2)
	.setMaxShardsPerNode(2)
	.setRouterField("myOwnField")
	.setAutoAddReplicas(true);
	CollectionAdminResponse response = createCollectionRequest.process(cloudClient);

	assertEquals(0, response.getStatus());
	assertTrue(response.isSuccess());
	waitForRecoveriesToFinish(collection1, false);

	String collection2 = "solrj_collection2";
	createCollectionRequest = CollectionAdminRequest.createCollection(collection2,"conf1",2,2)
	.setMaxShardsPerNode(2)
	.setRouterField("myOwnField")
	.setAutoAddReplicas(false);
	CollectionAdminResponse response2 = createCollectionRequest.process(getCommonCloudSolrClient());

	assertEquals(0, response2.getStatus());
	assertTrue(response2.isSuccess());

	waitForRecoveriesToFinish(collection2, false);

	String collection3 = "solrj_collection3";
	createCollectionRequest = CollectionAdminRequest.createCollection(collection3,"conf1",5,1)
	.setMaxShardsPerNode(1)
	.setRouterField("myOwnField")
	.setAutoAddReplicas(true);
	CollectionAdminResponse response3 = createCollectionRequest.process(getCommonCloudSolrClient());

	assertEquals(0, response3.getStatus());
	assertTrue(response3.isSuccess());

	waitForRecoveriesToFinish(collection3, false);

	// a collection has only 1 replica per a shard
	String collection4 = "solrj_collection4";
	createCollectionRequest = CollectionAdminRequest.createCollection(collection4,"conf1",5,1)
	.setMaxShardsPerNode(5)
	.setRouterField("text")
	.setAutoAddReplicas(true);
	CollectionAdminResponse response4 = createCollectionRequest.process(getCommonCloudSolrClient());

	assertEquals(0, response4.getStatus());
	assertTrue(response4.isSuccess());

	waitForRecoveriesToFinish(collection4, false);

	// all collections
	String[] collections = {collection1, collection2, collection3, collection4};

	// add some documents to collection4
	final int numDocs = 100;
	addDocs(collection4, numDocs, false); // indexed but not committed

	// no result because not committed yet
	queryAndAssertResultSize(collection4, 0, 10000);

	assertUlogDir(collections);

	jettys.get(1).stop();
	jettys.get(2).stop();

	Thread.sleep(5000);

	assertSliceAndReplicaCount(collection1, 2, 2, 120000);

	assertEquals(4, ClusterStateUtil.getLiveAndActiveReplicaCount(cloudClient.getZkStateReader(), collection1));
	assertTrue(ClusterStateUtil.getLiveAndActiveReplicaCount(cloudClient.getZkStateReader(), collection2) < 4);

	// collection3 has maxShardsPerNode=1, there are 4 standard jetties and one control jetty and 2 nodes stopped
	ClusterStateUtil.waitForLiveAndActiveReplicaCount(cloudClient.getZkStateReader(), collection3, 3, 30000);

	// collection4 has maxShardsPerNode=5 and setMaxShardsPerNode=5
	ClusterStateUtil.waitForLiveAndActiveReplicaCount(cloudClient.getZkStateReader(), collection4, 5, 30000);

	// all docs should be queried after failover
	cloudClient.commit(); // to query all docs
	assertSliceAndReplicaCount(collection4, 5, 1, 120000);
	queryAndAssertResultSize(collection4, numDocs, 10000);

	// collection1 should still be at 4
	assertEquals(4, ClusterStateUtil.getLiveAndActiveReplicaCount(cloudClient.getZkStateReader(), collection1));
	// and collection2 less than 4
	assertTrue(ClusterStateUtil.getLiveAndActiveReplicaCount(cloudClient.getZkStateReader(), collection2) < 4);

	assertUlogDir(collections);

	boolean allowOverseerRestart = random().nextBoolean();
	List<JettySolrRunner> stoppedJetties = allowOverseerRestart
	? jettys.stream().filter(jettySolrRunner -> random().nextBoolean()).collect(Collectors.toList()) : notOverseerJetties();
	ChaosMonkey.stop(stoppedJetties);
	controlJetty.stop();

	assertTrue("Timeout waiting for all not live", waitingForReplicasNotLive(cloudClient.getZkStateReader(), 45000, stoppedJetties));

	ChaosMonkey.start(stoppedJetties);
	controlJetty.start();

	assertSliceAndReplicaCount(collection1, 2, 2, 120000);
	assertSliceAndReplicaCount(collection3, 5, 1, 120000);

	// all docs should be queried
	assertSliceAndReplicaCount(collection4, 5, 1, 120000);
	queryAndAssertResultSize(collection4, numDocs, 10000);

	assertUlogDir(collections);

	int jettyIndex = random().nextInt(jettys.size());
	jettys.get(jettyIndex).stop();
	jettys.get(jettyIndex).start();

	assertSliceAndReplicaCount(collection1, 2, 2, 120000);

	assertUlogDir(collections);

	assertSliceAndReplicaCount(collection3, 5, 1, 120000);
	assertSliceAndReplicaCount(collection4, 5, 1, 120000);
	}

	private void queryAndAssertResultSize(String collection, int expectedResultSize, int timeoutMS)
	throws SolrServerException, IOException, InterruptedException {
	long startTimestamp = System.nanoTime();

	long actualResultSize = 0;
	while(true) {
	if (System.nanoTime() - startTimestamp > TimeUnit.MILLISECONDS.toNanos(timeoutMS) \|\| actualResultSize > expectedResultSize) {
	fail("expected: " + expectedResultSize + ", actual: " + actualResultSize);
	}
	SolrParams queryAll = new SolrQuery(":");
	cloudClient.setDefaultCollection(collection);
	try {
	QueryResponse queryResponse = cloudClient.query(queryAll);
	actualResultSize = queryResponse.getResults().getNumFound();
	if(expectedResultSize == actualResultSize) {
	return;
	}
	} catch (SolrServerException \| IOException e) {
	log.warn("Querying solr threw an exception. This can be expected to happen during restarts.", e);
	}

	Thread.sleep(1000);
	}
	}

	private void addDocs(String collection, int numDocs, boolean commit) throws SolrServerException, IOException {
	for (int docId = 1; docId <= numDocs; docId++) {
	SolrInputDocument doc = new SolrInputDocument();
	doc.addField("id", docId);
	doc.addField("text", "shard" + docId % 5);
	cloudClient.setDefaultCollection(collection);
	cloudClient.add(doc);
	}
	if (commit) {
	cloudClient.commit();
	}
	}

	/**
	* After failover, ulogDir should not be changed.
	*/
	private void assertUlogDir(String... collections) {
	for (String collection : collections) {
	Collection<Slice> slices = cloudClient.getZkStateReader().getClusterState().getCollection(collection).getSlices();
	for (Slice slice : slices) {
	for (Replica replica : slice.getReplicas()) {
	Map<String, Object> properties = replica.getProperties();
	String coreName = replica.getCoreName();
	String curUlogDir = (String) properties.get(CoreDescriptor.CORE_ULOGDIR);
	String prevUlogDir = collectionUlogDirMap.get(coreName);
	if (curUlogDir != null) {
	if (prevUlogDir == null) {
	collectionUlogDirMap.put(coreName, curUlogDir);
	} else {
	assertEquals(prevUlogDir, curUlogDir);
	}
	}
	}
	}
	}
	}

	// Overseer failover is not currently guaranteed with MoveReplica or Policy Framework
	private List<JettySolrRunner> notOverseerJetties() throws IOException, SolrServerException {
	CollectionAdminResponse response = CollectionAdminRequest.getOverseerStatus().process(cloudClient);
	String overseerNode = (String) response.getResponse().get("leader");
	return jettys.stream().filter(jetty -> !(jetty.getCoreContainer() != null && overseerNode.equals(jetty.getNodeName())))
	.collect(Collectors.toList());
	}

	private boolean waitingForReplicasNotLive(ZkStateReader zkStateReader, int timeoutInMs, List<JettySolrRunner> jetties) {
	Set<String> nodeNames = jetties.stream()
	.filter(jetty -> jetty.getCoreContainer() != null)
	.map(JettySolrRunner::getNodeName)
	.collect(Collectors.toSet());
	long timeout = System.nanoTime()
	+ TimeUnit.NANOSECONDS.convert(timeoutInMs, TimeUnit.MILLISECONDS);
	boolean success = false;
	while (!success && System.nanoTime() < timeout) {
	success = true;
	ClusterState clusterState = zkStateReader.getClusterState();
	if (clusterState != null) {
	Map<String, DocCollection> collections = clusterState.getCollectionsMap();
	for (Map.Entry<String, DocCollection> entry : collections.entrySet()) {
	DocCollection docCollection = entry.getValue();
	Collection<Slice> slices = docCollection.getSlices();
	for (Slice slice : slices) {
	// only look at active shards
	if (slice.getState() == Slice.State.ACTIVE) {
	Collection<Replica> replicas = slice.getReplicas();
	for (Replica replica : replicas) {
	if (nodeNames.contains(replica.getNodeName())) {
	boolean live = clusterState.liveNodesContain(replica
	.getNodeName());
	if (live) {
	success = false;
	}
	}
	}
	}
	}
	}
	if (!success) {
	try {
	Thread.sleep(500);
	} catch (InterruptedException e) {
	Thread.currentThread().interrupt();
	throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "Interrupted");
	}
	}
	}
	}

	return success;
	}

	private void assertSliceAndReplicaCount(String collection, int numSlices, int numReplicas, int timeOutInMs) throws InterruptedException {
	TimeOut timeOut = new TimeOut(timeOutInMs, TimeUnit.MILLISECONDS, TimeSource.NANO_TIME);
	while (!timeOut.hasTimedOut()) {
	ClusterState clusterState = cloudClient.getZkStateReader().getClusterState();
	Collection<Slice> slices = clusterState.getCollection(collection).getActiveSlices();
	if (slices.size() == numSlices) {
	boolean isMatch = true;
	for (Slice slice : slices) {
	int count = 0;
	for (Replica replica : slice.getReplicas()) {
	if (replica.getState() == Replica.State.ACTIVE && clusterState.liveNodesContain(replica.getNodeName())) {
	count++;
	}
	}
	if (count < numReplicas) {
	isMatch = false;
	}
	}
	if (isMatch) return;
	}
	Thread.sleep(200);
	}
	fail("Expected numSlices=" + numSlices + " numReplicas=" + numReplicas + " but found " + cloudClient.getZkStateReader().getClusterState().getCollection(collection) + " with /live_nodes: " + cloudClient.getZkStateReader().getClusterState().getLiveNodes());
	}

	}