You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by ab...@apache.org on 2017/06/12 17:38:08 UTC
lucene-solr:jira/solr-10704: SOLR-10704 Wait until all leader
replicas are recovered.
Repository: lucene-solr
Updated Branches:
refs/heads/jira/solr-10704 [created] 1f04fd59e
SOLR-10704 Wait until all leader replicas are recovered.
Project: http://git-wip-us.apache.org/repos/asf/lucene-solr/repo
Commit: http://git-wip-us.apache.org/repos/asf/lucene-solr/commit/1f04fd59
Tree: http://git-wip-us.apache.org/repos/asf/lucene-solr/tree/1f04fd59
Diff: http://git-wip-us.apache.org/repos/asf/lucene-solr/diff/1f04fd59
Branch: refs/heads/jira/solr-10704
Commit: 1f04fd59eeed914edd14d76e63eb247e4711512f
Parents: b2ea95c
Author: Andrzej Bialecki <ab...@apache.org>
Authored: Mon Jun 12 19:37:20 2017 +0200
Committer: Andrzej Bialecki <ab...@apache.org>
Committed: Mon Jun 12 19:37:20 2017 +0200
----------------------------------------------------------------------
.../org/apache/solr/cloud/ReplaceNodeCmd.java | 86 +++++++++++++++++++-
.../org/apache/solr/cloud/ReplaceNodeTest.java | 9 +-
2 files changed, 89 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/1f04fd59/solr/core/src/java/org/apache/solr/cloud/ReplaceNodeCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/ReplaceNodeCmd.java b/solr/core/src/java/org/apache/solr/cloud/ReplaceNodeCmd.java
index e4240be..c046d75 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ReplaceNodeCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ReplaceNodeCmd.java
@@ -20,15 +20,18 @@ package org.apache.solr.cloud;
import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.cloud.ClusterState;
+import org.apache.solr.common.cloud.CollectionStateWatcher;
import org.apache.solr.common.cloud.DocCollection;
import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.cloud.Slice;
@@ -70,13 +73,32 @@ public class ReplaceNodeCmd implements OverseerCollectionMessageHandler.Cmd {
throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "Target Node: " + target + " is not live");
}
List<ZkNodeProps> sourceReplicas = getReplicasOfNode(source, clusterState);
-
+ // how many leaders are we moving?
+ int numLeaders = 0;
+ for (ZkNodeProps props : sourceReplicas) {
+ if (props.getBool(ZkStateReader.LEADER_PROP, false)) {
+ numLeaders++;
+ }
+ }
+ // map of shardId_replicaId to watchers
+ Map<String, RecoveryWatcher> watchers = new HashMap<>();
List<ZkNodeProps> createdReplicas = new ArrayList<>();
AtomicBoolean anyOneFailed = new AtomicBoolean(false);
CountDownLatch countDownLatch = new CountDownLatch(sourceReplicas.size());
+ CountDownLatch replicasToRecover = new CountDownLatch(numLeaders);
+
for (ZkNodeProps sourceReplica : sourceReplicas) {
+ if (sourceReplica.getBool(ZkStateReader.LEADER_PROP, false)) {
+ String shardId = sourceReplica.getStr(SHARD_ID_PROP);
+ String replicaId = sourceReplica.getStr(ZkStateReader.REPLICA_PROP);
+ String collectionId = sourceReplica.getStr(COLLECTION_PROP);
+ String key = shardId + "_" + replicaId;
+ RecoveryWatcher watcher = new RecoveryWatcher(collectionId, shardId, replicaId, replicasToRecover);
+ watchers.put(key, watcher);
+ zkStateReader.registerCollectionStateWatcher(collectionId, watcher);
+ }
NamedList nl = new NamedList();
log.info("Going to create replica for collection={} shard={} on node={}", sourceReplica.getStr(COLLECTION_PROP), sourceReplica.getStr(SHARD_ID_PROP), target);
ZkNodeProps msg = sourceReplica.plus("parallel", String.valueOf(parallel)).plus(CoreAdminParams.NODE, target);
@@ -106,10 +128,26 @@ public class ReplaceNodeCmd implements OverseerCollectionMessageHandler.Cmd {
}
}
- log.debug("Waiting for replace node action to complete");
- countDownLatch.await(5, TimeUnit.MINUTES);
- log.debug("Finished waiting for replace node action to complete");
+ log.debug("Waiting for replicas to be added");
+ if (!countDownLatch.await(5, TimeUnit.MINUTES)) {
+ log.info("Timed out waiting for replicas to be added");
+ anyOneFailed.set(true);
+ } else {
+ log.debug("Finished waiting for replicas to be added");
+ }
+ // now wait for leader replicas to recover
+ log.debug("Waiting for " + numLeaders + " leader replicas to recover");
+ if (!replicasToRecover.await(5, TimeUnit.MINUTES)) {
+ log.info("Timed out waiting for " + replicasToRecover.getCount() + " leader replicas to recover");
+ anyOneFailed.set(true);
+ } else {
+ log.debug("Finished waiting for leader replicas to recover");
+ }
+ // remove the watchers, we're done either way
+ for (RecoveryWatcher watcher : watchers.values()) {
+ zkStateReader.removeCollectionStateWatcher(watcher.collectionId, watcher);
+ }
if (anyOneFailed.get()) {
log.info("Failed to create some replicas. Cleaning up all replicas on target node");
CountDownLatch cleanupLatch = new CountDownLatch(createdReplicas.size());
@@ -155,6 +193,7 @@ public class ReplaceNodeCmd implements OverseerCollectionMessageHandler.Cmd {
ZkStateReader.CORE_NAME_PROP, replica.getCoreName(),
ZkStateReader.REPLICA_PROP, replica.getName(),
ZkStateReader.REPLICA_TYPE, replica.getType().name(),
+ ZkStateReader.LEADER_PROP, String.valueOf(replica.equals(slice.getLeader())),
CoreAdminParams.NODE, source);
sourceReplicas.add(props);
}
@@ -163,4 +202,43 @@ public class ReplaceNodeCmd implements OverseerCollectionMessageHandler.Cmd {
}
return sourceReplicas;
}
+
+ // we use this watcher to wait for replicas to recover
+ private static class RecoveryWatcher implements CollectionStateWatcher {
+ String collectionId;
+ String shardId;
+ String replicaId;
+ CountDownLatch countDownLatch;
+
+ RecoveryWatcher(String collectionId, String shardId, String replicaId, CountDownLatch countDownLatch) {
+ this.collectionId = collectionId;
+ this.shardId = shardId;
+ this.replicaId = replicaId;
+ this.countDownLatch = countDownLatch;
+ }
+
+ @Override
+ public boolean onStateChanged(Set<String> liveNodes, DocCollection collectionState) {
+ if (collectionState == null) { // collection has been deleted - don't wait
+ countDownLatch.countDown();
+ return true;
+ }
+ Slice slice = collectionState.getSlice(shardId);
+ if (slice == null) { // shard has been removed - don't wait
+ countDownLatch.countDown();
+ return true;
+ }
+ for (Replica replica : slice.getReplicas()) {
+ if (!replica.getName().equals(replicaId)) { // another replica exists - check that it's active
+ // check its state
+ if (replica.getState().equals(Replica.State.ACTIVE)) { // recovered - stop waiting
+ countDownLatch.countDown();
+ return true;
+ }
+ }
+ }
+ // set the watch again to wait for the new replica to recover
+ return false;
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/1f04fd59/solr/core/src/test/org/apache/solr/cloud/ReplaceNodeTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/ReplaceNodeTest.java b/solr/core/src/test/org/apache/solr/cloud/ReplaceNodeTest.java
index ecfd3ee..d7fae92 100644
--- a/solr/core/src/test/org/apache/solr/cloud/ReplaceNodeTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/ReplaceNodeTest.java
@@ -67,11 +67,16 @@ public class ReplaceNodeTest extends SolrCloudTestCase {
CollectionAdminRequest.Create create;
// NOTE: always using the createCollection that takes in 'int' for all types of replicas, so we never
// have to worry about null checking when comparing the Create command with the final Slices
- create = pickRandom(CollectionAdminRequest.createCollection(coll, "conf1", 5, 2,0,0),
+ create = pickRandom(
+ CollectionAdminRequest.createCollection(coll, "conf1", 5, 2,0,0),
CollectionAdminRequest.createCollection(coll, "conf1", 5, 1,1,0),
CollectionAdminRequest.createCollection(coll, "conf1", 5, 0,1,1),
CollectionAdminRequest.createCollection(coll, "conf1", 5, 1,0,1),
- CollectionAdminRequest.createCollection(coll, "conf1", 5, 0,2,0));
+ CollectionAdminRequest.createCollection(coll, "conf1", 5, 0,2,0),
+ // check also replicationFactor 1
+ CollectionAdminRequest.createCollection(coll, "conf1", 5, 1,0,0),
+ CollectionAdminRequest.createCollection(coll, "conf1", 5, 0,1,0)
+ );
create.setCreateNodeSet(StrUtils.join(l, ',')).setMaxShardsPerNode(3);
cloudClient.request(create);
log.info("excluded_node : {} ", emptyNode);