lucene-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From markrmil...@apache.org
Subject svn commit: r1401798 - in /lucene/dev/trunk/solr: ./ core/src/java/org/apache/solr/client/solrj/embedded/ core/src/java/org/apache/solr/cloud/ core/src/java/org/apache/solr/handler/component/ core/src/java/org/apache/solr/update/ core/src/test/org/apac...
Date Wed, 24 Oct 2012 17:52:16 GMT
Author: markrmiller
Date: Wed Oct 24 17:52:15 2012
New Revision: 1401798

URL: http://svn.apache.org/viewvc?rev=1401798&view=rev
Log:
SOLR-3933: Distributed commits are not guaranteed to be ordered within a request.

SOLR-3939: An empty or just replicated index cannot become the leader of a shard after a leader
goes down.
  
SOLR-3971: A collection that is created with numShards=1 turns into a numShards=2 collection
after starting up a second core and not specifying numShards.

SOLR-3932: SolrCmdDistributorTest either takes 3 seconds or 3 minutes.

Modified:
    lucene/dev/trunk/solr/CHANGES.txt
    lucene/dev/trunk/solr/core/src/java/org/apache/solr/client/solrj/embedded/JettySolrRunner.java
    lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/ElectionContext.java
    lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/Overseer.java
    lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java
    lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/SyncStrategy.java
    lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/ZkController.java
    lucene/dev/trunk/solr/core/src/java/org/apache/solr/handler/component/RealTimeGetComponent.java
    lucene/dev/trunk/solr/core/src/java/org/apache/solr/update/PeerSync.java
    lucene/dev/trunk/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java
    lucene/dev/trunk/solr/core/src/test/org/apache/solr/cloud/BasicDistributedZk2Test.java
    lucene/dev/trunk/solr/core/src/test/org/apache/solr/cloud/BasicDistributedZkTest.java
    lucene/dev/trunk/solr/core/src/test/org/apache/solr/update/SolrCmdDistributorTest.java
    lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java

Modified: lucene/dev/trunk/solr/CHANGES.txt
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/CHANGES.txt?rev=1401798&r1=1401797&r2=1401798&view=diff
==============================================================================
--- lucene/dev/trunk/solr/CHANGES.txt (original)
+++ lucene/dev/trunk/solr/CHANGES.txt Wed Oct 24 17:52:15 2012
@@ -88,6 +88,16 @@ Bug Fixes
 * SOLR-3961: Fixed error using LimitTokenCountFilterFactory 
   (Jack Krupansky, hossman)
 
+* SOLR-3933: Distributed commits are not guaranteed to be ordered within a 
+  request. (Mark Miller)
+
+* SOLR-3939: An empty or just replicated index cannot become the leader of a
+  shard after a leader goes down. (Joel Bernstein, yonik, Mark Miller)
+  
+* SOLR-3971: A collection that is created with numShards=1 turns into a 
+  numShards=2 collection after starting up a second core and not specifying 
+  numShards. (Mark Miller)
+
 Other Changes
 ----------------------
 
@@ -100,6 +110,9 @@ Other Changes
 * SOLR-3966: Eliminate superfluous warning from LanguageIdentifierUpdateProcessor
   (Markus Jelsma via hossman)
 
+* SOLR-3932: SolrCmdDistributorTest either takes 3 seconds or 3 minutes.
+  (yonik, Mark Miller)  
+
 ==================  4.0.0 ==================
 
 Versions of Major Components

Modified: lucene/dev/trunk/solr/core/src/java/org/apache/solr/client/solrj/embedded/JettySolrRunner.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/core/src/java/org/apache/solr/client/solrj/embedded/JettySolrRunner.java?rev=1401798&r1=1401797&r2=1401798&view=diff
==============================================================================
--- lucene/dev/trunk/solr/core/src/java/org/apache/solr/client/solrj/embedded/JettySolrRunner.java
(original)
+++ lucene/dev/trunk/solr/core/src/java/org/apache/solr/client/solrj/embedded/JettySolrRunner.java
Wed Oct 24 17:52:15 2012
@@ -35,6 +35,7 @@ import org.eclipse.jetty.server.Connecto
 import org.eclipse.jetty.server.Server;
 import org.eclipse.jetty.server.bio.SocketConnector;
 import org.eclipse.jetty.server.handler.GzipHandler;
+import org.eclipse.jetty.server.nio.SelectChannelConnector;
 import org.eclipse.jetty.server.session.HashSessionIdManager;
 import org.eclipse.jetty.servlet.FilterHolder;
 import org.eclipse.jetty.servlet.ServletContextHandler;
