solr/core/src/test/org/apache/solr/cloud/PeerSyncReplicationTest.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.nio.file.Files;
 import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;

 import com.codahale.metrics.Counter;
 import com.codahale.metrics.Metric;
 import com.codahale.metrics.MetricRegistry;
 import org.apache.commons.lang3.RandomStringUtils;
 import org.apache.lucene.util.LuceneTestCase.Slow;
 import org.apache.solr.client.solrj.SolrQuery;
 import org.apache.solr.client.solrj.SolrServerException;
 import org.apache.solr.client.solrj.request.UpdateRequest;
 import org.apache.solr.cloud.ZkTestServer.LimitViolationAction;
 import org.apache.solr.common.SolrInputDocument;
 import org.apache.solr.common.cloud.ClusterState;
 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.ModifiableSolrParams;
 import org.apache.solr.common.util.TimeSource;
 import org.apache.solr.core.CoreContainer;
 import org.apache.solr.metrics.SolrMetricManager;
 import org.apache.solr.util.TimeOut;
 import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import static java.util.Collections.singletonList;
 import static java.util.concurrent.TimeUnit.SECONDS;

 /**
  * Test PeerSync when a node restarts and documents are indexed when node was down.
  *
  * This test is modeled after SyncSliceTest
  */
 @Slow
 public class PeerSyncReplicationTest extends AbstractFullDistribZkTestBase {

   private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());

   private boolean success = false;
   int docId = 0;

   List<CloudJettyRunner> nodesDown = new ArrayList<>();

   @Override
   public void distribTearDown() throws Exception {
     if (!success) {
       printLayoutOnTearDown = true;
     }
     System.clearProperty("distribUpdateSoTimeout");
     System.clearProperty("solr.directoryFactory");
     System.clearProperty("solr.ulog.numRecordsToKeep");
     System.clearProperty("tests.zk.violationReportAction");
     super.distribTearDown();
   }

   public PeerSyncReplicationTest() {
     super();
     sliceCount = 1;
     fixShardCount(3);
   }

   protected String getCloudSolrConfig() {
     return "solrconfig-tlog.xml";
   }

   @Override
   public void distribSetUp() throws Exception {
     // set socket timeout small, so replica won't be put into LIR state when they restart
     System.setProperty("distribUpdateSoTimeout", "3000");
     // tlog gets deleted after node restarts if we use CachingDirectoryFactory.
     // make sure that tlog stays intact after we restart a node
     System.setProperty("solr.directoryFactory", "solr.StandardDirectoryFactory");
     System.setProperty("solr.ulog.numRecordsToKeep", "1000");
     System.setProperty("tests.zk.violationReportAction", LimitViolationAction.IGNORE.toString());
     super.distribSetUp();
   }

   @Test
   //commented 2-Aug-2018 @BadApple(bugUrl="https://issues.apache.org/jira/browse/SOLR-12028")
   public void test() throws Exception {
     handle.clear();
     handle.put("timestamp", SKIPVAL);

     waitForThingsToLevelOut(30);

     del("*:*");

     // index enough docs and commit to establish frame of reference for PeerSync
     for (int i = 0; i < 100; i++) {
       indexDoc(id, docId, i1, 50, tlong, 50, t1,
           "document number " + docId++);
     }
     commit();
     waitForThingsToLevelOut(30);

     try {
       checkShardConsistency(false, true);

       long cloudClientDocs = cloudClient.query(new SolrQuery("*:*")).getResults().getNumFound();
       assertEquals(docId, cloudClientDocs);

       CloudJettyRunner initialLeaderJetty = shardToLeaderJetty.get("shard1");
       List<CloudJettyRunner> otherJetties = getOtherAvailableJetties(initialLeaderJetty);
       CloudJettyRunner neverLeader = otherJetties.get(otherJetties.size() - 1);
       otherJetties.remove(neverLeader) ;

       // first shutdown a node that will never be a leader
       forceNodeFailures(singletonList(neverLeader));

       // node failure and recovery via PeerSync
       log.info("Forcing PeerSync");
       CloudJettyRunner nodePeerSynced = forceNodeFailureAndDoPeerSync(false);

       // add a few more docs
       indexDoc(id, docId, i1, 50, tlong, 50, t1,
           "document number " + docId++);
       indexDoc(id, docId, i1, 50, tlong, 50, t1,
           "document number " + docId++);
       commit();

       cloudClientDocs = cloudClient.query(new SolrQuery("*:*")).getResults().getNumFound();
       assertEquals(docId, cloudClientDocs);

       // now shutdown all other nodes except for 'nodeShutDownForFailure'
       otherJetties.remove(nodePeerSynced);
       forceNodeFailures(otherJetties);
       waitForThingsToLevelOut(30);
       checkShardConsistency(false, true);

       // now shutdown the original leader
       log.info("Now shutting down initial leader");
       forceNodeFailures(singletonList(initialLeaderJetty));
       log.info("Updating mappings from zk");
       waitForNewLeader(cloudClient, "shard1", (Replica) initialLeaderJetty.client.info, new TimeOut(15, SECONDS, TimeSource.NANO_TIME));
       updateMappingsFromZk(jettys, clients, true);
       assertEquals("PeerSynced node did not become leader", nodePeerSynced, shardToLeaderJetty.get("shard1"));

       // bring up node that was down all along, and let it PeerSync from the node that was forced to PeerSynce
       bringUpDeadNodeAndEnsureNoReplication(neverLeader, false);
       waitTillNodesActive();

       checkShardConsistency(false, true);


       // bring back all the nodes including initial leader
       // (commented as reports Maximum concurrent create/delete watches above limit violation and reports thread leaks)
       /*for(int i = 0 ; i < nodesDown.size(); i++) {
         bringUpDeadNodeAndEnsureNoReplication(shardToLeaderJetty.get("shard1"), neverLeader, false);
       }
       checkShardConsistency(false, true);*/

       // make sure leader has not changed after bringing initial leader back
       assertEquals(nodePeerSynced, shardToLeaderJetty.get("shard1"));

       // assert metrics
       SolrMetricManager manager = nodePeerSynced.jetty.getCoreContainer().getMetricManager();
       MetricRegistry registry = null;
       for (String name : manager.registryNames()) {
         if (name.startsWith("solr.core.collection1")) {
           registry = manager.registry(name);
           break;
         }
       }
       assertNotNull(registry);
       Map<String, Metric> metrics = registry.getMetrics();
       assertTrue("REPLICATION.peerSync.time present", metrics.containsKey("REPLICATION.peerSync.time"));
       assertTrue("REPLICATION.peerSync.errors present", metrics.containsKey("REPLICATION.peerSync.errors"));

       Counter counter = (Counter)metrics.get("REPLICATION.peerSync.errors");
       assertEquals(0L, counter.getCount());
       success = true;
     } finally {
       System.clearProperty("solr.disableFingerprint");
     }
   }

   class IndexInBackGround extends Thread {
     private int numDocs;
     private CloudJettyRunner runner;

     public IndexInBackGround(int numDocs, CloudJettyRunner nodeToBringUp) {
       super(getClassName());
       this.numDocs = numDocs;
       this.runner = nodeToBringUp;
     }

     public void run() {
       try {
         // If we don't wait for cores get loaded, the leader may put this replica into LIR state
         waitForCoreLoading();
         for (int i = 0; i < numDocs; i++) {
           indexDoc(id, docId, i1, 50, tlong, 50, t1, "document number " + docId);
           docId++;
           // slow down adds, to get documents indexed while in PeerSync
           Thread.sleep(100);
         }
       } catch (Exception e) {
         log.error("Error indexing doc in background", e);
         //Throwing an error here will kill the thread
       }
     }

     private void waitForCoreLoading() throws InterruptedException {
       while (true) {
         if (runner.jetty.getCoreContainer() != null) {
           CoreContainer cc = runner.jetty.getCoreContainer();
           cc.waitForLoadingCoresToFinish(20000);
           break;
         }
         Thread.sleep(100);
       }
     }
   }


   private void forceNodeFailures(List<CloudJettyRunner> replicasToShutDown) throws Exception {
     for (CloudJettyRunner replicaToShutDown : replicasToShutDown) {
       replicaToShutDown.jetty.stop();
     }

     int totalDown = 0;

     Set<CloudJettyRunner> jetties = new HashSet<>();
     jetties.addAll(shardToJetty.get("shard1"));

     if (replicasToShutDown != null) {
       jetties.removeAll(replicasToShutDown);
       totalDown += replicasToShutDown.size();
     }

     jetties.removeAll(nodesDown);
     totalDown += nodesDown.size();

     assertEquals(getShardCount() - totalDown, jetties.size());

     nodesDown.addAll(replicasToShutDown);
   }


   private CloudJettyRunner forceNodeFailureAndDoPeerSync(boolean disableFingerprint)
       throws Exception {
     // kill non leader - new leader could have all the docs or be missing one
     CloudJettyRunner leaderJetty = shardToLeaderJetty.get("shard1");

     List<CloudJettyRunner> nonLeaderJetties = getOtherAvailableJetties(leaderJetty);
     CloudJettyRunner replicaToShutDown = nonLeaderJetties.get(random().nextInt(nonLeaderJetties.size())); // random non leader node

     forceNodeFailures(Arrays.asList(replicaToShutDown));

     // two docs need to be sync'd back when replica restarts
     indexDoc(id, docId, i1, 50, tlong, 50, t1,
         "document number " + docId++);
     indexDoc(id, docId, i1, 50, tlong, 50, t1,
         "document number " + docId++);
     commit();

     bringUpDeadNodeAndEnsureNoReplication(replicaToShutDown, disableFingerprint);

     return replicaToShutDown;
   }


   private void bringUpDeadNodeAndEnsureNoReplication(CloudJettyRunner nodeToBringUp, boolean disableFingerprint)
       throws Exception {
     // disable fingerprint check if needed
     System.setProperty("solr.disableFingerprint", String.valueOf(disableFingerprint));
     // we wait a little bit, so socket between leader -> replica will be timeout
     Thread.sleep(3000);
     IndexInBackGround iib = new IndexInBackGround(50, nodeToBringUp);
     iib.start();

     // bring back dead node and ensure it recovers
     nodeToBringUp.jetty.start();

     nodesDown.remove(nodeToBringUp);

     waitTillNodesActive();
     waitForThingsToLevelOut(30);

     Set<CloudJettyRunner> jetties = new HashSet<>();
     jetties.addAll(shardToJetty.get("shard1"));
     jetties.removeAll(nodesDown);
     assertEquals(getShardCount() - nodesDown.size(), jetties.size());

     waitForThingsToLevelOut(30);

     iib.join();

     cloudClient.commit();

     checkShardConsistency(false, false);

     long cloudClientDocs = cloudClient.query(new SolrQuery("*:*")).getResults().getNumFound();
     assertEquals(docId, cloudClientDocs);

     // if there was no replication, we should not have replication.properties file
     String replicationProperties = nodeToBringUp.jetty.getSolrHome() + "/cores/" + DEFAULT_TEST_COLLECTION_NAME + "/data/replication.properties";
     assertTrue("PeerSync failed. Had to fail back to replication", Files.notExists(Paths.get(replicationProperties)));
   }


   private void waitTillNodesActive() throws Exception {
     for (int i = 0; i < 60; i++) {
       Thread.sleep(3000);
       ZkStateReader zkStateReader = cloudClient.getZkStateReader();
       ClusterState clusterState = zkStateReader.getClusterState();
       DocCollection collection1 = clusterState.getCollection("collection1");
       Slice slice = collection1.getSlice("shard1");
       Collection<Replica> replicas = slice.getReplicas();
       boolean allActive = true;

       Collection<String> nodesDownNames =
           nodesDown.stream()
               .map(n -> n.coreNodeName)
               .collect(Collectors.toList());

       Collection<Replica> replicasToCheck =
           replicas.stream()
               .filter(r -> !nodesDownNames.contains(r.getName()))
               .collect(Collectors.toList());

       for (Replica replica : replicasToCheck) {
         if (!clusterState.liveNodesContain(replica.getNodeName()) || replica.getState() != Replica.State.ACTIVE) {
           allActive = false;
           break;
         }
       }
       if (allActive) {
         return;
       }
     }
     printLayout();
     fail("timeout waiting to see all nodes active");
   }


   private List<CloudJettyRunner> getOtherAvailableJetties(CloudJettyRunner leader) {
     List<CloudJettyRunner> candidates = new ArrayList<>();
     candidates.addAll(shardToJetty.get("shard1"));

     if (leader != null) {
       candidates.remove(leader);
     }

     candidates.removeAll(nodesDown);

     return candidates;
   }


   protected void indexDoc(Object... fields) throws IOException,
       SolrServerException {
     SolrInputDocument doc = new SolrInputDocument();

     addFields(doc, fields);
     addFields(doc, "rnd_s", RandomStringUtils.random(random().nextInt(100) + 100));

     UpdateRequest ureq = new UpdateRequest();
     ureq.add(doc);
     ModifiableSolrParams params = new ModifiableSolrParams();
     ureq.setParams(params);
     ureq.process(cloudClient);
   }

   // skip the randoms - they can deadlock...
   @Override
   protected void indexr(Object... fields) throws Exception {
     SolrInputDocument doc = new SolrInputDocument();
     addFields(doc, fields);
     addFields(doc, "rnd_b", true);
     indexDoc(doc);
   }

 }
	/*
	* 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.nio.file.Files;
	import java.nio.file.Paths;
	import java.util.ArrayList;
	import java.util.Arrays;
	import java.util.Collection;
	import java.util.HashSet;
	import java.util.List;
	import java.util.Map;
	import java.util.Set;
	import java.util.stream.Collectors;

	import com.codahale.metrics.Counter;
	import com.codahale.metrics.Metric;
	import com.codahale.metrics.MetricRegistry;
	import org.apache.commons.lang3.RandomStringUtils;
	import org.apache.lucene.util.LuceneTestCase.Slow;
	import org.apache.solr.client.solrj.SolrQuery;
	import org.apache.solr.client.solrj.SolrServerException;
	import org.apache.solr.client.solrj.request.UpdateRequest;
	import org.apache.solr.cloud.ZkTestServer.LimitViolationAction;
	import org.apache.solr.common.SolrInputDocument;
	import org.apache.solr.common.cloud.ClusterState;
	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.ModifiableSolrParams;
	import org.apache.solr.common.util.TimeSource;
	import org.apache.solr.core.CoreContainer;
	import org.apache.solr.metrics.SolrMetricManager;
	import org.apache.solr.util.TimeOut;
	import org.junit.Test;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import static java.util.Collections.singletonList;
	import static java.util.concurrent.TimeUnit.SECONDS;

	/**
	* Test PeerSync when a node restarts and documents are indexed when node was down.
	*
	* This test is modeled after SyncSliceTest
	*/
	@Slow
	public class PeerSyncReplicationTest extends AbstractFullDistribZkTestBase {

	private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());

	private boolean success = false;
	int docId = 0;

	List<CloudJettyRunner> nodesDown = new ArrayList<>();

	@Override
	public void distribTearDown() throws Exception {
	if (!success) {
	printLayoutOnTearDown = true;
	}
	System.clearProperty("distribUpdateSoTimeout");
	System.clearProperty("solr.directoryFactory");
	System.clearProperty("solr.ulog.numRecordsToKeep");
	System.clearProperty("tests.zk.violationReportAction");
	super.distribTearDown();
	}

	public PeerSyncReplicationTest() {
	super();
	sliceCount = 1;
	fixShardCount(3);
	}

	protected String getCloudSolrConfig() {
	return "solrconfig-tlog.xml";
	}

	@Override
	public void distribSetUp() throws Exception {
	// set socket timeout small, so replica won't be put into LIR state when they restart
	System.setProperty("distribUpdateSoTimeout", "3000");
	// tlog gets deleted after node restarts if we use CachingDirectoryFactory.
	// make sure that tlog stays intact after we restart a node
	System.setProperty("solr.directoryFactory", "solr.StandardDirectoryFactory");
	System.setProperty("solr.ulog.numRecordsToKeep", "1000");
	System.setProperty("tests.zk.violationReportAction", LimitViolationAction.IGNORE.toString());
	super.distribSetUp();
	}

	@Test
	//commented 2-Aug-2018 @BadApple(bugUrl="https://issues.apache.org/jira/browse/SOLR-12028")
	public void test() throws Exception {
	handle.clear();
	handle.put("timestamp", SKIPVAL);

	waitForThingsToLevelOut(30);

	del(":");

	// index enough docs and commit to establish frame of reference for PeerSync
	for (int i = 0; i < 100; i++) {
	indexDoc(id, docId, i1, 50, tlong, 50, t1,
	"document number " + docId++);
	}
	commit();
	waitForThingsToLevelOut(30);

	try {
	checkShardConsistency(false, true);

	long cloudClientDocs = cloudClient.query(new SolrQuery(":")).getResults().getNumFound();
	assertEquals(docId, cloudClientDocs);

	CloudJettyRunner initialLeaderJetty = shardToLeaderJetty.get("shard1");
	List<CloudJettyRunner> otherJetties = getOtherAvailableJetties(initialLeaderJetty);
	CloudJettyRunner neverLeader = otherJetties.get(otherJetties.size() - 1);
	otherJetties.remove(neverLeader) ;

	// first shutdown a node that will never be a leader
	forceNodeFailures(singletonList(neverLeader));

	// node failure and recovery via PeerSync
	log.info("Forcing PeerSync");
	CloudJettyRunner nodePeerSynced = forceNodeFailureAndDoPeerSync(false);

	// add a few more docs
	indexDoc(id, docId, i1, 50, tlong, 50, t1,
	"document number " + docId++);
	indexDoc(id, docId, i1, 50, tlong, 50, t1,
	"document number " + docId++);
	commit();

	cloudClientDocs = cloudClient.query(new SolrQuery(":")).getResults().getNumFound();
	assertEquals(docId, cloudClientDocs);

	// now shutdown all other nodes except for 'nodeShutDownForFailure'
	otherJetties.remove(nodePeerSynced);
	forceNodeFailures(otherJetties);
	waitForThingsToLevelOut(30);
	checkShardConsistency(false, true);

	// now shutdown the original leader
	log.info("Now shutting down initial leader");
	forceNodeFailures(singletonList(initialLeaderJetty));
	log.info("Updating mappings from zk");
	waitForNewLeader(cloudClient, "shard1", (Replica) initialLeaderJetty.client.info, new TimeOut(15, SECONDS, TimeSource.NANO_TIME));
	updateMappingsFromZk(jettys, clients, true);
	assertEquals("PeerSynced node did not become leader", nodePeerSynced, shardToLeaderJetty.get("shard1"));

	// bring up node that was down all along, and let it PeerSync from the node that was forced to PeerSynce
	bringUpDeadNodeAndEnsureNoReplication(neverLeader, false);
	waitTillNodesActive();

	checkShardConsistency(false, true);


	// bring back all the nodes including initial leader
	// (commented as reports Maximum concurrent create/delete watches above limit violation and reports thread leaks)
	/*for(int i = 0 ; i < nodesDown.size(); i++) {
	bringUpDeadNodeAndEnsureNoReplication(shardToLeaderJetty.get("shard1"), neverLeader, false);
	}
	checkShardConsistency(false, true);*/

	// make sure leader has not changed after bringing initial leader back
	assertEquals(nodePeerSynced, shardToLeaderJetty.get("shard1"));

	// assert metrics
	SolrMetricManager manager = nodePeerSynced.jetty.getCoreContainer().getMetricManager();
	MetricRegistry registry = null;
	for (String name : manager.registryNames()) {
	if (name.startsWith("solr.core.collection1")) {
	registry = manager.registry(name);
	break;
	}
	}
	assertNotNull(registry);
	Map<String, Metric> metrics = registry.getMetrics();
	assertTrue("REPLICATION.peerSync.time present", metrics.containsKey("REPLICATION.peerSync.time"));
	assertTrue("REPLICATION.peerSync.errors present", metrics.containsKey("REPLICATION.peerSync.errors"));

	Counter counter = (Counter)metrics.get("REPLICATION.peerSync.errors");
	assertEquals(0L, counter.getCount());
	success = true;
	} finally {
	System.clearProperty("solr.disableFingerprint");
	}
	}

	class IndexInBackGround extends Thread {
	private int numDocs;
	private CloudJettyRunner runner;

	public IndexInBackGround(int numDocs, CloudJettyRunner nodeToBringUp) {
	super(getClassName());
	this.numDocs = numDocs;
	this.runner = nodeToBringUp;
	}

	public void run() {
	try {
	// If we don't wait for cores get loaded, the leader may put this replica into LIR state
	waitForCoreLoading();
	for (int i = 0; i < numDocs; i++) {
	indexDoc(id, docId, i1, 50, tlong, 50, t1, "document number " + docId);
	docId++;
	// slow down adds, to get documents indexed while in PeerSync
	Thread.sleep(100);
	}
	} catch (Exception e) {
	log.error("Error indexing doc in background", e);
	//Throwing an error here will kill the thread
	}
	}

	private void waitForCoreLoading() throws InterruptedException {
	while (true) {
	if (runner.jetty.getCoreContainer() != null) {
	CoreContainer cc = runner.jetty.getCoreContainer();
	cc.waitForLoadingCoresToFinish(20000);
	break;
	}
	Thread.sleep(100);
	}
	}
	}


	private void forceNodeFailures(List<CloudJettyRunner> replicasToShutDown) throws Exception {
	for (CloudJettyRunner replicaToShutDown : replicasToShutDown) {
	replicaToShutDown.jetty.stop();
	}

	int totalDown = 0;

	Set<CloudJettyRunner> jetties = new HashSet<>();
	jetties.addAll(shardToJetty.get("shard1"));

	if (replicasToShutDown != null) {
	jetties.removeAll(replicasToShutDown);
	totalDown += replicasToShutDown.size();
	}

	jetties.removeAll(nodesDown);
	totalDown += nodesDown.size();

	assertEquals(getShardCount() - totalDown, jetties.size());

	nodesDown.addAll(replicasToShutDown);
	}



	private CloudJettyRunner forceNodeFailureAndDoPeerSync(boolean disableFingerprint)
	throws Exception {
	// kill non leader - new leader could have all the docs or be missing one
	CloudJettyRunner leaderJetty = shardToLeaderJetty.get("shard1");

	List<CloudJettyRunner> nonLeaderJetties = getOtherAvailableJetties(leaderJetty);
	CloudJettyRunner replicaToShutDown = nonLeaderJetties.get(random().nextInt(nonLeaderJetties.size())); // random non leader node

	forceNodeFailures(Arrays.asList(replicaToShutDown));

	// two docs need to be sync'd back when replica restarts
	indexDoc(id, docId, i1, 50, tlong, 50, t1,
	"document number " + docId++);
	indexDoc(id, docId, i1, 50, tlong, 50, t1,
	"document number " + docId++);
	commit();

	bringUpDeadNodeAndEnsureNoReplication(replicaToShutDown, disableFingerprint);

	return replicaToShutDown;
	}


	private void bringUpDeadNodeAndEnsureNoReplication(CloudJettyRunner nodeToBringUp, boolean disableFingerprint)
	throws Exception {
	// disable fingerprint check if needed
	System.setProperty("solr.disableFingerprint", String.valueOf(disableFingerprint));
	// we wait a little bit, so socket between leader -> replica will be timeout
	Thread.sleep(3000);
	IndexInBackGround iib = new IndexInBackGround(50, nodeToBringUp);
	iib.start();

	// bring back dead node and ensure it recovers
	nodeToBringUp.jetty.start();

	nodesDown.remove(nodeToBringUp);

	waitTillNodesActive();
	waitForThingsToLevelOut(30);

	Set<CloudJettyRunner> jetties = new HashSet<>();
	jetties.addAll(shardToJetty.get("shard1"));
	jetties.removeAll(nodesDown);
	assertEquals(getShardCount() - nodesDown.size(), jetties.size());

	waitForThingsToLevelOut(30);

	iib.join();

	cloudClient.commit();

	checkShardConsistency(false, false);

	long cloudClientDocs = cloudClient.query(new SolrQuery(":")).getResults().getNumFound();
	assertEquals(docId, cloudClientDocs);

	// if there was no replication, we should not have replication.properties file
	String replicationProperties = nodeToBringUp.jetty.getSolrHome() + "/cores/" + DEFAULT_TEST_COLLECTION_NAME + "/data/replication.properties";
	assertTrue("PeerSync failed. Had to fail back to replication", Files.notExists(Paths.get(replicationProperties)));
	}



	private void waitTillNodesActive() throws Exception {
	for (int i = 0; i < 60; i++) {
	Thread.sleep(3000);
	ZkStateReader zkStateReader = cloudClient.getZkStateReader();
	ClusterState clusterState = zkStateReader.getClusterState();
	DocCollection collection1 = clusterState.getCollection("collection1");
	Slice slice = collection1.getSlice("shard1");
	Collection<Replica> replicas = slice.getReplicas();
	boolean allActive = true;

	Collection<String> nodesDownNames =
	nodesDown.stream()
	.map(n -> n.coreNodeName)
	.collect(Collectors.toList());

	Collection<Replica> replicasToCheck =
	replicas.stream()
	.filter(r -> !nodesDownNames.contains(r.getName()))
	.collect(Collectors.toList());

	for (Replica replica : replicasToCheck) {
	if (!clusterState.liveNodesContain(replica.getNodeName()) \|\| replica.getState() != Replica.State.ACTIVE) {
	allActive = false;
	break;
	}
	}
	if (allActive) {
	return;
	}
	}
	printLayout();
	fail("timeout waiting to see all nodes active");
	}



	private List<CloudJettyRunner> getOtherAvailableJetties(CloudJettyRunner leader) {
	List<CloudJettyRunner> candidates = new ArrayList<>();
	candidates.addAll(shardToJetty.get("shard1"));

	if (leader != null) {
	candidates.remove(leader);
	}

	candidates.removeAll(nodesDown);

	return candidates;
	}



	protected void indexDoc(Object... fields) throws IOException,
	SolrServerException {
	SolrInputDocument doc = new SolrInputDocument();

	addFields(doc, fields);
	addFields(doc, "rnd_s", RandomStringUtils.random(random().nextInt(100) + 100));

	UpdateRequest ureq = new UpdateRequest();
	ureq.add(doc);
	ModifiableSolrParams params = new ModifiableSolrParams();
	ureq.setParams(params);
	ureq.process(cloudClient);
	}

	// skip the randoms - they can deadlock...
	@Override
	protected void indexr(Object... fields) throws Exception {
	SolrInputDocument doc = new SolrInputDocument();
	addFields(doc, fields);
	addFields(doc, "rnd_b", true);
	indexDoc(doc);
	}

	}