SOLR-5625: Add testing around SolrCmdDistributor

git-svn-id: https://svn.apache.org/repos/asf/lucene/dev/trunk@1561077 13f79535-47bb-0310-9956-ffa450edef68

solr/core/src/test/org/apache/solr/cloud/FullSolrCloudDistribCmdsTest.java

@@ -18,6 +18,8 @@ package org.apache.solr.cloud;
  */
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
 
 import org.apache.lucene.util.LuceneTestCase.Slow;
 import org.apache.solr.client.solrj.SolrQuery;
@@ -25,6 +27,7 @@ import org.apache.solr.client.solrj.SolrServer;
 import org.apache.solr.client.solrj.SolrServerException;
 import org.apache.solr.client.solrj.impl.ConcurrentUpdateSolrServer;
 import org.apache.solr.client.solrj.impl.HttpSolrServer;
+import org.apache.solr.client.solrj.request.QueryRequest;
 import org.apache.solr.client.solrj.request.UpdateRequest;
 import org.apache.solr.client.solrj.response.QueryResponse;
 import org.apache.solr.common.SolrDocument;
@@ -34,11 +37,11 @@ import org.apache.solr.common.cloud.SolrZkClient;
 import org.apache.solr.common.cloud.ZkNodeProps;
 import org.apache.solr.common.cloud.ZkStateReader;
 import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.params.CollectionParams.CollectionAction;
 import org.apache.solr.update.VersionInfo;
 import org.apache.solr.update.processor.DistributedUpdateProcessor;
 import org.apache.zookeeper.CreateMode;
 import org.junit.BeforeClass;