@@ -92,6 +93,7 @@ public class JettySolrRunner {
   private void init(String solrHome, String context, int port, boolean stopAtShutdown) {
     this.context = context;
     server = new Server(port);
+
     this.solrHome = solrHome;
     this.stopAtShutdown = stopAtShutdown;
     server.setStopAtShutdown(stopAtShutdown);
@@ -100,32 +102,45 @@ public class JettySolrRunner {
     }
     System.setProperty("solr.solr.home", solrHome);
     if (System.getProperty("jetty.testMode") != null) {
-      // SelectChannelConnector connector = new SelectChannelConnector();
-      // Normal SocketConnector is what solr's example server uses by default
-      SocketConnector connector = new SocketConnector();
+      SelectChannelConnector connector = new SelectChannelConnector();
       connector.setPort(port);
       connector.setReuseAddress(true);
-      if (!stopAtShutdown) {
-        QueuedThreadPool threadPool = (QueuedThreadPool) connector
-            .getThreadPool();
-        if (threadPool != null) {
+      connector.setLowResourcesMaxIdleTime(1500);
+      QueuedThreadPool threadPool = (QueuedThreadPool) connector
+          .getThreadPool();
+      if (threadPool != null) {
+        threadPool.setMaxThreads(10000);
+        threadPool.setMaxIdleTimeMs(5000);
+        if (!stopAtShutdown) {
           threadPool.setMaxStopTimeMs(100);
         }
       }
+      
       server.setConnectors(new Connector[] {connector});
       server.setSessionIdManager(new HashSessionIdManager(new Random()));
     } else {
-      if (!stopAtShutdown) {
-        for (Connector connector : server.getConnectors()) {
-          if (connector instanceof SocketConnector) {
-            QueuedThreadPool threadPool = (QueuedThreadPool) ((SocketConnector) connector)
-                .getThreadPool();
-            if (threadPool != null) {
-              threadPool.setMaxStopTimeMs(100);
-            }
+      
+      for (Connector connector : server.getConnectors()) {
+        QueuedThreadPool threadPool = null;
+        if (connector instanceof SocketConnector) {
+          threadPool = (QueuedThreadPool) ((SocketConnector) connector)
+              .getThreadPool();
+        }
+        if (connector instanceof SelectChannelConnector) {
+          threadPool = (QueuedThreadPool) ((SelectChannelConnector) connector)
+              .getThreadPool();
+        }
+        
+        if (threadPool != null) {
+          threadPool.setMaxThreads(10000);
+          threadPool.setMaxIdleTimeMs(5000);
+          if (!stopAtShutdown) {
+            threadPool.setMaxStopTimeMs(100);
           }
         }
+        
       }
+
     }
 
     // Initialize the servlets

Modified: lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/ElectionContext.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/ElectionContext.java?rev=1401798&r1=1401797&r2=1401798&view=diff
==============================================================================
--- lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/ElectionContext.java (original)
+++ lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/ElectionContext.java Wed Oct
24 17:52:15 2012
@@ -14,6 +14,7 @@ import org.apache.solr.common.cloud.ZkNo
 import org.apache.solr.common.cloud.ZkStateReader;
 import org.apache.solr.core.CoreContainer;
 import org.apache.solr.core.SolrCore;
+import org.apache.solr.update.UpdateLog;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.KeeperException;
 import org.slf4j.Logger;
@@ -162,6 +163,10 @@ final class ShardLeaderElectionContext e
       }
       
       log.info("I may be the new leader - try and sync");
+      
+      UpdateLog ulog = core.getUpdateHandler().getUpdateLog();
+ 
+      
       // we are going to attempt to be the leader
       // first cancel any current recovery
       core.getUpdateHandler().getSolrCoreState().cancelRecovery();
@@ -173,6 +178,14 @@ final class ShardLeaderElectionContext e
         success = false;
       }
       
+      if (!success && ulog.getRecentUpdates().getVersions(1).isEmpty()) {
+        // we failed sync, but we have no versions - we can't sync in that case
+        // - we were active
+        // before, so become leader anyway
+        log.info("We failed sync, but we have no versions - we can't sync in that case -
we were active before, so become leader anyway");
+        success = true;
+      }
+      
       // if !success but no one else is in active mode,
       // we are the leader anyway
       // TODO: should we also be leader if there is only one other active?
@@ -308,13 +321,7 @@ final class ShardLeaderElectionContext e
       return;
     }
     
-    log.info("There is a better leader candidate than us - going back into recovery");
-    
-    try {
-      zkController.publish(core.getCoreDescriptor(), ZkStateReader.DOWN);
-    } catch (Throwable t) {
-      SolrException.log(log, "Error trying to publish down state", t);
-    }
+    log.info("There may be a better leader candidate than us - going back into recovery");
     
     cancelElection();
     

Modified: lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/Overseer.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/Overseer.java?rev=1401798&r1=1401797&r2=1401798&view=diff
==============================================================================
--- lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/Overseer.java (original)
+++ lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/Overseer.java Wed Oct 24 17:52:15
2012
@@ -17,14 +17,11 @@ package org.apache.solr.cloud;
  * the License.
  */
 
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Map.Entry;
 
-import org.apache.noggit.JSONUtil;
 import org.apache.solr.common.SolrException;
 import org.apache.solr.common.cloud.ClusterState;
 import org.apache.solr.common.cloud.ClosableThread;
@@ -210,11 +207,11 @@ public class Overseer {
       private ClusterState updateState(ClusterState state, final ZkNodeProps message) {
         final String collection = message.getStr(ZkStateReader.COLLECTION_PROP);
         final String zkCoreNodeName = message.getStr(ZkStateReader.NODE_NAME_PROP) + "_"
+ message.getStr(ZkStateReader.CORE_NAME_PROP);
-        final Integer numShards = message.getStr(ZkStateReader.NUM_SHARDS_PROP)!=null?Integer.parseInt(message.getStr(ZkStateReader.NUM_SHARDS_PROP)):null;
-        
+        Integer numShards = message.getStr(ZkStateReader.NUM_SHARDS_PROP)!=null?Integer.parseInt(message.getStr(ZkStateReader.NUM_SHARDS_PROP)):null;
+        log.info("Update state numShards={} message={}", numShards, message);
         //collection does not yet exist, create placeholders if num shards is specified
-        if (!state.getCollections().contains(collection)
-            && numShards!=null) {
+        boolean collectionExists = state.getCollections().contains(collection);
+        if (!collectionExists && numShards!=null) {
           state = createCollection(state, collection, numShards);
         }
         
@@ -227,6 +224,10 @@ public class Overseer {
         }
         if(sliceName == null) {
           //request new shardId 
+          if (collectionExists) {
+            // use existing numShards
+            numShards = state.getCollectionStates().get(collection).size();
+          }
           sliceName = AssignShard.assignShard(collection, state, numShards);
         }
 
@@ -269,6 +270,8 @@ public class Overseer {
       }
 
       private ClusterState createCollection(ClusterState state, String collectionName, int
numShards) {
+        log.info("Create collection {} with numShards {}", collectionName, numShards);
+        
         HashPartitioner hp = new HashPartitioner();
         List<HashPartitioner.Range> ranges = hp.partitionRange(numShards, hp.fullRange());
 

Modified: lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java?rev=1401798&r1=1401797&r2=1401798&view=diff
==============================================================================
--- lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java (original)
+++ lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java Wed Oct
24 17:52:15 2012
@@ -348,7 +348,7 @@ public class RecoveryStrategy extends Th
           // System.out.println("Attempting to PeerSync from " + leaderUrl
           // + " i am:" + zkController.getNodeName());
           PeerSync peerSync = new PeerSync(core,
-              Collections.singletonList(leaderUrl), ulog.numRecordsToKeep);
+              Collections.singletonList(leaderUrl), ulog.numRecordsToKeep, false, false);
           peerSync.setStartingVersions(recentVersions);
           boolean syncSuccess = peerSync.sync();
           if (syncSuccess) {
@@ -443,7 +443,7 @@ public class RecoveryStrategy extends Th
         // Or do a fall off retry...
         try {
 
-          log.error("Recovery failed - trying again... core=" + coreName);
+          log.error("Recovery failed - trying again... (" + retries + ") core=" + coreName);
           
           if (isClosed()) {
             retries = INTERRUPTED;
@@ -451,7 +451,7 @@ public class RecoveryStrategy extends Th
           
           retries++;
           if (retries >= MAX_RETRIES) {
-            if (retries == INTERRUPTED) {
+            if (retries >= INTERRUPTED) {
               SolrException.log(log, "Recovery failed - interrupted. core="
                   + coreName);
               try {
@@ -463,7 +463,7 @@ public class RecoveryStrategy extends Th
               }
             } else {
               SolrException.log(log,
-                  "Recovery failed - max retries exceeded. core=" + coreName);
+                  "Recovery failed - max retries exceeded (" + retries + "). core=" + coreName);
               try {
                 recoveryFailed(core, zkController, baseUrl, coreZkNodeName,
                     core.getCoreDescriptor());
@@ -482,6 +482,7 @@ public class RecoveryStrategy extends Th
         try {
           // start at 1 sec and work up to a couple min
           double loopCount = Math.min(Math.pow(2, retries), 600); 
+          log.info("Wait {} seconds before trying to recover again ({})", loopCount, retries);
           for (int i = 0; i < loopCount; i++) {
             if (isClosed()) break; // check if someone closed us
             Thread.sleep(STARTING_RECOVERY_DELAY);

Modified: lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/SyncStrategy.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/SyncStrategy.java?rev=1401798&r1=1401797&r2=1401798&view=diff
==============================================================================
--- lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/SyncStrategy.java (original)
+++ lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/SyncStrategy.java Wed Oct 24
17:52:15 2012
@@ -176,7 +176,7 @@ public class SyncStrategy {
     // if we can't reach a replica for sync, we still consider the overall sync a success
     // TODO: as an assurance, we should still try and tell the sync nodes that we couldn't
reach
     // to recover once more?
-    PeerSync peerSync = new PeerSync(core, syncWith, core.getUpdateHandler().getUpdateLog().numRecordsToKeep,
true);
+    PeerSync peerSync = new PeerSync(core, syncWith, core.getUpdateHandler().getUpdateLog().numRecordsToKeep,
true, true);
     return peerSync.sync();
   }
   

Modified: lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/ZkController.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/ZkController.java?rev=1401798&r1=1401797&r2=1401798&view=diff
==============================================================================
--- lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/ZkController.java (original)
+++ lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/ZkController.java Wed Oct 24
17:52:15 2012
@@ -781,6 +781,7 @@ public final class ZkController {
     //System.out.println(Thread.currentThread().getStackTrace()[3]);
     Integer numShards = cd.getCloudDescriptor().getNumShards();
     if (numShards == null) { //XXX sys prop hack
+      log.info("numShards not found on descriptor - reading it from system property");
       numShards = Integer.getInteger(ZkStateReader.NUM_SHARDS_PROP);
     }
     

Modified: lucene/dev/trunk/solr/core/src/java/org/apache/solr/handler/component/RealTimeGetComponent.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/core/src/java/org/apache/solr/handler/component/RealTimeGetComponent.java?rev=1401798&r1=1401797&r2=1401798&view=diff
==============================================================================
--- lucene/dev/trunk/solr/core/src/java/org/apache/solr/handler/component/RealTimeGetComponent.java
(original)
+++ lucene/dev/trunk/solr/core/src/java/org/apache/solr/handler/component/RealTimeGetComponent.java
Wed Oct 24 17:52:15 2012
@@ -525,7 +525,7 @@ public class RealTimeGetComponent extend
     
     boolean cantReachIsSuccess = rb.req.getParams().getBool("cantReachIsSuccess", false);
     
-    PeerSync peerSync = new PeerSync(rb.req.getCore(), replicas, nVersions, cantReachIsSuccess);
+    PeerSync peerSync = new PeerSync(rb.req.getCore(), replicas, nVersions, cantReachIsSuccess,
true);
     boolean success = peerSync.sync();
     
     // TODO: more complex response?

Modified: lucene/dev/trunk/solr/core/src/java/org/apache/solr/update/PeerSync.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/core/src/java/org/apache/solr/update/PeerSync.java?rev=1401798&r1=1401797&r2=1401798&view=diff
==============================================================================
--- lucene/dev/trunk/solr/core/src/java/org/apache/solr/update/PeerSync.java (original)
+++ lucene/dev/trunk/solr/core/src/java/org/apache/solr/update/PeerSync.java Wed Oct 24 17:52:15
2012
@@ -19,6 +19,7 @@ package org.apache.solr.update;
 
 import java.io.IOException;
 import java.net.ConnectException;
+import java.net.SocketException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Comparator;
@@ -79,6 +80,7 @@ public class PeerSync  {
   private long ourLowThreshold;  // 20th percentile
   private long ourHighThreshold; // 80th percentile
   private boolean cantReachIsSuccess;
+  private boolean getNoVersionsIsSuccess;
   private static final HttpClient client;
   static {
     ModifiableSolrParams params = new ModifiableSolrParams();
@@ -129,14 +131,15 @@ public class PeerSync  {
   }
 
   public PeerSync(SolrCore core, List<String> replicas, int nUpdates) {
-    this(core, replicas, nUpdates, false);
+    this(core, replicas, nUpdates, false, true);
   }
   
-  public PeerSync(SolrCore core, List<String> replicas, int nUpdates, boolean cantReachIsSuccess)
{
+  public PeerSync(SolrCore core, List<String> replicas, int nUpdates, boolean cantReachIsSuccess,
boolean getNoVersionsIsSuccess) {
     this.replicas = replicas;
     this.nUpdates = nUpdates;
     this.maxUpdates = nUpdates;
     this.cantReachIsSuccess = cantReachIsSuccess;
+    this.getNoVersionsIsSuccess = getNoVersionsIsSuccess;
 
     
     uhandler = core.getUpdateHandler();
@@ -301,7 +304,7 @@ public class PeerSync  {
         Throwable solrException = ((SolrServerException) srsp.getException())
             .getRootCause();
         if (solrException instanceof ConnectException || solrException instanceof ConnectTimeoutException
-            || solrException instanceof NoHttpResponseException) {
+            || solrException instanceof NoHttpResponseException || solrException instanceof
SocketException) {
           log.warn(msg() + " couldn't connect to " + srsp.getShardAddress() + ", counting
as success");
 
           return true;
@@ -343,7 +346,7 @@ public class PeerSync  {
     log.info(msg() + " Received " + otherVersions.size() + " versions from " + sreq.shards[0]
);
 
     if (otherVersions.size() == 0) {
-      return true;
+      return getNoVersionsIsSuccess; 
     }
     
     boolean completeList = otherVersions.size() < nUpdates;  // do we have their complete
list of updates?

Modified: lucene/dev/trunk/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java?rev=1401798&r1=1401797&r2=1401798&view=diff
==============================================================================
--- lucene/dev/trunk/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java (original)
+++ lucene/dev/trunk/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java Wed
Oct 24 17:52:15 2012
@@ -101,7 +101,6 @@ public class SolrCmdDistributor {
   
   public void finish() {
 
-    // piggyback on any outstanding adds or deletes if possible.
     flushAdds(1);
     flushDeletes(1);
 
@@ -150,6 +149,12 @@ public class SolrCmdDistributor {
   
   public void distribCommit(CommitUpdateCommand cmd, List<Node> nodes,
       ModifiableSolrParams params) throws IOException {
+    
+    // make sure we are ordered
+    flushAdds(1);
+    flushDeletes(1);
+
+    
     // Wait for all outstanding responses to make sure that a commit
     // can't sneak in ahead of adds or deletes we already sent.
     // We could do this on a per-server basis, but it's more complex
@@ -163,7 +168,7 @@ public class SolrCmdDistributor {
     
     addCommit(ureq, cmd);
     
-    log.info("Distrib commit to:" + nodes);
+    log.info("Distrib commit to:" + nodes + " params:" + params);
     
     for (Node node : nodes) {
       submit(ureq, node);

Modified: lucene/dev/trunk/solr/core/src/test/org/apache/solr/cloud/BasicDistributedZk2Test.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/core/src/test/org/apache/solr/cloud/BasicDistributedZk2Test.java?rev=1401798&r1=1401797&r2=1401798&view=diff
==============================================================================
--- lucene/dev/trunk/solr/core/src/test/org/apache/solr/cloud/BasicDistributedZk2Test.java
(original)
+++ lucene/dev/trunk/solr/core/src/test/org/apache/solr/cloud/BasicDistributedZk2Test.java
Wed Oct 24 17:52:15 2012
@@ -59,11 +59,11 @@ public class BasicDistributedZk2Test ext
           "now is the time for all good men", "foo_f", 1.414f, "foo_b", "true",
           "foo_d", 1.414d);
       
-      // make sure we are in a steady state...
-      waitForRecoveriesToFinish(false);
-      
       commit();
       
+      // make sure we are in a steady state...
+      waitForRecoveriesToFinish(false);
+
       assertDocCounts(false);
       
       indexAbunchOfDocs();
@@ -203,6 +203,9 @@ public class BasicDistributedZk2Test ext
     }
     
     commit();
+    
+    printLayout();
+    
     query("q", "*:*", "sort", "n_tl1 desc");
     
     // long cloudClientDocs = cloudClient.query(new

Modified: lucene/dev/trunk/solr/core/src/test/org/apache/solr/cloud/BasicDistributedZkTest.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/core/src/test/org/apache/solr/cloud/BasicDistributedZkTest.java?rev=1401798&r1=1401797&r2=1401798&view=diff
==============================================================================
--- lucene/dev/trunk/solr/core/src/test/org/apache/solr/cloud/BasicDistributedZkTest.java
(original)
+++ lucene/dev/trunk/solr/core/src/test/org/apache/solr/cloud/BasicDistributedZkTest.java
Wed Oct 24 17:52:15 2012
@@ -27,6 +27,7 @@ import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Random;
 import java.util.Set;
 import java.util.concurrent.Callable;
 import java.util.concurrent.CompletionService;
@@ -66,9 +67,11 @@ import org.apache.solr.common.cloud.ZkNo
 import org.apache.solr.common.cloud.ZkStateReader;
 import org.apache.solr.common.params.CollectionParams.CollectionAction;
 import org.apache.solr.common.params.CommonParams;
+import org.apache.solr.common.params.CoreAdminParams;
 import org.apache.solr.common.params.ModifiableSolrParams;
 import org.apache.solr.common.params.UpdateParams;
 import org.apache.solr.common.util.NamedList;
+import org.apache.solr.update.DirectUpdateHandler2;
 import org.apache.solr.update.SolrCmdDistributor.Request;
 import org.apache.solr.util.DefaultSolrThreadFactory;
 import org.junit.Before;
@@ -124,7 +127,7 @@ public class BasicDistributedZkTest exte
     fixShardCount = true;
     
     sliceCount = 2;
-    shardCount = 3;
+    shardCount = 4;
     completionService = new ExecutorCompletionService<Request>(executor);
     pending = new HashSet<Future<Request>>();
     
@@ -319,17 +322,18 @@ public class BasicDistributedZkTest exte
 
     // would be better if these where all separate tests - but much, much
     // slower
-    doOptimisticLockingAndUpdating();
-    testMultipleCollections();
-    testANewCollectionInOneInstance();
-    testSearchByCollectionName();
-    testANewCollectionInOneInstanceWithManualShardAssignement();
-    testNumberOfCommitsWithCommitAfterAdd();
-
-    testUpdateProcessorsRunOnlyOnce("distrib-dup-test-chain-explicit");
-    testUpdateProcessorsRunOnlyOnce("distrib-dup-test-chain-implicit");
-
-    testCollectionsAPI();
+//    doOptimisticLockingAndUpdating();
+//    testMultipleCollections();
+//    testANewCollectionInOneInstance();
+//    testSearchByCollectionName();
+//    testANewCollectionInOneInstanceWithManualShardAssignement();
+//    testNumberOfCommitsWithCommitAfterAdd();
+//
+//    testUpdateProcessorsRunOnlyOnce("distrib-dup-test-chain-explicit");
+//    testUpdateProcessorsRunOnlyOnce("distrib-dup-test-chain-implicit");
+//
+//    testCollectionsAPI();
+    testCoreUnloadAndLeaders();
     
     // Thread.sleep(10000000000L);
     if (DEBUG) {
@@ -337,6 +341,215 @@ public class BasicDistributedZkTest exte
     }
   }
 
+  /**
+   * @throws Exception on any problem
+   */
+  private void testCoreUnloadAndLeaders() throws Exception {
+    // create a new collection collection
+    SolrServer client = clients.get(0);
+    String url1 = getBaseUrl(client);
+    HttpSolrServer server = new HttpSolrServer(url1);
+    
+    Create createCmd = new Create();
+    createCmd.setCoreName("unloadcollection1");
+    createCmd.setCollection("unloadcollection");
+    createCmd.setNumShards(1);
+    String core1DataDir = dataDir.getAbsolutePath() + File.separator + System.currentTimeMillis()
+ "unloadcollection1" + "_1n";
+    createCmd.setDataDir(core1DataDir);
+    server.request(createCmd);
+    
+    zkStateReader.updateClusterState(true);
+
+    int slices = zkStateReader.getClusterState().getCollectionStates().get("unloadcollection").size();
+    assertEquals(1, slices);
+    
+    client = clients.get(1);
+    String url2 = getBaseUrl(client);
+    server = new HttpSolrServer(url2);
+    
+    createCmd = new Create();
+    createCmd.setCoreName("unloadcollection2");
+    createCmd.setCollection("unloadcollection");
+    String core2dataDir = dataDir.getAbsolutePath() + File.separator + System.currentTimeMillis()
+ "unloadcollection1" + "_2n";
+    createCmd.setDataDir(core2dataDir);
+    server.request(createCmd);
+    
+    zkStateReader.updateClusterState(true);
+    slices = zkStateReader.getClusterState().getCollectionStates().get("unloadcollection").size();
+    assertEquals(1, slices);
+    
+    waitForRecoveriesToFinish("unloadcollection", zkStateReader, false);
+    
+    ZkCoreNodeProps leaderProps = getLeaderUrlFromZk("unloadcollection", "shard1");
+    
+    Random random = random();
+    HttpSolrServer collectionClient;
+    if (random.nextBoolean()) {
+      collectionClient = new HttpSolrServer(leaderProps.getCoreUrl());
+      // lets try and use the solrj client to index and retrieve a couple
+      // documents
+      SolrInputDocument doc1 = getDoc(id, 6, i1, -600, tlong, 600, t1,
+          "humpty dumpy sat on a wall");
+      SolrInputDocument doc2 = getDoc(id, 7, i1, -600, tlong, 600, t1,
+          "humpty dumpy3 sat on a walls");
+      SolrInputDocument doc3 = getDoc(id, 8, i1, -600, tlong, 600, t1,
+          "humpty dumpy2 sat on a walled");
+      collectionClient.add(doc1);
+      collectionClient.add(doc2);
+      collectionClient.add(doc3);
+      collectionClient.commit();
+    }
+
+    // create another replica for our collection
+    client = clients.get(2);
+    String url3 = getBaseUrl(client);
+    server = new HttpSolrServer(url3);
+    
+    createCmd = new Create();
+    createCmd.setCoreName("unloadcollection3");
+    createCmd.setCollection("unloadcollection");
+    String core3dataDir = dataDir.getAbsolutePath() + File.separator + System.currentTimeMillis()
+ "unloadcollection" + "_3n";
+    createCmd.setDataDir(core3dataDir);
+    server.request(createCmd);
+    
+    Thread.sleep(1000);
+    
+    waitForRecoveriesToFinish("unloadcollection", zkStateReader, false);
+    
+    // so that we start with some versions when we reload...
+    DirectUpdateHandler2.commitOnClose = false;
+    
+    HttpSolrServer addClient = new HttpSolrServer(url3 + "/unloadcollection3");
+    // add a few docs
+    for (int x = 20; x < 100; x++) {
+      SolrInputDocument doc1 = getDoc(id, x, i1, -600, tlong, 600, t1,
+          "humpty dumpy sat on a wall");
+      addClient.add(doc1);
+    }
+
+    // don't commit so they remain in the tran log
+    //collectionClient.commit();
+    
+    // unload the leader
+    collectionClient = new HttpSolrServer(leaderProps.getBaseUrl());
+    
+    Unload unloadCmd = new Unload(false);
+    unloadCmd.setCoreName(leaderProps.getCoreName());
+    ModifiableSolrParams p = (ModifiableSolrParams) unloadCmd.getParams();
+
+    collectionClient.request(unloadCmd);
+
+//    Thread.currentThread().sleep(500);
+//    printLayout();
+    
+    int tries = 20;
+    while (leaderProps.getCoreUrl().equals(zkStateReader.getLeaderUrl("unloadcollection",
"shard1", 15000))) {
+      Thread.sleep(100);
+      if (tries-- == 0) {
+        fail("Leader never changed");
+      }
+    }
+    
+    // ensure there is a leader
+    zkStateReader.getLeaderProps("unloadcollection", "shard1", 15000);
+    
+    addClient = new HttpSolrServer(url2 + "/unloadcollection2");
+    // add a few docs while the leader is down
+    for (int x = 101; x < 200; x++) {
+      SolrInputDocument doc1 = getDoc(id, x, i1, -600, tlong, 600, t1,
+          "humpty dumpy sat on a wall");
+      addClient.add(doc1);
+    }
+    
+    
+    // create another replica for our collection
+    client = clients.get(3);
+    String url4 = getBaseUrl(client);
+    server = new HttpSolrServer(url4);
+    
+    createCmd = new Create();
+    createCmd.setCoreName("unloadcollection4");
+    createCmd.setCollection("unloadcollection");
+    String core4dataDir = dataDir.getAbsolutePath() + File.separator + System.currentTimeMillis()
+ "unloadcollection" + "_4n";
+    createCmd.setDataDir(core4dataDir);
+    server.request(createCmd);
+    
+    Thread.sleep(1000);
+    
+    waitForRecoveriesToFinish("unloadcollection", zkStateReader, false);
+    
+    // unload the leader again
+    leaderProps = getLeaderUrlFromZk("unloadcollection", "shard1");
+    collectionClient = new HttpSolrServer(leaderProps.getBaseUrl());
+    
+    unloadCmd = new Unload(false);
+    unloadCmd.setCoreName(leaderProps.getCoreName());
+    p = (ModifiableSolrParams) unloadCmd.getParams();
+    collectionClient.request(unloadCmd);
+    
+    tries = 20;
+    while (leaderProps.getCoreUrl().equals(zkStateReader.getLeaderUrl("unloadcollection",
"shard1", 15000))) {
+      Thread.sleep(100);
+      if (tries-- == 0) {
+        fail("Leader never changed");
+      }
+    }
+    
+    zkStateReader.getLeaderProps("unloadcollection", "shard1", 15000);
+    
+    
+    // set this back
+    DirectUpdateHandler2.commitOnClose = true;
+    
+    // bring the downed leader back as replica
+    server = new HttpSolrServer(leaderProps.getBaseUrl());
+    
+    createCmd = new Create();
+    createCmd.setCoreName(leaderProps.getCoreName());
+    createCmd.setCollection("unloadcollection");
+    createCmd.setDataDir(core1DataDir);
+    server.request(createCmd);
+    
+    Thread.sleep(1000);
+    
+    waitForRecoveriesToFinish("unloadcollection", zkStateReader, false);
+    
+    
+    server = new HttpSolrServer(url1 + "/unloadcollection");
+   // System.out.println(server.query(new SolrQuery("*:*")).getResults().getNumFound());
+    server = new HttpSolrServer(url2 + "/unloadcollection");
+    server.commit();
+    SolrQuery q = new SolrQuery("*:*");
+    q.set("distrib", false);
+    long found1 = server.query(q).getResults().getNumFound();
+    server = new HttpSolrServer(url3 + "/unloadcollection");
+    server.commit();
+    q = new SolrQuery("*:*");
+    q.set("distrib", false);
+    long found3 = server.query(q).getResults().getNumFound();
+    server = new HttpSolrServer(url4 + "/unloadcollection");
+    server.commit();
+    q = new SolrQuery("*:*");
+    q.set("distrib", false);
+    long found4 = server.query(q).getResults().getNumFound();
+    
+    // all 3 shards should now have the same number of docs
+    assertEquals(found1, found3);
+    assertEquals(found3, found4);
+    
+  }
+
+
+  private String getBaseUrl(SolrServer client) {
+    String url2 = ((HttpSolrServer) client).getBaseURL()
+        .substring(
+            0,
+            ((HttpSolrServer) client).getBaseURL().length()
+                - DEFAULT_COLLECTION.length() -1);
+    return url2;
+  }
+
+
   private void testCollectionsAPI() throws Exception {
  
     // TODO: fragile - because we dont pass collection.confName, it will only
@@ -347,32 +560,13 @@ public class BasicDistributedZkTest exte
     // create new collections rapid fire
     Map<String,List<Integer>> collectionInfos = new HashMap<String,List<Integer>>();
     int cnt = atLeast(3);
+    
     for (int i = 0; i < cnt; i++) {
-      ModifiableSolrParams params = new ModifiableSolrParams();
-      params.set("action", CollectionAction.CREATE.toString());
-      int numShards = _TestUtil.nextInt(random(), 0, shardCount) + 1;
-      int numReplicas = _TestUtil.nextInt(random(), 0, 5) + 1;
-      params.set("numShards", numShards);
-      params.set(OverseerCollectionProcessor.REPLICATION_FACTOR, numReplicas);
-      String collectionName = "awholynewcollection_" + i;
-      int clientIndex = random().nextInt(2);
-      List<Integer> list = new ArrayList<Integer>();
-      list.add(numShards);
-      list.add(numReplicas);
-      collectionInfos.put(collectionName, list);
-      params.set("name", collectionName);
-      SolrRequest request = new QueryRequest(params);
-      request.setPath("/admin/collections");
-   
-      final String baseUrl = ((HttpSolrServer) clients.get(clientIndex)).getBaseURL().substring(
-          0,
-          ((HttpSolrServer) clients.get(clientIndex)).getBaseURL().length()
-              - DEFAULT_COLLECTION.length() - 1);
-      
-      createNewSolrServer("", baseUrl).request(request);
+      createCollection(collectionInfos, i,
+          _TestUtil.nextInt(random(), 0, shardCount) + 1,
+          _TestUtil.nextInt(random(), 0, 5) + 1);
     }
     
-    
     Set<Entry<String,List<Integer>>> collectionInfosEntrySet = collectionInfos.entrySet();
     for (Entry<String,List<Integer>> entry : collectionInfosEntrySet) {
       String collection = entry.getKey();
@@ -387,8 +581,8 @@ public class BasicDistributedZkTest exte
       waitForNon403or404or503(collectionClient);
     }
     
-    for (int i = 0; i < cnt; i++) {
-      waitForRecoveriesToFinish("awholynewcollection_" + i, zkStateReader, false);
+    for (int j = 0; j < cnt; j++) {
+      waitForRecoveriesToFinish("awholynewcollection_" + j, zkStateReader, false);
     }
     
     List<String> collectionNameList = new ArrayList<String>();
@@ -400,7 +594,7 @@ public class BasicDistributedZkTest exte
     HttpSolrServer collectionClient = new HttpSolrServer(url);
     
     
-    // lets try and use the solrj client to index and retrieve a couple documents
+    // lets try and use the solrj client to index a couple documents
     SolrInputDocument doc1 = getDoc(id, 6, i1, -600, tlong, 600, t1,
         "humpty dumpy sat on a wall");
     SolrInputDocument doc2 = getDoc(id, 7, i1, -600, tlong, 600, t1,
@@ -442,6 +636,9 @@ public class BasicDistributedZkTest exte
     boolean allTimesAreCorrect = waitForReloads(collectionName, urlToTimeBefore);
     assertTrue("some core start times did not change on reload", allTimesAreCorrect);
     
+    
+    waitForRecoveriesToFinish("awholynewcollection_" + (cnt - 1), zkStateReader, false);
+    
     // remove a collection
     params = new ModifiableSolrParams();
     params.set("action", CollectionAction.DELETE.toString());
@@ -453,7 +650,35 @@ public class BasicDistributedZkTest exte
     
     // ensure its out of the state
     checkForMissingCollection(collectionName);
+    
+    //collectionNameList.remove(collectionName);
+
+  }
 
+
+  protected void createCollection(Map<String,List<Integer>> collectionInfos,
+      int i, int numShards, int numReplicas) throws SolrServerException, IOException {
+    ModifiableSolrParams params = new ModifiableSolrParams();
+    params.set("action", CollectionAction.CREATE.toString());
+
+    params.set("numShards", numShards);
+    params.set(OverseerCollectionProcessor.REPLICATION_FACTOR, numReplicas);
+    String collectionName = "awholynewcollection_" + i;
+    int clientIndex = random().nextInt(2);
+    List<Integer> list = new ArrayList<Integer>();
+    list.add(numShards);
+    list.add(numReplicas);
+    collectionInfos.put(collectionName, list);
+    params.set("name", collectionName);
+    SolrRequest request = new QueryRequest(params);
+    request.setPath("/admin/collections");
+  
+    final String baseUrl = ((HttpSolrServer) clients.get(clientIndex)).getBaseURL().substring(
+        0,
+        ((HttpSolrServer) clients.get(clientIndex)).getBaseURL().length()
+            - DEFAULT_COLLECTION.length() - 1);
+    
+    createNewSolrServer("", baseUrl).request(request);
   }
 
   private boolean waitForReloads(String collectionName, Map<String,Long> urlToTimeBefore)
throws SolrServerException, IOException {
@@ -537,6 +762,15 @@ public class BasicDistributedZkTest exte
     
     throw new RuntimeException("Could not find a live node for collection:" + collection);
   }
+  
+  private ZkCoreNodeProps getLeaderUrlFromZk(String collection, String slice) {
+    ClusterState clusterState = solrj.getZkStateReader().getClusterState();
+    ZkNodeProps leader = clusterState.getLeader(collection, slice);
+    if (leader == null) {
+      throw new RuntimeException("Could not find leader:" + collection + " " + slice);
+    }
+    return new ZkCoreNodeProps(leader);
+  }
 
   private void waitForNon403or404or503(HttpSolrServer collectionClient)
       throws Exception {
@@ -1099,5 +1333,8 @@ public class BasicDistributedZkTest exte
     }
     System.clearProperty("numShards");
     System.clearProperty("zkHost");
+    
+    // insurance
+    DirectUpdateHandler2.commitOnClose = true;
   }
 }

Modified: lucene/dev/trunk/solr/core/src/test/org/apache/solr/update/SolrCmdDistributorTest.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/core/src/test/org/apache/solr/update/SolrCmdDistributorTest.java?rev=1401798&r1=1401797&r2=1401798&view=diff
==============================================================================
--- lucene/dev/trunk/solr/core/src/test/org/apache/solr/update/SolrCmdDistributorTest.java
(original)
+++ lucene/dev/trunk/solr/core/src/test/org/apache/solr/update/SolrCmdDistributorTest.java
Wed Oct 24 17:52:15 2012
@@ -24,6 +24,7 @@ import java.util.List;
 import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.solr.BaseDistributedSearchTestCase;
 import org.apache.solr.client.solrj.SolrQuery;
@@ -38,9 +39,15 @@ import org.apache.solr.common.cloud.ZkSt
 import org.apache.solr.common.params.ModifiableSolrParams;
 import org.apache.solr.common.util.ExecutorUtil;
 import org.apache.solr.common.util.NamedList;
+import org.apache.solr.core.CoreContainer;
+import org.apache.solr.core.SolrCore;
+import org.apache.solr.core.SolrEventListener;
+import org.apache.solr.search.SolrIndexSearcher;
+import org.apache.solr.servlet.SolrDispatchFilter;
 import org.apache.solr.update.SolrCmdDistributor.Node;
 import org.apache.solr.update.SolrCmdDistributor.Response;
 import org.apache.solr.update.SolrCmdDistributor.StdNode;
+import org.apache.solr.update.processor.DistributedUpdateProcessor;
 import org.apache.solr.util.DefaultSolrThreadFactory;
 
 public class SolrCmdDistributorTest extends BaseDistributedSearchTestCase {
@@ -92,6 +99,7 @@ public class SolrCmdDistributorTest exte
     SolrCmdDistributor cmdDistrib = new SolrCmdDistributor(5, executor);
     
     ModifiableSolrParams params = new ModifiableSolrParams();
+
     List<Node> nodes = new ArrayList<Node>();
 
     ZkNodeProps nodeProps = new ZkNodeProps(ZkStateReader.BASE_URL_PROP,
@@ -103,12 +111,17 @@ public class SolrCmdDistributorTest exte
     
     AddUpdateCommand cmd = new AddUpdateCommand(null);
     cmd.solrDoc = sdoc("id", 1);
+    params = new ModifiableSolrParams();
+
     cmdDistrib.distribAdd(cmd, nodes, params);
     
     CommitUpdateCommand ccmd = new CommitUpdateCommand(null, false);
+    params = new ModifiableSolrParams();
+    params.set(DistributedUpdateProcessor.COMMIT_END_POINT, true);
     cmdDistrib.distribCommit(ccmd, nodes, params);
     cmdDistrib.finish();
 
+    
     Response response = cmdDistrib.getResponse();
     
     assertEquals(response.errors.toString(), 0, response.errors.size());
@@ -125,18 +138,26 @@ public class SolrCmdDistributorTest exte
     // add another 2 docs to control and 3 to client
     cmdDistrib = new SolrCmdDistributor(5, executor);
     cmd.solrDoc = sdoc("id", 2);
+    params = new ModifiableSolrParams();
+    params.set(DistributedUpdateProcessor.COMMIT_END_POINT, true);
     cmdDistrib.distribAdd(cmd, nodes, params);
     
     AddUpdateCommand cmd2 = new AddUpdateCommand(null);
     cmd2.solrDoc = sdoc("id", 3);
 
+    params = new ModifiableSolrParams();
+    params.set(DistributedUpdateProcessor.COMMIT_END_POINT, true);
     cmdDistrib.distribAdd(cmd2, nodes, params);
     
     AddUpdateCommand cmd3 = new AddUpdateCommand(null);
     cmd3.solrDoc = sdoc("id", 4);
     
+    params = new ModifiableSolrParams();
+    params.set(DistributedUpdateProcessor.COMMIT_END_POINT, true);
     cmdDistrib.distribAdd(cmd3, Collections.singletonList(nodes.get(1)), params);
     
+    params = new ModifiableSolrParams();
+    params.set(DistributedUpdateProcessor.COMMIT_END_POINT, true);
     cmdDistrib.distribCommit(ccmd, nodes, params);
     cmdDistrib.finish();
     response = cmdDistrib.getResponse();
@@ -156,9 +177,18 @@ public class SolrCmdDistributorTest exte
     DeleteUpdateCommand dcmd = new DeleteUpdateCommand(null);
     dcmd.id = "2";
     
+    
+
     cmdDistrib = new SolrCmdDistributor(5, executor);
+    
+    params = new ModifiableSolrParams();
+    params.set(DistributedUpdateProcessor.COMMIT_END_POINT, true);
+    
     cmdDistrib.distribDelete(dcmd, nodes, params);
     
+    params = new ModifiableSolrParams();
+    params.set(DistributedUpdateProcessor.COMMIT_END_POINT, true);
+    
     cmdDistrib.distribCommit(ccmd, nodes, params);
     cmdDistrib.finish();
 
@@ -184,7 +214,7 @@ public class SolrCmdDistributorTest exte
     
     cmdDistrib = new SolrCmdDistributor(5, executor);
     
-    int cnt = atLeast(201);
+    int cnt = atLeast(303);
     for (int i = 0; i < cnt; i++) {
       nodes.clear();
       for (SolrServer c : clients) {
@@ -194,13 +224,13 @@ public class SolrCmdDistributorTest exte
         HttpSolrServer httpClient = (HttpSolrServer) c;
         nodeProps = new ZkNodeProps(ZkStateReader.BASE_URL_PROP,
             httpClient.getBaseURL(), ZkStateReader.CORE_NAME_PROP, "");
-        System.out.println("node props:" + nodeProps);
         nodes.add(new StdNode(new ZkCoreNodeProps(nodeProps)));
 
       }
       AddUpdateCommand c = new AddUpdateCommand(null);
       c.solrDoc = sdoc("id", id++);
       if (nodes.size() > 0) {
+        params = new ModifiableSolrParams();
         cmdDistrib.distribAdd(c, nodes, params);
       }
     }
@@ -214,11 +244,37 @@ public class SolrCmdDistributorTest exte
       
       nodes.add(new StdNode(new ZkCoreNodeProps(nodeProps)));
     }
+    
+    final AtomicInteger commits = new AtomicInteger();
+    for(JettySolrRunner jetty : jettys) {
+      CoreContainer cores = ((SolrDispatchFilter) jetty.getDispatchFilter().getFilter()).getCores();
+      SolrCore core = cores.getCore("collection1");
+      try {
+        core.getUpdateHandler().registerCommitCallback(new SolrEventListener() {
+          @Override
+          public void init(NamedList args) {}
+          @Override
+          public void postSoftCommit() {}
+          @Override
+          public void postCommit() {
+            commits.incrementAndGet();
+          }
+          @Override
+          public void newSearcher(SolrIndexSearcher newSearcher,
+              SolrIndexSearcher currentSearcher) {}
+        });
+      } finally {
+        core.close();
+      }
+    }
+    params = new ModifiableSolrParams();
+    params.set(DistributedUpdateProcessor.COMMIT_END_POINT, true);
 
     cmdDistrib.distribCommit(ccmd, nodes, params);
     
-    
     cmdDistrib.finish();
+
+    assertEquals(shardCount, commits.get());
     
     for (SolrServer c : clients) {
       NamedList<Object> resp = c.request(new LukeRequest());

Modified: lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java?rev=1401798&r1=1401797&r2=1401798&view=diff
==============================================================================
--- lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java (original)
+++ lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java Wed
Oct 24 17:52:15 2012
@@ -186,7 +186,7 @@ public class ZkStateReader {
           if (EventType.None.equals(event.getType())) {
             return;
           }
-          log.info("A cluster state change has occurred - updating...");
+          log.info("A cluster state change has occurred - updating... ({})", ZkStateReader.this.clusterState.getLiveNodes().size());
           try {
             
             // delayed approach
@@ -235,13 +235,13 @@ public class ZkStateReader {
               if (EventType.None.equals(event.getType())) {
                 return;
               }
-              log.info("Updating live nodes");
               try {
                 // delayed approach
                 // ZkStateReader.this.updateClusterState(false, true);
                 synchronized (ZkStateReader.this.getUpdateLock()) {
                   List<String> liveNodes = zkClient.getChildren(
                       LIVE_NODES_ZKNODE, this, true);
+                  log.info("Updating live nodes... ({})", liveNodes.size());
                   Set<String> liveNodesSet = new HashSet<String>();
                   liveNodesSet.addAll(liveNodes);
                   ClusterState clusterState = new ClusterState(
@@ -296,7 +296,7 @@ public class ZkStateReader {
           
           clusterState = ClusterState.load(zkClient, liveNodesSet);
         } else {
-          log.info("Updating live nodes from ZooKeeper... ");
+          log.info("Updating live nodes from ZooKeeper... ({})", liveNodesSet.size());
           clusterState = new ClusterState(
               ZkStateReader.this.clusterState.getZkClusterStateVersion(), liveNodesSet,
               ZkStateReader.this.clusterState.getCollectionStates());



Mime
View raw message