You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by tf...@apache.org on 2017/03/08 22:31:48 UTC
[5/6] lucene-solr:jira/solr-10233: First commit for SOLR-10233
First commit for SOLR-10233
Many nocommits and TODOs. Not all tests passing.
Project: http://git-wip-us.apache.org/repos/asf/lucene-solr/repo
Commit: http://git-wip-us.apache.org/repos/asf/lucene-solr/commit/043d5cb2
Tree: http://git-wip-us.apache.org/repos/asf/lucene-solr/tree/043d5cb2
Diff: http://git-wip-us.apache.org/repos/asf/lucene-solr/diff/043d5cb2
Branch: refs/heads/jira/solr-10233
Commit: 043d5cb23ce59104abd6bb682eeabf63be51225f
Parents: cbcf892
Author: Tomas Fernandez Lobbe <tf...@apache.org>
Authored: Wed Mar 8 13:56:18 2017 -0800
Committer: Tomas Fernandez Lobbe <tf...@apache.org>
Committed: Wed Mar 8 13:56:18 2017 -0800
----------------------------------------------------------------------
.../org/apache/solr/cloud/AddReplicaCmd.java | 9 +-
.../src/java/org/apache/solr/cloud/Assign.java | 4 +-
.../org/apache/solr/cloud/CloudDescriptor.java | 17 +-
.../apache/solr/cloud/CreateCollectionCmd.java | 35 +-
.../org/apache/solr/cloud/ElectionContext.java | 5 +-
.../cloud/OverseerCollectionMessageHandler.java | 82 +++-
.../org/apache/solr/cloud/RecoveryStrategy.java | 2 +-
.../apache/solr/cloud/ReplicateFromLeader.java | 35 +-
.../cloud/ReplicateOnlyRecoveryStrategy.java | 363 ++++++++++++++
.../org/apache/solr/cloud/SplitShardCmd.java | 2 +-
.../org/apache/solr/cloud/ZkController.java | 48 +-
.../solr/cloud/overseer/ReplicaMutator.java | 7 +-
.../solr/cloud/overseer/SliceMutator.java | 28 +-
.../solr/cloud/overseer/ZkStateWriter.java | 1 +
.../apache/solr/cloud/rule/ReplicaAssigner.java | 8 +-
.../org/apache/solr/core/CoreContainer.java | 41 +-
.../org/apache/solr/handler/IndexFetcher.java | 17 +-
.../solr/handler/admin/CollectionsHandler.java | 15 +-
.../solr/handler/admin/CoreAdminHandler.java | 3 +
.../handler/component/RealTimeGetComponent.java | 5 +
.../solr/update/DefaultSolrCoreState.java | 24 +-
.../solr/update/DirectUpdateHandler2.java | 3 +-
.../org/apache/solr/update/UpdateHandler.java | 12 +-
.../processor/DistributedUpdateProcessor.java | 11 +-
.../cloud-minimal/conf/solrconfig.xml | 2 +-
.../apache/solr/cloud/TestPassiveReplica.java | 484 +++++++++++++++++++
.../solr/client/solrj/impl/CloudSolrClient.java | 3 +-
.../solrj/request/CollectionAdminRequest.java | 84 +++-
.../apache/solr/common/cloud/DocCollection.java | 10 +-
.../org/apache/solr/common/cloud/Replica.java | 32 ++
.../org/apache/solr/common/cloud/Slice.java | 21 +-
.../apache/solr/common/cloud/ZkStateReader.java | 12 +-
.../solr/common/params/CoreAdminParams.java | 5 +
.../java/org/apache/solr/SolrTestCaseJ4.java | 2 +-
.../java/org/apache/solr/cloud/ChaosMonkey.java | 4 +-
35 files changed, 1276 insertions(+), 160 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/043d5cb2/solr/core/src/java/org/apache/solr/cloud/AddReplicaCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/AddReplicaCmd.java b/solr/core/src/java/org/apache/solr/cloud/AddReplicaCmd.java
index 6bb3350..4420a92 100644
--- a/solr/core/src/java/org/apache/solr/cloud/AddReplicaCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/AddReplicaCmd.java
@@ -72,6 +72,7 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
String node = message.getStr(CoreAdminParams.NODE);
String shard = message.getStr(SHARD_ID_PROP);
String coreName = message.getStr(CoreAdminParams.NAME);
+ Replica.Type replicaType = Replica.Type.valueOf(message.getStr(ZkStateReader.REPLICA_TYPE, Replica.Type.REALTIME.name()));
boolean parallel = message.getBool("parallel", false);
if (StringUtils.isBlank(coreName)) {
coreName = message.getStr(CoreAdminParams.PROPERTY_PREFIX + CoreAdminParams.NAME);
@@ -93,7 +94,7 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
// Kind of unnecessary, but it does put the logic of whether to override maxShardsPerNode in one place.
if (!skipCreateReplicaInClusterState) {
node = getNodesForNewReplicas(clusterState, collection, shard, 1, node,
- ocmh.overseer.getZkController().getCoreContainer()).get(0).nodeName;
+ ocmh.overseer.getZkController().getCoreContainer()).get(0).nodeName;// TODO: use replica type in this logic too
}
log.info("Node Identified {} for creating new replica", node);
@@ -101,7 +102,7 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "Node: " + node + " is not live");
}
if (coreName == null) {
- coreName = Assign.buildCoreName(coll, shard);
+ coreName = Assign.buildCoreName(coll, shard, replicaType);
} else if (!skipCreateReplicaInClusterState) {
//Validate that the core name is unique in that collection
for (Slice slice : coll.getSlices()) {
@@ -126,7 +127,8 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
ZkStateReader.CORE_NAME_PROP, coreName,
ZkStateReader.STATE_PROP, Replica.State.DOWN.toString(),
ZkStateReader.BASE_URL_PROP, zkStateReader.getBaseUrlForNodeName(node),
- ZkStateReader.NODE_NAME_PROP, node);
+ ZkStateReader.NODE_NAME_PROP, node,
+ ZkStateReader.REPLICA_TYPE, replicaType.name());
Overseer.getStateUpdateQueue(zkStateReader.getZkClient()).offer(Utils.toJSON(props));
}
params.set(CoreAdminParams.CORE_NODE_NAME,
@@ -142,6 +144,7 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
params.set(CoreAdminParams.NAME, coreName);
params.set(COLL_CONF, configName);
params.set(CoreAdminParams.COLLECTION, collection);
+ params.set(CoreAdminParams.REPLICA_TYPE, replicaType.name());
if (shard != null) {
params.set(CoreAdminParams.SHARD, shard);
} else if (routeKey != null) {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/043d5cb2/solr/core/src/java/org/apache/solr/cloud/Assign.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/Assign.java b/solr/core/src/java/org/apache/solr/cloud/Assign.java
index ba03ccd..c404226 100644
--- a/solr/core/src/java/org/apache/solr/cloud/Assign.java
+++ b/solr/core/src/java/org/apache/solr/cloud/Assign.java
@@ -108,7 +108,7 @@ public class Assign {
return returnShardId;
}
- static String buildCoreName(DocCollection collection, String shard) {
+ static String buildCoreName(DocCollection collection, String shard, Replica.Type type) {
Slice slice = collection.getSlice(shard);
int replicaNum = slice.getReplicas().size();
for (; ; ) {
@@ -123,7 +123,7 @@ public class Assign {
if (exists) replicaNum++;
else break;
}
- return collection.getName() + "_" + shard + "_replica" + replicaNum;
+ return collection.getName() + "_" + shard + "_replica_" + type.name().substring(0,1).toLowerCase() + replicaNum;
}
static class ReplicaCount {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/043d5cb2/solr/core/src/java/org/apache/solr/cloud/CloudDescriptor.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/CloudDescriptor.java b/solr/core/src/java/org/apache/solr/cloud/CloudDescriptor.java
index fdc7b02..d34d1ee 100644
--- a/solr/core/src/java/org/apache/solr/cloud/CloudDescriptor.java
+++ b/solr/core/src/java/org/apache/solr/cloud/CloudDescriptor.java
@@ -43,6 +43,13 @@ public class CloudDescriptor {
volatile Replica.State lastPublished = Replica.State.ACTIVE;
public static final String NUM_SHARDS = "numShards";
+
+ public static final String REPLICA_TYPE = "replicaType";
+
+ /**
+ * The type of replica this core hosts
+ */
+ private final Replica.Type replicaType;
public CloudDescriptor(String coreName, Properties props, CoreDescriptor cd) {
this.cd = cd;
@@ -56,7 +63,7 @@ public class CloudDescriptor {
if (Strings.isNullOrEmpty(nodeName))
this.nodeName = null;
this.numShards = PropertiesUtil.toInteger(props.getProperty(CloudDescriptor.NUM_SHARDS), null);
-
+ this.replicaType = Replica.Type.valueOf(props.getProperty(CloudDescriptor.REPLICA_TYPE, Replica.Type.REALTIME.name()));
for (String propName : props.stringPropertyNames()) {
if (propName.startsWith(ZkController.COLLECTION_PARAM_PREFIX)) {
collectionParams.put(propName.substring(ZkController.COLLECTION_PARAM_PREFIX.length()), props.getProperty(propName));
@@ -64,6 +71,10 @@ public class CloudDescriptor {
}
}
+ public boolean requiresTransactionLog() {
+ return this.replicaType != Replica.Type.PASSIVE;
+ }
+
public Replica.State getLastPublished() {
return lastPublished;
}
@@ -135,4 +146,8 @@ public class CloudDescriptor {
if(nodeName==null) cd.getPersistableStandardProperties().remove(CoreDescriptor.CORE_NODE_NAME);
else cd.getPersistableStandardProperties().setProperty(CoreDescriptor.CORE_NODE_NAME, nodeName);
}
+
+ public Replica.Type getReplicaType() {
+ return replicaType;
+ }
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/043d5cb2/solr/core/src/java/org/apache/solr/cloud/CreateCollectionCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/CreateCollectionCmd.java b/solr/core/src/java/org/apache/solr/cloud/CreateCollectionCmd.java
index a1bb70e..d9564b5 100644
--- a/solr/core/src/java/org/apache/solr/cloud/CreateCollectionCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/CreateCollectionCmd.java
@@ -62,6 +62,8 @@ import static org.apache.solr.cloud.OverseerCollectionMessageHandler.NUM_SLICES;
import static org.apache.solr.cloud.OverseerCollectionMessageHandler.RANDOM;
import static org.apache.solr.common.cloud.ZkStateReader.MAX_SHARDS_PER_NODE;
import static org.apache.solr.common.cloud.ZkStateReader.REPLICATION_FACTOR;
+import static org.apache.solr.common.cloud.ZkStateReader.PASSIVE_REPLICAS;
+import static org.apache.solr.common.cloud.ZkStateReader.REALTIME_REPLICAS;
import static org.apache.solr.common.params.CollectionParams.CollectionAction.ADDREPLICA;
import static org.apache.solr.common.params.CommonAdminParams.ASYNC;
import static org.apache.solr.common.params.CommonParams.NAME;
@@ -96,7 +98,9 @@ public class CreateCollectionCmd implements Cmd {
// look at the replication factor and see if it matches reality
// if it does not, find best nodes to create more cores
- int repFactor = message.getInt(REPLICATION_FACTOR, 1);
+ int numRealtimeReplicas = message.getInt(REALTIME_REPLICAS, message.getInt(REPLICATION_FACTOR, 1));
+ int numPassiveReplicas = message.getInt(PASSIVE_REPLICAS, 0);
+ //TODO: Add active/append replicas
ShardHandler shardHandler = ocmh.shardHandlerFactory.getShardHandler();
final String async = message.getStr(ASYNC);
@@ -114,10 +118,10 @@ public class CreateCollectionCmd implements Cmd {
ClusterStateMutator.getShardNames(numSlices, shardNames);
}
- int maxShardsPerNode = message.getInt(MAX_SHARDS_PER_NODE, 1);
+ int maxShardsPerNode = message.getInt(MAX_SHARDS_PER_NODE, 100); //nocommit
- if (repFactor <= 0) {
- throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, REPLICATION_FACTOR + " must be greater than 0");
+ if (numRealtimeReplicas <= 0) {
+ throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, REALTIME_REPLICAS + " must be greater than 0");
}
if (numSlices <= 0) {
@@ -135,11 +139,11 @@ public class CreateCollectionCmd implements Cmd {
positionVsNodes = new HashMap<>();
} else {
- if (repFactor > nodeList.size()) {
+ if (numRealtimeReplicas > nodeList.size()) {
log.warn("Specified "
- + REPLICATION_FACTOR
+ + REALTIME_REPLICAS
+ " of "
- + repFactor
+ + numRealtimeReplicas
+ " on collection "
+ collectionName
+ " is higher than or equal to the number of Solr instances currently live or live and part of your " + CREATE_NODE_SET + "("
@@ -148,19 +152,20 @@ public class CreateCollectionCmd implements Cmd {
}
int maxShardsAllowedToCreate = maxShardsPerNode * nodeList.size();
- int requestedShardsToCreate = numSlices * repFactor;
+ int requestedShardsToCreate = numSlices * (numRealtimeReplicas + numPassiveReplicas);
if (maxShardsAllowedToCreate < requestedShardsToCreate) {
throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "Cannot create collection " + collectionName + ". Value of "
+ MAX_SHARDS_PER_NODE + " is " + maxShardsPerNode
+ ", and the number of nodes currently live or live and part of your "+CREATE_NODE_SET+" is " + nodeList.size()
+ ". This allows a maximum of " + maxShardsAllowedToCreate
+ " to be created. Value of " + NUM_SLICES + " is " + numSlices
- + " and value of " + REPLICATION_FACTOR + " is " + repFactor
+ + ", value of " + REALTIME_REPLICAS + " is " + numRealtimeReplicas
+ + " and value of " + PASSIVE_REPLICAS + " is " + numPassiveReplicas
+ ". This requires " + requestedShardsToCreate
+ " shards to be created (higher than the allowed number)");
}
- positionVsNodes = ocmh.identifyNodes(clusterState, nodeList, message, shardNames, repFactor);
+ positionVsNodes = ocmh.identifyNodes(clusterState, nodeList, message, shardNames, numRealtimeReplicas, 0, numPassiveReplicas);
}
ZkStateReader zkStateReader = ocmh.zkStateReader;
@@ -201,12 +206,13 @@ public class CreateCollectionCmd implements Cmd {
log.debug(formatString("Creating SolrCores for new collection {0}, shardNames {1} , replicationFactor : {2}",
- collectionName, shardNames, repFactor));
+ collectionName, shardNames, numRealtimeReplicas));
Map<String,ShardRequest> coresToCreate = new LinkedHashMap<>();
for (Map.Entry<ReplicaAssigner.Position, String> e : positionVsNodes.entrySet()) {
ReplicaAssigner.Position position = e.getKey();
String nodeName = e.getValue();
- String coreName = collectionName + "_" + position.shard + "_replica" + (position.index + 1);
+ // TODO: Adding the suffix is great for debugging, but may be an issue if at some point we want to support a way to change replica type
+ String coreName = collectionName + "_" + position.shard + "_replica" + position.suffix + (position.index + 1);
log.debug(formatString("Creating core {0} as part of shard {1} of collection {2} on {3}"
, coreName, position.shard, collectionName, nodeName));
@@ -221,7 +227,8 @@ public class CreateCollectionCmd implements Cmd {
ZkStateReader.SHARD_ID_PROP, position.shard,
ZkStateReader.CORE_NAME_PROP, coreName,
ZkStateReader.STATE_PROP, Replica.State.DOWN.toString(),
- ZkStateReader.BASE_URL_PROP, baseUrl);
+ ZkStateReader.BASE_URL_PROP, baseUrl,
+ ZkStateReader.REPLICA_TYPE, position.type.name());
Overseer.getStateUpdateQueue(zkStateReader.getZkClient()).offer(Utils.toJSON(props));
}
@@ -235,6 +242,8 @@ public class CreateCollectionCmd implements Cmd {
params.set(CoreAdminParams.SHARD, position.shard);
params.set(ZkStateReader.NUM_SHARDS_PROP, numSlices);
params.set(CoreAdminParams.NEW_COLLECTION, "true");
+ // This is used to tell the CoreAdminHandler that the new core doesn't need a tlog in case of passive replicas
+ params.set(CoreAdminParams.REPLICA_TYPE, position.type.name());
if (async != null) {
String coreAdminAsyncId = async + Math.abs(System.nanoTime());
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/043d5cb2/solr/core/src/java/org/apache/solr/cloud/ElectionContext.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/ElectionContext.java b/solr/core/src/java/org/apache/solr/cloud/ElectionContext.java
index 223a539..4aa721d 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ElectionContext.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ElectionContext.java
@@ -20,6 +20,7 @@ import java.io.Closeable;
import java.io.IOException;
import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
+import java.util.EnumSet;
import java.util.List;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
@@ -619,7 +620,7 @@ final class ShardLeaderElectionContext extends ShardLeaderElectionContextBase {
}
// on startup and after connection timeout, wait for all known shards
- if (found >= slices.getReplicasMap().size()) {
+ if (found >= slices.getReplicas(EnumSet.of(Replica.Type.APPEND, Replica.Type.REALTIME)).size()) {
log.info("Enough replicas found to continue.");
return true;
} else {
@@ -627,7 +628,7 @@ final class ShardLeaderElectionContext extends ShardLeaderElectionContextBase {
log.info("Waiting until we see more replicas up for shard {}: total={}"
+ " found={}"
+ " timeoutin={}ms",
- shardId, slices.getReplicasMap().size(), found,
+ shardId, slices.getReplicas(EnumSet.of(Replica.Type.APPEND, Replica.Type.REALTIME)).size(), found,
TimeUnit.MILLISECONDS.convert(timeoutAt - System.nanoTime(), TimeUnit.NANOSECONDS));
}
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/043d5cb2/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionMessageHandler.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionMessageHandler.java b/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionMessageHandler.java
index 4d64a00..9319e8d 100644
--- a/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionMessageHandler.java
+++ b/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionMessageHandler.java
@@ -16,6 +16,50 @@
*/
package org.apache.solr.cloud;
+import static org.apache.solr.common.cloud.DocCollection.SNITCH;
+import static org.apache.solr.common.cloud.ZkStateReader.BASE_URL_PROP;
+import static org.apache.solr.common.cloud.ZkStateReader.COLLECTION_PROP;
+import static org.apache.solr.common.cloud.ZkStateReader.CORE_NAME_PROP;
+import static org.apache.solr.common.cloud.ZkStateReader.CORE_NODE_NAME_PROP;
+import static org.apache.solr.common.cloud.ZkStateReader.ELECTION_NODE_PROP;
+import static org.apache.solr.common.cloud.ZkStateReader.PROPERTY_PROP;
+import static org.apache.solr.common.cloud.ZkStateReader.PROPERTY_VALUE_PROP;
+import static org.apache.solr.common.cloud.ZkStateReader.REJOIN_AT_HEAD_PROP;
+import static org.apache.solr.common.cloud.ZkStateReader.REPLICA_PROP;
+import static org.apache.solr.common.cloud.ZkStateReader.SHARD_ID_PROP;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.ADDREPLICA;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.ADDREPLICAPROP;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.ADDROLE;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.BACKUP;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.BALANCESHARDUNIQUE;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.CREATE;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.CREATEALIAS;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.CREATESHARD;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.CREATESNAPSHOT;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.DELETE;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.DELETEALIAS;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.DELETENODE;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.DELETEREPLICA;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.DELETEREPLICAPROP;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.DELETESHARD;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.DELETESNAPSHOT;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.MIGRATE;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.MIGRATESTATEFORMAT;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.MOCK_COLL_TASK;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.MOCK_REPLICA_TASK;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.MOCK_SHARD_TASK;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.MODIFYCOLLECTION;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.OVERSEERSTATUS;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.REBALANCELEADERS;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.RELOAD;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.REMOVEROLE;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.REPLACENODE;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.RESTORE;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.SPLITSHARD;
+import static org.apache.solr.common.params.CommonAdminParams.ASYNC;
+import static org.apache.solr.common.params.CommonParams.NAME;
+import static org.apache.solr.common.util.Utils.makeMap;
+
import java.io.Closeable;
import java.io.IOException;
import java.lang.invoke.MethodHandles;
@@ -33,7 +77,6 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
-import com.google.common.collect.ImmutableMap;
import org.apache.commons.lang.StringUtils;
import org.apache.solr.client.solrj.SolrResponse;
import org.apache.solr.client.solrj.SolrServerException;
@@ -79,21 +122,7 @@ import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static org.apache.solr.common.cloud.DocCollection.SNITCH;
-import static org.apache.solr.common.cloud.ZkStateReader.BASE_URL_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.COLLECTION_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.CORE_NAME_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.CORE_NODE_NAME_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.ELECTION_NODE_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.PROPERTY_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.PROPERTY_VALUE_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.REJOIN_AT_HEAD_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.REPLICA_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.SHARD_ID_PROP;
-import static org.apache.solr.common.params.CollectionParams.CollectionAction.*;
-import static org.apache.solr.common.params.CommonAdminParams.ASYNC;
-import static org.apache.solr.common.params.CommonParams.NAME;
-import static org.apache.solr.common.util.Utils.makeMap;
+import com.google.common.collect.ImmutableMap;
/**
* A {@link OverseerMessageHandler} that handles Collections API related
@@ -126,6 +155,7 @@ public class OverseerCollectionMessageHandler implements OverseerMessageHandler
static final String SKIP_CREATE_REPLICA_IN_CLUSTER_STATE = "skipCreateReplicaInClusterState";
+ //nocommit: review
public static final Map<String, Object> COLL_PROPS = Collections.unmodifiableMap(makeMap(
ROUTER, DocRouter.DEFAULT_NAME,
ZkStateReader.REPLICATION_FACTOR, "1",
@@ -699,14 +729,24 @@ public class OverseerCollectionMessageHandler implements OverseerMessageHandler
List<String> nodeList,
ZkNodeProps message,
List<String> shardNames,
- int repFactor) throws IOException {
+ int numRealtimeReplicas,
+ int numActiveReplicas,
+ int numPassiveReplicas) throws IOException {
List<Map> rulesMap = (List) message.get("rule");
if (rulesMap == null) {
int i = 0;
Map<Position, String> result = new HashMap<>();
for (String aShard : shardNames) {
- for (int j = 0; j < repFactor; j++){
- result.put(new Position(aShard, j), nodeList.get(i % nodeList.size()));
+ for (int j = 0; j < numRealtimeReplicas; j++){
+ result.put(new Position(aShard, j, Replica.Type.REALTIME), nodeList.get(i % nodeList.size()));
+ i++;
+ }
+ for (int j = 0; j < numActiveReplicas; j++){
+ result.put(new Position(aShard, j, Replica.Type.APPEND), nodeList.get(i % nodeList.size()));
+ i++;
+ }
+ for (int j = 0; j < numPassiveReplicas; j++){
+ result.put(new Position(aShard, j, Replica.Type.PASSIVE), nodeList.get(i % nodeList.size()));
i++;
}
}
@@ -718,7 +758,7 @@ public class OverseerCollectionMessageHandler implements OverseerMessageHandler
Map<String, Integer> sharVsReplicaCount = new HashMap<>();
- for (String shard : shardNames) sharVsReplicaCount.put(shard, repFactor);
+ for (String shard : shardNames) sharVsReplicaCount.put(shard, numRealtimeReplicas); // TODO: We probably want to separate different replica types here, not just add them
ReplicaAssigner replicaAssigner = new ReplicaAssigner(rules,
sharVsReplicaCount,
(List<Map>) message.get(SNITCH),
@@ -749,6 +789,8 @@ public class OverseerCollectionMessageHandler implements OverseerMessageHandler
if (result.size() == coreNames.size()) {
return result;
+ } else {
+ log.debug("Expecting {} cores but found {}", coreNames.size(), result.size());
}
if (timeout.hasTimedOut()) {
throw new SolrException(ErrorCode.SERVER_ERROR, "Timed out waiting to see all replicas: " + coreNames + " in cluster state.");
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/043d5cb2/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java b/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java
index 52efff9..cd13e24 100644
--- a/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java
+++ b/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java
@@ -461,7 +461,7 @@ public class RecoveryStrategy extends Thread implements Closeable {
LOG.info("Registering as Active after recovery.");
try {
if (onlyLeaderIndexes) {
- zkController.startReplicationFromLeader(coreName);
+ zkController.startReplicationFromLeader(coreName, true);
}
zkController.publish(core.getCoreDescriptor(), Replica.State.ACTIVE);
} catch (Exception e) {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/043d5cb2/solr/core/src/java/org/apache/solr/cloud/ReplicateFromLeader.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/ReplicateFromLeader.java b/solr/core/src/java/org/apache/solr/cloud/ReplicateFromLeader.java
index 48aaced..7188d28 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ReplicateFromLeader.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ReplicateFromLeader.java
@@ -49,7 +49,12 @@ public class ReplicateFromLeader {
this.coreName = coreName;
}
- public void startReplication() throws InterruptedException {
+ /**
+ * Start a replication handler thread that will periodically pull indices from the shard leader
+ * @param switchTransactionLog if true, ReplicationHandler will rotate the transaction log once
+ * the replication is done
+ */
+ public void startReplication(boolean switchTransactionLog) throws InterruptedException {
try (SolrCore core = cc.getCore(coreName)) {
if (core == null) {
if (cc.isShutDown()) {
@@ -78,19 +83,21 @@ public class ReplicateFromLeader {
}
replicationProcess = new ReplicationHandler();
- replicationProcess.setPollListener((solrCore, pollSuccess) -> {
- if (pollSuccess) {
- String commitVersion = getCommitVersion(core);
- if (commitVersion == null) return;
- if (Long.parseLong(commitVersion) == lastVersion) return;
- UpdateLog updateLog = solrCore.getUpdateHandler().getUpdateLog();
- SolrQueryRequest req = new LocalSolrQueryRequest(core,
- new ModifiableSolrParams());
- CommitUpdateCommand cuc = new CommitUpdateCommand(req, false);
- cuc.setVersion(Long.parseLong(commitVersion));
- updateLog.copyOverOldUpdates(cuc);
- }
- });
+ if (switchTransactionLog) {
+ replicationProcess.setPollListener((solrCore, pollSuccess) -> {
+ if (pollSuccess) {
+ String commitVersion = getCommitVersion(core);
+ if (commitVersion == null) return;
+ if (Long.parseLong(commitVersion) == lastVersion) return;
+ UpdateLog updateLog = solrCore.getUpdateHandler().getUpdateLog();
+ SolrQueryRequest req = new LocalSolrQueryRequest(core,
+ new ModifiableSolrParams());
+ CommitUpdateCommand cuc = new CommitUpdateCommand(req, false);
+ cuc.setVersion(Long.parseLong(commitVersion));
+ updateLog.copyOverOldUpdates(cuc);
+ }
+ });
+ }
replicationProcess.init(replicationConfig);
replicationProcess.inform(core);
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/043d5cb2/solr/core/src/java/org/apache/solr/cloud/ReplicateOnlyRecoveryStrategy.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/ReplicateOnlyRecoveryStrategy.java b/solr/core/src/java/org/apache/solr/cloud/ReplicateOnlyRecoveryStrategy.java
new file mode 100644
index 0000000..8077d20
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/cloud/ReplicateOnlyRecoveryStrategy.java
@@ -0,0 +1,363 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+import java.lang.invoke.MethodHandles;
+import java.util.Arrays;
+
+import org.apache.lucene.search.MatchAllDocsQuery;
+import org.apache.lucene.store.Directory;
+import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.impl.HttpSolrClient;
+import org.apache.solr.client.solrj.request.AbstractUpdateRequest;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+import org.apache.solr.common.SolrException;
+import org.apache.solr.common.SolrException.ErrorCode;
+import org.apache.solr.common.cloud.Replica;
+import org.apache.solr.common.cloud.ZkCoreNodeProps;
+import org.apache.solr.common.cloud.ZkNodeProps;
+import org.apache.solr.common.cloud.ZkStateReader;
+import org.apache.solr.common.cloud.ZooKeeperException;
+import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.params.UpdateParams;
+import org.apache.solr.core.CoreContainer;
+import org.apache.solr.core.CoreDescriptor;
+import org.apache.solr.core.DirectoryFactory.DirContext;
+import org.apache.solr.core.SolrCore;
+import org.apache.solr.handler.ReplicationHandler;
+import org.apache.solr.logging.MDCLoggingContext;
+import org.apache.solr.request.SolrRequestHandler;
+import org.apache.solr.search.SolrIndexSearcher;
+import org.apache.solr.update.processor.DistributedUpdateProcessor;
+import org.apache.solr.util.RefCounted;
+import org.apache.zookeeper.KeeperException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ReplicateOnlyRecoveryStrategy extends Thread implements Closeable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ private static final int MAX_RETRIES = 500;
+ private static final int STARTING_RECOVERY_DELAY = 5000;
+
+ public static interface RecoveryListener {
+ public void recovered();
+ public void failed();
+ }
+
+ private volatile boolean close = false;
+
+ private RecoveryStrategy.RecoveryListener recoveryListener;
+ private ZkController zkController;
+ private String baseUrl;
+ private String coreZkNodeName;
+ private ZkStateReader zkStateReader;
+ private volatile String coreName;
+ private int retries;
+ private boolean recoveringAfterStartup;
+ private CoreContainer cc;
+
+ // this should only be used from SolrCoreState
+ public ReplicateOnlyRecoveryStrategy(CoreContainer cc, CoreDescriptor cd, RecoveryStrategy.RecoveryListener recoveryListener) {
+ this.cc = cc;
+ this.coreName = cd.getName();
+ this.recoveryListener = recoveryListener;
+ setName("ReplicateOnlyRecoveryThread-"+this.coreName);
+ zkController = cc.getZkController();
+ zkStateReader = zkController.getZkStateReader();
+ baseUrl = zkController.getBaseUrl();
+ coreZkNodeName = cd.getCloudDescriptor().getCoreNodeName();
+ }
+
+ public void setRecoveringAfterStartup(boolean recoveringAfterStartup) {
+ this.recoveringAfterStartup = recoveringAfterStartup;
+ }
+
+ // make sure any threads stop retrying
+ @Override
+ public void close() {
+ close = true;
+ LOG.warn("Stopping recovery for core=[{}] coreNodeName=[{}]", coreName, coreZkNodeName);
+ }
+
+ private void recoveryFailed(final SolrCore core,
+ final ZkController zkController, final String baseUrl,
+ final String shardZkNodeName, final CoreDescriptor cd) throws KeeperException, InterruptedException {
+ SolrException.log(LOG, "Recovery failed - I give up.");
+ try {
+ zkController.publish(cd, Replica.State.RECOVERY_FAILED);
+ } finally {
+ close();
+ recoveryListener.failed();
+ }
+ }
+
+ private void replicate(String nodeName, SolrCore core, ZkNodeProps leaderprops)
+ throws SolrServerException, IOException {
+
+ ZkCoreNodeProps leaderCNodeProps = new ZkCoreNodeProps(leaderprops);
+ String leaderUrl = leaderCNodeProps.getCoreUrl();
+
+ LOG.info("Attempting to replicate from [{}].", leaderUrl);
+
+ // send commit
+ commitOnLeader(leaderUrl);
+
+ // use rep handler directly, so we can do this sync rather than async
+ SolrRequestHandler handler = core.getRequestHandler(ReplicationHandler.PATH);
+ ReplicationHandler replicationHandler = (ReplicationHandler) handler;
+
+ if (replicationHandler == null) {
+ throw new SolrException(ErrorCode.SERVICE_UNAVAILABLE,
+ "Skipping recovery, no " + ReplicationHandler.PATH + " handler found");
+ }
+
+ ModifiableSolrParams solrParams = new ModifiableSolrParams();
+ solrParams.set(ReplicationHandler.MASTER_URL, leaderUrl);
+
+ if (isClosed()) return; // we check closed on return
+ boolean success = replicationHandler.doFetch(solrParams, false);
+
+ if (!success) {
+ throw new SolrException(ErrorCode.SERVER_ERROR, "Replication for recovery failed.");
+ }
+
+ // solrcloud_debug
+ if (LOG.isDebugEnabled()) {
+ try {
+ RefCounted<SolrIndexSearcher> searchHolder = core
+ .getNewestSearcher(false);
+ SolrIndexSearcher searcher = searchHolder.get();
+ Directory dir = core.getDirectoryFactory().get(core.getIndexDir(), DirContext.META_DATA, null);
+ try {
+ LOG.debug(core.getCoreDescriptor().getCoreContainer()
+ .getZkController().getNodeName()
+ + " replicated "
+ + searcher.search(new MatchAllDocsQuery(), 1).totalHits
+ + " from "
+ + leaderUrl
+ + " gen:"
+ + (core.getDeletionPolicy().getLatestCommit() != null ? "null" : core.getDeletionPolicy().getLatestCommit().getGeneration())
+ + " data:" + core.getDataDir()
+ + " index:" + core.getIndexDir()
+ + " newIndex:" + core.getNewIndexDir()
+ + " files:" + Arrays.asList(dir.listAll()));
+ } finally {
+ core.getDirectoryFactory().release(dir);
+ searchHolder.decref();
+ }
+ } catch (Exception e) {
+ LOG.debug("Error in solrcloud_debug block", e);
+ }
+ }
+
+ }
+
+ private void commitOnLeader(String leaderUrl) throws SolrServerException,
+ IOException {
+ try (HttpSolrClient client = new HttpSolrClient.Builder(leaderUrl).build()) {
+ client.setConnectionTimeout(30000);
+ UpdateRequest ureq = new UpdateRequest();
+ ureq.setParams(new ModifiableSolrParams());
+ ureq.getParams().set(DistributedUpdateProcessor.COMMIT_END_POINT, true);
+ ureq.getParams().set(UpdateParams.OPEN_SEARCHER, false);
+ ureq.setAction(AbstractUpdateRequest.ACTION.COMMIT, false, true).process(client);
+ }
+ }
+
+ @Override
+ public void run() {
+
+ // set request info for logging
+ try (SolrCore core = cc.getCore(coreName)) {
+
+ if (core == null) {
+ SolrException.log(LOG, "SolrCore not found - cannot recover:" + coreName);
+ return;
+ }
+ MDCLoggingContext.setCore(core);
+
+ try {
+ doRecovery(core);
+ } catch (InterruptedException e) {
+ // nocommit: InterruptedException should't really happen now
+ Thread.currentThread().interrupt();
+ SolrException.log(LOG, "", e);
+ throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR, "", e);
+ } catch (Exception e) {
+ LOG.error("", e);
+ throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR, "", e);
+ }
+ } finally {
+ MDCLoggingContext.clear();
+ }
+ }
+
+ // TODO: perhaps make this grab a new core each time through the loop to handle core reloads?
+ public void doRecovery(SolrCore core) throws KeeperException, InterruptedException {
+ boolean successfulRecovery = false;
+
+// if (core.getUpdateHandler().getUpdateLog() != null) {
+// SolrException.log(LOG, "'replicate-only' recovery strategy should only be used if no update logs are present, but this core has one: "
+// + core.getUpdateHandler().getUpdateLog());
+// return;
+// }
+ while (!successfulRecovery && !isInterrupted() && !isClosed()) { // don't use interruption or it will close channels though
+ try {
+ CloudDescriptor cloudDesc = core.getCoreDescriptor().getCloudDescriptor();
+ ZkNodeProps leaderprops = zkStateReader.getLeaderRetry(
+ cloudDesc.getCollectionName(), cloudDesc.getShardId());
+ final String leaderBaseUrl = leaderprops.getStr(ZkStateReader.BASE_URL_PROP);
+ final String leaderCoreName = leaderprops.getStr(ZkStateReader.CORE_NAME_PROP);
+
+ String leaderUrl = ZkCoreNodeProps.getCoreUrl(leaderBaseUrl, leaderCoreName);
+
+ String ourUrl = ZkCoreNodeProps.getCoreUrl(baseUrl, coreName);
+
+ boolean isLeader = leaderUrl.equals(ourUrl); //TODO: We can probably delete most of this code if we say this strategy can only be used for passive replicas
+ if (isLeader && !cloudDesc.isLeader()) {
+ throw new SolrException(ErrorCode.SERVER_ERROR, "Cloud state still says we are leader.");
+ }
+ if (cloudDesc.isLeader()) {
+ // we are now the leader - no one else must have been suitable
+ LOG.warn("We have not yet recovered - but we are now the leader!");
+ LOG.info("Finished recovery process.");
+ zkController.publish(core.getCoreDescriptor(), Replica.State.ACTIVE);
+ return;
+ }
+
+
+ LOG.info("Publishing state of core [{}] as recovering, leader is [{}] and I am [{}]", core.getName(), leaderUrl,
+ ourUrl);
+ zkController.publish(core.getCoreDescriptor(), Replica.State.RECOVERING);
+
+ if (isClosed()) {
+ LOG.info("Recovery for core {} has been closed", core.getName());
+ break;
+ }
+
+ if (isClosed()) {
+ LOG.info("Recovery for core {} has been closed", core.getName());
+ break;
+ }
+ LOG.info("Starting Replication Recovery.");
+
+ try {
+ LOG.info("Stopping background replicate from leader process");
+ zkController.stopReplicationFromLeader(coreName);
+ replicate(zkController.getNodeName(), core, leaderprops);
+
+ if (isClosed()) {
+ LOG.info("Recovery for core {} has been closed", core.getName());
+ break;
+ }
+
+ LOG.info("Replication Recovery was successful.");
+ successfulRecovery = true;
+ } catch (Exception e) {
+ SolrException.log(LOG, "Error while trying to recover", e);
+ }
+
+ } catch (Exception e) {
+ SolrException.log(LOG, "Error while trying to recover. core=" + coreName, e);
+ } finally {
+ if (successfulRecovery) {
+ LOG.info("Restaring background replicate from leader process");
+ zkController.startReplicationFromLeader(coreName, false);
+ LOG.info("Registering as Active after recovery.");
+ try {
+ zkController.publish(core.getCoreDescriptor(), Replica.State.ACTIVE);
+ } catch (Exception e) {
+ LOG.error("Could not publish as ACTIVE after succesful recovery", e);
+ successfulRecovery = false;
+ }
+
+ if (successfulRecovery) {
+ close = true;
+ recoveryListener.recovered();
+ }
+ }
+ }
+
+ if (!successfulRecovery) {
+ // lets pause for a moment and we need to try again...
+ // TODO: we don't want to retry for some problems?
+ // Or do a fall off retry...
+ try {
+
+ if (isClosed()) {
+ LOG.info("Recovery for core {} has been closed", core.getName());
+ break;
+ }
+
+ LOG.error("Recovery failed - trying again... (" + retries + ")");
+
+ retries++;
+ if (retries >= MAX_RETRIES) {
+ SolrException.log(LOG, "Recovery failed - max retries exceeded (" + retries + ").");
+ try {
+ recoveryFailed(core, zkController, baseUrl, coreZkNodeName, core.getCoreDescriptor());
+ } catch (Exception e) {
+ SolrException.log(LOG, "Could not publish that recovery failed", e);
+ }
+ break;
+ }
+ } catch (Exception e) {
+ SolrException.log(LOG, "An error has occurred during recovery", e);
+ }
+
+ try {
+ // Wait an exponential interval between retries, start at 5 seconds and work up to a minute.
+ // If we're at attempt >= 4, there's no point computing pow(2, retries) because the result
+ // will always be the minimum of the two (12). Since we sleep at 5 seconds sub-intervals in
+ // order to check if we were closed, 12 is chosen as the maximum loopCount (5s * 12 = 1m).
+ double loopCount = retries < 4 ? Math.min(Math.pow(2, retries), 12) : 12;
+ LOG.info("Wait [{}] seconds before trying to recover again (attempt={})", loopCount, retries);
+ for (int i = 0; i < loopCount; i++) {
+ if (isClosed()) {
+ LOG.info("Recovery for core {} has been closed", core.getName());
+ break; // check if someone closed us
+ }
+ Thread.sleep(STARTING_RECOVERY_DELAY);
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOG.warn("Recovery was interrupted.", e);
+ close = true;
+ }
+ }
+
+ }
+
+ // if replay was skipped (possibly to due pulling a full index from the leader),
+ // then we still need to update version bucket seeds after recovery
+ if (successfulRecovery) {
+ LOG.info("Updating version bucket highest from index after successful recovery.");
+ core.seedVersionBuckets();
+ }
+
+ LOG.info("Finished recovery process, successful=[{}]", Boolean.toString(successfulRecovery));
+ }
+
+ public boolean isClosed() {
+ return close;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/043d5cb2/solr/core/src/java/org/apache/solr/cloud/SplitShardCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/SplitShardCmd.java b/solr/core/src/java/org/apache/solr/cloud/SplitShardCmd.java
index 5a099e1..837af79 100644
--- a/solr/core/src/java/org/apache/solr/cloud/SplitShardCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/SplitShardCmd.java
@@ -385,7 +385,7 @@ public class SplitShardCmd implements Cmd {
Map<ReplicaAssigner.Position, String> nodeMap = ocmh.identifyNodes(clusterState,
new ArrayList<>(clusterState.getLiveNodes()),
new ZkNodeProps(collection.getProperties()),
- subSlices, repFactor - 1);
+ subSlices, repFactor - 1, 0, 0);
List<Map<String, Object>> replicas = new ArrayList<>((repFactor - 1) * 2);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/043d5cb2/solr/core/src/java/org/apache/solr/cloud/ZkController.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/ZkController.java b/solr/core/src/java/org/apache/solr/cloud/ZkController.java
index d7ae437..ce37de5 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ZkController.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ZkController.java
@@ -62,6 +62,7 @@ import org.apache.solr.common.cloud.DefaultZkCredentialsProvider;
import org.apache.solr.common.cloud.DocCollection;
import org.apache.solr.common.cloud.OnReconnect;
import org.apache.solr.common.cloud.Replica;
+import org.apache.solr.common.cloud.Replica.Type;
import org.apache.solr.common.cloud.Slice;
import org.apache.solr.common.cloud.SolrZkClient;
import org.apache.solr.common.cloud.ZkACLProvider;
@@ -86,7 +87,6 @@ import org.apache.solr.core.CoreContainer;
import org.apache.solr.core.CoreDescriptor;
import org.apache.solr.core.SolrCore;
import org.apache.solr.logging.MDCLoggingContext;
-import org.apache.solr.update.CommitUpdateCommand;
import org.apache.solr.update.UpdateLog;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
@@ -883,12 +883,17 @@ public class ZkController {
try {
// If we're a preferred leader, insert ourselves at the head of the queue
boolean joinAtHead = false;
- Replica replica = zkStateReader.getClusterState().getReplica(desc.getCloudDescriptor().getCollectionName(),
- coreZkNodeName);
+ Replica replica = zkStateReader.getClusterState().getReplica(collection, coreZkNodeName);
if (replica != null) {
joinAtHead = replica.getBool(SliceMutator.PREFERRED_LEADER_PROP, false);
}
- joinElection(desc, afterExpiration, joinAtHead);
+ //TODO WHy would replica be null?
+ if (replica == null || replica.getType() != Type.PASSIVE) {
+ joinElection(desc, afterExpiration, joinAtHead);
+ } else if (replica.getType() == Type.PASSIVE) {
+ log.debug("Replica {} skipping election because replica is passive", coreZkNodeName);
+ startReplicationFromLeader(coreName, false);
+ }
} catch (InterruptedException e) {
// Restore the interrupted status
Thread.currentThread().interrupt();
@@ -905,6 +910,7 @@ public class ZkController {
String ourUrl = ZkCoreNodeProps.getCoreUrl(baseUrl, coreName);
log.debug("We are " + ourUrl + " and leader is " + leaderUrl);
boolean isLeader = leaderUrl.equals(ourUrl);
+ assert !(isLeader && zkStateReader.getClusterState().getCollection(collection).getReplica(coreZkNodeName).getType() == Type.PASSIVE): "Passive replica became leader!";
try (SolrCore core = cc.getCore(desc.getName())) {
@@ -944,11 +950,12 @@ public class ZkController {
= checkRecovery(recoverReloadedCores, isLeader, skipRecovery, collection, coreZkNodeName, core, cc, afterExpiration);
if (!didRecovery) {
if (isReplicaInOnlyLeaderIndexes) {
- startReplicationFromLeader(coreName);
+ startReplicationFromLeader(coreName, true);
}
publish(desc, Replica.State.ACTIVE);
}
+
core.getCoreDescriptor().getCloudDescriptor().setHasRegistered(true);
}
@@ -960,14 +967,18 @@ public class ZkController {
}
}
- public void startReplicationFromLeader(String coreName) throws InterruptedException {
+ public void startReplicationFromLeader(String coreName, boolean switchTransactionLog) throws InterruptedException {
+ log.info(coreName + " starting replication from leader");
ReplicateFromLeader replicateFromLeader = new ReplicateFromLeader(cc, coreName);
if (replicateFromLeaders.putIfAbsent(coreName, replicateFromLeader) == null) {
- replicateFromLeader.startReplication();
+ replicateFromLeader.startReplication(switchTransactionLog);
+ } else {
+ log.warn("A replicate from leader instance already exists for core {}", coreName);
}
}
public void stopReplicationFromLeader(String coreName) {
+ log.info(coreName + " stopping replication from leader");
ReplicateFromLeader replicateFromLeader = replicateFromLeaders.remove(coreName);
if (replicateFromLeader != null) {
replicateFromLeader.stopReplication();
@@ -1189,6 +1200,7 @@ public class ZkController {
// If the leader initiated recovery, then verify that this replica has performed
// recovery as requested before becoming active; don't even look at lirState if going down
if (state != Replica.State.DOWN) {
+ //nocommit check for readOnly case
final Replica.State lirState = getLeaderInitiatedRecoveryState(collection, shardId, coreNodeName);
if (lirState != null) {
if (state == Replica.State.ACTIVE) {
@@ -1269,12 +1281,16 @@ public class ZkController {
assert false : "No collection was specified [" + collection + "]";
return;
}
+ Replica replica = zkStateReader.getClusterState().getReplica(collection, coreNodeName);
+
+ if (replica == null || replica.getType() != Type.PASSIVE) {
+ ElectionContext context = electionContexts.remove(new ContextKey(collection, coreNodeName));
- ElectionContext context = electionContexts.remove(new ContextKey(collection, coreNodeName));
-
- if (context != null) {
- context.cancelElection();
+ if (context != null) {
+ context.cancelElection();
+ }
}
+// //TODO: Do we need to stop replication for type==append?
CloudDescriptor cloudDescriptor = cd.getCloudDescriptor();
zkStateReader.unregisterCore(cloudDescriptor.getCollectionName());
@@ -1357,6 +1373,7 @@ public class ZkController {
final String shardId = zkStateReader.getClusterState().getShardId(cd.getCollectionName(), getNodeName(), cd.getName());
if (shardId != null) {
cd.getCloudDescriptor().setShardId(shardId);
+ log.debug("Shard ID is {} for core {} ", shardId, cd.getName());
return;
}
try {
@@ -1456,6 +1473,7 @@ public class ZkController {
return true;
}
errorMessage.set("coreNodeName " + coreNodeName + " exists, but does not match expected node or core name");
+ log.debug("coreNodeName " + coreNodeName + " exists, but does not match expected node or core name");
return false;
});
} catch (TimeoutException e) {
@@ -2410,11 +2428,9 @@ public class ZkController {
for (Slice slice : slices) {
Collection<Replica> replicas = slice.getReplicas();
- for (Replica replica : replicas) {
- if (replica.getName().equals(
- dcore.getCloudDescriptor().getCoreNodeName())) {
- return true;
- }
+ Replica r = slice.getReplica(dcore.getCloudDescriptor().getCoreNodeName());
+ if (r != null) {
+ return true;
}
}
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/043d5cb2/solr/core/src/java/org/apache/solr/cloud/overseer/ReplicaMutator.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/overseer/ReplicaMutator.java b/solr/core/src/java/org/apache/solr/cloud/overseer/ReplicaMutator.java
index f03eeeb..c467405 100644
--- a/solr/core/src/java/org/apache/solr/cloud/overseer/ReplicaMutator.java
+++ b/solr/core/src/java/org/apache/solr/cloud/overseer/ReplicaMutator.java
@@ -271,11 +271,12 @@ public class ReplicaMutator {
replicaProps.putAll(message.getProperties());
if (slice != null) {
- Replica oldReplica = slice.getReplicasMap().get(coreNodeName);
+ Replica oldReplica = slice.getReplica(coreNodeName);
if (oldReplica != null) {
if (oldReplica.containsKey(ZkStateReader.LEADER_PROP)) {
replicaProps.put(ZkStateReader.LEADER_PROP, oldReplica.get(ZkStateReader.LEADER_PROP));
}
+ replicaProps.put(ZkStateReader.REPLICA_TYPE, oldReplica.getType().toString());
// Move custom props over.
for (Map.Entry<String, Object> ent : oldReplica.getProperties().entrySet()) {
if (ent.getKey().startsWith(COLL_PROP_PREFIX)) {
@@ -311,6 +312,8 @@ public class ReplicaMutator {
Replica replica = new Replica(coreNodeName, replicaProps);
+
+ log.debug("Will update state for replica: " + replica);
Map<String, Object> sliceProps = null;
Map<String, Replica> replicas;
@@ -328,11 +331,11 @@ public class ReplicaMutator {
sliceProps.put(ZkStateReader.STATE_PROP, shardState);
sliceProps.put(Slice.PARENT, shardParent);
}
-
replicas.put(replica.getName(), replica);
slice = new Slice(sliceName, replicas, sliceProps);
DocCollection newCollection = CollectionMutator.updateSlice(collectionName, collection, slice);
+ log.debug("Collection is now: " + newCollection);
return new ZkWriteCommand(collectionName, newCollection);
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/043d5cb2/solr/core/src/java/org/apache/solr/cloud/overseer/SliceMutator.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/overseer/SliceMutator.java b/solr/core/src/java/org/apache/solr/cloud/overseer/SliceMutator.java
index ec2ce2e..160bf0c 100644
--- a/solr/core/src/java/org/apache/solr/cloud/overseer/SliceMutator.java
+++ b/solr/core/src/java/org/apache/solr/cloud/overseer/SliceMutator.java
@@ -16,14 +16,16 @@
*/
package org.apache.solr.cloud.overseer;
-import java.lang.invoke.MethodHandles;
+import static org.apache.solr.cloud.OverseerCollectionMessageHandler.COLL_PROP_PREFIX;
+import static org.apache.solr.cloud.overseer.CollectionMutator.checkCollectionKeyExistence;
+import static org.apache.solr.common.util.Utils.makeMap;
+import java.lang.invoke.MethodHandles;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;
-import com.google.common.collect.ImmutableSet;
import org.apache.solr.cloud.Assign;
import org.apache.solr.cloud.Overseer;
import org.apache.solr.common.cloud.ClusterState;
@@ -37,9 +39,7 @@ import org.apache.solr.common.cloud.ZkStateReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static org.apache.solr.cloud.OverseerCollectionMessageHandler.COLL_PROP_PREFIX;
-import static org.apache.solr.cloud.overseer.CollectionMutator.checkCollectionKeyExistence;
-import static org.apache.solr.common.util.Utils.makeMap;
+import com.google.common.collect.ImmutableSet;
public class SliceMutator {
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
@@ -66,14 +66,16 @@ public class SliceMutator {
log.error("Invalid Collection/Slice {}/{} ", coll, slice);
return ZkStateWriter.NO_OP;
}
-
String coreNodeName = Assign.assignNode(collection);
- Replica replica = new Replica(coreNodeName,
+// Replica replica = new Replica(coreNodeName,
+ // coreNodeName overlaps?
+ Replica replica = new Replica(message.getStr(ZkStateReader.CORE_NAME_PROP),
makeMap(
ZkStateReader.CORE_NAME_PROP, message.getStr(ZkStateReader.CORE_NAME_PROP),
ZkStateReader.BASE_URL_PROP, message.getStr(ZkStateReader.BASE_URL_PROP),
ZkStateReader.STATE_PROP, message.getStr(ZkStateReader.STATE_PROP),
- ZkStateReader.NODE_NAME_PROP, message.getStr(ZkStateReader.NODE_NAME_PROP)));
+ ZkStateReader.NODE_NAME_PROP, message.getStr(ZkStateReader.NODE_NAME_PROP),
+ ZkStateReader.REPLICA_TYPE, message.get(ZkStateReader.REPLICA_TYPE)));
return new ZkWriteCommand(coll, updateReplica(collection, sl, replica.getName(), replica));
}
@@ -248,13 +250,15 @@ public class SliceMutator {
}
public static DocCollection updateReplica(DocCollection collection, final Slice slice, String coreNodeName, final Replica replica) {
- Map<String, Replica> copy = slice.getReplicasCopy();
+ Map<String, Replica> replicasCopy = slice.getReplicasCopy();
if (replica == null) {
- copy.remove(coreNodeName);
+ replicasCopy.remove(coreNodeName);
} else {
- copy.put(replica.getName(), replica);
+ replicasCopy.put(replica.getName(), replica);
}
- Slice newSlice = new Slice(slice.getName(), copy, slice.getProperties());
+ Slice newSlice = new Slice(slice.getName(), replicasCopy, slice.getProperties());
+ log.info("Old Slice: " + slice);
+ log.info("New Slice: " + newSlice);
return CollectionMutator.updateSlice(collection.getName(), collection, newSlice);
}
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/043d5cb2/solr/core/src/java/org/apache/solr/cloud/overseer/ZkStateWriter.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/overseer/ZkStateWriter.java b/solr/core/src/java/org/apache/solr/cloud/overseer/ZkStateWriter.java
index 23fb56c..a906b86 100644
--- a/solr/core/src/java/org/apache/solr/cloud/overseer/ZkStateWriter.java
+++ b/solr/core/src/java/org/apache/solr/cloud/overseer/ZkStateWriter.java
@@ -268,6 +268,7 @@ public class ZkStateWriter {
}
}
+ log.debug("New Cluster State is: " + clusterState);
return clusterState;
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/043d5cb2/solr/core/src/java/org/apache/solr/cloud/rule/ReplicaAssigner.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/rule/ReplicaAssigner.java b/solr/core/src/java/org/apache/solr/cloud/rule/ReplicaAssigner.java
index 3eab8b4..3c46743 100644
--- a/solr/core/src/java/org/apache/solr/cloud/rule/ReplicaAssigner.java
+++ b/solr/core/src/java/org/apache/solr/cloud/rule/ReplicaAssigner.java
@@ -62,10 +62,14 @@ public class ReplicaAssigner {
public static class Position implements Comparable<Position> {
public final String shard;
public final int index;
+ public final Replica.Type type;
+ public final String suffix;
- public Position(String shard, int replicaIdx) {
+ public Position(String shard, int replicaIdx, Replica.Type type) {
this.shard = shard;
this.index = replicaIdx;
+ this.suffix = '_' + type.name().substring(0,1).toLowerCase();
+ this.type = type;
}
@Override
@@ -188,7 +192,7 @@ public class ReplicaAssigner {
List<Position> positions = new ArrayList<>();
for (int pos : p) {
for (int j = 0; j < shardVsReplicaCount.get(shardNames.get(pos)); j++) {
- positions.add(new Position(shardNames.get(pos), j));
+ positions.add(new Position(shardNames.get(pos), j, Replica.Type.REALTIME));
}
}
Collections.sort(positions);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/043d5cb2/solr/core/src/java/org/apache/solr/core/CoreContainer.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/core/CoreContainer.java b/solr/core/src/java/org/apache/solr/core/CoreContainer.java
index 0de671e..19f824a 100644
--- a/solr/core/src/java/org/apache/solr/core/CoreContainer.java
+++ b/solr/core/src/java/org/apache/solr/core/CoreContainer.java
@@ -16,6 +16,17 @@
*/
package org.apache.solr.core;
+import static java.util.Objects.requireNonNull;
+import static org.apache.solr.common.params.CommonParams.AUTHC_PATH;
+import static org.apache.solr.common.params.CommonParams.AUTHZ_PATH;
+import static org.apache.solr.common.params.CommonParams.COLLECTIONS_HANDLER_PATH;
+import static org.apache.solr.common.params.CommonParams.CONFIGSETS_HANDLER_PATH;
+import static org.apache.solr.common.params.CommonParams.CORES_HANDLER_PATH;
+import static org.apache.solr.common.params.CommonParams.INFO_HANDLER_PATH;
+import static org.apache.solr.common.params.CommonParams.METRICS_PATH;
+import static org.apache.solr.common.params.CommonParams.ZK_PATH;
+import static org.apache.solr.security.AuthenticationPlugin.AUTHENTICATION_PLUGIN_PROP;
+
import java.io.IOException;
import java.lang.invoke.MethodHandles;
import java.nio.file.Path;
@@ -36,9 +47,6 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
-import com.codahale.metrics.Gauge;
-import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.Maps;
import org.apache.http.auth.AuthSchemeProvider;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.config.Lookup;
@@ -57,6 +65,7 @@ import org.apache.solr.common.SolrException;
import org.apache.solr.common.SolrException.ErrorCode;
import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.cloud.Replica.State;
+import org.apache.solr.common.cloud.DocCollection;
import org.apache.solr.common.util.ExecutorUtil;
import org.apache.solr.common.util.IOUtils;
import org.apache.solr.common.util.Utils;
@@ -95,16 +104,9 @@ import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static java.util.Objects.requireNonNull;
-import static org.apache.solr.common.params.CommonParams.AUTHC_PATH;
-import static org.apache.solr.common.params.CommonParams.AUTHZ_PATH;
-import static org.apache.solr.common.params.CommonParams.COLLECTIONS_HANDLER_PATH;
-import static org.apache.solr.common.params.CommonParams.CONFIGSETS_HANDLER_PATH;
-import static org.apache.solr.common.params.CommonParams.CORES_HANDLER_PATH;
-import static org.apache.solr.common.params.CommonParams.INFO_HANDLER_PATH;
-import static org.apache.solr.common.params.CommonParams.METRICS_PATH;
-import static org.apache.solr.common.params.CommonParams.ZK_PATH;
-import static org.apache.solr.security.AuthenticationPlugin.AUTHENTICATION_PLUGIN_PROP;
+import com.codahale.metrics.Gauge;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Maps;
/**
*
@@ -1138,10 +1140,13 @@ public class CoreContainer {
SolrCore newCore = core.reload(coreConfig);
registerCore(cd.getName(), newCore, false, false);
if (getZkController() != null) {
- boolean onlyLeaderIndexes = getZkController().getClusterState().getCollection(cd.getCollectionName()).getRealtimeReplicas() == 1;
- if (onlyLeaderIndexes && !cd.getCloudDescriptor().isLeader()) {
+ DocCollection docCollection = getZkController().getClusterState().getCollection(cd.getCollectionName());
+ boolean onlyLeaderIndexes = docCollection.getRealtimeReplicas() == 1;
+ Replica replica = docCollection.getReplica(cd.getCloudDescriptor().getCoreNodeName());
+ assert replica != null;
+ if (onlyLeaderIndexes && !cd.getCloudDescriptor().isLeader()) { //TODO: needed here?
getZkController().stopReplicationFromLeader(core.getName());
- getZkController().startReplicationFromLeader(newCore.getName());
+ getZkController().startReplicationFromLeader(newCore.getName(), true);
}
}
} catch (SolrCoreState.CoreIsClosedException e) {
@@ -1228,6 +1233,10 @@ public class CoreContainer {
if (zkSys.getZkController() != null) {
// cancel recovery in cloud mode
core.getSolrCoreState().cancelRecovery();
+ if (core.getCoreDescriptor().getCloudDescriptor().getReplicaType() == Replica.Type.PASSIVE) { // TODO: Also for Replica.Type.ACTIVE?
+ // Stop replication if this is part of a passive replica before closing the code
+ zkSys.getZkController().stopReplicationFromLeader(name);
+ }
}
core.unloadOnClose(deleteIndexDir, deleteDataDir, deleteInstanceDir);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/043d5cb2/solr/core/src/java/org/apache/solr/handler/IndexFetcher.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/handler/IndexFetcher.java b/solr/core/src/java/org/apache/solr/handler/IndexFetcher.java
index a07496f..435ddbe 100644
--- a/solr/core/src/java/org/apache/solr/handler/IndexFetcher.java
+++ b/solr/core/src/java/org/apache/solr/handler/IndexFetcher.java
@@ -304,17 +304,28 @@ public class IndexFetcher {
// when we are a bit more confident we may want to try a partial replication
// if the error is connection related or something, but we have to be careful
forceReplication = true;
+ LOG.info("Last replication failed, so I'll force replication");
}
try {
if (fetchFromLeader) {
+ assert !solrCore.isClosed(): "Replication should be stopped before closing the core";
Replica replica = getLeaderReplica();
CloudDescriptor cd = solrCore.getCoreDescriptor().getCloudDescriptor();
if (cd.getCoreNodeName().equals(replica.getName())) {
return false;
}
- masterUrl = replica.getCoreUrl();
- LOG.info("Updated masterUrl to " + masterUrl);
+ if (replica.getState() != Replica.State.ACTIVE) {
+ LOG.info("Replica {} is leader but it's state is {}, skipping replication", replica.getName(), replica.getState());
+ return false;
+ }
+ if (!replica.getCoreUrl().equals(masterUrl)) {
+ masterUrl = replica.getCoreUrl();
+ LOG.info("Updated masterUrl to {}", masterUrl);
+ // TODO: Do we need to set forceReplication = true?
+ } else {
+ LOG.debug("masterUrl didn't change");
+ }
}
//get the current 'replicateable' index version in the master
NamedList response;
@@ -355,6 +366,7 @@ public class IndexFetcher {
if (forceReplication && commit.getGeneration() != 0) {
// since we won't get the files for an empty index,
// we just clear ours and commit
+ LOG.info("New index in Master. Deleting mine...");
RefCounted<IndexWriter> iw = solrCore.getUpdateHandler().getSolrCoreState().getIndexWriter(solrCore);
try {
iw.get().deleteAll();
@@ -367,6 +379,7 @@ public class IndexFetcher {
//there is nothing to be replicated
successfulInstall = true;
+ LOG.debug("Nothing to replicate, master's version is 0");
return true;
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/043d5cb2/solr/core/src/java/org/apache/solr/handler/admin/CollectionsHandler.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/handler/admin/CollectionsHandler.java b/solr/core/src/java/org/apache/solr/handler/admin/CollectionsHandler.java
index 2e17af6..a08131b 100644
--- a/solr/core/src/java/org/apache/solr/handler/admin/CollectionsHandler.java
+++ b/solr/core/src/java/org/apache/solr/handler/admin/CollectionsHandler.java
@@ -112,15 +112,7 @@ import static org.apache.solr.common.cloud.DocCollection.DOC_ROUTER;
import static org.apache.solr.common.cloud.DocCollection.RULE;
import static org.apache.solr.common.cloud.DocCollection.SNITCH;
import static org.apache.solr.common.cloud.DocCollection.STATE_FORMAT;
-import static org.apache.solr.common.cloud.ZkStateReader.AUTO_ADD_REPLICAS;
-import static org.apache.solr.common.cloud.ZkStateReader.COLLECTION_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.MAX_SHARDS_PER_NODE;
-import static org.apache.solr.common.cloud.ZkStateReader.REALTIME_REPLICAS;
-import static org.apache.solr.common.cloud.ZkStateReader.PROPERTY_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.PROPERTY_VALUE_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.REPLICATION_FACTOR;
-import static org.apache.solr.common.cloud.ZkStateReader.REPLICA_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.SHARD_ID_PROP;
+import static org.apache.solr.common.cloud.ZkStateReader.*;
import static org.apache.solr.common.params.CollectionAdminParams.COUNT_PROP;
import static org.apache.solr.common.params.CollectionParams.CollectionAction.*;
import static org.apache.solr.common.params.CommonAdminParams.ASYNC;
@@ -406,6 +398,7 @@ public class CollectionsHandler extends RequestHandlerBase implements Permission
AUTO_ADD_REPLICAS,
RULE,
SNITCH,
+ PASSIVE_REPLICAS,
REALTIME_REPLICAS);
if (props.get(STATE_FORMAT) == null) {
@@ -620,7 +613,8 @@ public class CollectionsHandler extends RequestHandlerBase implements Permission
_ROUTE_,
CoreAdminParams.NAME,
INSTANCE_DIR,
- DATA_DIR);
+ DATA_DIR,
+ REPLICA_TYPE);
return copyPropertiesWithPrefix(req.getParams(), props, COLL_PROP_PREFIX);
}),
OVERSEERSTATUS_OP(OVERSEERSTATUS, (req, rsp, h) -> (Map) new LinkedHashMap<>()),
@@ -1074,6 +1068,7 @@ public class CollectionsHandler extends RequestHandlerBase implements Permission
}
+ // TODO: Maybe add REALTIME_REPLICAS HERE?
public static final List<String> MODIFIABLE_COLL_PROPS = Arrays.asList(
RULE,
SNITCH,
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/043d5cb2/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminHandler.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminHandler.java b/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminHandler.java
index 275ec18..b5e0e73 100644
--- a/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminHandler.java
+++ b/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminHandler.java
@@ -171,7 +171,9 @@ public class CoreAdminHandler extends RequestHandlerBase implements PermissionNa
final CallInfo callInfo = new CallInfo(this, req, rsp, op);
if (taskId == null) {
+ log.info("Starting Operation: " + req);
callInfo.call();
+ log.info("Done with Operation: " + req);
} else {
try {
MDC.put("CoreAdminHandler.asyncId", taskId);
@@ -227,6 +229,7 @@ public class CoreAdminHandler extends RequestHandlerBase implements PermissionNa
.put(CoreAdminParams.ROLES, CoreDescriptor.CORE_ROLES)
.put(CoreAdminParams.CORE_NODE_NAME, CoreDescriptor.CORE_NODE_NAME)
.put(ZkStateReader.NUM_SHARDS_PROP, CloudDescriptor.NUM_SHARDS)
+ .put(CoreAdminParams.REPLICA_TYPE, CloudDescriptor.REPLICA_TYPE)
.build();
protected static Map<String, String> buildCoreParams(SolrParams params) {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/043d5cb2/solr/core/src/java/org/apache/solr/handler/component/RealTimeGetComponent.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/handler/component/RealTimeGetComponent.java b/solr/core/src/java/org/apache/solr/handler/component/RealTimeGetComponent.java
index 123abea..c60f520 100644
--- a/solr/core/src/java/org/apache/solr/handler/component/RealTimeGetComponent.java
+++ b/solr/core/src/java/org/apache/solr/handler/component/RealTimeGetComponent.java
@@ -103,6 +103,11 @@ public class RealTimeGetComponent extends SearchComponent
SolrQueryResponse rsp = rb.rsp;
SolrParams params = req.getParams();
+ if (req.getCore().getCoreDescriptor().getCloudDescriptor() != null
+ && !req.getCore().getCoreDescriptor().getCloudDescriptor().requiresTransactionLog()) {
+ return;
+ }
+
if (!params.getBool(COMPONENT_NAME, true)) {
return;
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/043d5cb2/solr/core/src/java/org/apache/solr/update/DefaultSolrCoreState.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/update/DefaultSolrCoreState.java b/solr/core/src/java/org/apache/solr/update/DefaultSolrCoreState.java
index 59e35f0..82adbdd 100644
--- a/solr/core/src/java/org/apache/solr/update/DefaultSolrCoreState.java
+++ b/solr/core/src/java/org/apache/solr/update/DefaultSolrCoreState.java
@@ -16,6 +16,7 @@
*/
package org.apache.solr.update;
+import java.io.Closeable;
import java.io.IOException;
import java.lang.invoke.MethodHandles;
import java.util.concurrent.ExecutionException;
@@ -32,8 +33,9 @@ import org.apache.lucene.index.MergePolicy;
import org.apache.lucene.search.Sort;
import org.apache.solr.cloud.ActionThrottle;
import org.apache.solr.cloud.RecoveryStrategy;
-import org.apache.solr.common.SolrException.ErrorCode;
+import org.apache.solr.cloud.ReplicateOnlyRecoveryStrategy;
import org.apache.solr.common.SolrException;
+import org.apache.solr.common.SolrException.ErrorCode;
import org.apache.solr.core.CoreContainer;
import org.apache.solr.core.CoreDescriptor;
import org.apache.solr.core.DirectoryFactory;
@@ -64,7 +66,8 @@ public final class DefaultSolrCoreState extends SolrCoreState implements Recover
private SolrIndexWriter indexWriter = null;
private DirectoryFactory directoryFactory;
- private volatile RecoveryStrategy recoveryStrat;
+// private volatile RecoveryStrategy recoveryStrat; //nocommit: Make interface
+ private volatile Thread recoveryStrat;
private volatile boolean lastReplicationSuccess = true;
@@ -310,9 +313,16 @@ public final class DefaultSolrCoreState extends SolrCoreState implements Recover
recoveryThrottle.minimumWaitBetweenActions();
recoveryThrottle.markAttemptingAction();
- recoveryStrat = new RecoveryStrategy(cc, cd, DefaultSolrCoreState.this);
- recoveryStrat.setRecoveringAfterStartup(recoveringAfterStartup);
- Future<?> future = cc.getUpdateShardHandler().getRecoveryExecutor().submit(recoveryStrat);
+ Future<?> future = null;
+ if (cd.getCloudDescriptor().requiresTransactionLog()) {
+ RecoveryStrategy recoveryStrat = new RecoveryStrategy(cc, cd, DefaultSolrCoreState.this);
+ recoveryStrat.setRecoveringAfterStartup(recoveringAfterStartup);
+ future = cc.getUpdateShardHandler().getRecoveryExecutor().submit(recoveryStrat);
+ } else {
+ ReplicateOnlyRecoveryStrategy recoveryStrat = new ReplicateOnlyRecoveryStrategy(cc, cd, DefaultSolrCoreState.this);
+ recoveryStrat.setRecoveringAfterStartup(recoveringAfterStartup);
+ future = cc.getUpdateShardHandler().getRecoveryExecutor().submit(recoveryStrat);
+ }
try {
future.get();
} catch (InterruptedException e) {
@@ -352,9 +362,11 @@ public final class DefaultSolrCoreState extends SolrCoreState implements Recover
public void cancelRecovery() {
if (recoveryStrat != null) {
try {
- recoveryStrat.close();
+ ((Closeable)recoveryStrat).close();
} catch (NullPointerException e) {
// okay
+ } catch (IOException e) {
+ // okay
}
}
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/043d5cb2/solr/core/src/java/org/apache/solr/update/DirectUpdateHandler2.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/update/DirectUpdateHandler2.java b/solr/core/src/java/org/apache/solr/update/DirectUpdateHandler2.java
index abb5512..31cb782 100644
--- a/solr/core/src/java/org/apache/solr/update/DirectUpdateHandler2.java
+++ b/solr/core/src/java/org/apache/solr/update/DirectUpdateHandler2.java
@@ -137,7 +137,8 @@ public class DirectUpdateHandler2 extends UpdateHandler implements SolrCoreState
}
public DirectUpdateHandler2(SolrCore core, UpdateHandler updateHandler) {
- super(core, updateHandler.getUpdateLog());
+// TODO super(core, updateHandler.getUpdateLog(), core.getCoreDescriptor().getCloudDescriptor().requiresTransactionLog());
+ super(core, updateHandler.getUpdateLog(), true);
solrCoreState = core.getSolrCoreState();
UpdateHandlerInfo updateHandlerInfo = core.getSolrConfig()
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/043d5cb2/solr/core/src/java/org/apache/solr/update/UpdateHandler.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/update/UpdateHandler.java b/solr/core/src/java/org/apache/solr/update/UpdateHandler.java
index 1cf8a3f..f08bb17 100644
--- a/solr/core/src/java/org/apache/solr/update/UpdateHandler.java
+++ b/solr/core/src/java/org/apache/solr/update/UpdateHandler.java
@@ -91,10 +91,16 @@ public abstract class UpdateHandler implements SolrInfoMBean {
}
public UpdateHandler(SolrCore core) {
- this(core, null);
+// this(core, null, core.getCoreDescriptor().getCloudDescriptor().requiresTransactionLog());//nocommit, handle non-cloud case
+ this(core, null, true);//nocommit, handle non-cloud case
}
-
+
public UpdateHandler(SolrCore core, UpdateLog updateLog) {
+// this(core, updateLog, core.getCoreDescriptor().getCloudDescriptor().requiresTransactionLog());//nocommit, handle non-cloud case
+ this(core, updateLog, true);//nocommit, handle non-cloud case
+ }
+
+ public UpdateHandler(SolrCore core, UpdateLog updateLog, boolean needsUpdateLog) {
this.core=core;
idField = core.getLatestSchema().getUniqueKeyField();
idFieldType = idField!=null ? idField.getType() : null;
@@ -102,7 +108,7 @@ public abstract class UpdateHandler implements SolrInfoMBean {
PluginInfo ulogPluginInfo = core.getSolrConfig().getPluginInfo(UpdateLog.class.getName());
- if (updateLog == null && ulogPluginInfo != null && ulogPluginInfo.isEnabled()) {
+ if (updateLog == null && ulogPluginInfo != null && ulogPluginInfo.isEnabled() && needsUpdateLog) {
String dataDir = (String)ulogPluginInfo.initArgs.get("dir");
String ulogDir = core.getCoreDescriptor().getUlogDir();
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/043d5cb2/solr/core/src/java/org/apache/solr/update/processor/DistributedUpdateProcessor.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/update/processor/DistributedUpdateProcessor.java b/solr/core/src/java/org/apache/solr/update/processor/DistributedUpdateProcessor.java
index 08ede72..18d0ff9 100644
--- a/solr/core/src/java/org/apache/solr/update/processor/DistributedUpdateProcessor.java
+++ b/solr/core/src/java/org/apache/solr/update/processor/DistributedUpdateProcessor.java
@@ -24,6 +24,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
+import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -1885,7 +1886,7 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
zkCheck();
nodes = getCollectionUrls(req, req.getCore().getCoreDescriptor()
- .getCloudDescriptor().getCollectionName());
+ .getCloudDescriptor().getCollectionName(), EnumSet.of(Replica.Type.APPEND,Replica.Type.REALTIME));
if (isLeader && nodes.size() == 1) {
singleLeader = true;
}
@@ -1905,7 +1906,7 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
doLocalCommit(cmd);
} else {
assert TestInjection.waitForInSyncWithLeader(req.getCore(),
- zkController, collection, cloudDesc.getShardId());
+ zkController, collection, cloudDesc.getShardId()): "Core " + req.getCore() + " not in sync with leader";
}
} catch (InterruptedException e) {
throw new SolrException(ErrorCode.SERVICE_UNAVAILABLE, "Exception finding leader for shard " + cloudDesc.getShardId(), e);
@@ -1959,7 +1960,7 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
- private List<Node> getCollectionUrls(SolrQueryRequest req, String collection) {
+ private List<Node> getCollectionUrls(SolrQueryRequest req, String collection, EnumSet<Replica.Type> types) {
ClusterState clusterState = req.getCore().getCoreDescriptor()
.getCoreContainer().getZkController().getClusterState();
Map<String,Slice> slices = clusterState.getSlicesMap(collection);
@@ -1974,6 +1975,10 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
Map<String,Replica> shardMap = replicas.getReplicasMap();
for (Entry<String,Replica> entry : shardMap.entrySet()) {
+ if (!types.contains(entry.getValue().getType())) {
+ log.info("getCollectionUrls: Skipping replica " + entry.getValue().getName());//nocommit: too verbose
+ continue;
+ }
ZkCoreNodeProps nodeProps = new ZkCoreNodeProps(entry.getValue());
if (clusterState.liveNodesContain(nodeProps.getNodeName())) {
urls.add(new StdNode(nodeProps, collection, replicas.getName()));
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/043d5cb2/solr/core/src/test-files/solr/configsets/cloud-minimal/conf/solrconfig.xml
----------------------------------------------------------------------
diff --git a/solr/core/src/test-files/solr/configsets/cloud-minimal/conf/solrconfig.xml b/solr/core/src/test-files/solr/configsets/cloud-minimal/conf/solrconfig.xml
index 059e58f..8da7d28 100644
--- a/solr/core/src/test-files/solr/configsets/cloud-minimal/conf/solrconfig.xml
+++ b/solr/core/src/test-files/solr/configsets/cloud-minimal/conf/solrconfig.xml
@@ -33,7 +33,7 @@
<commitWithin>
<softCommit>${solr.commitwithin.softcommit:true}</softCommit>
</commitWithin>
- <updateLog></updateLog>
+ <updateLog class="${solr.ulog:solr.UpdateLog}"></updateLog>
</updateHandler>
<requestHandler name="/select" class="solr.SearchHandler">