solr/core/src/test/org/apache/solr/cloud/TestTlogReplayVsRecovery.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.net.URI;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;

 import org.apache.lucene.util.LuceneTestCase.AwaitsFix;

 import org.apache.solr.JSONTestUtil;
 import org.apache.solr.client.solrj.SolrServerException;
 import org.apache.solr.client.solrj.cloud.SocketProxy;
 import org.apache.solr.client.solrj.embedded.JettySolrRunner;
 import org.apache.solr.client.solrj.impl.HttpSolrClient;
 import org.apache.solr.client.solrj.request.CollectionAdminRequest;
 import org.apache.solr.client.solrj.request.QueryRequest;
 import org.apache.solr.client.solrj.response.QueryResponse;
 import org.apache.solr.client.solrj.response.RequestStatusState;
 import org.apache.solr.common.SolrInputDocument;
 import org.apache.solr.common.cloud.Replica;
 import org.apache.solr.util.TestInjection;
 import org.junit.After;
 import org.junit.Before;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 @AwaitsFix(bugUrl="https://issues.apache.org/jira/browse/SOLR-13486;https://issues.apache.org/jira/browse/SOLR-14183")
 public class TestTlogReplayVsRecovery extends SolrCloudTestCase {

   private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
   private static final String COLLECTION = "collecion_with_slow_tlog_recovery";

   private JettySolrRunner NODE0;
   private JettySolrRunner NODE1;
   private Map<JettySolrRunner, SocketProxy> proxies;
   private Map<URI, JettySolrRunner> jettys;

   // we want to ensure there is tlog replay on the leader after we restart it,
   // so in addition to not committing the docs we add during network partition
   // we also want to ensure that our leader doesn't do a "Commit on close"
   //
   // TODO: once SOLR-13486 is fixed, we should randomize this...
   private static final boolean TEST_VALUE_FOR_SKIP_COMMIT_ON_CLOSE = true;

   @Before
   public void setupCluster() throws Exception {
     TestInjection.skipIndexWriterCommitOnClose = TEST_VALUE_FOR_SKIP_COMMIT_ON_CLOSE;

     System.setProperty("solr.directoryFactory", "solr.StandardDirectoryFactory");
     System.setProperty("solr.ulog.numRecordsToKeep", "1000");
     System.setProperty("leaderVoteWait", "60000");

     configureCluster(2)
         .addConfig("conf", configset("cloud-minimal"))
         .configure();

     NODE0 = cluster.getJettySolrRunner(0);
     NODE1 = cluster.getJettySolrRunner(1);

     // Add proxies
     proxies = new HashMap<>(cluster.getJettySolrRunners().size());
     jettys = new HashMap<>();
     for (JettySolrRunner jetty:cluster.getJettySolrRunners()) {
       SocketProxy proxy = new SocketProxy();
       jetty.setProxyPort(proxy.getListenPort());
       cluster.stopJettySolrRunner(jetty);//TODO: Can we avoid this restart
       cluster.startJettySolrRunner(jetty);
       proxy.open(jetty.getBaseUrl().toURI());
       if (log.isInfoEnabled()) {
         log.info("Adding proxy for URL: {}. Proxy: {}", jetty.getBaseUrl(), proxy.getUrl());
       }
       proxies.put(jetty, proxy);
       jettys.put(proxy.getUrl(), jetty);
     }
   }

   @After
   public void tearDownCluster() throws Exception {
     TestInjection.reset();

     if (null != proxies) {
       for (SocketProxy proxy : proxies.values()) {
         proxy.close();
       }
       proxies = null;
     }
     jettys = null;
     System.clearProperty("solr.directoryFactory");
     System.clearProperty("solr.ulog.numRecordsToKeep");
     System.clearProperty("leaderVoteWait");

     shutdownCluster();
   }

   public void testManyDocsInTlogReplayWhileReplicaIsTryingToRecover() throws Exception {
     // TODO: One the basic problem in SOLR-13486 is fixed, this test can be made more robust by:
     // 1) randomizing the number of committedDocs (pre net split) & uncommittedDocs (post net split)
     //    to trigger diff recovery strategies & shutdown behavior
     // 2) replace "committedDocs + uncommittedDocs" with 4 variables:
     //    a: docs committed before network split (add + commit)
     //    b: docs not committed before network split (add w/o commit)
     //    c: docs committed after network split (add + commit)
     //    d: docs not committed after network split (add w/o commit)
     final int committedDocs = 3;
     final int uncommittedDocs = 50;

     log.info("Create Collection...");
     assertEquals(RequestStatusState.COMPLETED,
                  CollectionAdminRequest.createCollection(COLLECTION, 1, 2)
                  .setCreateNodeSet("")
                  .processAndWait(cluster.getSolrClient(), DEFAULT_TIMEOUT));
     assertEquals(RequestStatusState.COMPLETED,
                  CollectionAdminRequest.addReplicaToShard(COLLECTION, "shard1")
                  .setNode(NODE0.getNodeName())
                  .processAndWait(cluster.getSolrClient(), DEFAULT_TIMEOUT));

     waitForState("Timeout waiting for shard leader", COLLECTION, clusterShape(1, 1));

     assertEquals(RequestStatusState.COMPLETED,
                  CollectionAdminRequest.addReplicaToShard(COLLECTION, "shard1")
                  .setNode(NODE1.getNodeName())
                  .processAndWait(cluster.getSolrClient(), DEFAULT_TIMEOUT));

     cluster.waitForActiveCollection(COLLECTION, 1, 2);

     waitForState("Timeout waiting for 1x2 collection", COLLECTION, clusterShape(1, 2));

     final Replica leader = getCollectionState(COLLECTION).getSlice("shard1").getLeader();
     assertEquals("Sanity check failed", NODE0.getNodeName(), leader.getNodeName());

     log.info("Add and commit {} docs...", committedDocs);
     addDocs(true, committedDocs, 1);
     assertDocsExistInBothReplicas(1, committedDocs);

     log.info("Partition nodes...");
     proxies.get(NODE0).close();
     proxies.get(NODE1).close();

     log.info("Adding {} (uncommitted) docs during network partition....", uncommittedDocs);
     addDocs(false, uncommittedDocs, committedDocs + 1);

     log.info("Stopping leader node...");
     assertEquals("Something broke our expected skipIndexWriterCommitOnClose",
                  TEST_VALUE_FOR_SKIP_COMMIT_ON_CLOSE, TestInjection.skipIndexWriterCommitOnClose);
     NODE0.stop();
     cluster.waitForJettyToStop(NODE0);

     log.info("Un-Partition replica (NODE1)...");
     proxies.get(NODE1).reopen();

     waitForState("Timeout waiting for leader goes DOWN", COLLECTION, (liveNodes, collectionState)
                  -> collectionState.getReplica(leader.getName()).getState() == Replica.State.DOWN);

     // Sanity check that a new (out of sync) replica doesn't come up in our place...
     expectThrows(TimeoutException.class,
                  "Did not time out waiting for new leader, out of sync replica became leader",
                  () -> {
                    cluster.getSolrClient().waitForState(COLLECTION, 10, TimeUnit.SECONDS, (state) -> {
             Replica newLeader = state.getSlice("shard1").getLeader();
             if (newLeader != null && !newLeader.getName().equals(leader.getName()) && newLeader.getState() == Replica.State.ACTIVE) {
               // this is is the bad case, our "bad" state was found before timeout
               log.error("WTF: New Leader={}", newLeader);
               return true;
             }
             return false; // still no bad state, wait for timeout
           });
       });

     log.info("Enabling TestInjection.updateLogReplayRandomPause");
     TestInjection.updateLogReplayRandomPause = "true:100";

     log.info("Un-Partition & restart leader (NODE0)...");
     proxies.get(NODE0).reopen();
     NODE0.start();

     log.info("Waiting for all nodes and active collection...");

     cluster.waitForAllNodes(30);;
     waitForState("Timeout waiting for leader", COLLECTION, (liveNodes, collectionState) -> {
       Replica newLeader = collectionState.getLeader("shard1");
       return newLeader != null && newLeader.getName().equals(leader.getName());
     });
     waitForState("Timeout waiting for active collection", COLLECTION, clusterShape(1, 2));

     cluster.waitForActiveCollection(COLLECTION, 1, 2);

     log.info("Check docs on both replicas...");
     assertDocsExistInBothReplicas(1, committedDocs + uncommittedDocs);

     log.info("Test ok, delete collection...");
     CollectionAdminRequest.deleteCollection(COLLECTION).process(cluster.getSolrClient());
   }

   /**
    * Adds the specified number of docs directly to the leader,
    * using increasing docIds begining with startId.  Commits if and only if the boolean is true.
    */
   private void addDocs(final boolean commit, final int numDocs, final int startId) throws SolrServerException, IOException {

     List<SolrInputDocument> docs = new ArrayList<>(numDocs);
     for (int i = 0; i < numDocs; i++) {
       int id = startId + i;
       docs.add(new SolrInputDocument("id", String.valueOf(id), "fieldName_s", String.valueOf(id)));
     }
     // For simplicity, we always add out docs directly to NODE0
     // (where the leader should be) and bypass the proxy...
     try (HttpSolrClient client = getHttpSolrClient(NODE0.getBaseUrl().toString())) {
       assertEquals(0, client.add(COLLECTION, docs).getStatus());
       if (commit) {
         assertEquals(0, client.commit(COLLECTION).getStatus());
       }
     }
   }

   /**
    * uses distrib=false RTG requests to verify that every doc between firstDocId and lastDocId
    * (inclusive) can be found on both the leader and the replica
    */
   private void assertDocsExistInBothReplicas(int firstDocId,
                                              int lastDocId) throws Exception {
     try (HttpSolrClient leaderSolr = getHttpSolrClient(NODE0.getBaseUrl().toString());
          HttpSolrClient replicaSolr = getHttpSolrClient(NODE1.getBaseUrl().toString())) {
       for (int d = firstDocId; d <= lastDocId; d++) {
         String docId = String.valueOf(d);
         assertDocExists("leader", leaderSolr, docId);
         assertDocExists("replica", replicaSolr, docId);
       }
     }
   }

   /**
    * uses distrib=false RTG requests to verify that the specified docId can be found using the
    * specified solr client
    */
   private void assertDocExists(final String clientName, final HttpSolrClient client, final String docId) throws Exception {
     final QueryResponse rsp = (new QueryRequest(params("qt", "/get",
                                                        "id", docId,
                                                        "_trace", clientName,
                                                        "distrib", "false")))
       .process(client, COLLECTION);
     assertEquals(0, rsp.getStatus());

     String match = JSONTestUtil.matchObj("/id", rsp.getResponse().get("doc"), docId);
     assertTrue("Doc with id=" + docId + " not found in " + clientName
                + " due to: " + match + "; rsp="+rsp, match == null);
   }

 }
	/*
	* 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.net.URI;
	import java.util.ArrayList;
	import java.util.HashMap;
	import java.util.List;
	import java.util.Map;
	import java.util.concurrent.TimeUnit;
	import java.util.concurrent.TimeoutException;

	import org.apache.lucene.util.LuceneTestCase.AwaitsFix;

	import org.apache.solr.JSONTestUtil;
	import org.apache.solr.client.solrj.SolrServerException;
	import org.apache.solr.client.solrj.cloud.SocketProxy;
	import org.apache.solr.client.solrj.embedded.JettySolrRunner;
	import org.apache.solr.client.solrj.impl.HttpSolrClient;
	import org.apache.solr.client.solrj.request.CollectionAdminRequest;
	import org.apache.solr.client.solrj.request.QueryRequest;
	import org.apache.solr.client.solrj.response.QueryResponse;
	import org.apache.solr.client.solrj.response.RequestStatusState;
	import org.apache.solr.common.SolrInputDocument;
	import org.apache.solr.common.cloud.Replica;
	import org.apache.solr.util.TestInjection;
	import org.junit.After;
	import org.junit.Before;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	@AwaitsFix(bugUrl="https://issues.apache.org/jira/browse/SOLR-13486;https://issues.apache.org/jira/browse/SOLR-14183")
	public class TestTlogReplayVsRecovery extends SolrCloudTestCase {

	private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
	private static final String COLLECTION = "collecion_with_slow_tlog_recovery";

	private JettySolrRunner NODE0;
	private JettySolrRunner NODE1;
	private Map<JettySolrRunner, SocketProxy> proxies;
	private Map<URI, JettySolrRunner> jettys;

	// we want to ensure there is tlog replay on the leader after we restart it,
	// so in addition to not committing the docs we add during network partition
	// we also want to ensure that our leader doesn't do a "Commit on close"
	//
	// TODO: once SOLR-13486 is fixed, we should randomize this...
	private static final boolean TEST_VALUE_FOR_SKIP_COMMIT_ON_CLOSE = true;

	@Before
	public void setupCluster() throws Exception {
	TestInjection.skipIndexWriterCommitOnClose = TEST_VALUE_FOR_SKIP_COMMIT_ON_CLOSE;

	System.setProperty("solr.directoryFactory", "solr.StandardDirectoryFactory");
	System.setProperty("solr.ulog.numRecordsToKeep", "1000");
	System.setProperty("leaderVoteWait", "60000");

	configureCluster(2)
	.addConfig("conf", configset("cloud-minimal"))
	.configure();

	NODE0 = cluster.getJettySolrRunner(0);
	NODE1 = cluster.getJettySolrRunner(1);

	// Add proxies
	proxies = new HashMap<>(cluster.getJettySolrRunners().size());
	jettys = new HashMap<>();
	for (JettySolrRunner jetty:cluster.getJettySolrRunners()) {
	SocketProxy proxy = new SocketProxy();
	jetty.setProxyPort(proxy.getListenPort());
	cluster.stopJettySolrRunner(jetty);//TODO: Can we avoid this restart
	cluster.startJettySolrRunner(jetty);
	proxy.open(jetty.getBaseUrl().toURI());
	if (log.isInfoEnabled()) {
	log.info("Adding proxy for URL: {}. Proxy: {}", jetty.getBaseUrl(), proxy.getUrl());
	}
	proxies.put(jetty, proxy);
	jettys.put(proxy.getUrl(), jetty);
	}
	}

	@After
	public void tearDownCluster() throws Exception {
	TestInjection.reset();

	if (null != proxies) {
	for (SocketProxy proxy : proxies.values()) {
	proxy.close();
	}
	proxies = null;
	}
	jettys = null;
	System.clearProperty("solr.directoryFactory");
	System.clearProperty("solr.ulog.numRecordsToKeep");
	System.clearProperty("leaderVoteWait");

	shutdownCluster();
	}

	public void testManyDocsInTlogReplayWhileReplicaIsTryingToRecover() throws Exception {
	// TODO: One the basic problem in SOLR-13486 is fixed, this test can be made more robust by:
	// 1) randomizing the number of committedDocs (pre net split) & uncommittedDocs (post net split)
	// to trigger diff recovery strategies & shutdown behavior
	// 2) replace "committedDocs + uncommittedDocs" with 4 variables:
	// a: docs committed before network split (add + commit)
	// b: docs not committed before network split (add w/o commit)
	// c: docs committed after network split (add + commit)
	// d: docs not committed after network split (add w/o commit)
	final int committedDocs = 3;
	final int uncommittedDocs = 50;

	log.info("Create Collection...");
	assertEquals(RequestStatusState.COMPLETED,
	CollectionAdminRequest.createCollection(COLLECTION, 1, 2)
	.setCreateNodeSet("")
	.processAndWait(cluster.getSolrClient(), DEFAULT_TIMEOUT));
	assertEquals(RequestStatusState.COMPLETED,
	CollectionAdminRequest.addReplicaToShard(COLLECTION, "shard1")
	.setNode(NODE0.getNodeName())
	.processAndWait(cluster.getSolrClient(), DEFAULT_TIMEOUT));

	waitForState("Timeout waiting for shard leader", COLLECTION, clusterShape(1, 1));

	assertEquals(RequestStatusState.COMPLETED,
	CollectionAdminRequest.addReplicaToShard(COLLECTION, "shard1")
	.setNode(NODE1.getNodeName())
	.processAndWait(cluster.getSolrClient(), DEFAULT_TIMEOUT));

	cluster.waitForActiveCollection(COLLECTION, 1, 2);

	waitForState("Timeout waiting for 1x2 collection", COLLECTION, clusterShape(1, 2));

	final Replica leader = getCollectionState(COLLECTION).getSlice("shard1").getLeader();
	assertEquals("Sanity check failed", NODE0.getNodeName(), leader.getNodeName());

	log.info("Add and commit {} docs...", committedDocs);
	addDocs(true, committedDocs, 1);
	assertDocsExistInBothReplicas(1, committedDocs);

	log.info("Partition nodes...");
	proxies.get(NODE0).close();
	proxies.get(NODE1).close();

	log.info("Adding {} (uncommitted) docs during network partition....", uncommittedDocs);
	addDocs(false, uncommittedDocs, committedDocs + 1);

	log.info("Stopping leader node...");
	assertEquals("Something broke our expected skipIndexWriterCommitOnClose",
	TEST_VALUE_FOR_SKIP_COMMIT_ON_CLOSE, TestInjection.skipIndexWriterCommitOnClose);
	NODE0.stop();
	cluster.waitForJettyToStop(NODE0);

	log.info("Un-Partition replica (NODE1)...");
	proxies.get(NODE1).reopen();

	waitForState("Timeout waiting for leader goes DOWN", COLLECTION, (liveNodes, collectionState)
	-> collectionState.getReplica(leader.getName()).getState() == Replica.State.DOWN);

	// Sanity check that a new (out of sync) replica doesn't come up in our place...
	expectThrows(TimeoutException.class,
	"Did not time out waiting for new leader, out of sync replica became leader",
	() -> {
	cluster.getSolrClient().waitForState(COLLECTION, 10, TimeUnit.SECONDS, (state) -> {
	Replica newLeader = state.getSlice("shard1").getLeader();
	if (newLeader != null && !newLeader.getName().equals(leader.getName()) && newLeader.getState() == Replica.State.ACTIVE) {
	// this is is the bad case, our "bad" state was found before timeout
	log.error("WTF: New Leader={}", newLeader);
	return true;
	}
	return false; // still no bad state, wait for timeout
	});
	});

	log.info("Enabling TestInjection.updateLogReplayRandomPause");
	TestInjection.updateLogReplayRandomPause = "true:100";

	log.info("Un-Partition & restart leader (NODE0)...");
	proxies.get(NODE0).reopen();
	NODE0.start();

	log.info("Waiting for all nodes and active collection...");

	cluster.waitForAllNodes(30);;
	waitForState("Timeout waiting for leader", COLLECTION, (liveNodes, collectionState) -> {
	Replica newLeader = collectionState.getLeader("shard1");
	return newLeader != null && newLeader.getName().equals(leader.getName());
	});
	waitForState("Timeout waiting for active collection", COLLECTION, clusterShape(1, 2));

	cluster.waitForActiveCollection(COLLECTION, 1, 2);

	log.info("Check docs on both replicas...");
	assertDocsExistInBothReplicas(1, committedDocs + uncommittedDocs);

	log.info("Test ok, delete collection...");
	CollectionAdminRequest.deleteCollection(COLLECTION).process(cluster.getSolrClient());
	}

	/**
	* Adds the specified number of docs directly to the leader,
	* using increasing docIds begining with startId. Commits if and only if the boolean is true.
	*/
	private void addDocs(final boolean commit, final int numDocs, final int startId) throws SolrServerException, IOException {

	List<SolrInputDocument> docs = new ArrayList<>(numDocs);
	for (int i = 0; i < numDocs; i++) {
	int id = startId + i;
	docs.add(new SolrInputDocument("id", String.valueOf(id), "fieldName_s", String.valueOf(id)));
	}
	// For simplicity, we always add out docs directly to NODE0
	// (where the leader should be) and bypass the proxy...
	try (HttpSolrClient client = getHttpSolrClient(NODE0.getBaseUrl().toString())) {
	assertEquals(0, client.add(COLLECTION, docs).getStatus());
	if (commit) {
	assertEquals(0, client.commit(COLLECTION).getStatus());
	}
	}
	}

	/**
	* uses distrib=false RTG requests to verify that every doc between firstDocId and lastDocId
	* (inclusive) can be found on both the leader and the replica
	*/
	private void assertDocsExistInBothReplicas(int firstDocId,
	int lastDocId) throws Exception {
	try (HttpSolrClient leaderSolr = getHttpSolrClient(NODE0.getBaseUrl().toString());
	HttpSolrClient replicaSolr = getHttpSolrClient(NODE1.getBaseUrl().toString())) {
	for (int d = firstDocId; d <= lastDocId; d++) {
	String docId = String.valueOf(d);
	assertDocExists("leader", leaderSolr, docId);
	assertDocExists("replica", replicaSolr, docId);
	}
	}
	}

	/**
	* uses distrib=false RTG requests to verify that the specified docId can be found using the
	* specified solr client
	*/
	private void assertDocExists(final String clientName, final HttpSolrClient client, final String docId) throws Exception {
	final QueryResponse rsp = (new QueryRequest(params("qt", "/get",
	"id", docId,
	"_trace", clientName,
	"distrib", "false")))
	.process(client, COLLECTION);
	assertEquals(0, rsp.getStatus());

	String match = JSONTestUtil.matchObj("/id", rsp.getResponse().get("doc"), docId);
	assertTrue("Doc with id=" + docId + " not found in " + clientName
	+ " due to: " + match + "; rsp="+rsp, match == null);
	}

	}