-import org.junit.Ignore;
 
 /**
  * Super basic testing, no shard restarting or anything.
@@ -49,12 +52,14 @@ public class FullSolrCloudDistribCmdsTest extends AbstractFullDistribZkTestBase
   
   @BeforeClass
   public static void beforeSuperClass() {
+    schemaString = "schema15.xml";      // we need a string id
   }
   
   public FullSolrCloudDistribCmdsTest() {
     super();
-    shardCount = 4;
+    fixShardCount = true;
-    sliceCount = 2;
+    shardCount = 6;
+    sliceCount = 3;
   }
   
   @Override
@@ -124,21 +129,28 @@ public class FullSolrCloudDistribCmdsTest extends AbstractFullDistribZkTestBase
     
     docId = testIndexQueryDeleteHierarchical(docId);
     
-    testIndexingWithSuss();
+    docId = testIndexingDocPerRequestWithHttpSolrServer(docId);
+    
+    testIndexingWithSuss(docId);
     
     // TODO: testOptimisticUpdate(results);
     
     testDeleteByQueryDistrib();
     
-    testThatCantForwardToLeaderFails();
+    docId = testThatCantForwardToLeaderFails(docId);
+    
+    
+    docId = testIndexingBatchPerRequestWithHttpSolrServer(docId);
   }
 
-  private void testThatCantForwardToLeaderFails() throws Exception {
+  private long testThatCantForwardToLeaderFails(long docId) throws Exception {
     ZkStateReader zkStateReader = cloudClient.getZkStateReader();
     ZkNodeProps props = zkStateReader.getLeaderRetry(DEFAULT_COLLECTION, "shard1");
     
     chaosMonkey.stopShard("shard1");
-
+    
+    Thread.sleep(1000);
+    
     // fake that the leader is still advertised
     String leaderPath = ZkStateReader.getShardLeadersPath(DEFAULT_COLLECTION, "shard1");
     SolrZkClient zkClient = new SolrZkClient(zkServer.getZkAddress(), 10000);
@@ -146,9 +158,9 @@ public class FullSolrCloudDistribCmdsTest extends AbstractFullDistribZkTestBase
     try {
       zkClient.makePath(leaderPath, ZkStateReader.toJSON(props),
           CreateMode.EPHEMERAL, true);
-      for (int i = 200; i < 210; i++) {
+      for (int i = 0; i < 200; i++) {
         try {
-          index_specific(cloudClient, id, i);
+          index_specific(shardToJetty.get("shard2").get(0).client.solrClient, id, docId++);
         } catch (SolrException e) {
           // expected
           fails++;
@@ -162,8 +174,9 @@ public class FullSolrCloudDistribCmdsTest extends AbstractFullDistribZkTestBase
     } finally {
       zkClient.close();
     }
-    
+
     assertTrue("A whole shard is down - some of these should fail", fails > 0);
+    return docId;
   }
 
   private long addTwoDocsInOneRequest(long docId) throws
@@ -171,14 +184,8 @@ public class FullSolrCloudDistribCmdsTest extends AbstractFullDistribZkTestBase
     QueryResponse results;
     UpdateRequest uReq;
     uReq = new UpdateRequest();
-    //uReq.setParam(UpdateParams.UPDATE_CHAIN, DISTRIB_UPDATE_CHAIN);
+    docId = addDoc(docId, uReq);
-    SolrInputDocument doc1 = new SolrInputDocument();
+    docId = addDoc(docId, uReq);
-
-    addFields(doc1, "id", docId++);
-    uReq.add(doc1);
-    SolrInputDocument doc2 = new SolrInputDocument();
-    addFields(doc2, "id", docId++);
-    uReq.add(doc2);
     
     uReq.process(cloudClient);
     uReq.process(controlClient);
@@ -307,23 +314,146 @@ public class FullSolrCloudDistribCmdsTest extends AbstractFullDistribZkTestBase
     return docId;
   }
   
-  private void testIndexingWithSuss() throws Exception {
+  
+  private long testIndexingDocPerRequestWithHttpSolrServer(long docId) throws Exception {
+    int docs = random().nextInt(TEST_NIGHTLY ? 4013 : 97) + 1;
+    for (int i = 0; i < docs; i++) {
+      UpdateRequest uReq;
+      uReq = new UpdateRequest();
+      docId = addDoc(docId, uReq);
+      
+      uReq.process(cloudClient);
+      uReq.process(controlClient);
+      
+    }
+    commit();
+    
+    checkShardConsistency();
+    assertDocCounts(VERBOSE);
+    
+    return docId++;
+  }
+  
+  private long testIndexingBatchPerRequestWithHttpSolrServer(long docId) throws Exception {
+    
+    // remove collection
+    ModifiableSolrParams params = new ModifiableSolrParams();
+    params.set("action", CollectionAction.DELETE.toString());
+    params.set("name", "collection1");
+    QueryRequest request = new QueryRequest(params);
+    request.setPath("/admin/collections");
+    
+  
+    cloudClient.request(request);
+    
+    controlClient.deleteByQuery("*:*");
+    controlClient.commit();
+    
+    // somtimes we use an oversharded collection
+    createCollection(null, "collection2", 36, 3, 100000, cloudClient, null, "conf1");
+    cloudClient.setDefaultCollection("collection2");
+    waitForRecoveriesToFinish("collection2", false);
+    
+    class IndexThread extends Thread {
+      Integer name;
+      
+      public IndexThread(Integer name) {
+        this.name = name;
+      }
+      
+      @Override
+      public void run() {
+        int rnds = random().nextInt(TEST_NIGHTLY ? 25 : 3) + 1;
+        for (int i = 0; i < rnds; i++) {
+          UpdateRequest uReq;
+          uReq = new UpdateRequest();
+          int cnt = random().nextInt(TEST_NIGHTLY ? 3313 : 350) + 1;
+          for (int j = 0; j <cnt; j++) {
+            addDoc("thread" + name + "_" + i + "_" + j, uReq);
+          }
+          
+          try {
+            uReq.process(cloudClient);
+            uReq.process(controlClient);
+          } catch (SolrServerException e) {
+            throw new RuntimeException(e);
+          } catch (IOException e) {
+            throw new RuntimeException(e);
+          }
+
+          
+        }
+      }
+    };
+    List<Thread> threads = new ArrayList<Thread>();
+
+    int nthreads = random().nextInt(TEST_NIGHTLY ? 4 : 2) + 1;
+    for (int i = 0; i < nthreads; i++) {
+      IndexThread thread = new IndexThread(i);
+      threads.add(thread);
+      thread.start();
+    }
+    
+    for (Thread thread : threads) {
+      thread.join();
+    }
+    
+    commit();
+    
+    waitForRecoveriesToFinish("collection2", false);
+    
+    printLayout();
+    
+    SolrQuery query = new SolrQuery("*:*");
+    long controlCount = controlClient.query(query).getResults()
+        .getNumFound();
+    long cloudCount = cloudClient.query(query).getResults().getNumFound();
+
+    
+    compareResults(controlCount, cloudCount);
+    
+    assertEquals("Control does not match cloud", controlCount, cloudCount);
+    System.out.println("DOCS:" + controlCount);
+
+    return docId;
+  }
+
+  private long addDoc(long docId, UpdateRequest uReq) {
+    addDoc(Long.toString(docId++), uReq);
+    return docId;
+  }
+  
+  private long addDoc(String docId, UpdateRequest uReq) {
+    SolrInputDocument doc1 = new SolrInputDocument();
+    
+    uReq.add(doc1);
+    addFields(doc1, "id", docId, "text_t", "some text so that it not's negligent work to parse this doc, even though it's still a pretty short doc");
+    return -1;
+  }
+  
+  private long testIndexingWithSuss(long docId) throws Exception {
     ConcurrentUpdateSolrServer suss = new ConcurrentUpdateSolrServer(
-        ((HttpSolrServer) clients.get(0)).getBaseURL(), 3, 1);
+        ((HttpSolrServer) clients.get(0)).getBaseURL(), 10, 2);
+    QueryResponse results = query(cloudClient);
+    long beforeCount = results.getResults().getNumFound();
+    int cnt = TEST_NIGHTLY ? 2933 : 313;
     try {
       suss.setConnectionTimeout(15000);
-      suss.setSoTimeout(30000);
+      for (int i = 0; i < cnt; i++) {
-      for (int i = 100; i < 150; i++) {
+        index_specific(suss, id, docId++, "text_t", "some text so that it not's negligent work to parse this doc, even though it's still a pretty short doc");
-        index_specific(suss, id, i);
       }
       suss.blockUntilFinished();
       
       commit();
-      
+
       checkShardConsistency();
+      assertDocCounts(VERBOSE);
     } finally {
       suss.shutdown();
     }
+    results = query(cloudClient);
+    assertEquals(beforeCount + cnt, results.getResults().getNumFound());
+    return docId;
   }
   
   private void testOptimisticUpdate(QueryResponse results) throws Exception {
@@ -375,5 +505,9 @@ public class FullSolrCloudDistribCmdsTest extends AbstractFullDistribZkTestBase
   public void tearDown() throws Exception {
     super.tearDown();
   }
+  
+  protected SolrInputDocument addRandFields(SolrInputDocument sdoc) {
+    return sdoc;
+  }
 
 }