You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by sh...@apache.org on 2018/10/03 11:06:56 UTC
[1/5] lucene-solr:branch_7x: SOLR-9317: ADDREPLICA command should be
able to add more than one replica to a collection, shard at a time.
Repository: lucene-solr
Updated Branches:
refs/heads/branch_7x 29a99a69a -> 5d3ceed43
SOLR-9317: ADDREPLICA command should be able to add more than one replica to a collection,shard at a time.
The API now supports 'nrtReplicas', 'tlogReplicas', 'pullReplicas' parameters as well 'createNodeSet' parameter. As part of this change, the CREATESHARD API now delegates placing replicas entirely to the ADDREPLICA command and uses the new parameters to add all the replicas in one API call.
(cherry picked from commit 4bcace571ee1e512b2ca4aa3d93bc7bd522b55fe)
Project: http://git-wip-us.apache.org/repos/asf/lucene-solr/repo
Commit: http://git-wip-us.apache.org/repos/asf/lucene-solr/commit/0d7627f2
Tree: http://git-wip-us.apache.org/repos/asf/lucene-solr/tree/0d7627f2
Diff: http://git-wip-us.apache.org/repos/asf/lucene-solr/diff/0d7627f2
Branch: refs/heads/branch_7x
Commit: 0d7627f29bfb72a693f46a5ba23ed8c5ceb3e089
Parents: 29a99a6
Author: Shalin Shekhar Mangar <sh...@apache.org>
Authored: Fri Sep 21 15:12:21 2018 +0530
Committer: Shalin Shekhar Mangar <sh...@apache.org>
Committed: Wed Oct 3 15:56:18 2018 +0530
----------------------------------------------------------------------
solr/CHANGES.txt | 5 +
.../cloud/api/collections/AddReplicaCmd.java | 334 +++++++++++++------
.../solr/cloud/api/collections/Assign.java | 22 +-
.../cloud/api/collections/CreateShardCmd.java | 157 +++------
.../cloud/api/collections/MoveReplicaCmd.java | 2 +-
.../OverseerCollectionMessageHandler.java | 2 +-
.../cloud/api/collections/ReplaceNodeCmd.java | 2 +-
.../solr/handler/admin/CollectionsHandler.java | 9 +-
.../org/apache/solr/cloud/AddReplicaTest.java | 90 ++++-
.../CollectionTooManyReplicasTest.java | 2 +-
.../cloud/api/collections/ShardSplitTest.java | 2 +-
.../sim/SimClusterStateProvider.java | 93 ++++--
.../autoscaling/sim/TestSimPolicyCloud.java | 2 +-
solr/solr-ref-guide/src/collections-api.adoc | 68 +++-
.../solrj/request/CollectionAdminRequest.java | 53 +++
15 files changed, 557 insertions(+), 286 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/0d7627f2/solr/CHANGES.txt
----------------------------------------------------------------------
diff --git a/solr/CHANGES.txt b/solr/CHANGES.txt
index bb0846c..b14e9ad 100644
--- a/solr/CHANGES.txt
+++ b/solr/CHANGES.txt
@@ -35,6 +35,11 @@ New Features
error. Previously, the collapsing behavior was unreliable and undefined despite no explicit error.
(Munendra S N, David Smiley)
+* SOLR-9317: ADDREPLICA command should be able to add more than one replica to a collection,shard at a time.
+ The API now supports 'nrtReplicas', 'tlogReplicas', 'pullReplicas' parameters as well 'createNodeSet' parameter.
+ As part of this change, the CREATESHARD API now delegates placing replicas entirely to the ADDREPLICA command
+ and uses the new parameters to add all the replicas in one API call. (shalin)
+
* SOLR-11522: /autoscaling/suggestions now include rebalance options as well even if there are no violations (noble)
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/0d7627f2/solr/core/src/java/org/apache/solr/cloud/api/collections/AddReplicaCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/api/collections/AddReplicaCmd.java b/solr/core/src/java/org/apache/solr/cloud/api/collections/AddReplicaCmd.java
index c9dbaec..f128c2e 100644
--- a/solr/core/src/java/org/apache/solr/cloud/api/collections/AddReplicaCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/api/collections/AddReplicaCmd.java
@@ -20,14 +20,17 @@ package org.apache.solr.cloud.api.collections;
import java.io.IOException;
import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
+import java.util.EnumMap;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
import org.apache.commons.lang.StringUtils;
import org.apache.solr.client.solrj.cloud.SolrCloudManager;
@@ -41,6 +44,7 @@ import org.apache.solr.common.SolrException;
import org.apache.solr.common.cloud.ClusterState;
import org.apache.solr.common.cloud.DocCollection;
import org.apache.solr.common.cloud.Replica;
+import org.apache.solr.common.cloud.ReplicaPosition;
import org.apache.solr.common.cloud.Slice;
import org.apache.solr.common.cloud.ZkNodeProps;
import org.apache.solr.common.cloud.ZkStateReader;
@@ -54,10 +58,14 @@ import org.apache.solr.handler.component.ShardHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import static org.apache.solr.cloud.api.collections.OverseerCollectionMessageHandler.CREATE_NODE_SET;
import static org.apache.solr.cloud.api.collections.OverseerCollectionMessageHandler.SKIP_CREATE_REPLICA_IN_CLUSTER_STATE;
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.NRT_REPLICAS;
+import static org.apache.solr.common.cloud.ZkStateReader.PULL_REPLICAS;
import static org.apache.solr.common.cloud.ZkStateReader.SHARD_ID_PROP;
+import static org.apache.solr.common.cloud.ZkStateReader.TLOG_REPLICAS;
import static org.apache.solr.common.params.CollectionAdminParams.COLL_CONF;
import static org.apache.solr.common.params.CollectionAdminParams.WITH_COLLECTION;
import static org.apache.solr.common.params.CollectionParams.CollectionAction.ADDREPLICA;
@@ -79,28 +87,116 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
addReplica(state, message, results, null);
}
- ZkNodeProps addReplica(ClusterState clusterState, ZkNodeProps message, NamedList results, Runnable onComplete)
+ List<ZkNodeProps> addReplica(ClusterState clusterState, ZkNodeProps message, NamedList results, Runnable onComplete)
throws IOException, InterruptedException {
log.debug("addReplica() : {}", Utils.toJSONString(message));
String collectionName = message.getStr(COLLECTION_PROP);
+ String shard = message.getStr(SHARD_ID_PROP);
+
DocCollection coll = clusterState.getCollection(collectionName);
+ if (coll == null) {
+ throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "Collection: " + collectionName + " does not exist");
+ }
+ if (coll.getSlice(shard) == null) {
+ throw new SolrException(SolrException.ErrorCode.BAD_REQUEST,
+ "Collection: " + collectionName + " shard: " + shard + " does not exist");
+ }
boolean waitForFinalState = message.getBool(WAIT_FOR_FINAL_STATE, false);
boolean skipCreateReplicaInClusterState = message.getBool(SKIP_CREATE_REPLICA_IN_CLUSTER_STATE, false);
final String asyncId = message.getStr(ASYNC);
- AtomicReference<PolicyHelper.SessionWrapper> sessionWrapper = new AtomicReference<>();
- message = assignReplicaDetails(ocmh.cloudManager, clusterState, message, sessionWrapper);
-
String node = message.getStr(CoreAdminParams.NODE);
- String shard = message.getStr(SHARD_ID_PROP);
- String coreName = message.getStr(CoreAdminParams.NAME);
- String coreNodeName = message.getStr(CoreAdminParams.CORE_NODE_NAME);
+ String createNodeSetStr = message.getStr(CREATE_NODE_SET);
+
+ if (node != null && createNodeSetStr != null) {
+ throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "Both 'node' and 'createNodeSet' parameters cannot be specified together.");
+ }
+
int timeout = message.getInt(TIMEOUT, 10 * 60); // 10 minutes
- Replica.Type replicaType = Replica.Type.valueOf(message.getStr(ZkStateReader.REPLICA_TYPE, Replica.Type.NRT.name()).toUpperCase(Locale.ROOT));
boolean parallel = message.getBool("parallel", false);
+ Replica.Type replicaType = Replica.Type.valueOf(message.getStr(ZkStateReader.REPLICA_TYPE, Replica.Type.NRT.name()).toUpperCase(Locale.ROOT));
+ EnumMap<Replica.Type, Integer> replicaTypesVsCount = new EnumMap<>(Replica.Type.class);
+ replicaTypesVsCount.put(Replica.Type.NRT, message.getInt(NRT_REPLICAS, replicaType == Replica.Type.NRT ? 1 : 0));
+ replicaTypesVsCount.put(Replica.Type.TLOG, message.getInt(TLOG_REPLICAS, replicaType == Replica.Type.TLOG ? 1 : 0));
+ replicaTypesVsCount.put(Replica.Type.PULL, message.getInt(PULL_REPLICAS, replicaType == Replica.Type.PULL ? 1 : 0));
+
+ int totalReplicas = 0;
+ for (Map.Entry<Replica.Type, Integer> entry : replicaTypesVsCount.entrySet()) {
+ totalReplicas += entry.getValue();
+ }
+ if (totalReplicas > 1) {
+ if (message.getStr(CoreAdminParams.NAME) != null) {
+ throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "Cannot create " + totalReplicas + " replicas if 'name' parameter is specified");
+ }
+ if (message.getStr(CoreAdminParams.CORE_NODE_NAME) != null) {
+ throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "Cannot create " + totalReplicas + " replicas if 'coreNodeName' parameter is specified");
+ }
+ }
+
+ AtomicReference<PolicyHelper.SessionWrapper> sessionWrapper = new AtomicReference<>();
+ List<CreateReplica> createReplicas = buildReplicaPositions(ocmh.cloudManager, clusterState, collectionName, message, replicaTypesVsCount, sessionWrapper)
+ .stream()
+ .map(replicaPosition -> assignReplicaDetails(ocmh.cloudManager, clusterState, message, replicaPosition))
+ .collect(Collectors.toList());
+
+ ShardHandler shardHandler = ocmh.shardHandlerFactory.getShardHandler();
+ ZkStateReader zkStateReader = ocmh.zkStateReader;
+ // For tracking async calls.
+ Map<String,String> requestMap = new HashMap<>();
+
+ for (CreateReplica createReplica : createReplicas) {
+ assert createReplica.coreName != null;
+ ModifiableSolrParams params = getReplicaParams(clusterState, message, results, collectionName, coll, skipCreateReplicaInClusterState, asyncId, shardHandler, createReplica);
+ ocmh.sendShardRequest(createReplica.node, params, shardHandler, asyncId, requestMap);
+ }
+
+ Runnable runnable = () -> {
+ ocmh.processResponses(results, shardHandler, true, "ADDREPLICA failed to create replica", asyncId, requestMap);
+ for (CreateReplica replica : createReplicas) {
+ ocmh.waitForCoreNodeName(collectionName, replica.node, replica.coreName);
+ }
+
+ if (sessionWrapper.get() != null) {
+ sessionWrapper.get().release();
+ }
+ if (onComplete != null) onComplete.run();
+ };
+
+ if (!parallel || waitForFinalState) {
+ if (waitForFinalState) {
+ SolrCloseableLatch latch = new SolrCloseableLatch(totalReplicas, ocmh);
+ ActiveReplicaWatcher watcher = new ActiveReplicaWatcher(collectionName, null,
+ createReplicas.stream().map(createReplica -> createReplica.coreName).collect(Collectors.toList()), latch);
+ try {
+ zkStateReader.registerCollectionStateWatcher(collectionName, watcher);
+ runnable.run();
+ if (!latch.await(timeout, TimeUnit.SECONDS)) {
+ throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "Timeout waiting " + timeout + " seconds for replica to become active.");
+ }
+ } finally {
+ zkStateReader.removeCollectionStateWatcher(collectionName, watcher);
+ }
+ } else {
+ runnable.run();
+ }
+ } else {
+ ocmh.tpe.submit(runnable);
+ }
+
+ return createReplicas.stream()
+ .map(createReplica -> new ZkNodeProps(
+ ZkStateReader.COLLECTION_PROP, createReplica.collectionName,
+ ZkStateReader.SHARD_ID_PROP, createReplica.sliceName,
+ ZkStateReader.CORE_NAME_PROP, createReplica.coreName,
+ ZkStateReader.NODE_NAME_PROP, createReplica.node
+ ))
+ .collect(Collectors.toList());
+ }
+
+ private ModifiableSolrParams getReplicaParams(ClusterState clusterState, ZkNodeProps message, NamedList results, String collectionName, DocCollection coll, boolean skipCreateReplicaInClusterState, String asyncId, ShardHandler shardHandler, CreateReplica createReplica) throws IOException, InterruptedException {
if (coll.getStr(WITH_COLLECTION) != null) {
String withCollectionName = coll.getStr(WITH_COLLECTION);
DocCollection withCollection = clusterState.getCollection(withCollectionName);
@@ -109,14 +205,14 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
}
String withCollectionShard = withCollection.getActiveSlices().iterator().next().getName();
- List<Replica> replicas = withCollection.getReplicas(node);
+ List<Replica> replicas = withCollection.getReplicas(createReplica.node);
if (replicas == null || replicas.isEmpty()) {
// create a replica of withCollection on the identified node before proceeding further
ZkNodeProps props = new ZkNodeProps(
Overseer.QUEUE_OPERATION, ADDREPLICA.toString(),
ZkStateReader.COLLECTION_PROP, withCollectionName,
ZkStateReader.SHARD_ID_PROP, withCollectionShard,
- "node", node,
+ "node", createReplica.node,
CommonAdminParams.WAIT_FOR_FINAL_STATE, Boolean.TRUE.toString()); // set to true because we want `withCollection` to be ready after this collection is created
addReplica(clusterState, props, results, null);
}
@@ -130,14 +226,14 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
ZkNodeProps props = new ZkNodeProps(
Overseer.QUEUE_OPERATION, ADDREPLICA.toLower(),
ZkStateReader.COLLECTION_PROP, collectionName,
- ZkStateReader.SHARD_ID_PROP, shard,
- ZkStateReader.CORE_NAME_PROP, coreName,
+ ZkStateReader.SHARD_ID_PROP, createReplica.sliceName,
+ ZkStateReader.CORE_NAME_PROP, createReplica.coreName,
ZkStateReader.STATE_PROP, Replica.State.DOWN.toString(),
- ZkStateReader.BASE_URL_PROP, zkStateReader.getBaseUrlForNodeName(node),
- ZkStateReader.NODE_NAME_PROP, node,
- ZkStateReader.REPLICA_TYPE, replicaType.name());
- if (coreNodeName != null) {
- props = props.plus(ZkStateReader.CORE_NODE_NAME_PROP, coreNodeName);
+ ZkStateReader.BASE_URL_PROP, zkStateReader.getBaseUrlForNodeName(createReplica.node),
+ ZkStateReader.NODE_NAME_PROP, createReplica.node,
+ ZkStateReader.REPLICA_TYPE, createReplica.replicaType.name());
+ if (createReplica.coreNodeName != null) {
+ props = props.plus(ZkStateReader.CORE_NODE_NAME_PROP, createReplica.coreNodeName);
}
try {
Overseer.getStateUpdateQueue(zkStateReader.getZkClient()).offer(Utils.toJSON(props));
@@ -146,7 +242,7 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
}
}
params.set(CoreAdminParams.CORE_NODE_NAME,
- ocmh.waitToSeeReplicasInState(collectionName, Collections.singletonList(coreName)).get(coreName).getName());
+ ocmh.waitToSeeReplicasInState(collectionName, Collections.singletonList(createReplica.coreName)).get(createReplica.coreName).getName());
}
String configName = zkStateReader.readConfigName(collectionName);
@@ -156,12 +252,12 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
String instanceDir = message.getStr(CoreAdminParams.INSTANCE_DIR);
params.set(CoreAdminParams.ACTION, CoreAdminParams.CoreAdminAction.CREATE.toString());
- params.set(CoreAdminParams.NAME, coreName);
+ params.set(CoreAdminParams.NAME, createReplica.coreName);
params.set(COLL_CONF, configName);
params.set(CoreAdminParams.COLLECTION, collectionName);
- params.set(CoreAdminParams.REPLICA_TYPE, replicaType.name());
- if (shard != null) {
- params.set(CoreAdminParams.SHARD, shard);
+ params.set(CoreAdminParams.REPLICA_TYPE, createReplica.replicaType.name());
+ if (createReplica.sliceName != null) {
+ params.set(CoreAdminParams.SHARD, createReplica.sliceName);
} else if (routeKey != null) {
Collection<Slice> slices = coll.getRouter().getSearchSlicesSingle(routeKey, null, coll);
if (slices.isEmpty()) {
@@ -181,108 +277,34 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
if (instanceDir != null) {
params.set(CoreAdminParams.INSTANCE_DIR, instanceDir);
}
- if (coreNodeName != null) {
- params.set(CoreAdminParams.CORE_NODE_NAME, coreNodeName);
+ if (createReplica.coreNodeName != null) {
+ params.set(CoreAdminParams.CORE_NODE_NAME, createReplica.coreNodeName);
}
ocmh.addPropertyParams(message, params);
- // For tracking async calls.
- Map<String,String> requestMap = new HashMap<>();
- ShardHandler shardHandler = ocmh.shardHandlerFactory.getShardHandler();
-
- ocmh.sendShardRequest(node, params, shardHandler, asyncId, requestMap);
-
- final String fnode = node;
- final String fcoreName = coreName;
-
- Runnable runnable = () -> {
- ocmh.processResponses(results, shardHandler, true, "ADDREPLICA failed to create replica", asyncId, requestMap);
- ocmh.waitForCoreNodeName(collectionName, fnode, fcoreName);
- if (sessionWrapper.get() != null) {
- sessionWrapper.get().release();
- }
- if (onComplete != null) onComplete.run();
- };
-
- if (!parallel || waitForFinalState) {
- if (waitForFinalState) {
- SolrCloseableLatch latch = new SolrCloseableLatch(1, ocmh);
- ActiveReplicaWatcher watcher = new ActiveReplicaWatcher(collectionName, null, Collections.singletonList(coreName), latch);
- try {
- zkStateReader.registerCollectionStateWatcher(collectionName, watcher);
- runnable.run();
- if (!latch.await(timeout, TimeUnit.SECONDS)) {
- throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "Timeout waiting " + timeout + " seconds for replica to become active.");
- }
- } finally {
- zkStateReader.removeCollectionStateWatcher(collectionName, watcher);
- }
- } else {
- runnable.run();
- }
- } else {
- ocmh.tpe.submit(runnable);
- }
-
-
- return new ZkNodeProps(
- ZkStateReader.COLLECTION_PROP, collectionName,
- ZkStateReader.SHARD_ID_PROP, shard,
- ZkStateReader.CORE_NAME_PROP, coreName,
- ZkStateReader.NODE_NAME_PROP, node
- );
+ return params;
}
- public static ZkNodeProps assignReplicaDetails(SolrCloudManager cloudManager, ClusterState clusterState,
- ZkNodeProps message, AtomicReference<PolicyHelper.SessionWrapper> sessionWrapper) throws IOException, InterruptedException {
+ public static CreateReplica assignReplicaDetails(SolrCloudManager cloudManager, ClusterState clusterState,
+ ZkNodeProps message, ReplicaPosition replicaPosition) {
boolean skipCreateReplicaInClusterState = message.getBool(SKIP_CREATE_REPLICA_IN_CLUSTER_STATE, false);
String collection = message.getStr(COLLECTION_PROP);
- String node = message.getStr(CoreAdminParams.NODE);
+ String node = replicaPosition.node;
String shard = message.getStr(SHARD_ID_PROP);
String coreName = message.getStr(CoreAdminParams.NAME);
String coreNodeName = message.getStr(CoreAdminParams.CORE_NODE_NAME);
- Replica.Type replicaType = Replica.Type.valueOf(message.getStr(ZkStateReader.REPLICA_TYPE, Replica.Type.NRT.name()).toUpperCase(Locale.ROOT));
+ Replica.Type replicaType = replicaPosition.type;
+
if (StringUtils.isBlank(coreName)) {
coreName = message.getStr(CoreAdminParams.PROPERTY_PREFIX + CoreAdminParams.NAME);
}
- DocCollection coll = clusterState.getCollection(collection);
- if (coll == null) {
- throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "Collection: " + collection + " does not exist");
- }
- if (coll.getSlice(shard) == null) {
- throw new SolrException(SolrException.ErrorCode.BAD_REQUEST,
- "Collection: " + collection + " shard: " + shard + " does not exist");
- }
-
- // Kind of unnecessary, but it does put the logic of whether to override maxShardsPerNode in one place.
- if (!skipCreateReplicaInClusterState) {
- if (CloudUtil.usePolicyFramework(coll, cloudManager)) {
- if (node == null) {
- if(coll.getPolicyName() != null) message.getProperties().put(Policy.POLICY, coll.getPolicyName());
- node = Assign.identifyNodes(cloudManager,
- clusterState,
- Collections.emptyList(),
- collection,
- message,
- Collections.singletonList(shard),
- replicaType == Replica.Type.NRT ? 1 : 0,
- replicaType == Replica.Type.TLOG ? 1 : 0,
- replicaType == Replica.Type.PULL ? 1 : 0
- ).get(0).node;
- sessionWrapper.set(PolicyHelper.getLastSessionWrapper(true));
- }
- } else {
- node = Assign.getNodesForNewReplicas(clusterState, collection, shard, 1, node,
- cloudManager).get(0).nodeName;// TODO: use replica type in this logic too
- }
- }
log.info("Node Identified {} for creating new replica of shard {}", node, shard);
-
if (!clusterState.liveNodesContain(node)) {
throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "Node: " + node + " is not live");
}
+ DocCollection coll = clusterState.getCollection(collection);
if (coreName == null) {
coreName = Assign.buildSolrCoreName(cloudManager.getDistribStateManager(), coll, shard, replicaType);
} else if (!skipCreateReplicaInClusterState) {
@@ -297,11 +319,103 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
}
}
}
- if (coreNodeName != null) {
- message = message.plus(CoreAdminParams.CORE_NODE_NAME, coreNodeName);
+ return new CreateReplica(collection, shard, node, replicaType, coreName, coreNodeName);
+ }
+
+ public static List<ReplicaPosition> buildReplicaPositions(SolrCloudManager cloudManager, ClusterState clusterState,
+ String collectionName, ZkNodeProps message,
+ EnumMap<Replica.Type, Integer> replicaTypeVsCount,
+ AtomicReference< PolicyHelper.SessionWrapper> sessionWrapper) throws IOException, InterruptedException {
+ boolean skipCreateReplicaInClusterState = message.getBool(SKIP_CREATE_REPLICA_IN_CLUSTER_STATE, false);
+ String sliceName = message.getStr(SHARD_ID_PROP);
+ DocCollection collection = clusterState.getCollection(collectionName);
+
+ int numNrtReplicas = replicaTypeVsCount.get(Replica.Type.NRT);
+ int numPullReplicas = replicaTypeVsCount.get(Replica.Type.PULL);
+ int numTlogReplicas = replicaTypeVsCount.get(Replica.Type.TLOG);
+ int totalReplicas = numNrtReplicas + numPullReplicas + numTlogReplicas;
+
+ String node = message.getStr(CoreAdminParams.NODE);
+ Object createNodeSetStr = message.get(OverseerCollectionMessageHandler.CREATE_NODE_SET);
+ if (createNodeSetStr == null) {
+ if (node != null) {
+ message.getProperties().put(OverseerCollectionMessageHandler.CREATE_NODE_SET, node);
+ createNodeSetStr = node;
+ }
}
- message = message.plus(CoreAdminParams.NAME, coreName);
- message = message.plus(CoreAdminParams.NODE, node);
- return message;
+
+ List<ReplicaPosition> positions = null;
+ if (!skipCreateReplicaInClusterState) {
+ if (CloudUtil.usePolicyFramework(collection, cloudManager)) {
+ if (node == null) {
+ if (collection.getPolicyName() != null) message.getProperties().put(Policy.POLICY, collection.getPolicyName());
+ positions = Assign.identifyNodes(cloudManager,
+ clusterState,
+ Assign.getLiveOrLiveAndCreateNodeSetList(clusterState.getLiveNodes(), message, OverseerCollectionMessageHandler.RANDOM),
+ collection.getName(),
+ message,
+ Collections.singletonList(sliceName),
+ numNrtReplicas,
+ numTlogReplicas,
+ numPullReplicas);
+ sessionWrapper.set(PolicyHelper.getLastSessionWrapper(true));
+ }
+ } else {
+ List<Assign.ReplicaCount> sortedNodeList = Assign.getNodesForNewReplicas(clusterState, collection.getName(), sliceName, numNrtReplicas,
+ numTlogReplicas, numPullReplicas, createNodeSetStr, cloudManager);
+ int i = 0;
+ positions = new ArrayList<>();
+ for (Map.Entry<Replica.Type, Integer> e : replicaTypeVsCount.entrySet()) {
+ for (int j = 0; j < e.getValue(); j++) {
+ positions.add(new ReplicaPosition(sliceName, j + 1, e.getKey(), sortedNodeList.get(i % sortedNodeList.size()).nodeName));
+ i++;
+ }
+ }
+ }
+ }
+
+ if (positions == null) {
+ assert node != null;
+ if (node == null) {
+ // in case asserts are disabled
+ throw new SolrException(SolrException.ErrorCode.SERVER_ERROR,
+ "A node should have been identified to add replica but wasn't. Please inform solr developers at SOLR-9317");
+ }
+ // it is unlikely that multiple replicas have been requested to be created on
+ // the same node, but we've got to accommodate.
+ positions = new ArrayList<>(totalReplicas);
+ int i = 0;
+ for (Map.Entry<Replica.Type, Integer> entry : replicaTypeVsCount.entrySet()) {
+ for (int j = 0; j < entry.getValue(); j++) {
+ positions.add(new ReplicaPosition(sliceName, i++, entry.getKey(), node));
+ }
+ }
+ }
+ return positions;
}
+
+ /**
+ * A data structure to keep all information required to create a new replica in one place.
+ * Think of it as a typed ZkNodeProps for replica creation.
+ *
+ * This is <b>not</b> a public API and can be changed at any time without notice.
+ */
+ public static class CreateReplica {
+ public final String collectionName;
+ public final String sliceName;
+ public final String node;
+ public final Replica.Type replicaType;
+ public String coreName;
+ public String coreNodeName;
+
+ CreateReplica(String collectionName, String sliceName, String node, Replica.Type replicaType, String coreName, String coreNodeName) {
+ this.collectionName = collectionName;
+ this.sliceName = sliceName;
+ this.node = node;
+ this.replicaType = replicaType;
+ this.coreName = coreName;
+ this.coreNodeName = coreNodeName;
+ }
+ }
+
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/0d7627f2/solr/core/src/java/org/apache/solr/cloud/api/collections/Assign.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/api/collections/Assign.java b/solr/core/src/java/org/apache/solr/cloud/api/collections/Assign.java
index d323510..42de84a 100644
--- a/solr/core/src/java/org/apache/solr/cloud/api/collections/Assign.java
+++ b/solr/core/src/java/org/apache/solr/cloud/api/collections/Assign.java
@@ -273,7 +273,7 @@ public class Assign {
} else {
if (numTlogReplicas + numPullReplicas != 0 && rulesMap != null) {
throw new SolrException(SolrException.ErrorCode.BAD_REQUEST,
- Replica.Type.TLOG + " or " + Replica.Type.PULL + " replica types not supported with placement rules or cluster policies");
+ Replica.Type.TLOG + " or " + Replica.Type.PULL + " replica types not supported with placement rules");
}
}
@@ -322,11 +322,11 @@ public class Assign {
// Gets a list of candidate nodes to put the required replica(s) on. Throws errors if not enough replicas
// could be created on live nodes given maxShardsPerNode, Replication factor (if from createShard) etc.
public static List<ReplicaCount> getNodesForNewReplicas(ClusterState clusterState, String collectionName,
- String shard, int nrtReplicas,
+ String shard, int nrtReplicas, int tlogReplicas, int pullReplicas,
Object createNodeSet, SolrCloudManager cloudManager) throws IOException, InterruptedException, AssignmentException {
- log.debug("getNodesForNewReplicas() shard: {} , replicas : {} , createNodeSet {}", shard, nrtReplicas, createNodeSet );
+ log.debug("getNodesForNewReplicas() shard: {} , nrtReplicas : {} , tlogReplicas: {} , pullReplicas: {} , createNodeSet {}", shard, nrtReplicas, tlogReplicas, pullReplicas, createNodeSet );
DocCollection coll = clusterState.getCollection(collectionName);
- Integer maxShardsPerNode = coll.getMaxShardsPerNode();
+ Integer maxShardsPerNode = coll.getMaxShardsPerNode() == -1 ? Integer.MAX_VALUE : coll.getMaxShardsPerNode();
List<String> createNodeList = null;
if (createNodeSet instanceof List) {
@@ -338,15 +338,15 @@ public class Assign {
HashMap<String, ReplicaCount> nodeNameVsShardCount = getNodeNameVsShardCount(collectionName, clusterState, createNodeList);
if (createNodeList == null) { // We only care if we haven't been told to put new replicas on specific nodes.
- int availableSlots = 0;
+ long availableSlots = 0;
for (Map.Entry<String, ReplicaCount> ent : nodeNameVsShardCount.entrySet()) {
//ADDREPLICA can put more than maxShardsPerNode on an instance, so this test is necessary.
if (maxShardsPerNode > ent.getValue().thisCollectionNodes) {
availableSlots += (maxShardsPerNode - ent.getValue().thisCollectionNodes);
}
}
- if (availableSlots < nrtReplicas) {
- throw new SolrException(SolrException.ErrorCode.BAD_REQUEST,
+ if (availableSlots < nrtReplicas + tlogReplicas + pullReplicas) {
+ throw new AssignmentException(
String.format(Locale.ROOT, "Cannot create %d new replicas for collection %s given the current number of live nodes and a maxShardsPerNode of %d",
nrtReplicas, collectionName, maxShardsPerNode));
}
@@ -355,13 +355,17 @@ public class Assign {
List l = (List) coll.get(DocCollection.RULE);
List<ReplicaPosition> replicaPositions = null;
if (l != null) {
+ if (tlogReplicas + pullReplicas > 0) {
+ throw new AssignmentException(Replica.Type.TLOG + " or " + Replica.Type.PULL +
+ " replica types not supported with placement rules");
+ }
// TODO: make it so that this method doesn't require access to CC
replicaPositions = getNodesViaRules(clusterState, shard, nrtReplicas, cloudManager, coll, createNodeList, l);
}
String policyName = coll.getStr(POLICY);
AutoScalingConfig autoScalingConfig = cloudManager.getDistribStateManager().getAutoScalingConfig();
if (policyName != null || !autoScalingConfig.getPolicy().getClusterPolicy().isEmpty()) {
- replicaPositions = Assign.getPositionsUsingPolicy(collectionName, Collections.singletonList(shard), nrtReplicas, 0, 0,
+ replicaPositions = Assign.getPositionsUsingPolicy(collectionName, Collections.singletonList(shard), nrtReplicas, tlogReplicas, pullReplicas,
policyName, cloudManager, createNodeList);
}
@@ -461,7 +465,7 @@ public class Assign {
return nodeNameVsShardCount;
}
DocCollection coll = clusterState.getCollection(collectionName);
- Integer maxShardsPerNode = coll.getMaxShardsPerNode();
+ int maxShardsPerNode = coll.getMaxShardsPerNode() == -1 ? Integer.MAX_VALUE : coll.getMaxShardsPerNode();
Map<String, DocCollection> collections = clusterState.getCollectionsMap();
for (Map.Entry<String, DocCollection> entry : collections.entrySet()) {
DocCollection c = entry.getValue();
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/0d7627f2/solr/core/src/java/org/apache/solr/cloud/api/collections/CreateShardCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/api/collections/CreateShardCmd.java b/solr/core/src/java/org/apache/solr/cloud/api/collections/CreateShardCmd.java
index 802583c..e7f35f1 100644
--- a/solr/core/src/java/org/apache/solr/cloud/api/collections/CreateShardCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/api/collections/CreateShardCmd.java
@@ -17,32 +17,17 @@
package org.apache.solr.cloud.api.collections;
-import java.io.IOException;
import java.lang.invoke.MethodHandles;
-import java.util.ArrayList;
-import java.util.Collections;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
-import com.google.common.collect.ImmutableMap;
-import org.apache.solr.client.solrj.cloud.autoscaling.Policy;
-import org.apache.solr.client.solrj.cloud.autoscaling.PolicyHelper;
-import org.apache.solr.client.solrj.cloud.SolrCloudManager;
-import org.apache.solr.cloud.CloudUtil;
import org.apache.solr.cloud.Overseer;
-import org.apache.solr.common.SolrCloseableLatch;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.cloud.ClusterState;
import org.apache.solr.common.cloud.DocCollection;
-import org.apache.solr.common.cloud.Replica;
-import org.apache.solr.common.cloud.ReplicaPosition;
import org.apache.solr.common.cloud.ZkNodeProps;
import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.params.CommonAdminParams;
-import org.apache.solr.common.params.CoreAdminParams;
import org.apache.solr.common.util.NamedList;
import org.apache.solr.common.util.SimpleOrderedMap;
import org.apache.solr.common.util.Utils;
@@ -77,114 +62,60 @@ public class CreateShardCmd implements OverseerCollectionMessageHandler.Cmd {
DocCollection collection = clusterState.getCollection(collectionName);
- ZkStateReader zkStateReader = ocmh.zkStateReader;
- AtomicReference<PolicyHelper.SessionWrapper> sessionWrapper = new AtomicReference<>();
- SolrCloseableLatch countDownLatch;
- try {
- List<ReplicaPosition> positions = buildReplicaPositions(ocmh.cloudManager, clusterState, collectionName, message, sessionWrapper);
- Overseer.getStateUpdateQueue(zkStateReader.getZkClient()).offer(Utils.toJSON(message));
- // wait for a while until we see the shard
- ocmh.waitForNewShard(collectionName, sliceName);
-
- String async = message.getStr(ASYNC);
- countDownLatch = new SolrCloseableLatch(positions.size(), ocmh);
- for (ReplicaPosition position : positions) {
- String nodeName = position.node;
- String coreName = Assign.buildSolrCoreName(ocmh.cloudManager.getDistribStateManager(), collection, sliceName, position.type);
- log.info("Creating replica " + coreName + " as part of slice " + sliceName + " of collection " + collectionName
- + " on " + nodeName);
-
- // Need to create new params for each request
- ZkNodeProps addReplicasProps = new ZkNodeProps(
- COLLECTION_PROP, collectionName,
- SHARD_ID_PROP, sliceName,
- ZkStateReader.REPLICA_TYPE, position.type.name(),
- CoreAdminParams.NODE, nodeName,
- CoreAdminParams.NAME, coreName,
- CommonAdminParams.WAIT_FOR_FINAL_STATE, Boolean.toString(waitForFinalState));
- Map<String, Object> propertyParams = new HashMap<>();
- ocmh.addPropertyParams(message, propertyParams);
- addReplicasProps = addReplicasProps.plus(propertyParams);
- if (async != null) addReplicasProps.getProperties().put(ASYNC, async);
- final NamedList addResult = new NamedList();
- ocmh.addReplica(zkStateReader.getClusterState(), addReplicasProps, addResult, () -> {
- countDownLatch.countDown();
- Object addResultFailure = addResult.get("failure");
- if (addResultFailure != null) {
- SimpleOrderedMap failure = (SimpleOrderedMap) results.get("failure");
- if (failure == null) {
- failure = new SimpleOrderedMap();
- results.add("failure", failure);
- }
- failure.addAll((NamedList) addResultFailure);
- } else {
- SimpleOrderedMap success = (SimpleOrderedMap) results.get("success");
- if (success == null) {
- success = new SimpleOrderedMap();
- results.add("success", success);
- }
- success.addAll((NamedList) addResult.get("success"));
- }
- });
- }
- } finally {
- if (sessionWrapper.get() != null) sessionWrapper.get().release();
- }
-
- log.debug("Waiting for create shard action to complete");
- countDownLatch.await(5, TimeUnit.MINUTES);
- log.debug("Finished waiting for create shard action to complete");
-
- log.info("Finished create command on all shards for collection: " + collectionName);
-
- }
-
- public static List<ReplicaPosition> buildReplicaPositions(SolrCloudManager cloudManager, ClusterState clusterState,
- String collectionName, ZkNodeProps message, AtomicReference< PolicyHelper.SessionWrapper> sessionWrapper) throws IOException, InterruptedException {
- String sliceName = message.getStr(SHARD_ID_PROP);
- DocCollection collection = clusterState.getCollection(collectionName);
-
int numNrtReplicas = message.getInt(NRT_REPLICAS, message.getInt(REPLICATION_FACTOR, collection.getInt(NRT_REPLICAS, collection.getInt(REPLICATION_FACTOR, 1))));
int numPullReplicas = message.getInt(PULL_REPLICAS, collection.getInt(PULL_REPLICAS, 0));
int numTlogReplicas = message.getInt(TLOG_REPLICAS, collection.getInt(TLOG_REPLICAS, 0));
- int totalReplicas = numNrtReplicas + numPullReplicas + numTlogReplicas;
if (numNrtReplicas + numTlogReplicas <= 0) {
throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, NRT_REPLICAS + " + " + TLOG_REPLICAS + " must be greater than 0");
}
- Object createNodeSetStr = message.get(OverseerCollectionMessageHandler.CREATE_NODE_SET);
-
- boolean usePolicyFramework = CloudUtil.usePolicyFramework(collection, cloudManager);
- List<ReplicaPosition> positions;
- if (usePolicyFramework) {
- if (collection.getPolicyName() != null) message.getProperties().put(Policy.POLICY, collection.getPolicyName());
- positions = Assign.identifyNodes(cloudManager,
- clusterState,
- Assign.getLiveOrLiveAndCreateNodeSetList(clusterState.getLiveNodes(), message, OverseerCollectionMessageHandler.RANDOM),
- collection.getName(),
- message,
- Collections.singletonList(sliceName),
- numNrtReplicas,
- numTlogReplicas,
- numPullReplicas);
- sessionWrapper.set(PolicyHelper.getLastSessionWrapper(true));
- } else {
- List<Assign.ReplicaCount> sortedNodeList = Assign.getNodesForNewReplicas(clusterState, collection.getName(), sliceName, totalReplicas,
- createNodeSetStr, cloudManager);
- int i = 0;
- positions = new ArrayList<>();
- for (Map.Entry<Replica.Type, Integer> e : ImmutableMap.of(Replica.Type.NRT, numNrtReplicas,
- Replica.Type.TLOG, numTlogReplicas,
- Replica.Type.PULL, numPullReplicas
- ).entrySet()) {
- for (int j = 0; j < e.getValue(); j++) {
- positions.add(new ReplicaPosition(sliceName, j + 1, e.getKey(), sortedNodeList.get(i % sortedNodeList.size()).nodeName));
- i++;
+ ZkStateReader zkStateReader = ocmh.zkStateReader;
+ Overseer.getStateUpdateQueue(zkStateReader.getZkClient()).offer(Utils.toJSON(message));
+ // wait for a while until we see the shard
+ ocmh.waitForNewShard(collectionName, sliceName);
+ String async = message.getStr(ASYNC);
+ ZkNodeProps addReplicasProps = new ZkNodeProps(
+ COLLECTION_PROP, collectionName,
+ SHARD_ID_PROP, sliceName,
+ ZkStateReader.NRT_REPLICAS, String.valueOf(numNrtReplicas),
+ ZkStateReader.TLOG_REPLICAS, String.valueOf(numTlogReplicas),
+ ZkStateReader.PULL_REPLICAS, String.valueOf(numPullReplicas),
+ OverseerCollectionMessageHandler.CREATE_NODE_SET, message.getStr(OverseerCollectionMessageHandler.CREATE_NODE_SET),
+ CommonAdminParams.WAIT_FOR_FINAL_STATE, Boolean.toString(waitForFinalState));
+
+ Map<String, Object> propertyParams = new HashMap<>();
+ ocmh.addPropertyParams(message, propertyParams);
+ addReplicasProps = addReplicasProps.plus(propertyParams);
+ if (async != null) addReplicasProps.getProperties().put(ASYNC, async);
+ final NamedList addResult = new NamedList();
+ try {
+ ocmh.addReplica(zkStateReader.getClusterState(), addReplicasProps, addResult, () -> {
+ Object addResultFailure = addResult.get("failure");
+ if (addResultFailure != null) {
+ SimpleOrderedMap failure = (SimpleOrderedMap) results.get("failure");
+ if (failure == null) {
+ failure = new SimpleOrderedMap();
+ results.add("failure", failure);
+ }
+ failure.addAll((NamedList) addResultFailure);
+ } else {
+ SimpleOrderedMap success = (SimpleOrderedMap) results.get("success");
+ if (success == null) {
+ success = new SimpleOrderedMap();
+ results.add("success", success);
+ }
+ success.addAll((NamedList) addResult.get("success"));
}
- }
+ });
+ } catch (Assign.AssignmentException e) {
+ // clean up the slice that we created
+ ZkNodeProps deleteShard = new ZkNodeProps(COLLECTION_PROP, collectionName, SHARD_ID_PROP, sliceName, ASYNC, async);
+ new DeleteShardCmd(ocmh).call(clusterState, deleteShard, results);
+ throw e;
}
- return positions;
+
+ log.info("Finished create command on all shards for collection: " + collectionName);
}
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/0d7627f2/solr/core/src/java/org/apache/solr/cloud/api/collections/MoveReplicaCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/api/collections/MoveReplicaCmd.java b/solr/core/src/java/org/apache/solr/cloud/api/collections/MoveReplicaCmd.java
index 2df0f77..6071b1b 100644
--- a/solr/core/src/java/org/apache/solr/cloud/api/collections/MoveReplicaCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/api/collections/MoveReplicaCmd.java
@@ -268,7 +268,7 @@ public class MoveReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
NamedList addResult = new NamedList();
SolrCloseableLatch countDownLatch = new SolrCloseableLatch(1, ocmh);
ActiveReplicaWatcher watcher = null;
- ZkNodeProps props = ocmh.addReplica(clusterState, addReplicasProps, addResult, null);
+ ZkNodeProps props = ocmh.addReplica(clusterState, addReplicasProps, addResult, null).get(0);
log.debug("props " + props);
if (replica.equals(slice.getLeader()) || waitForFinalState) {
watcher = new ActiveReplicaWatcher(coll.getName(), null, Collections.singletonList(newCoreName), countDownLatch);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/0d7627f2/solr/core/src/java/org/apache/solr/cloud/api/collections/OverseerCollectionMessageHandler.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/api/collections/OverseerCollectionMessageHandler.java b/solr/core/src/java/org/apache/solr/cloud/api/collections/OverseerCollectionMessageHandler.java
index 25b630e..59480cd 100644
--- a/solr/core/src/java/org/apache/solr/cloud/api/collections/OverseerCollectionMessageHandler.java
+++ b/solr/core/src/java/org/apache/solr/cloud/api/collections/OverseerCollectionMessageHandler.java
@@ -709,7 +709,7 @@ public class OverseerCollectionMessageHandler implements OverseerMessageHandler,
}
}
- ZkNodeProps addReplica(ClusterState clusterState, ZkNodeProps message, NamedList results, Runnable onComplete)
+ List<ZkNodeProps> addReplica(ClusterState clusterState, ZkNodeProps message, NamedList results, Runnable onComplete)
throws Exception {
return ((AddReplicaCmd) commandMap.get(ADDREPLICA)).addReplica(clusterState, message, results, onComplete);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/0d7627f2/solr/core/src/java/org/apache/solr/cloud/api/collections/ReplaceNodeCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/api/collections/ReplaceNodeCmd.java b/solr/core/src/java/org/apache/solr/cloud/api/collections/ReplaceNodeCmd.java
index d08b519..a09eec3 100644
--- a/solr/core/src/java/org/apache/solr/cloud/api/collections/ReplaceNodeCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/api/collections/ReplaceNodeCmd.java
@@ -140,7 +140,7 @@ public class ReplaceNodeCmd implements OverseerCollectionMessageHandler.Cmd {
log.debug("Successfully created replica for collection={} shard={} on node={}",
sourceReplica.getStr(COLLECTION_PROP), sourceReplica.getStr(SHARD_ID_PROP), target);
}
- });
+ }).get(0);
if (addedReplica != null) {
createdReplicas.add(addedReplica);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/0d7627f2/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 3a46b2b..aef2448 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
@@ -694,6 +694,9 @@ public class CollectionsHandler extends RequestHandlerBase implements Permission
throw new SolrException(ErrorCode.BAD_REQUEST, "shards can be added only to 'implicit' collections");
copy(req.getParams(), map,
REPLICATION_FACTOR,
+ NRT_REPLICAS,
+ TLOG_REPLICAS,
+ PULL_REPLICAS,
CREATE_NODE_SET,
WAIT_FOR_FINAL_STATE);
return copyPropertiesWithPrefix(req.getParams(), map, COLL_PROP_PREFIX);
@@ -828,7 +831,11 @@ public class CollectionsHandler extends RequestHandlerBase implements Permission
DATA_DIR,
ULOG_DIR,
REPLICA_TYPE,
- WAIT_FOR_FINAL_STATE);
+ WAIT_FOR_FINAL_STATE,
+ NRT_REPLICAS,
+ TLOG_REPLICAS,
+ PULL_REPLICAS,
+ CREATE_NODE_SET);
return copyPropertiesWithPrefix(req.getParams(), props, COLL_PROP_PREFIX);
}),
OVERSEERSTATUS_OP(OVERSEERSTATUS, (req, rsp, h) -> (Map) new LinkedHashMap<>()),
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/0d7627f2/solr/core/src/test/org/apache/solr/cloud/AddReplicaTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/AddReplicaTest.java b/solr/core/src/test/org/apache/solr/cloud/AddReplicaTest.java
index f311a12..7b558c0 100644
--- a/solr/core/src/test/org/apache/solr/cloud/AddReplicaTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/AddReplicaTest.java
@@ -17,7 +17,10 @@
package org.apache.solr.cloud;
import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
import java.util.Collection;
+import java.util.EnumSet;
+import java.util.List;
import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.apache.solr.client.solrj.request.CollectionAdminRequest;
@@ -26,11 +29,15 @@ import org.apache.solr.common.cloud.ClusterState;
import org.apache.solr.common.cloud.DocCollection;
import org.apache.solr.common.cloud.Replica;
import org.apache.solr.util.LogLevel;
+import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import static org.apache.solr.client.solrj.response.RequestStatusState.COMPLETED;
+import static org.apache.solr.client.solrj.response.RequestStatusState.FAILED;
+
/**
*
*/
@@ -45,10 +52,75 @@ public class AddReplicaTest extends SolrCloudTestCase {
.configure();
}
+ @Before
+ public void setUp() throws Exception {
+ super.setUp();
+ cluster.deleteAllCollections();
+ }
+
+ @Test
+ public void testAddMultipleReplicas() throws Exception {
+ cluster.waitForAllNodes(5);
+ String collection = "testAddMultipleReplicas";
+ CloudSolrClient cloudClient = cluster.getSolrClient();
+
+ CollectionAdminRequest.Create create = CollectionAdminRequest.createCollection(collection, "conf1", 1, 1);
+ create.setMaxShardsPerNode(2);
+ cloudClient.request(create);
+
+ CollectionAdminRequest.AddReplica addReplica = CollectionAdminRequest.addReplicaToShard(collection, "shard1")
+ .setNrtReplicas(1)
+ .setTlogReplicas(1)
+ .setPullReplicas(1);
+ RequestStatusState status = addReplica.processAndWait(collection + "_xyz1", cloudClient, 120);
+ assertEquals(COMPLETED, status);
+ DocCollection docCollection = cloudClient.getZkStateReader().getClusterState().getCollectionOrNull(collection);
+ assertNotNull(docCollection);
+ assertEquals(4, docCollection.getReplicas().size());
+ assertEquals(2, docCollection.getReplicas(EnumSet.of(Replica.Type.NRT)).size());
+ assertEquals(1, docCollection.getReplicas(EnumSet.of(Replica.Type.TLOG)).size());
+ assertEquals(1, docCollection.getReplicas(EnumSet.of(Replica.Type.PULL)).size());
+
+ // try to add 5 more replicas which should fail because numNodes(4)*maxShardsPerNode(2)=8 and 4 replicas already exist
+ addReplica = CollectionAdminRequest.addReplicaToShard(collection, "shard1")
+ .setNrtReplicas(3)
+ .setTlogReplicas(1)
+ .setPullReplicas(1);
+ status = addReplica.processAndWait(collection + "_xyz1", cloudClient, 120);
+ assertEquals(FAILED, status);
+ docCollection = cloudClient.getZkStateReader().getClusterState().getCollectionOrNull(collection);
+ assertNotNull(docCollection);
+ // sanity check that everything is as before
+ assertEquals(4, docCollection.getReplicas().size());
+ assertEquals(2, docCollection.getReplicas(EnumSet.of(Replica.Type.NRT)).size());
+ assertEquals(1, docCollection.getReplicas(EnumSet.of(Replica.Type.TLOG)).size());
+ assertEquals(1, docCollection.getReplicas(EnumSet.of(Replica.Type.PULL)).size());
+
+ // but adding any number of replicas is supported if an explicit create node set is specified
+ // so test that as well
+ List<String> createNodeSet = new ArrayList<>(2);
+ createNodeSet.add(cluster.getRandomJetty(random()).getNodeName());
+ createNodeSet.add(cluster.getRandomJetty(random()).getNodeName());
+ addReplica = CollectionAdminRequest.addReplicaToShard(collection, "shard1")
+ .setNrtReplicas(3)
+ .setTlogReplicas(1)
+ .setPullReplicas(1)
+ .setCreateNodeSet(String.join(",", createNodeSet));
+ status = addReplica.processAndWait(collection + "_xyz1", cloudClient, 120);
+ assertEquals(COMPLETED, status);
+ docCollection = cloudClient.getZkStateReader().getClusterState().getCollectionOrNull(collection);
+ assertNotNull(docCollection);
+ // sanity check that everything is as before
+ assertEquals(9, docCollection.getReplicas().size());
+ assertEquals(5, docCollection.getReplicas(EnumSet.of(Replica.Type.NRT)).size());
+ assertEquals(2, docCollection.getReplicas(EnumSet.of(Replica.Type.TLOG)).size());
+ assertEquals(2, docCollection.getReplicas(EnumSet.of(Replica.Type.PULL)).size());
+ }
+
@Test
//commented 2-Aug-2018 @BadApple(bugUrl="https://issues.apache.org/jira/browse/SOLR-12028") // 09-Apr-2018
public void test() throws Exception {
- cluster.waitForAllNodes(5000);
+ cluster.waitForAllNodes(5);
String collection = "addreplicatest_coll";
CloudSolrClient cloudClient = cluster.getSolrClient();
@@ -65,16 +137,16 @@ public class AddReplicaTest extends SolrCloudTestCase {
addReplica.processAsync("000", cloudClient);
CollectionAdminRequest.RequestStatus requestStatus = CollectionAdminRequest.requestStatus("000");
CollectionAdminRequest.RequestStatusResponse rsp = requestStatus.process(cloudClient);
- assertTrue(rsp.getRequestStatus() != RequestStatusState.COMPLETED);
+ assertNotSame(rsp.getRequestStatus(), COMPLETED);
// wait for async request success
boolean success = false;
for (int i = 0; i < 200; i++) {
rsp = requestStatus.process(cloudClient);
- if (rsp.getRequestStatus() == RequestStatusState.COMPLETED) {
+ if (rsp.getRequestStatus() == COMPLETED) {
success = true;
break;
}
- assertFalse(rsp.toString(), rsp.getRequestStatus() == RequestStatusState.FAILED);
+ assertNotSame(rsp.toString(), rsp.getRequestStatus(), RequestStatusState.FAILED);
Thread.sleep(500);
}
assertTrue(success);
@@ -82,23 +154,23 @@ public class AddReplicaTest extends SolrCloudTestCase {
replicas2.removeAll(replicas);
assertEquals(1, replicas2.size());
Replica r = replicas2.iterator().next();
- assertTrue(r.toString(), r.getState() != Replica.State.ACTIVE);
+ assertNotSame(r.toString(), r.getState(), Replica.State.ACTIVE);
// use waitForFinalState
addReplica.setWaitForFinalState(true);
addReplica.processAsync("001", cloudClient);
requestStatus = CollectionAdminRequest.requestStatus("001");
rsp = requestStatus.process(cloudClient);
- assertTrue(rsp.getRequestStatus() != RequestStatusState.COMPLETED);
+ assertNotSame(rsp.getRequestStatus(), COMPLETED);
// wait for async request success
success = false;
for (int i = 0; i < 200; i++) {
rsp = requestStatus.process(cloudClient);
- if (rsp.getRequestStatus() == RequestStatusState.COMPLETED) {
+ if (rsp.getRequestStatus() == COMPLETED) {
success = true;
break;
}
- assertFalse(rsp.toString(), rsp.getRequestStatus() == RequestStatusState.FAILED);
+ assertNotSame(rsp.toString(), rsp.getRequestStatus(), RequestStatusState.FAILED);
Thread.sleep(500);
}
assertTrue(success);
@@ -114,7 +186,7 @@ public class AddReplicaTest extends SolrCloudTestCase {
if (replica.getName().equals(replica2)) {
continue; // may be still recovering
}
- assertTrue(coll.toString() + "\n" + replica.toString(), replica.getState() == Replica.State.ACTIVE);
+ assertSame(coll.toString() + "\n" + replica.toString(), replica.getState(), Replica.State.ACTIVE);
}
}
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/0d7627f2/solr/core/src/test/org/apache/solr/cloud/api/collections/CollectionTooManyReplicasTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/api/collections/CollectionTooManyReplicasTest.java b/solr/core/src/test/org/apache/solr/cloud/api/collections/CollectionTooManyReplicasTest.java
index f68fa9e..09a119b 100644
--- a/solr/core/src/test/org/apache/solr/cloud/api/collections/CollectionTooManyReplicasTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/api/collections/CollectionTooManyReplicasTest.java
@@ -158,7 +158,7 @@ public class CollectionTooManyReplicasTest extends SolrCloudTestCase {
assertTrue("Should have gotten the right error message back",
e2.getMessage().contains("given the current number of live nodes and a maxShardsPerNode of"));
- // And finally, insure that there are all the replcias we expect. We should have shards 1, 2 and 4 and each
+ // And finally, ensure that there are all the replicas we expect. We should have shards 1, 2 and 4 and each
// should have exactly two replicas
waitForState("Expected shards shardstart, 1, 2 and 4, each with two active replicas", collectionName, (n, c) -> {
return DocCollection.isFullyActive(n, c, 4, 2);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/0d7627f2/solr/core/src/test/org/apache/solr/cloud/api/collections/ShardSplitTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/api/collections/ShardSplitTest.java b/solr/core/src/test/org/apache/solr/cloud/api/collections/ShardSplitTest.java
index 39ae618..cd87bb5 100644
--- a/solr/core/src/test/org/apache/solr/cloud/api/collections/ShardSplitTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/api/collections/ShardSplitTest.java
@@ -542,7 +542,7 @@ public class ShardSplitTest extends AbstractFullDistribZkTestBase {
@Test
public void testSplitShardWithRule() throws Exception {
- doSplitShardWithRule(SolrIndexSplitter.SplitMethod.LINK);
+ doSplitShardWithRule(SolrIndexSplitter.SplitMethod.REWRITE);
}
@Test
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/0d7627f2/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java
index c739fd9..637f5bf 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java
@@ -23,6 +23,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
+import java.util.EnumMap;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
@@ -434,23 +435,46 @@ public class SimClusterStateProvider implements ClusterStateProvider {
ClusterState clusterState = getClusterState();
DocCollection coll = clusterState.getCollection(message.getStr(ZkStateReader.COLLECTION_PROP));
AtomicReference<PolicyHelper.SessionWrapper> sessionWrapper = new AtomicReference<>();
- message = AddReplicaCmd.assignReplicaDetails(cloudManager, clusterState, message, sessionWrapper);
+
+ Replica.Type replicaType = Replica.Type.valueOf(message.getStr(ZkStateReader.REPLICA_TYPE, Replica.Type.NRT.name()).toUpperCase(Locale.ROOT));
+ EnumMap<Replica.Type, Integer> replicaTypesVsCount = new EnumMap<>(Replica.Type.class);
+ replicaTypesVsCount.put(Replica.Type.NRT, message.getInt(NRT_REPLICAS, replicaType == Replica.Type.NRT ? 1 : 0));
+ replicaTypesVsCount.put(Replica.Type.TLOG, message.getInt(TLOG_REPLICAS, replicaType == Replica.Type.TLOG ? 1 : 0));
+ replicaTypesVsCount.put(Replica.Type.PULL, message.getInt(PULL_REPLICAS, replicaType == Replica.Type.PULL ? 1 : 0));
+
+ int totalReplicas = 0;
+ for (Map.Entry<Replica.Type, Integer> entry : replicaTypesVsCount.entrySet()) {
+ totalReplicas += entry.getValue();
+ }
+ if (totalReplicas > 1) {
+ if (message.getStr(CoreAdminParams.NAME) != null) {
+ throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "Cannot create " + totalReplicas + " replicas if 'name' parameter is specified");
+ }
+ if (message.getStr(CoreAdminParams.CORE_NODE_NAME) != null) {
+ throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "Cannot create " + totalReplicas + " replicas if 'coreNodeName' parameter is specified");
+ }
+ }
+
+ List<ReplicaPosition> replicaPositions = AddReplicaCmd.buildReplicaPositions(cloudManager, clusterState, coll.getName(), message, replicaTypesVsCount, sessionWrapper);
+ for (ReplicaPosition replicaPosition : replicaPositions) {
+ AddReplicaCmd.CreateReplica createReplica = AddReplicaCmd.assignReplicaDetails(cloudManager, clusterState, message, replicaPosition);
+ if (message.getStr(CoreAdminParams.CORE_NODE_NAME) == null) {
+ createReplica.coreNodeName = Assign.assignCoreNodeName(stateManager, coll);
+ }
+ ReplicaInfo ri = new ReplicaInfo(
+ createReplica.coreNodeName,
+ createReplica.coreName,
+ createReplica.collectionName,
+ createReplica.sliceName,
+ createReplica.replicaType,
+ createReplica.node,
+ message.getProperties()
+ );
+ simAddReplica(ri.getNode(), ri, true);
+ }
if (sessionWrapper.get() != null) {
sessionWrapper.get().release();
}
- if (message.getStr(CoreAdminParams.CORE_NODE_NAME) == null) {
- message = message.plus(CoreAdminParams.CORE_NODE_NAME, Assign.assignCoreNodeName(stateManager, coll));
- }
- ReplicaInfo ri = new ReplicaInfo(
- message.getStr(CoreAdminParams.CORE_NODE_NAME),
- message.getStr(CoreAdminParams.NAME),
- message.getStr(ZkStateReader.COLLECTION_PROP),
- message.getStr(ZkStateReader.SHARD_ID_PROP),
- Replica.Type.valueOf(message.getStr(ZkStateReader.REPLICA_TYPE, Replica.Type.NRT.name()).toUpperCase(Locale.ROOT)),
- message.getStr(CoreAdminParams.NODE),
- message.getProperties()
- );
- simAddReplica(message.getStr(CoreAdminParams.NODE), ri, true);
results.add("success", "");
}
@@ -1069,31 +1093,30 @@ public class SimClusterStateProvider implements ClusterStateProvider {
.filter(e -> !e.getKey().equals("replicas"))
.forEach(e -> props.put(e.getKey(), e.getValue()));
// 2. create new replicas
- AtomicReference<PolicyHelper.SessionWrapper> sessionWrapper = new AtomicReference<>();
- List<ReplicaPosition> positions = CreateShardCmd.buildReplicaPositions(cloudManager, clusterState, collectionName,
- message, sessionWrapper);
- if (sessionWrapper.get() != null) {
- sessionWrapper.get().release();
- }
- AtomicInteger replicaNum = new AtomicInteger(1);
- positions.forEach(pos -> {
- Map<String, Object> replicaProps = new HashMap<>();
- replicaProps.put(ZkStateReader.SHARD_ID_PROP, pos.shard);
- replicaProps.put(ZkStateReader.NODE_NAME_PROP, pos.node);
- replicaProps.put(ZkStateReader.REPLICA_TYPE, pos.type.toString());
- replicaProps.put(ZkStateReader.BASE_URL_PROP, Utils.getBaseUrlForNodeName(pos.node, "http"));
- String coreName = String.format(Locale.ROOT, "%s_%s_replica_%s%s", collectionName, pos.shard, pos.type.name().substring(0,1).toLowerCase(Locale.ROOT),
- replicaNum.getAndIncrement());
+ EnumMap<Replica.Type, Integer> replicaTypesVsCount = new EnumMap<>(Replica.Type.class);
+ int numNrtReplicas = message.getInt(NRT_REPLICAS, message.getInt(REPLICATION_FACTOR, collection.getInt(NRT_REPLICAS, collection.getInt(REPLICATION_FACTOR, 1))));
+ int numTlogReplicas = message.getInt(TLOG_REPLICAS, message.getInt(TLOG_REPLICAS, collection.getInt(TLOG_REPLICAS, 0)));
+ int numPullReplicas = message.getInt(PULL_REPLICAS, message.getInt(PULL_REPLICAS, collection.getInt(PULL_REPLICAS, 0)));
+ replicaTypesVsCount.put(Replica.Type.NRT, numNrtReplicas);
+ replicaTypesVsCount.put(Replica.Type.TLOG, numTlogReplicas);
+ replicaTypesVsCount.put(Replica.Type.PULL, numPullReplicas);
+
+ ZkNodeProps addReplicasProps = new ZkNodeProps(
+ COLLECTION_PROP, collectionName,
+ SHARD_ID_PROP, sliceName,
+ ZkStateReader.NRT_REPLICAS, String.valueOf(replicaTypesVsCount.get(Replica.Type.NRT)),
+ ZkStateReader.TLOG_REPLICAS, String.valueOf(replicaTypesVsCount.get(Replica.Type.TLOG)),
+ ZkStateReader.PULL_REPLICAS, String.valueOf(replicaTypesVsCount.get(Replica.Type.PULL)),
+ OverseerCollectionMessageHandler.CREATE_NODE_SET, message.getStr(OverseerCollectionMessageHandler.CREATE_NODE_SET)
+ );
+
try {
- replicaProps.put(ZkStateReader.CORE_NAME_PROP, coreName);
- ReplicaInfo ri = new ReplicaInfo("core_node" + Assign.incAndGetId(stateManager, collectionName, 0),
- coreName, collectionName, pos.shard, pos.type, pos.node, replicaProps);
- simAddReplica(pos.node, ri, false);
+ simAddReplica(addReplicasProps, results);
} catch (Exception e) {
throw new RuntimeException(e);
}
- });
- Map<String, Object> colProps = collProperties.computeIfAbsent(collectionName, c -> new ConcurrentHashMap<>());
+
+ collProperties.computeIfAbsent(collectionName, c -> new ConcurrentHashMap<>());
simRunLeaderElection(Collections.singleton(collectionName), true);
results.add("success", "");
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/0d7627f2/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestSimPolicyCloud.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestSimPolicyCloud.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestSimPolicyCloud.java
index 7e9da4e..86624b8 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestSimPolicyCloud.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestSimPolicyCloud.java
@@ -236,7 +236,7 @@ public class TestSimPolicyCloud extends SimSolrCloudTestCase {
String pullNodeName = nodes.get(1);
int pullPort = (Integer)cluster.getSimNodeStateProvider().simGetNodeValue(pullNodeName, ImplicitSnitch.PORT);
- String tlogNodeName = nodes.get(1);
+ String tlogNodeName = nodes.get(2);
int tlogPort = (Integer)cluster.getSimNodeStateProvider().simGetNodeValue(tlogNodeName, ImplicitSnitch.PORT);
log.info("NRT {} PULL {} , TLOG {} ", nrtNodeName, pullNodeName, tlogNodeName);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/0d7627f2/solr/solr-ref-guide/src/collections-api.adoc
----------------------------------------------------------------------
diff --git a/solr/solr-ref-guide/src/collections-api.adoc b/solr/solr-ref-guide/src/collections-api.adoc
index e4230dd..d601069 100644
--- a/solr/solr-ref-guide/src/collections-api.adoc
+++ b/solr/solr-ref-guide/src/collections-api.adoc
@@ -396,6 +396,10 @@ Use SPLITSHARD for collections created with the 'compositeId' router (`router.ke
`/admin/collections?action=CREATESHARD&shard=_shardName_&collection=_name_`
+The default values for `replicationFactor` or `nrtReplicas`, `tlogReplicas`, `pullReplicas` from the collection is used to determine the number of replicas to be created for the new shard. This can be customized by explicitly passing the corresponding parameters to the request.
+
+The API uses the Autoscaling framework to find the best possible nodes in the cluster when an Autoscaling preferences or policy is configured. Refer to <<solrcloud-autoscaling-policy-preferences.adoc#solrcloud-autoscaling-policy-preferences,Autoscaling Policy and Preferences>> section for more details.
+
=== CREATESHARD Parameters
`collection`::
@@ -409,6 +413,15 @@ Allows defining the nodes to spread the new collection across. If not provided,
+
The format is a comma-separated list of node_names, such as `localhost:8983_solr,localhost:8984_solr,localhost:8985_solr`.
+`nrtReplicas`::
+The number of `nrt` replicas that should be created for the new shard (optional, the defaults for the collection is used if omitted)
+
+`tlogReplicas`::
+The number of `tlog` replicas that should be created for the new shard (optional, the defaults for the collection is used if omitted)
+
+`pullReplicas`::
+The number of `pull` replicas that should be created for the new shard (optional, the defaults for the collection is used if omitted)
+
`property._name_=_value_`::
Set core property _name_ to _value_. See the section <<defining-core-properties.adoc#defining-core-properties,Defining core.properties>> for details on supported properties and values.
@@ -1016,9 +1029,9 @@ http://localhost:8983/solr/admin/collections?action=DELETEREPLICA&collection=tes
[[addreplica]]
== ADDREPLICA: Add Replica
-Add a replica to a shard in a collection. The node name can be specified if the replica is to be created in a specific node.
+Add one or more replicas to a shard in a collection. The node name can be specified if the replica is to be created in a specific node. Otherwise, a set of nodes can be specified and the most suitable ones among them will be chosen to create the replica(s).
-The API uses the Autoscaling framework to find nodes that can satisfy the disk requirements for the new replica but only when an Autoscaling policy is configured. Refer to <<solrcloud-autoscaling-policy-preferences.adoc#solrcloud-autoscaling-policy-preferences,Autoscaling Policy and Preferences>> section for more details.
+The API uses the Autoscaling framework to find nodes that can satisfy the disk requirements for the new replica(s) but only when an Autoscaling preferences or policy is configured. Refer to <<solrcloud-autoscaling-policy-preferences.adoc#solrcloud-autoscaling-policy-preferences,Autoscaling Policy and Preferences>> section for more details.
`/admin/collections?action=ADDREPLICA&collection=_collection_&shard=_shard_&node=_nodeName_`
@@ -1038,7 +1051,14 @@ If the exact shard name is not known, users may pass the `\_route_` value and th
Ignored if the `shard` parameter is also specified.
`node`::
-The name of the node where the replica should be created.
+The name of the node where the replica should be created (optional)
+
+`createNodeSet`::
+A comma-separated list of nodes among which the best ones will be chosen to place the replicas (optional)
++
+The format is a comma-separated list of node_names, such as `localhost:8983_solr,localhost:8984_solr,localhost:8985_solr`.
+
+If neither `node`, nor `createNodeSet` is specified then the best node(s) from among all the live nodes in the cluster are chosen.
`instanceDir`::
The instanceDir for the core that will be created.
@@ -1057,6 +1077,15 @@ The type of replica to create. These possible values are allowed:
+
See the section <<shards-and-indexing-data-in-solrcloud.adoc#types-of-replicas,Types of Replicas>> for more information about replica type options.
+`nrtReplicas`::
+The number of `nrt` replicas that should be created (optional, defaults to 1 if `type` is `nrt` otherwise 0).
+
+`tlogReplicas`::
+The number of `tlog` replicas that should be created (optional, defaults to 1 if `type` is `tlog` otherwise 0).
+
+`pullReplicas`::
+The number of `pull` replicas that should be created (optional, defaults to 1 if `type` is `pull` otherwise 0).
+
`property._name_=_value_`::
Set core property _name_ to _value_. See <<defining-core-properties.adoc#defining-core-properties,Defining core.properties>> for details about supported properties and values.
@@ -1096,6 +1125,39 @@ http://localhost:8983/solr/admin/collections?action=ADDREPLICA&collection=test2&
</response>
----
+[source,text]
+----
+http://localhost:8983/solr/admin/collections?action=addreplica&collection=gettingstarted&shard=shard1&tlogReplicas=1&pullReplicas=1
+----
+
+*Output*
+
+[source,json]
+----
+{
+ "responseHeader": {
+ "status": 0,
+ "QTime": 784
+ },
+ "success": {
+ "127.0.1.1:7574_solr": {
+ "responseHeader": {
+ "status": 0,
+ "QTime": 257
+ },
+ "core": "gettingstarted_shard1_replica_p11"
+ },
+ "127.0.1.1:8983_solr": {
+ "responseHeader": {
+ "status": 0,
+ "QTime": 295
+ },
+ "core": "gettingstarted_shard1_replica_t10"
+ }
+ }
+}
+----
+
[[clusterprop]]
== CLUSTERPROP: Cluster Properties
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/0d7627f2/solr/solrj/src/java/org/apache/solr/client/solrj/request/CollectionAdminRequest.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/request/CollectionAdminRequest.java b/solr/solrj/src/java/org/apache/solr/client/solrj/request/CollectionAdminRequest.java
index 4cf68d4..0f6de19 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/request/CollectionAdminRequest.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/request/CollectionAdminRequest.java
@@ -57,7 +57,10 @@ 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.ZkStateReader.AUTO_ADD_REPLICAS;
import static org.apache.solr.common.cloud.ZkStateReader.MAX_SHARDS_PER_NODE;
+import static org.apache.solr.common.cloud.ZkStateReader.NRT_REPLICAS;
+import static org.apache.solr.common.cloud.ZkStateReader.PULL_REPLICAS;
import static org.apache.solr.common.cloud.ZkStateReader.REPLICATION_FACTOR;
+import static org.apache.solr.common.cloud.ZkStateReader.TLOG_REPLICAS;
import static org.apache.solr.common.params.CollectionAdminParams.COLL_CONF;
import static org.apache.solr.common.params.CollectionAdminParams.COLOCATED_WITH;
import static org.apache.solr.common.params.CollectionAdminParams.COUNT_PROP;
@@ -1648,6 +1651,8 @@ public abstract class CollectionAdminRequest<T extends CollectionAdminResponse>
protected String ulogDir;
protected Properties properties;
protected Replica.Type type;
+ protected Integer nrtReplicas, tlogReplicas, pullReplicas;
+ protected String createNodeSet;
private AddReplica(String collection, String shard, String routeKey, Replica.Type type) {
super(CollectionAction.ADDREPLICA);
@@ -1727,6 +1732,42 @@ public abstract class CollectionAdminRequest<T extends CollectionAdminResponse>
return shard;
}
+ public Integer getNrtReplicas() {
+ return nrtReplicas;
+ }
+
+ public AddReplica setNrtReplicas(Integer nrtReplicas) {
+ this.nrtReplicas = nrtReplicas;
+ return this;
+ }
+
+ public Integer getTlogReplicas() {
+ return tlogReplicas;
+ }
+
+ public AddReplica setTlogReplicas(Integer tlogReplicas) {
+ this.tlogReplicas = tlogReplicas;
+ return this;
+ }
+
+ public Integer getPullReplicas() {
+ return pullReplicas;
+ }
+
+ public AddReplica setPullReplicas(Integer pullReplicas) {
+ this.pullReplicas = pullReplicas;
+ return this;
+ }
+
+ public String getCreateNodeSet() {
+ return createNodeSet;
+ }
+
+ public AddReplica setCreateNodeSet(String createNodeSet) {
+ this.createNodeSet = createNodeSet;
+ return this;
+ }
+
@Override
public SolrParams getParams() {
ModifiableSolrParams params = new ModifiableSolrParams(super.getParams());
@@ -1759,6 +1800,18 @@ public abstract class CollectionAdminRequest<T extends CollectionAdminResponse>
if (properties != null) {
addProperties(params, properties);
}
+ if (nrtReplicas != null) {
+ params.add(NRT_REPLICAS, String.valueOf(nrtReplicas));
+ }
+ if (tlogReplicas != null) {
+ params.add(TLOG_REPLICAS, String.valueOf(tlogReplicas));
+ }
+ if (pullReplicas != null) {
+ params.add(PULL_REPLICAS, String.valueOf(pullReplicas));
+ }
+ if (createNodeSet != null) {
+ params.add(CREATE_NODE_SET_PARAM, createNodeSet);
+ }
return params;
}
[4/5] lucene-solr:branch_7x: SOLR-12648: Autoscaling framework based
replica placement is not used unless a policy is specified or non-empty
cluster policy exists
Posted by sh...@apache.org.
SOLR-12648: Autoscaling framework based replica placement is not used unless a policy is specified or non-empty cluster policy exists
(cherry picked from commit 65105aa81b1ac61b5e163ca249419184433233df)
Project: http://git-wip-us.apache.org/repos/asf/lucene-solr/repo
Commit: http://git-wip-us.apache.org/repos/asf/lucene-solr/commit/a4beebd4
Tree: http://git-wip-us.apache.org/repos/asf/lucene-solr/tree/a4beebd4
Diff: http://git-wip-us.apache.org/repos/asf/lucene-solr/diff/a4beebd4
Branch: refs/heads/branch_7x
Commit: a4beebd402b86a26e5205f5b49e44238134a2265
Parents: 17cb1b1
Author: Shalin Shekhar Mangar <sh...@apache.org>
Authored: Wed Oct 3 15:52:51 2018 +0530
Committer: Shalin Shekhar Mangar <sh...@apache.org>
Committed: Wed Oct 3 15:57:32 2018 +0530
----------------------------------------------------------------------
solr/CHANGES.txt | 3 +
.../java/org/apache/solr/cloud/CloudUtil.java | 49 +++-
.../solr/cloud/api/collections/Assign.java | 3 +-
.../api/collections/CreateCollectionCmd.java | 4 +-
.../client/solrj/cloud/autoscaling/Policy.java | 20 +-
.../solrj/cloud/autoscaling/Preference.java | 8 +
.../solrj/cloud/autoscaling/TestPolicy.java | 237 +++++++++++++++++++
7 files changed, 308 insertions(+), 16 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a4beebd4/solr/CHANGES.txt
----------------------------------------------------------------------
diff --git a/solr/CHANGES.txt b/solr/CHANGES.txt
index 85112e5..9c55193 100644
--- a/solr/CHANGES.txt
+++ b/solr/CHANGES.txt
@@ -61,6 +61,9 @@ Bug Fixes
* SOLR-12776: Setting of TMP in solr.cmd causes invisibility of Solr to JDK tools (Petr Bodnar via Erick Erickson)
+* SOLR-12648: Autoscaling framework based replica placement is not used unless a policy is specified or
+ non-empty cluster policy exists. (shalin)
+
================== 7.5.0 ==================
Consult the LUCENE_CHANGES.txt file for additional, low level, changes in this release.
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a4beebd4/solr/core/src/java/org/apache/solr/cloud/CloudUtil.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/CloudUtil.java b/solr/core/src/java/org/apache/solr/cloud/CloudUtil.java
index 13734f6..26ba1b8 100644
--- a/solr/core/src/java/org/apache/solr/cloud/CloudUtil.java
+++ b/solr/core/src/java/org/apache/solr/cloud/CloudUtil.java
@@ -23,10 +23,13 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
import org.apache.commons.io.FileUtils;
-import org.apache.solr.client.solrj.cloud.autoscaling.AutoScalingConfig;
import org.apache.solr.client.solrj.cloud.SolrCloudManager;
+import org.apache.solr.client.solrj.cloud.autoscaling.AutoScalingConfig;
+import org.apache.solr.client.solrj.cloud.autoscaling.Policy;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.SolrException.ErrorCode;
import org.apache.solr.common.cloud.ClusterState;
@@ -45,8 +48,8 @@ import org.slf4j.LoggerFactory;
public class CloudUtil {
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
-
-
+
+
/**
* See if coreNodeName has been taken over by another baseUrl and unload core
* + throw exception if it has been.
@@ -57,26 +60,26 @@ public class CloudUtil {
ZkController zkController = cc.getZkController();
String thisCnn = zkController.getCoreNodeName(desc);
String thisBaseUrl = zkController.getBaseUrl();
-
+
log.debug("checkSharedFSFailoverReplaced running for coreNodeName={} baseUrl={}", thisCnn, thisBaseUrl);
// if we see our core node name on a different base url, unload
- final DocCollection docCollection = zkController.getClusterState().getCollectionOrNull(desc.getCloudDescriptor().getCollectionName());
+ final DocCollection docCollection = zkController.getClusterState().getCollectionOrNull(desc.getCloudDescriptor().getCollectionName());
if (docCollection != null && docCollection.getSlicesMap() != null) {
Map<String,Slice> slicesMap = docCollection.getSlicesMap();
for (Slice slice : slicesMap.values()) {
for (Replica replica : slice.getReplicas()) {
-
+
String cnn = replica.getName();
String baseUrl = replica.getStr(ZkStateReader.BASE_URL_PROP);
log.debug("compare against coreNodeName={} baseUrl={}", cnn, baseUrl);
-
+
if (thisCnn != null && thisCnn.equals(cnn)
&& !thisBaseUrl.equals(baseUrl)) {
if (cc.getLoadedCoreNames().contains(desc.getName())) {
cc.unload(desc.getName());
}
-
+
try {
FileUtils.deleteDirectory(desc.getInstanceDir().toFile());
} catch (IOException e) {
@@ -144,9 +147,37 @@ public class CloudUtil {
}
+ /**
+ * <b>Note:</b> where possible, the {@link #usePolicyFramework(DocCollection, SolrCloudManager)} method should
+ * be used instead of this method
+ *
+ * @return true if autoscaling policy framework should be used for replica placement
+ */
+ public static boolean usePolicyFramework(SolrCloudManager cloudManager) throws IOException, InterruptedException {
+ Objects.requireNonNull(cloudManager, "The SolrCloudManager instance cannot be null");
+ return usePolicyFramework(Optional.empty(), cloudManager);
+ }
+
+ /**
+ * @return true if auto scaling policy framework should be used for replica placement
+ * for this collection, otherwise false
+ */
public static boolean usePolicyFramework(DocCollection collection, SolrCloudManager cloudManager)
throws IOException, InterruptedException {
+ Objects.requireNonNull(collection, "The DocCollection instance cannot be null");
+ Objects.requireNonNull(cloudManager, "The SolrCloudManager instance cannot be null");
+ return usePolicyFramework(Optional.of(collection), cloudManager);
+ }
+
+ private static boolean usePolicyFramework(Optional<DocCollection> collection, SolrCloudManager cloudManager) throws IOException, InterruptedException {
AutoScalingConfig autoScalingConfig = cloudManager.getDistribStateManager().getAutoScalingConfig();
- return !autoScalingConfig.getPolicy().getClusterPolicy().isEmpty() || collection.getPolicyName() != null;
+ // if no autoscaling configuration exists then obviously we cannot use the policy framework
+ if (autoScalingConfig.getPolicy().isEmpty()) return false;
+ // do custom preferences exist
+ if (!autoScalingConfig.getPolicy().getClusterPreferences().equals(Policy.DEFAULT_PREFERENCES)) return true;
+ // does a cluster policy exist
+ if (!autoScalingConfig.getPolicy().getClusterPolicy().isEmpty()) return true;
+ // finally we check if the current collection has a policy
+ return !collection.isPresent() || collection.get().getPolicyName() != null;
}
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a4beebd4/solr/core/src/java/org/apache/solr/cloud/api/collections/Assign.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/api/collections/Assign.java b/solr/core/src/java/org/apache/solr/cloud/api/collections/Assign.java
index 542ca1b..a24ad1a 100644
--- a/solr/core/src/java/org/apache/solr/cloud/api/collections/Assign.java
+++ b/solr/core/src/java/org/apache/solr/cloud/api/collections/Assign.java
@@ -41,6 +41,7 @@ import org.apache.solr.client.solrj.cloud.autoscaling.AutoScalingConfig;
import org.apache.solr.client.solrj.cloud.autoscaling.BadVersionException;
import org.apache.solr.client.solrj.cloud.autoscaling.PolicyHelper;
import org.apache.solr.client.solrj.cloud.autoscaling.VersionedData;
+import org.apache.solr.cloud.CloudUtil;
import org.apache.solr.cloud.rule.ReplicaAssigner;
import org.apache.solr.cloud.rule.Rule;
import org.apache.solr.common.SolrException;
@@ -583,7 +584,7 @@ public class Assign {
AutoScalingConfig autoScalingConfig = solrCloudManager.getDistribStateManager().getAutoScalingConfig();
StrategyType strategyType = null;
- if ((ruleMaps == null || ruleMaps.isEmpty()) && policyName == null && autoScalingConfig.getPolicy().getClusterPolicy().isEmpty()) {
+ if ((ruleMaps == null || ruleMaps.isEmpty()) && !CloudUtil.usePolicyFramework(collection, solrCloudManager)) {
strategyType = StrategyType.LEGACY;
} else if (ruleMaps != null && !ruleMaps.isEmpty()) {
strategyType = StrategyType.RULES;
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a4beebd4/solr/core/src/java/org/apache/solr/cloud/api/collections/CreateCollectionCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/api/collections/CreateCollectionCmd.java b/solr/core/src/java/org/apache/solr/cloud/api/collections/CreateCollectionCmd.java
index 542345d..212437e 100644
--- a/solr/core/src/java/org/apache/solr/cloud/api/collections/CreateCollectionCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/api/collections/CreateCollectionCmd.java
@@ -36,7 +36,6 @@ import java.util.concurrent.atomic.AtomicReference;
import org.apache.solr.client.solrj.cloud.DistribStateManager;
import org.apache.solr.client.solrj.cloud.SolrCloudManager;
import org.apache.solr.client.solrj.cloud.autoscaling.AlreadyExistsException;
-import org.apache.solr.client.solrj.cloud.autoscaling.AutoScalingConfig;
import org.apache.solr.client.solrj.cloud.autoscaling.BadVersionException;
import org.apache.solr.client.solrj.cloud.autoscaling.NotEmptyException;
import org.apache.solr.client.solrj.cloud.autoscaling.Policy;
@@ -134,8 +133,7 @@ public class CreateCollectionCmd implements OverseerCollectionMessageHandler.Cmd
String router = message.getStr("router.name", DocRouter.DEFAULT_NAME);
String policy = message.getStr(Policy.POLICY);
- AutoScalingConfig autoScalingConfig = ocmh.cloudManager.getDistribStateManager().getAutoScalingConfig();
- boolean usePolicyFramework = !autoScalingConfig.getPolicy().getClusterPolicy().isEmpty() || policy != null;
+ boolean usePolicyFramework = CloudUtil.usePolicyFramework(ocmh.cloudManager) || policy != null;
// fail fast if parameters are wrong or incomplete
List<String> shardNames = populateShardNames(message, router);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a4beebd4/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Policy.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Policy.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Policy.java
index b39951a..78043df 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Policy.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Policy.java
@@ -94,7 +94,15 @@ public class Policy implements MapWriter {
final List<Pair<String, Type>> params;
final List<String> perReplicaAttributes;
final int zkVersion;
+ /**
+ * True if cluster policy, preferences and custom policies are all non-existent
+ */
final boolean empty;
+ /**
+ * True if cluster preferences was originally empty, false otherwise. It is used to figure out if
+ * the current preferences were implicitly added or not.
+ */
+ final boolean emptyPreferences;
public Policy() {
this(Collections.emptyMap());
@@ -115,7 +123,8 @@ public class Policy implements MapWriter {
Preference preference = initialClusterPreferences.get(i);
preference.next = initialClusterPreferences.get(i + 1);
}
- if (initialClusterPreferences.isEmpty()) {
+ emptyPreferences = initialClusterPreferences.isEmpty();
+ if (emptyPreferences) {
initialClusterPreferences.addAll(DEFAULT_PREFERENCES);
}
this.clusterPreferences = Collections.unmodifiableList(initialClusterPreferences);
@@ -162,7 +171,8 @@ public class Policy implements MapWriter {
this.zkVersion = version;
this.policies = policies != null ? Collections.unmodifiableMap(policies) : Collections.emptyMap();
this.clusterPolicy = clusterPolicy != null ? Collections.unmodifiableList(clusterPolicy) : Collections.emptyList();
- this.clusterPreferences = clusterPreferences != null ? Collections.unmodifiableList(clusterPreferences) : DEFAULT_PREFERENCES;
+ this.emptyPreferences = clusterPreferences == null;
+ this.clusterPreferences = emptyPreferences ? DEFAULT_PREFERENCES : Collections.unmodifiableList(clusterPreferences);
this.params = Collections.unmodifiableList(
buildParams(this.clusterPreferences, this.clusterPolicy, this.policies).stream()
.map(s -> new Pair<>(s, VariableBase.getTagType(s)))
@@ -211,6 +221,10 @@ public class Policy implements MapWriter {
@Override
public void writeMap(EntryWriter ew) throws IOException {
+ // if we were initially empty then we don't want to persist any implicitly added
+ // policy or preferences
+ if (empty) return;
+
if (!policies.isEmpty()) {
ew.put(POLICIES, (MapWriter) ew1 -> {
for (Map.Entry<String, List<Clause>> e : policies.entrySet()) {
@@ -218,7 +232,7 @@ public class Policy implements MapWriter {
}
});
}
- if (!clusterPreferences.isEmpty()) {
+ if (!emptyPreferences && !clusterPreferences.isEmpty()) {
ew.put(CLUSTER_PREFERENCES, (IteratorWriter) iw -> {
for (Preference p : clusterPreferences) iw.add(p);
});
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a4beebd4/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Preference.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Preference.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Preference.java
index 422f987..127012a 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Preference.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Preference.java
@@ -19,6 +19,7 @@ package org.apache.solr.client.solrj.cloud.autoscaling;
import java.io.IOException;
import java.lang.invoke.MethodHandles;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -137,4 +138,11 @@ public class Preference implements MapWriter {
public String toString() {
return Utils.toJSONString(this);
}
+
+ /**
+ * @return an unmodifiable copy of the original map from which this object was constructed
+ */
+ public Map getOriginal() {
+ return Collections.unmodifiableMap(original);
+ }
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a4beebd4/solr/solrj/src/test/org/apache/solr/client/solrj/cloud/autoscaling/TestPolicy.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/test/org/apache/solr/client/solrj/cloud/autoscaling/TestPolicy.java b/solr/solrj/src/test/org/apache/solr/client/solrj/cloud/autoscaling/TestPolicy.java
index 29e45ed..23184a0 100644
--- a/solr/solrj/src/test/org/apache/solr/client/solrj/cloud/autoscaling/TestPolicy.java
+++ b/solr/solrj/src/test/org/apache/solr/client/solrj/cloud/autoscaling/TestPolicy.java
@@ -21,6 +21,7 @@ package org.apache.solr.client.solrj.cloud.autoscaling;
import java.io.IOException;
import java.io.StringWriter;
import java.lang.invoke.MethodHandles;
+import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@@ -33,6 +34,7 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
@@ -47,11 +49,15 @@ import org.apache.solr.client.solrj.cloud.autoscaling.Suggester.Hint;
import org.apache.solr.client.solrj.impl.ClusterStateProvider;
import org.apache.solr.client.solrj.impl.SolrClientNodeStateProvider;
import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.cloud.Overseer;
+import org.apache.solr.cloud.api.collections.Assign;
+import org.apache.solr.common.MapWriter;
import org.apache.solr.common.cloud.ClusterState;
import org.apache.solr.common.cloud.DocCollection;
import org.apache.solr.common.cloud.DocRouter;
import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.cloud.ReplicaPosition;
+import org.apache.solr.common.cloud.ZkNodeProps;
import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.cloud.rule.ImplicitSnitch;
import org.apache.solr.common.params.CollectionParams;
@@ -71,9 +77,13 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.solr.client.solrj.cloud.autoscaling.Policy.CLUSTER_PREFERENCES;
import static org.apache.solr.client.solrj.cloud.autoscaling.Variable.Type.CORES;
import static org.apache.solr.client.solrj.cloud.autoscaling.Variable.Type.FREEDISK;
import static org.apache.solr.client.solrj.cloud.autoscaling.Variable.Type.REPLICA;
+import static org.apache.solr.common.cloud.ZkStateReader.COLLECTION_PROP;
+import static org.apache.solr.common.cloud.ZkStateReader.REPLICA_TYPE;
+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.MOVEREPLICA;
@@ -3036,5 +3046,232 @@ public class TestPolicy extends SolrTestCaseJ4 {
}
+ public void testAutoscalingPreferencesUsedWithNoPolicy() throws IOException, InterruptedException {
+ String dataproviderdata = "{" +
+ " 'liveNodes': [" +
+ " 'node1:8983'," +
+ " 'node2:8984'," +
+ " 'node3:8985'" +
+ " ]," +
+ " 'replicaInfo': {" +
+ " 'node1:8983': {" +
+ " 'c1': {" +
+ " 's1': [" +
+ " {'r1': {'type': 'NRT', 'INDEX.sizeInGB':'1100'}}," +
+ " {'r2': {'type': 'NRT'}}" +
+ " ]," +
+ " 's2': [" +
+ " {'r1': {'type': 'NRT', 'INDEX.sizeInGB':'1100'}}," +
+ " {'r2': {'type': 'NRT'}}" +
+ " ]" +
+ " }" +
+ " }" +
+ " }," +
+ " 'nodeValues': {" +
+ " 'node1:8983': {" +
+ " 'cores': 4," +
+ " 'freedisk': 300," +
+ " 'totaldisk': 4700," +
+ " 'port': 8983" +
+ " }," +
+ " 'node2:8984': {" +
+ " 'cores': 0," +
+ " 'freedisk': 1000," +
+ " 'totaldisk': 1200," +
+ " 'port': 8984" +
+ " }," +
+ " 'node3:8985': {" +
+ " 'cores': 0," +
+ " 'freedisk': 1651," +
+ " 'totaldisk': 1700," +
+ " 'port': 8985" +
+ " }" +
+ " }," +
+ " 'autoscalingJson': {" +
+ " 'cluster-preferences': [" +
+ " { 'maximize': 'freedisk'}," +
+ " { 'minimize': 'cores', 'precision': 3}" +
+ " ]" +
+ " }" +
+ "}";
+
+ String clusterState = "{\n" +
+ " \"c1\" : {\n" +
+ " \"router\":{\"name\":\"compositeId\"},\n" +
+ " \"maxShardsPerNode\":-1,\n" +
+ " \"shards\" : {\n" +
+ " \"s1\" : {\n" +
+ " \"replicas\" : {\n" +
+ " \"r1\" : {\n" +
+ " \"type\" : \"NRT\",\n" +
+ " \"node_name\" : \"node1:8983\",\n" +
+ " \"state\" : \"active\",\n" +
+ " \"leader\" : \"true\"\n" +
+ " },\n" +
+ " \"r2\" : {\n" +
+ " \"type\" : \"NRT\",\n" +
+ " \"node_name\" : \"node1:8983\",\n" +
+ " \"state\" : \"active\"\n" +
+ " }\n" +
+ " }\n" +
+ " },\n" +
+ " \"s2\" : {\n" +
+ " \"replicas\" : {\n" +
+ " \"r1\" : {\n" +
+ " \"type\" : \"NRT\",\n" +
+ " \"node_name\" : \"node1:8983\",\n" +
+ " \"state\" : \"active\",\n" +
+ " \"leader\" : \"true\"\n" +
+ " },\n" +
+ " \"r2\" : {\n" +
+ " \"type\" : \"NRT\",\n" +
+ " \"node_name\" : \"node1:8983\",\n" +
+ " \"state\" : \"active\"\n" +
+ " }\n" +
+ " }\n" +
+ " }\n" +
+ " }\n" +
+ " }\n" +
+ "}";
+
+ Map m = (Map) Utils.fromJSONString(dataproviderdata);
+
+ Map replicaInfo = (Map) m.get("replicaInfo");
+ replicaInfo.forEach((node, val) -> {
+ Map m1 = (Map) val;
+ m1.forEach((coll, val2) -> {
+ Map m2 = (Map) val2;
+ m2.forEach((shard, val3) -> {
+ List l3 = (List) val3;
+ for (int i = 0; i < l3.size(); i++) {
+ Object o = l3.get(i);
+ Map m3 = (Map) o;
+ String name = m3.keySet().iterator().next().toString();
+ m3 = (Map) m3.get(name);
+ Replica.Type type = Replica.Type.get((String) m3.get("type"));
+ l3.set(i, new ReplicaInfo(name, name
+ , coll.toString(), shard.toString(), type, (String) node, m3));
+ }
+ });
+
+ });
+ });
+ AutoScalingConfig asc = m.containsKey("autoscalingJson") ? new AutoScalingConfig((Map<String, Object>) m.get("autoscalingJson")) : null;
+ DelegatingCloudManager cloudManager = new DelegatingCloudManager(null) {
+
+ @Override
+ public DistribStateManager getDistribStateManager() {
+ return new DelegatingDistribStateManager(null) {
+ @Override
+ public AutoScalingConfig getAutoScalingConfig() {
+ return asc;
+ }
+ };
+ }
+
+ @Override
+ public ClusterStateProvider getClusterStateProvider() {
+ return new DelegatingClusterStateProvider(null) {
+ @Override
+ public Set<String> getLiveNodes() {
+ return new HashSet<>((Collection<String>) m.get("liveNodes"));
+ }
+
+ @Override
+ public ClusterState getClusterState() throws IOException {
+ return ClusterState.load(0, clusterState.getBytes(Charset.forName("UTF-8")), getLiveNodes(), ZkStateReader.getCollectionPath("c1"));
+ }
+ };
+ }
+ @Override
+ public NodeStateProvider getNodeStateProvider() {
+ return new DelegatingNodeStateProvider(null) {
+ @Override
+ public Map<String, Object> getNodeValues(String node, Collection<String> tags) {
+ Map<String, Object> result = (Map<String, Object>) Utils.getObjectByPath(m, false, Arrays.asList("nodeValues", node));
+ return result == null ? Collections.emptyMap() : result;
+ }
+
+ @Override
+ public Map<String, Map<String, List<ReplicaInfo>>> getReplicaInfo(String node, Collection<String> keys) {
+ Map<String, Map<String, List<ReplicaInfo>>> result = (Map<String, Map<String, List<ReplicaInfo>>>) Utils.getObjectByPath(m, false, Arrays.asList("replicaInfo", node));
+ return result == null ? Collections.emptyMap() : result;
+ }
+ };
+ }
+ };
+
+ ZkNodeProps message = new ZkNodeProps(
+ Overseer.QUEUE_OPERATION, ADDREPLICA.toLower(),
+ COLLECTION_PROP, "c1",
+ SHARD_ID_PROP, "s1",
+ REPLICA_TYPE, Replica.Type.NRT.toString()
+ );
+
+ Assign.AssignRequest assignRequest = new Assign.AssignRequestBuilder()
+ .forCollection("c1")
+ .forShard(Collections.singletonList("s1"))
+ .assignNrtReplicas(1)
+ .build();
+ Assign.AssignStrategyFactory assignStrategyFactory = new Assign.AssignStrategyFactory(cloudManager);
+ ClusterState state = cloudManager.getClusterStateProvider().getClusterState();
+ DocCollection collection = state.getCollection("c1");
+ Assign.AssignStrategy assignStrategy = assignStrategyFactory.create(state, collection);
+ List<ReplicaPosition> replicaPositions = assignStrategy.assign(cloudManager, assignRequest);
+
+ assertEquals(1, replicaPositions.size());
+ ReplicaPosition replicaPosition = replicaPositions.get(0);
+ assertEquals("node3:8985", replicaPosition.node); // only node3:8985 has enough space to handle the new replica
+ assertEquals("s1", replicaPosition.shard); // sanity check
+ }
+
+ /**
+ * Tests that an empty policy should not persist implicitly added keys to MapWriter
+ * <p>
+ * The reason behind doing this is to ensure that implicitly added cluster preferences do not ever
+ * go to ZooKeeper so that we can decide whether to enable autoscaling policy framework or not.
+ *
+ * @see org.apache.solr.cloud.CloudUtil#usePolicyFramework(DocCollection, SolrCloudManager)
+ */
+ public void testPolicyMapWriterWithEmptyPreferences() throws IOException {
+ List<Map> defaultPreferences = Policy.DEFAULT_PREFERENCES
+ .stream().map(preference -> preference.getOriginal()).collect(Collectors.toList());
+
+ // first we create a completely empty policy
+ Policy policy = new Policy();
+ // sanity check that the default cluster preferences were added implicitly
+ assertNotNull(policy.getClusterPreferences());
+ // and they were the same as the default preferences
+ assertEquals(policy.getClusterPreferences().size(), defaultPreferences.size());
+ Set<String> writtenKeys = new HashSet<>();
+ policy.writeMap(new MapWriter.EntryWriter() {
+ @Override
+ public MapWriter.EntryWriter put(String k, Object v) throws IOException {
+ writtenKeys.add(k);
+ return this;
+ }
+ });
+ // but those implicitly added cluster preferences are never written by MapWriter
+ assertEquals(0, writtenKeys.size());
+
+ // reset
+ writtenKeys.clear();
+ // now we create a policy that only has cluster preferences which happen to be the same as the default
+ // preferences
+ policy = new Policy(Utils.makeMap(CLUSTER_PREFERENCES, defaultPreferences));
+ // sanity checks
+ assertNotNull(policy.getClusterPreferences());
+ assertEquals(policy.getClusterPreferences().size(), defaultPreferences.size());
+ policy.writeMap(new MapWriter.EntryWriter() {
+ @Override
+ public MapWriter.EntryWriter put(String k, Object v) throws IOException {
+ writtenKeys.add(k);
+ return this;
+ }
+ });
+ // since the user explicitly added those preferences, they should be written by MapWriter
+ assertEquals(1, writtenKeys.size());
+ assertTrue(writtenKeys.contains(CLUSTER_PREFERENCES));
+ }
}
[3/5] lucene-solr:branch_7x: SOLR-12756: Refactor Assign and extract
replica placement strategies out of it.
Posted by sh...@apache.org.
SOLR-12756: Refactor Assign and extract replica placement strategies out of it.
Now, assignment is done with the help of a builder class instead of calling a method with large number of arguments. The number of special cases that had to be handled have been cut down as well.
(cherry picked from commit c587410f99375005c680ece5e24a4dfd40d8d3eb)
Project: http://git-wip-us.apache.org/repos/asf/lucene-solr/repo
Commit: http://git-wip-us.apache.org/repos/asf/lucene-solr/commit/17cb1b17
Tree: http://git-wip-us.apache.org/repos/asf/lucene-solr/tree/17cb1b17
Diff: http://git-wip-us.apache.org/repos/asf/lucene-solr/diff/17cb1b17
Branch: refs/heads/branch_7x
Commit: 17cb1b17172926d0d9aed3dfd3b9adb90cf65e0f
Parents: f08fee2
Author: Shalin Shekhar Mangar <sh...@apache.org>
Authored: Thu Sep 27 16:15:38 2018 +0530
Committer: Shalin Shekhar Mangar <sh...@apache.org>
Committed: Wed Oct 3 15:57:14 2018 +0530
----------------------------------------------------------------------
solr/CHANGES.txt | 4 +
.../cloud/api/collections/AddReplicaCmd.java | 45 +--
.../solr/cloud/api/collections/Assign.java | 358 ++++++++++++-------
.../api/collections/CreateCollectionCmd.java | 34 +-
.../cloud/api/collections/ReplaceNodeCmd.java | 33 +-
.../solr/cloud/api/collections/RestoreCmd.java | 16 +-
.../cloud/api/collections/SplitShardCmd.java | 17 +-
.../solr/cloud/overseer/ReplicaMutator.java | 4 +-
.../CollectionTooManyReplicasTest.java | 8 +-
.../solr/cloud/autoscaling/TestPolicyCloud.java | 1 +
.../sim/SimClusterStateProvider.java | 24 +-
.../autoscaling/sim/TestSimPolicyCloud.java | 1 +
.../solr/common/cloud/ReplicaPosition.java | 2 +-
13 files changed, 331 insertions(+), 216 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/17cb1b17/solr/CHANGES.txt
----------------------------------------------------------------------
diff --git a/solr/CHANGES.txt b/solr/CHANGES.txt
index b14e9ad..85112e5 100644
--- a/solr/CHANGES.txt
+++ b/solr/CHANGES.txt
@@ -49,6 +49,10 @@ Other Changes
* SOLR-12762: Fix javadoc for SolrCloudTestCase.clusterShape() method and add a method that validates only against
Active slices (Anshum Gupta)
+* SOLR-12756: Refactor Assign and extract replica placement strategies out of it. Now, assignment is done with the help
+ of a builder class instead of calling a method with large number of arguments. The number of special cases that had
+ to be handled have been cut down as well. (shalin)
+
Bug Fixes
----------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/17cb1b17/solr/core/src/java/org/apache/solr/cloud/api/collections/AddReplicaCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/api/collections/AddReplicaCmd.java b/solr/core/src/java/org/apache/solr/cloud/api/collections/AddReplicaCmd.java
index f128c2e..6e851db 100644
--- a/solr/core/src/java/org/apache/solr/cloud/api/collections/AddReplicaCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/api/collections/AddReplicaCmd.java
@@ -34,10 +34,8 @@ import java.util.stream.Collectors;
import org.apache.commons.lang.StringUtils;
import org.apache.solr.client.solrj.cloud.SolrCloudManager;
-import org.apache.solr.client.solrj.cloud.autoscaling.Policy;
import org.apache.solr.client.solrj.cloud.autoscaling.PolicyHelper;
import org.apache.solr.cloud.ActiveReplicaWatcher;
-import org.apache.solr.cloud.CloudUtil;
import org.apache.solr.cloud.Overseer;
import org.apache.solr.common.SolrCloseableLatch;
import org.apache.solr.common.SolrException;
@@ -76,6 +74,12 @@ import static org.apache.solr.common.params.CommonAdminParams.WAIT_FOR_FINAL_STA
public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+ /**
+ * When AddReplica is called with this set to true, then we do not try to find node assignments
+ * for the add replica API. If set to true, a valid "node" should be specified.
+ */
+ public static final String SKIP_NODE_ASSIGNMENT = "skipNodeAssignment";
+
private final OverseerCollectionMessageHandler ocmh;
public AddReplicaCmd(OverseerCollectionMessageHandler ocmh) {
@@ -213,6 +217,8 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
ZkStateReader.COLLECTION_PROP, withCollectionName,
ZkStateReader.SHARD_ID_PROP, withCollectionShard,
"node", createReplica.node,
+ // since we already computed node assignments (which include assigning a node for this withCollection replica) we want to skip the assignment step
+ SKIP_NODE_ASSIGNMENT, "true",
CommonAdminParams.WAIT_FOR_FINAL_STATE, Boolean.TRUE.toString()); // set to true because we want `withCollection` to be ready after this collection is created
addReplica(clusterState, props, results, null);
}
@@ -300,7 +306,7 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
coreName = message.getStr(CoreAdminParams.PROPERTY_PREFIX + CoreAdminParams.NAME);
}
- log.info("Node Identified {} for creating new replica of shard {}", node, shard);
+ log.info("Node Identified {} for creating new replica of shard {} for collection {}", node, shard, collection);
if (!clusterState.liveNodesContain(node)) {
throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "Node: " + node + " is not live");
}
@@ -327,6 +333,7 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
EnumMap<Replica.Type, Integer> replicaTypeVsCount,
AtomicReference< PolicyHelper.SessionWrapper> sessionWrapper) throws IOException, InterruptedException {
boolean skipCreateReplicaInClusterState = message.getBool(SKIP_CREATE_REPLICA_IN_CLUSTER_STATE, false);
+ boolean skipNodeAssignment = message.getBool(SKIP_NODE_ASSIGNMENT, false);
String sliceName = message.getStr(SHARD_ID_PROP);
DocCollection collection = clusterState.getCollection(collectionName);
@@ -345,33 +352,11 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
}
List<ReplicaPosition> positions = null;
- if (!skipCreateReplicaInClusterState) {
- if (CloudUtil.usePolicyFramework(collection, cloudManager)) {
- if (node == null) {
- if (collection.getPolicyName() != null) message.getProperties().put(Policy.POLICY, collection.getPolicyName());
- positions = Assign.identifyNodes(cloudManager,
- clusterState,
- Assign.getLiveOrLiveAndCreateNodeSetList(clusterState.getLiveNodes(), message, OverseerCollectionMessageHandler.RANDOM),
- collection.getName(),
- message,
- Collections.singletonList(sliceName),
- numNrtReplicas,
- numTlogReplicas,
- numPullReplicas);
- sessionWrapper.set(PolicyHelper.getLastSessionWrapper(true));
- }
- } else {
- List<Assign.ReplicaCount> sortedNodeList = Assign.getNodesForNewReplicas(clusterState, collection.getName(), sliceName, numNrtReplicas,
- numTlogReplicas, numPullReplicas, createNodeSetStr, cloudManager);
- int i = 0;
- positions = new ArrayList<>();
- for (Map.Entry<Replica.Type, Integer> e : replicaTypeVsCount.entrySet()) {
- for (int j = 0; j < e.getValue(); j++) {
- positions.add(new ReplicaPosition(sliceName, j + 1, e.getKey(), sortedNodeList.get(i % sortedNodeList.size()).nodeName));
- i++;
- }
- }
- }
+ if (!skipCreateReplicaInClusterState && !skipNodeAssignment) {
+
+ positions = Assign.getNodesForNewReplicas(clusterState, collection.getName(), sliceName, numNrtReplicas,
+ numTlogReplicas, numPullReplicas, createNodeSetStr, cloudManager);
+ sessionWrapper.set(PolicyHelper.getLastSessionWrapper(true));
}
if (positions == null) {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/17cb1b17/solr/core/src/java/org/apache/solr/cloud/api/collections/Assign.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/api/collections/Assign.java b/solr/core/src/java/org/apache/solr/cloud/api/collections/Assign.java
index 9b33f52..542ca1b 100644
--- a/solr/core/src/java/org/apache/solr/cloud/api/collections/Assign.java
+++ b/solr/core/src/java/org/apache/solr/cloud/api/collections/Assign.java
@@ -21,12 +21,14 @@ import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
+import java.util.Comparator;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
+import java.util.Objects;
import java.util.Random;
import java.util.Set;
import java.util.stream.Collectors;
@@ -217,12 +219,7 @@ public class Assign {
}
public static List<String> getLiveOrLiveAndCreateNodeSetList(final Set<String> liveNodes, final ZkNodeProps message, final Random random) {
- // TODO: add smarter options that look at the current number of cores per
- // node?
- // for now we just go random (except when createNodeSet and createNodeSet.shuffle=false are passed in)
-
List<String> nodeList;
-
final String createNodeSetStr = message.getStr(CREATE_NODE_SET);
final List<String> createNodeList = (createNodeSetStr == null) ? null :
StrUtils.splitSmart((OverseerCollectionMessageHandler.CREATE_NODE_SET_EMPTY.equals(createNodeSetStr) ?
@@ -243,67 +240,6 @@ public class Assign {
return nodeList;
}
- public static List<ReplicaPosition> identifyNodes(SolrCloudManager cloudManager,
- ClusterState clusterState,
- List<String> nodeList,
- String collectionName,
- ZkNodeProps message,
- List<String> shardNames,
- int numNrtReplicas,
- int numTlogReplicas,
- int numPullReplicas) throws IOException, InterruptedException, AssignmentException {
- List<Map> rulesMap = (List) message.get("rule");
- String policyName = message.getStr(POLICY);
- AutoScalingConfig autoScalingConfig = cloudManager.getDistribStateManager().getAutoScalingConfig();
-
- if (rulesMap == null && policyName == null && autoScalingConfig.getPolicy().getClusterPolicy().isEmpty()) {
- log.debug("Identify nodes using default");
- int i = 0;
- List<ReplicaPosition> result = new ArrayList<>();
- for (String aShard : shardNames)
- for (Map.Entry<Replica.Type, Integer> e : ImmutableMap.of(Replica.Type.NRT, numNrtReplicas,
- Replica.Type.TLOG, numTlogReplicas,
- Replica.Type.PULL, numPullReplicas
- ).entrySet()) {
- for (int j = 0; j < e.getValue(); j++){
- result.add(new ReplicaPosition(aShard, j, e.getKey(), nodeList.get(i % nodeList.size())));
- i++;
- }
- }
- return result;
- } else {
- if (numTlogReplicas + numPullReplicas != 0 && rulesMap != null) {
- throw new SolrException(SolrException.ErrorCode.BAD_REQUEST,
- Replica.Type.TLOG + " or " + Replica.Type.PULL + " replica types not supported with placement rules");
- }
- }
-
- if (rulesMap != null && !rulesMap.isEmpty()) {
- List<Rule> rules = new ArrayList<>();
- for (Object map : rulesMap) rules.add(new Rule((Map) map));
- Map<String, Integer> sharVsReplicaCount = new HashMap<>();
-
- for (String shard : shardNames) sharVsReplicaCount.put(shard, numNrtReplicas);
- ReplicaAssigner replicaAssigner = new ReplicaAssigner(rules,
- sharVsReplicaCount,
- (List<Map>) message.get(SNITCH),
- new HashMap<>(),//this is a new collection. So, there are no nodes in any shard
- nodeList,
- cloudManager,
- clusterState);
-
- Map<ReplicaPosition, String> nodeMappings = replicaAssigner.getNodeMappings();
- return nodeMappings.entrySet().stream()
- .map(e -> new ReplicaPosition(e.getKey().shard, e.getKey().index, e.getKey().type, e.getValue()))
- .collect(Collectors.toList());
- } else {
- if (message.getStr(CREATE_NODE_SET) == null)
- nodeList = Collections.emptyList();// unless explicitly specified do not pass node list to Policy
- return getPositionsUsingPolicy(collectionName,
- shardNames, numNrtReplicas, numTlogReplicas, numPullReplicas, policyName, cloudManager, nodeList);
- }
- }
-
static class ReplicaCount {
public final String nodeName;
public int thisCollectionNodes = 0;
@@ -318,11 +254,11 @@ public class Assign {
}
}
- // Only called from createShard and addReplica (so far).
+ // Only called from addReplica (and by extension createShard) (so far).
//
// Gets a list of candidate nodes to put the required replica(s) on. Throws errors if not enough replicas
// could be created on live nodes given maxShardsPerNode, Replication factor (if from createShard) etc.
- public static List<ReplicaCount> getNodesForNewReplicas(ClusterState clusterState, String collectionName,
+ public static List<ReplicaPosition> getNodesForNewReplicas(ClusterState clusterState, String collectionName,
String shard, int nrtReplicas, int tlogReplicas, int pullReplicas,
Object createNodeSet, SolrCloudManager cloudManager) throws IOException, InterruptedException, AssignmentException {
log.debug("getNodesForNewReplicas() shard: {} , nrtReplicas : {} , tlogReplicas: {} , pullReplicas: {} , createNodeSet {}", shard, nrtReplicas, tlogReplicas, pullReplicas, createNodeSet );
@@ -331,13 +267,13 @@ public class Assign {
List<String> createNodeList = null;
if (createNodeSet instanceof List) {
- createNodeList = (List) createNodeSet;
+ createNodeList = (List<String>) createNodeSet;
} else {
// deduplicate
createNodeList = createNodeSet == null ? null : new ArrayList<>(new LinkedHashSet<>(StrUtils.splitSmart((String) createNodeSet, ",", true)));
}
- HashMap<String, ReplicaCount> nodeNameVsShardCount = getNodeNameVsShardCount(collectionName, clusterState, createNodeList);
+ HashMap<String, ReplicaCount> nodeNameVsShardCount = getNodeNameVsShardCount(collectionName, clusterState, createNodeList);
if (createNodeList == null) { // We only care if we haven't been told to put new replicas on specific nodes.
long availableSlots = 0;
@@ -349,40 +285,22 @@ public class Assign {
}
if (availableSlots < nrtReplicas + tlogReplicas + pullReplicas) {
throw new AssignmentException(
- String.format(Locale.ROOT, "Cannot create %d new replicas for collection %s given the current number of live nodes and a maxShardsPerNode of %d",
- nrtReplicas, collectionName, maxShardsPerNode));
+ String.format(Locale.ROOT, "Cannot create %d new replicas for collection %s given the current number of eligible live nodes %d and a maxShardsPerNode of %d",
+ nrtReplicas, collectionName, nodeNameVsShardCount.size(), maxShardsPerNode));
}
}
- List l = (List) coll.get(DocCollection.RULE);
- List<ReplicaPosition> replicaPositions = null;
- if (l != null) {
- if (tlogReplicas + pullReplicas > 0) {
- throw new AssignmentException(Replica.Type.TLOG + " or " + Replica.Type.PULL +
- " replica types not supported with placement rules");
- }
- // TODO: make it so that this method doesn't require access to CC
- replicaPositions = getNodesViaRules(clusterState, shard, nrtReplicas, cloudManager, coll, createNodeList, l);
- }
- String policyName = coll.getStr(POLICY);
- AutoScalingConfig autoScalingConfig = cloudManager.getDistribStateManager().getAutoScalingConfig();
- if (policyName != null || !autoScalingConfig.getPolicy().getClusterPolicy().isEmpty()) {
- replicaPositions = Assign.getPositionsUsingPolicy(collectionName, Collections.singletonList(shard), nrtReplicas, tlogReplicas, pullReplicas,
- policyName, cloudManager, createNodeList);
- }
-
- if(replicaPositions != null){
- List<ReplicaCount> repCounts = new ArrayList<>();
- for (ReplicaPosition p : replicaPositions) {
- repCounts.add(new ReplicaCount(p.node));
- }
- return repCounts;
- }
-
- ArrayList<ReplicaCount> sortedNodeList = new ArrayList<>(nodeNameVsShardCount.values());
- Collections.sort(sortedNodeList, (x, y) -> (x.weight() < y.weight()) ? -1 : ((x.weight() == y.weight()) ? 0 : 1));
- return sortedNodeList;
-
+ AssignRequest assignRequest = new AssignRequestBuilder()
+ .forCollection(collectionName)
+ .forShard(Collections.singletonList(shard))
+ .assignNrtReplicas(nrtReplicas)
+ .assignTlogReplicas(tlogReplicas)
+ .assignPullReplicas(pullReplicas)
+ .onNodes(createNodeList)
+ .build();
+ AssignStrategyFactory assignStrategyFactory = new AssignStrategyFactory(cloudManager);
+ AssignStrategy assignStrategy = assignStrategyFactory.create(clusterState, coll);
+ return assignStrategy.assign(cloudManager, assignRequest);
}
public static List<ReplicaPosition> getPositionsUsingPolicy(String collName, List<String> shardNames,
@@ -418,35 +336,7 @@ public class Assign {
}
}
- private static List<ReplicaPosition> getNodesViaRules(ClusterState clusterState, String shard, int numberOfNodes,
- SolrCloudManager cloudManager, DocCollection coll, List<String> createNodeList, List l) {
- ArrayList<Rule> rules = new ArrayList<>();
- for (Object o : l) rules.add(new Rule((Map) o));
- Map<String, Map<String, Integer>> shardVsNodes = new LinkedHashMap<>();
- for (Slice slice : coll.getSlices()) {
- LinkedHashMap<String, Integer> n = new LinkedHashMap<>();
- shardVsNodes.put(slice.getName(), n);
- for (Replica replica : slice.getReplicas()) {
- Integer count = n.get(replica.getNodeName());
- if (count == null) count = 0;
- n.put(replica.getNodeName(), ++count);
- }
- }
- List snitches = (List) coll.get(SNITCH);
- List<String> nodesList = createNodeList == null ?
- new ArrayList<>(clusterState.getLiveNodes()) :
- createNodeList;
- Map<ReplicaPosition, String> positions = new ReplicaAssigner(
- rules,
- Collections.singletonMap(shard, numberOfNodes),
- snitches,
- shardVsNodes,
- nodesList, cloudManager, clusterState).getNodeMappings();
-
- return positions.entrySet().stream().map(e -> e.getKey().setNode(e.getValue())).collect(Collectors.toList());// getReplicaCounts(positions);
- }
-
- private static HashMap<String, ReplicaCount> getNodeNameVsShardCount(String collectionName,
+ static HashMap<String, ReplicaCount> getNodeNameVsShardCount(String collectionName,
ClusterState clusterState, List<String> createNodeList) {
Set<String> nodes = clusterState.getLiveNodes();
@@ -477,7 +367,7 @@ public class Assign {
for (Replica replica : replicas) {
ReplicaCount count = nodeNameVsShardCount.get(replica.getNodeName());
if (count != null) {
- count.totalNodes++; // Used ot "weigh" whether this node should be used later.
+ count.totalNodes++; // Used to "weigh" whether this node should be used later.
if (entry.getKey().equals(collectionName)) {
count.thisCollectionNodes++;
if (count.thisCollectionNodes >= maxShardsPerNode) nodeNameVsShardCount.remove(replica.getNodeName());
@@ -513,4 +403,210 @@ public class Assign {
super(message, cause, enableSuppression, writableStackTrace);
}
}
+
+ public interface AssignStrategy {
+ List<ReplicaPosition> assign(SolrCloudManager solrCloudManager, AssignRequest assignRequest)
+ throws Assign.AssignmentException, IOException, InterruptedException;
+ }
+
+ public static class AssignRequest {
+ public String collectionName;
+ public List<String> shardNames;
+ public List<String> nodes;
+ public int numNrtReplicas;
+ public int numTlogReplicas;
+ public int numPullReplicas;
+
+ public AssignRequest(String collectionName, List<String> shardNames, List<String> nodes, int numNrtReplicas, int numTlogReplicas, int numPullReplicas) {
+ this.collectionName = collectionName;
+ this.shardNames = shardNames;
+ this.nodes = nodes;
+ this.numNrtReplicas = numNrtReplicas;
+ this.numTlogReplicas = numTlogReplicas;
+ this.numPullReplicas = numPullReplicas;
+ }
+ }
+
+ public static class AssignRequestBuilder {
+ private String collectionName;
+ private List<String> shardNames;
+ private List<String> nodes;
+ private int numNrtReplicas;
+ private int numTlogReplicas;
+ private int numPullReplicas;
+
+ public AssignRequestBuilder forCollection(String collectionName) {
+ this.collectionName = collectionName;
+ return this;
+ }
+
+ public AssignRequestBuilder forShard(List<String> shardNames) {
+ this.shardNames = shardNames;
+ return this;
+ }
+
+ public AssignRequestBuilder onNodes(List<String> nodes) {
+ this.nodes = nodes;
+ return this;
+ }
+
+ public AssignRequestBuilder assignNrtReplicas(int numNrtReplicas) {
+ this.numNrtReplicas = numNrtReplicas;
+ return this;
+ }
+
+ public AssignRequestBuilder assignTlogReplicas(int numTlogReplicas) {
+ this.numTlogReplicas = numTlogReplicas;
+ return this;
+ }
+
+ public AssignRequestBuilder assignPullReplicas(int numPullReplicas) {
+ this.numPullReplicas = numPullReplicas;
+ return this;
+ }
+
+ public AssignRequest build() {
+ Objects.requireNonNull(collectionName, "The collectionName cannot be null");
+ Objects.requireNonNull(shardNames, "The shard names cannot be null");
+ return new AssignRequest(collectionName, shardNames, nodes, numNrtReplicas,
+ numTlogReplicas, numPullReplicas);
+ }
+ }
+
+ public static class LegacyAssignStrategy implements AssignStrategy {
+ @Override
+ public List<ReplicaPosition> assign(SolrCloudManager solrCloudManager, AssignRequest assignRequest) throws Assign.AssignmentException, IOException, InterruptedException {
+ ClusterState clusterState = solrCloudManager.getClusterStateProvider().getClusterState();
+ List<String> nodeList = assignRequest.nodes;
+
+ HashMap<String, Assign.ReplicaCount> nodeNameVsShardCount = Assign.getNodeNameVsShardCount(assignRequest.collectionName, clusterState, assignRequest.nodes);
+ if (nodeList == null || nodeList.isEmpty()) {
+ ArrayList<Assign.ReplicaCount> sortedNodeList = new ArrayList<>(nodeNameVsShardCount.values());
+ sortedNodeList.sort(Comparator.comparingInt(Assign.ReplicaCount::weight));
+ nodeList = sortedNodeList.stream().map(replicaCount -> replicaCount.nodeName).collect(Collectors.toList());
+ }
+
+ int i = 0;
+ List<ReplicaPosition> result = new ArrayList<>();
+ for (String aShard : assignRequest.shardNames)
+ for (Map.Entry<Replica.Type, Integer> e : ImmutableMap.of(Replica.Type.NRT, assignRequest.numNrtReplicas,
+ Replica.Type.TLOG, assignRequest.numTlogReplicas,
+ Replica.Type.PULL, assignRequest.numPullReplicas
+ ).entrySet()) {
+ for (int j = 0; j < e.getValue(); j++) {
+ result.add(new ReplicaPosition(aShard, j, e.getKey(), nodeList.get(i % nodeList.size())));
+ i++;
+ }
+ }
+ return result;
+ }
+ }
+
+ public static class RulesBasedAssignStrategy implements AssignStrategy {
+ public List<Rule> rules;
+ public List snitches;
+ public ClusterState clusterState;
+
+ public RulesBasedAssignStrategy(List<Rule> rules, List snitches, ClusterState clusterState) {
+ this.rules = rules;
+ this.snitches = snitches;
+ this.clusterState = clusterState;
+ }
+
+ @Override
+ public List<ReplicaPosition> assign(SolrCloudManager solrCloudManager, AssignRequest assignRequest) throws Assign.AssignmentException, IOException, InterruptedException {
+ if (assignRequest.numTlogReplicas + assignRequest.numPullReplicas != 0) {
+ throw new Assign.AssignmentException(
+ Replica.Type.TLOG + " or " + Replica.Type.PULL + " replica types not supported with placement rules or cluster policies");
+ }
+
+ Map<String, Integer> shardVsReplicaCount = new HashMap<>();
+ for (String shard : assignRequest.shardNames) shardVsReplicaCount.put(shard, assignRequest.numNrtReplicas);
+
+ Map<String, Map<String, Integer>> shardVsNodes = new LinkedHashMap<>();
+ DocCollection docCollection = solrCloudManager.getClusterStateProvider().getClusterState().getCollectionOrNull(assignRequest.collectionName);
+ if (docCollection != null) {
+ for (Slice slice : docCollection.getSlices()) {
+ LinkedHashMap<String, Integer> n = new LinkedHashMap<>();
+ shardVsNodes.put(slice.getName(), n);
+ for (Replica replica : slice.getReplicas()) {
+ Integer count = n.get(replica.getNodeName());
+ if (count == null) count = 0;
+ n.put(replica.getNodeName(), ++count);
+ }
+ }
+ }
+
+ List<String> nodesList = assignRequest.nodes == null ? new ArrayList<>(clusterState.getLiveNodes()) : assignRequest.nodes;
+
+ ReplicaAssigner replicaAssigner = new ReplicaAssigner(rules,
+ shardVsReplicaCount,
+ snitches,
+ shardVsNodes,
+ nodesList,
+ solrCloudManager, clusterState);
+
+ Map<ReplicaPosition, String> nodeMappings = replicaAssigner.getNodeMappings();
+ return nodeMappings.entrySet().stream()
+ .map(e -> new ReplicaPosition(e.getKey().shard, e.getKey().index, e.getKey().type, e.getValue()))
+ .collect(Collectors.toList());
+ }
+ }
+
+ public static class PolicyBasedAssignStrategy implements AssignStrategy {
+ public String policyName;
+
+ public PolicyBasedAssignStrategy(String policyName) {
+ this.policyName = policyName;
+ }
+
+ @Override
+ public List<ReplicaPosition> assign(SolrCloudManager solrCloudManager, AssignRequest assignRequest) throws Assign.AssignmentException, IOException, InterruptedException {
+ return Assign.getPositionsUsingPolicy(assignRequest.collectionName,
+ assignRequest.shardNames, assignRequest.numNrtReplicas,
+ assignRequest.numTlogReplicas, assignRequest.numPullReplicas,
+ policyName, solrCloudManager, assignRequest.nodes);
+ }
+ }
+
+ public static class AssignStrategyFactory {
+ public SolrCloudManager solrCloudManager;
+
+ public AssignStrategyFactory(SolrCloudManager solrCloudManager) {
+ this.solrCloudManager = solrCloudManager;
+ }
+
+ public AssignStrategy create(ClusterState clusterState, DocCollection collection) throws IOException, InterruptedException {
+ List<Map> ruleMaps = (List<Map>) collection.get("rule");
+ String policyName = collection.getStr(POLICY);
+ List snitches = (List) collection.get(SNITCH);
+ AutoScalingConfig autoScalingConfig = solrCloudManager.getDistribStateManager().getAutoScalingConfig();
+
+ StrategyType strategyType = null;
+ if ((ruleMaps == null || ruleMaps.isEmpty()) && policyName == null && autoScalingConfig.getPolicy().getClusterPolicy().isEmpty()) {
+ strategyType = StrategyType.LEGACY;
+ } else if (ruleMaps != null && !ruleMaps.isEmpty()) {
+ strategyType = StrategyType.RULES;
+ } else {
+ strategyType = StrategyType.POLICY;
+ }
+
+ switch (strategyType) {
+ case LEGACY:
+ return new LegacyAssignStrategy();
+ case RULES:
+ List<Rule> rules = new ArrayList<>();
+ for (Object map : ruleMaps) rules.add(new Rule((Map) map));
+ return new RulesBasedAssignStrategy(rules, snitches, clusterState);
+ case POLICY:
+ return new PolicyBasedAssignStrategy(policyName);
+ default:
+ throw new Assign.AssignmentException("Unknown strategy type: " + strategyType);
+ }
+ }
+
+ private enum StrategyType {
+ LEGACY, RULES, POLICY;
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/17cb1b17/solr/core/src/java/org/apache/solr/cloud/api/collections/CreateCollectionCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/api/collections/CreateCollectionCmd.java b/solr/core/src/java/org/apache/solr/cloud/api/collections/CreateCollectionCmd.java
index 4f66ff9..542345d 100644
--- a/solr/core/src/java/org/apache/solr/cloud/api/collections/CreateCollectionCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/api/collections/CreateCollectionCmd.java
@@ -42,6 +42,7 @@ import org.apache.solr.client.solrj.cloud.autoscaling.NotEmptyException;
import org.apache.solr.client.solrj.cloud.autoscaling.Policy;
import org.apache.solr.client.solrj.cloud.autoscaling.PolicyHelper;
import org.apache.solr.client.solrj.cloud.autoscaling.VersionedData;
+import org.apache.solr.cloud.CloudUtil;
import org.apache.solr.cloud.Overseer;
import org.apache.solr.cloud.ZkController;
import org.apache.solr.cloud.overseer.ClusterStateMutator;
@@ -131,7 +132,6 @@ public class CreateCollectionCmd implements OverseerCollectionMessageHandler.Cmd
ocmh.validateConfigOrThrowSolrException(configName);
- List<String> nodeList = new ArrayList<>();
String router = message.getStr("router.name", DocRouter.DEFAULT_NAME);
String policy = message.getStr(Policy.POLICY);
AutoScalingConfig autoScalingConfig = ocmh.cloudManager.getDistribStateManager().getAutoScalingConfig();
@@ -177,10 +177,12 @@ public class CreateCollectionCmd implements OverseerCollectionMessageHandler.Cmd
throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "Could not fully create collection: " + collectionName);
}
+ // refresh cluster state
+ clusterState = ocmh.cloudManager.getClusterStateProvider().getClusterState();
+
List<ReplicaPosition> replicaPositions = null;
try {
- replicaPositions = buildReplicaPositions(ocmh.cloudManager, clusterState, message,
- nodeList, shardNames, sessionWrapper);
+ replicaPositions = buildReplicaPositions(ocmh.cloudManager, clusterState, clusterState.getCollection(collectionName), message, shardNames, sessionWrapper);
} catch (Assign.AssignmentException e) {
ZkNodeProps deleteMessage = new ZkNodeProps("name", collectionName);
new DeleteCollectionCmd(ocmh).call(clusterState, deleteMessage, results);
@@ -188,7 +190,7 @@ public class CreateCollectionCmd implements OverseerCollectionMessageHandler.Cmd
throw new SolrException(ErrorCode.SERVER_ERROR, e.getMessage(), e.getCause());
}
- if (nodeList.isEmpty()) {
+ if (replicaPositions.isEmpty()) {
log.debug("Finished create command for collection: {}", collectionName);
return;
}
@@ -333,8 +335,9 @@ public class CreateCollectionCmd implements OverseerCollectionMessageHandler.Cmd
}
public static List<ReplicaPosition> buildReplicaPositions(SolrCloudManager cloudManager, ClusterState clusterState,
+ DocCollection docCollection,
ZkNodeProps message,
- List<String> nodeList, List<String> shardNames,
+ List<String> shardNames,
AtomicReference<PolicyHelper.SessionWrapper> sessionWrapper) throws IOException, InterruptedException, Assign.AssignmentException {
final String collectionName = message.getStr(NAME);
// look at the replication factor and see if it matches reality
@@ -342,11 +345,9 @@ public class CreateCollectionCmd implements OverseerCollectionMessageHandler.Cmd
int numTlogReplicas = message.getInt(TLOG_REPLICAS, 0);
int numNrtReplicas = message.getInt(NRT_REPLICAS, message.getInt(REPLICATION_FACTOR, numTlogReplicas>0?0:1));
int numPullReplicas = message.getInt(PULL_REPLICAS, 0);
- AutoScalingConfig autoScalingConfig = cloudManager.getDistribStateManager().getAutoScalingConfig();
- String policy = message.getStr(Policy.POLICY);
- boolean usePolicyFramework = !autoScalingConfig.getPolicy().getClusterPolicy().isEmpty() || policy != null;
+ boolean usePolicyFramework = CloudUtil.usePolicyFramework(docCollection, cloudManager);
- Integer numSlices = shardNames.size();
+ int numSlices = shardNames.size();
int maxShardsPerNode = checkMaxShardsPerNode(message, usePolicyFramework);
// we need to look at every node and see how many cores it serves
@@ -354,7 +355,7 @@ public class CreateCollectionCmd implements OverseerCollectionMessageHandler.Cmd
// but (for now) require that each core goes on a distinct node.
List<ReplicaPosition> replicaPositions;
- nodeList.addAll(Assign.getLiveOrLiveAndCreateNodeSetList(clusterState.getLiveNodes(), message, OverseerCollectionMessageHandler.RANDOM));
+ List<String> nodeList = Assign.getLiveOrLiveAndCreateNodeSetList(clusterState.getLiveNodes(), message, OverseerCollectionMessageHandler.RANDOM);
if (nodeList.isEmpty()) {
log.warn("It is unusual to create a collection ("+collectionName+") without cores.");
@@ -387,8 +388,17 @@ public class CreateCollectionCmd implements OverseerCollectionMessageHandler.Cmd
+ ". This requires " + requestedShardsToCreate
+ " shards to be created (higher than the allowed number)");
}
- replicaPositions = Assign.identifyNodes(cloudManager
- , clusterState, nodeList, collectionName, message, shardNames, numNrtReplicas, numTlogReplicas, numPullReplicas);
+ Assign.AssignRequest assignRequest = new Assign.AssignRequestBuilder()
+ .forCollection(collectionName)
+ .forShard(shardNames)
+ .assignNrtReplicas(numNrtReplicas)
+ .assignTlogReplicas(numTlogReplicas)
+ .assignPullReplicas(numPullReplicas)
+ .onNodes(nodeList)
+ .build();
+ Assign.AssignStrategyFactory assignStrategyFactory = new Assign.AssignStrategyFactory(cloudManager);
+ Assign.AssignStrategy assignStrategy = assignStrategyFactory.create(clusterState, docCollection);
+ replicaPositions = assignStrategy.assign(cloudManager, assignRequest);
sessionWrapper.set(PolicyHelper.getLastSessionWrapper(true));
}
return replicaPositions;
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/17cb1b17/solr/core/src/java/org/apache/solr/cloud/api/collections/ReplaceNodeCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/api/collections/ReplaceNodeCmd.java b/solr/core/src/java/org/apache/solr/cloud/api/collections/ReplaceNodeCmd.java
index a09eec3..c622f0f 100644
--- a/solr/core/src/java/org/apache/solr/cloud/api/collections/ReplaceNodeCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/api/collections/ReplaceNodeCmd.java
@@ -104,20 +104,25 @@ public class ReplaceNodeCmd implements OverseerCollectionMessageHandler.Cmd {
try {
for (ZkNodeProps sourceReplica : sourceReplicas) {
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);
+ String sourceCollection = sourceReplica.getStr(COLLECTION_PROP);
+ log.info("Going to create replica for collection={} shard={} on node={}", sourceCollection, sourceReplica.getStr(SHARD_ID_PROP), target);
String targetNode = target;
if (targetNode == null) {
Replica.Type replicaType = Replica.Type.get(sourceReplica.getStr(ZkStateReader.REPLICA_TYPE));
- targetNode = Assign.identifyNodes(ocmh.cloudManager,
- clusterState,
- new ArrayList<>(ocmh.cloudManager.getClusterStateProvider().getLiveNodes()),
- sourceReplica.getStr(COLLECTION_PROP),
- message,
- Collections.singletonList(sourceReplica.getStr(SHARD_ID_PROP)),
- replicaType == Replica.Type.NRT ? 1: 0,
- replicaType == Replica.Type.TLOG ? 1 : 0,
- replicaType == Replica.Type.PULL ? 1 : 0
- ).get(0).node;
+ int numNrtReplicas = replicaType == Replica.Type.NRT ? 1 : 0;
+ int numTlogReplicas = replicaType == Replica.Type.TLOG ? 1 : 0;
+ int numPullReplicas = replicaType == Replica.Type.PULL ? 1 : 0;
+ Assign.AssignRequest assignRequest = new Assign.AssignRequestBuilder()
+ .forCollection(sourceCollection)
+ .forShard(Collections.singletonList(sourceReplica.getStr(SHARD_ID_PROP)))
+ .assignNrtReplicas(numNrtReplicas)
+ .assignTlogReplicas(numTlogReplicas)
+ .assignPullReplicas(numPullReplicas)
+ .onNodes(new ArrayList<>(ocmh.cloudManager.getClusterStateProvider().getLiveNodes()))
+ .build();
+ Assign.AssignStrategyFactory assignStrategyFactory = new Assign.AssignStrategyFactory(ocmh.cloudManager);
+ Assign.AssignStrategy assignStrategy = assignStrategyFactory.create(clusterState, clusterState.getCollection(sourceCollection));
+ targetNode = assignStrategy.assign(ocmh.cloudManager, assignRequest).get(0).node;
sessionWrapperRef.set(PolicyHelper.getLastSessionWrapper(true));
}
ZkNodeProps msg = sourceReplica.plus("parallel", String.valueOf(parallel)).plus(CoreAdminParams.NODE, targetNode);
@@ -127,7 +132,7 @@ public class ReplaceNodeCmd implements OverseerCollectionMessageHandler.Cmd {
countDownLatch.countDown();
if (nl.get("failure") != null) {
String errorString = String.format(Locale.ROOT, "Failed to create replica for collection=%s shard=%s" +
- " on node=%s", sourceReplica.getStr(COLLECTION_PROP), sourceReplica.getStr(SHARD_ID_PROP), target);
+ " on node=%s", sourceCollection, sourceReplica.getStr(SHARD_ID_PROP), target);
log.warn(errorString);
// one replica creation failed. Make the best attempt to
// delete all the replicas created so far in the target
@@ -138,7 +143,7 @@ public class ReplaceNodeCmd implements OverseerCollectionMessageHandler.Cmd {
}
} else {
log.debug("Successfully created replica for collection={} shard={} on node={}",
- sourceReplica.getStr(COLLECTION_PROP), sourceReplica.getStr(SHARD_ID_PROP), target);
+ sourceCollection, sourceReplica.getStr(SHARD_ID_PROP), target);
}
}).get(0);
@@ -147,7 +152,7 @@ public class ReplaceNodeCmd implements OverseerCollectionMessageHandler.Cmd {
if (sourceReplica.getBool(ZkStateReader.LEADER_PROP, false) || waitForFinalState) {
String shardName = sourceReplica.getStr(SHARD_ID_PROP);
String replicaName = sourceReplica.getStr(ZkStateReader.REPLICA_PROP);
- String collectionName = sourceReplica.getStr(COLLECTION_PROP);
+ String collectionName = sourceCollection;
String key = collectionName + "_" + replicaName;
CollectionStateWatcher watcher;
if (waitForFinalState) {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/17cb1b17/solr/core/src/java/org/apache/solr/cloud/api/collections/RestoreCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/api/collections/RestoreCmd.java b/solr/core/src/java/org/apache/solr/cloud/api/collections/RestoreCmd.java
index d082ac3..d100ce0 100644
--- a/solr/core/src/java/org/apache/solr/cloud/api/collections/RestoreCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/api/collections/RestoreCmd.java
@@ -232,11 +232,17 @@ public class RestoreCmd implements OverseerCollectionMessageHandler.Cmd {
PolicyHelper.SessionWrapper sessionWrapper = null;
try {
- List<ReplicaPosition> replicaPositions = Assign.identifyNodes(
- ocmh.cloudManager, clusterState,
- nodeList, restoreCollectionName,
- message, sliceNames,
- numNrtReplicas, numTlogReplicas, numPullReplicas);
+ Assign.AssignRequest assignRequest = new Assign.AssignRequestBuilder()
+ .forCollection(restoreCollectionName)
+ .forShard(sliceNames)
+ .assignNrtReplicas(numNrtReplicas)
+ .assignTlogReplicas(numTlogReplicas)
+ .assignPullReplicas(numPullReplicas)
+ .onNodes(nodeList)
+ .build();
+ Assign.AssignStrategyFactory assignStrategyFactory = new Assign.AssignStrategyFactory(ocmh.cloudManager);
+ Assign.AssignStrategy assignStrategy = assignStrategyFactory.create(clusterState, restoreCollection);
+ List<ReplicaPosition> replicaPositions = assignStrategy.assign(ocmh.cloudManager, assignRequest);
sessionWrapper = PolicyHelper.getLastSessionWrapper(true);
//Create one replica per shard and copy backed up data to it
for (Slice slice : restoreCollection.getSlices()) {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/17cb1b17/solr/core/src/java/org/apache/solr/cloud/api/collections/SplitShardCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/api/collections/SplitShardCmd.java b/solr/core/src/java/org/apache/solr/cloud/api/collections/SplitShardCmd.java
index a101c95..e2f4cef 100644
--- a/solr/core/src/java/org/apache/solr/cloud/api/collections/SplitShardCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/api/collections/SplitShardCmd.java
@@ -361,12 +361,17 @@ public class SplitShardCmd implements OverseerCollectionMessageHandler.Cmd {
}
t = timings.sub("identifyNodesForReplicas");
- List<ReplicaPosition> replicaPositions = Assign.identifyNodes(ocmh.cloudManager,
- clusterState,
- new ArrayList<>(clusterState.getLiveNodes()),
- collectionName,
- new ZkNodeProps(collection.getProperties()),
- subSlices, numNrt.get(), numTlog.get(), numPull.get());
+ Assign.AssignRequest assignRequest = new Assign.AssignRequestBuilder()
+ .forCollection(collectionName)
+ .forShard(subSlices)
+ .assignNrtReplicas(numNrt.get())
+ .assignTlogReplicas(numTlog.get())
+ .assignPullReplicas(numPull.get())
+ .onNodes(new ArrayList<>(clusterState.getLiveNodes()))
+ .build();
+ Assign.AssignStrategyFactory assignStrategyFactory = new Assign.AssignStrategyFactory(ocmh.cloudManager);
+ Assign.AssignStrategy assignStrategy = assignStrategyFactory.create(clusterState, collection);
+ List<ReplicaPosition> replicaPositions = assignStrategy.assign(ocmh.cloudManager, assignRequest);
sessionWrapper = PolicyHelper.getLastSessionWrapper(true);
t.stop();
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/17cb1b17/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 34843c1..6cbdbfb 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
@@ -282,9 +282,7 @@ public class ReplicaMutator {
Slice slice = collection != null ? collection.getSlice(sliceName) : null;
- Map<String, Object> replicaProps = new LinkedHashMap<>();
-
- replicaProps.putAll(message.getProperties());
+ Map<String, Object> replicaProps = new LinkedHashMap<>(message.getProperties());
if (slice != null) {
Replica oldReplica = slice.getReplica(coreNodeName);
if (oldReplica != null) {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/17cb1b17/solr/core/src/test/org/apache/solr/cloud/api/collections/CollectionTooManyReplicasTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/api/collections/CollectionTooManyReplicasTest.java b/solr/core/src/test/org/apache/solr/cloud/api/collections/CollectionTooManyReplicasTest.java
index 09a119b..1ac0cae 100644
--- a/solr/core/src/test/org/apache/solr/cloud/api/collections/CollectionTooManyReplicasTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/api/collections/CollectionTooManyReplicasTest.java
@@ -79,7 +79,7 @@ public class CollectionTooManyReplicasTest extends SolrCloudTestCase {
});
assertTrue("Should have gotten the right error message back",
- e.getMessage().contains("given the current number of live nodes and a maxShardsPerNode of"));
+ e.getMessage().contains("given the current number of eligible live nodes"));
// Oddly, we should succeed next just because setting property.name will not check for nodes being "full up"
@@ -106,7 +106,7 @@ public class CollectionTooManyReplicasTest extends SolrCloudTestCase {
});
assertTrue("Should have gotten the right error message back",
- e2.getMessage().contains("given the current number of live nodes and a maxShardsPerNode of"));
+ e2.getMessage().contains("given the current number of eligible live nodes"));
// wait for recoveries to finish, for a clean shutdown - see SOLR-9645
waitForState("Expected to see all replicas active", collectionName, (n, c) -> {
@@ -141,7 +141,7 @@ public class CollectionTooManyReplicasTest extends SolrCloudTestCase {
.process(cluster.getSolrClient());
});
assertTrue("Should have gotten the right error message back",
- e.getMessage().contains("given the current number of live nodes and a maxShardsPerNode of"));
+ e.getMessage().contains("given the current number of eligible live nodes"));
// Hmmm, providing a nodeset also overrides the checks for max replicas, so prove it.
List<String> nodes = getAllNodeNames(collectionName);
@@ -156,7 +156,7 @@ public class CollectionTooManyReplicasTest extends SolrCloudTestCase {
.process(cluster.getSolrClient());
});
assertTrue("Should have gotten the right error message back",
- e2.getMessage().contains("given the current number of live nodes and a maxShardsPerNode of"));
+ e2.getMessage().contains("given the current number of eligible live nodes"));
// And finally, ensure that there are all the replicas we expect. We should have shards 1, 2 and 4 and each
// should have exactly two replicas
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/17cb1b17/solr/core/src/test/org/apache/solr/cloud/autoscaling/TestPolicyCloud.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/TestPolicyCloud.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/TestPolicyCloud.java
index d1dcecf..bfd5878 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/TestPolicyCloud.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/TestPolicyCloud.java
@@ -325,6 +325,7 @@ public class TestPolicyCloud extends SolrCloudTestCase {
Utils.getObjectByPath(json, true, "cluster-policy[2]/port"));
CollectionAdminRequest.createCollectionWithImplicitRouter("policiesTest", "conf", "s1", 1, 1, 1)
+ .setMaxShardsPerNode(-1)
.process(cluster.getSolrClient());
DocCollection coll = getCollectionState("policiesTest");
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/17cb1b17/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java
index 637f5bf..ee39666 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java
@@ -777,7 +777,6 @@ public class SimClusterStateProvider implements ClusterStateProvider {
results.add(CoreAdminParams.REQUESTID, props.getStr(CommonAdminParams.ASYNC));
}
boolean waitForFinalState = props.getBool(CommonAdminParams.WAIT_FOR_FINAL_STATE, false);
- List<String> nodeList = new ArrayList<>();
final String collectionName = props.getStr(NAME);
log.debug("-- simCreateCollection {}, currentVersion={}", collectionName, clusterStateVersion);
@@ -856,8 +855,8 @@ public class SimClusterStateProvider implements ClusterStateProvider {
opDelay(collectionName, CollectionParams.CollectionAction.CREATE.name());
AtomicReference<PolicyHelper.SessionWrapper> sessionWrapper = new AtomicReference<>();
- List<ReplicaPosition> replicaPositions = CreateCollectionCmd.buildReplicaPositions(cloudManager, getClusterState(), props,
- nodeList, shardNames, sessionWrapper);
+ List<ReplicaPosition> replicaPositions = CreateCollectionCmd.buildReplicaPositions(cloudManager, getClusterState(), cmd.collection, props,
+ shardNames, sessionWrapper);
if (sessionWrapper.get() != null) {
sessionWrapper.get().release();
}
@@ -1168,13 +1167,18 @@ public class SimClusterStateProvider implements ClusterStateProvider {
SplitShardCmd.fillRanges(cloudManager, message, collection, parentSlice, subRanges, subSlices, subShardNames, true);
// add replicas for new subShards
int repFactor = parentSlice.getReplicas().size();
- List<ReplicaPosition> replicaPositions = Assign.identifyNodes(cloudManager,
- clusterState,
- new ArrayList<>(clusterState.getLiveNodes()),
- collectionName,
- new ZkNodeProps(collection.getProperties()),
- // reproduce the bug
- subSlices, repFactor, 0, 0);
+ Assign.AssignRequest assignRequest = new Assign.AssignRequestBuilder()
+ .forCollection(collectionName)
+ .forShard(subSlices)
+ .assignNrtReplicas(repFactor)
+ .assignTlogReplicas(0)
+ .assignPullReplicas(0)
+ .onNodes(new ArrayList<>(clusterState.getLiveNodes()))
+ .build();
+ Assign.AssignStrategyFactory assignStrategyFactory = new Assign.AssignStrategyFactory(cloudManager);
+ Assign.AssignStrategy assignStrategy = assignStrategyFactory.create(clusterState, collection);
+ // reproduce the bug
+ List<ReplicaPosition> replicaPositions = assignStrategy.assign(cloudManager, assignRequest);
PolicyHelper.SessionWrapper sessionWrapper = PolicyHelper.getLastSessionWrapper(true);
if (sessionWrapper != null) sessionWrapper.release();
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/17cb1b17/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestSimPolicyCloud.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestSimPolicyCloud.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestSimPolicyCloud.java
index 86624b8..379011d 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestSimPolicyCloud.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestSimPolicyCloud.java
@@ -257,6 +257,7 @@ public class TestSimPolicyCloud extends SimSolrCloudTestCase {
Utils.getObjectByPath(json, true, "cluster-policy[2]/port"));
CollectionAdminRequest.createCollectionWithImplicitRouter("policiesTest", "conf", "s1", 1, 1, 1)
+ .setMaxShardsPerNode(-1)
.process(solrClient);
CloudTestUtils.waitForState(cluster, "Timeout waiting for collection to become active", "policiesTest",
CloudTestUtils.clusterShape(1, 3, false, true));
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/17cb1b17/solr/solrj/src/java/org/apache/solr/common/cloud/ReplicaPosition.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/common/cloud/ReplicaPosition.java b/solr/solrj/src/java/org/apache/solr/common/cloud/ReplicaPosition.java
index 591a001..62d8761 100644
--- a/solr/solrj/src/java/org/apache/solr/common/cloud/ReplicaPosition.java
+++ b/solr/solrj/src/java/org/apache/solr/common/cloud/ReplicaPosition.java
@@ -40,7 +40,7 @@ public class ReplicaPosition implements Comparable<ReplicaPosition> {
public int compareTo(ReplicaPosition that) {
//this is to ensure that we try one replica from each shard first instead of
// all replicas from same shard
- return that.index > index ? -1 : that.index == index ? 0 : 1;
+ return Integer.compare(index, that.index);
}
@Override
[2/5] lucene-solr:branch_7x: SOLR-9317: Deduplicate node list before
trying to find if all nodes are up. The test is also fixed to not send
duplicate node names in the createNodeSet parameter.
Posted by sh...@apache.org.
SOLR-9317: Deduplicate node list before trying to find if all nodes are up. The test is also fixed to not send duplicate node names in the createNodeSet parameter.
(cherry picked from commit e437b2f1ed80af6bb8b909f64511fe43829eb67d)
Project: http://git-wip-us.apache.org/repos/asf/lucene-solr/repo
Commit: http://git-wip-us.apache.org/repos/asf/lucene-solr/commit/f08fee29
Tree: http://git-wip-us.apache.org/repos/asf/lucene-solr/tree/f08fee29
Diff: http://git-wip-us.apache.org/repos/asf/lucene-solr/diff/f08fee29
Branch: refs/heads/branch_7x
Commit: f08fee29726a06dcba096e1fe3dff32d7daf9833
Parents: 0d7627f
Author: Shalin Shekhar Mangar <sh...@apache.org>
Authored: Tue Sep 25 12:14:16 2018 +0530
Committer: Shalin Shekhar Mangar <sh...@apache.org>
Committed: Wed Oct 3 15:56:35 2018 +0530
----------------------------------------------------------------------
.../org/apache/solr/cloud/api/collections/Assign.java | 4 +++-
.../src/test/org/apache/solr/cloud/AddReplicaTest.java | 10 ++++++----
2 files changed, 9 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/f08fee29/solr/core/src/java/org/apache/solr/cloud/api/collections/Assign.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/api/collections/Assign.java b/solr/core/src/java/org/apache/solr/cloud/api/collections/Assign.java
index 42de84a..9b33f52 100644
--- a/solr/core/src/java/org/apache/solr/cloud/api/collections/Assign.java
+++ b/solr/core/src/java/org/apache/solr/cloud/api/collections/Assign.java
@@ -23,6 +23,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
@@ -332,7 +333,8 @@ public class Assign {
if (createNodeSet instanceof List) {
createNodeList = (List) createNodeSet;
} else {
- createNodeList = createNodeSet == null ? null : StrUtils.splitSmart((String) createNodeSet, ",", true);
+ // deduplicate
+ createNodeList = createNodeSet == null ? null : new ArrayList<>(new LinkedHashSet<>(StrUtils.splitSmart((String) createNodeSet, ",", true)));
}
HashMap<String, ReplicaCount> nodeNameVsShardCount = getNodeNameVsShardCount(collectionName, clusterState, createNodeList);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/f08fee29/solr/core/src/test/org/apache/solr/cloud/AddReplicaTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/AddReplicaTest.java b/solr/core/src/test/org/apache/solr/cloud/AddReplicaTest.java
index 7b558c0..6d24562 100644
--- a/solr/core/src/test/org/apache/solr/cloud/AddReplicaTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/AddReplicaTest.java
@@ -17,10 +17,9 @@
package org.apache.solr.cloud;
import java.lang.invoke.MethodHandles;
-import java.util.ArrayList;
import java.util.Collection;
import java.util.EnumSet;
-import java.util.List;
+import java.util.LinkedHashSet;
import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.apache.solr.client.solrj.request.CollectionAdminRequest;
@@ -98,9 +97,12 @@ public class AddReplicaTest extends SolrCloudTestCase {
// but adding any number of replicas is supported if an explicit create node set is specified
// so test that as well
- List<String> createNodeSet = new ArrayList<>(2);
- createNodeSet.add(cluster.getRandomJetty(random()).getNodeName());
+ LinkedHashSet<String> createNodeSet = new LinkedHashSet<>(2);
createNodeSet.add(cluster.getRandomJetty(random()).getNodeName());
+ while (true) {
+ String nodeName = cluster.getRandomJetty(random()).getNodeName();
+ if (createNodeSet.add(nodeName)) break;
+ }
addReplica = CollectionAdminRequest.addReplicaToShard(collection, "shard1")
.setNrtReplicas(3)
.setTlogReplicas(1)
[5/5] lucene-solr:branch_7x: SOLR-12750: Migrate API should lock the
collection instead of shard
Posted by sh...@apache.org.
SOLR-12750: Migrate API should lock the collection instead of shard
(cherry picked from commit 84683b6aa072459b364c62e71abd526cdcf8e117)
Project: http://git-wip-us.apache.org/repos/asf/lucene-solr/repo
Commit: http://git-wip-us.apache.org/repos/asf/lucene-solr/commit/5d3ceed4
Tree: http://git-wip-us.apache.org/repos/asf/lucene-solr/tree/5d3ceed4
Diff: http://git-wip-us.apache.org/repos/asf/lucene-solr/diff/5d3ceed4
Branch: refs/heads/branch_7x
Commit: 5d3ceed438bf7a367e9f13fc09e9242da60d246f
Parents: a4beebd
Author: Shalin Shekhar Mangar <sh...@apache.org>
Authored: Wed Oct 3 16:24:26 2018 +0530
Committer: Shalin Shekhar Mangar <sh...@apache.org>
Committed: Wed Oct 3 16:36:28 2018 +0530
----------------------------------------------------------------------
solr/CHANGES.txt | 2 ++
.../src/java/org/apache/solr/common/params/CollectionParams.java | 2 +-
2 files changed, 3 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/5d3ceed4/solr/CHANGES.txt
----------------------------------------------------------------------
diff --git a/solr/CHANGES.txt b/solr/CHANGES.txt
index 9c55193..ff0bf65 100644
--- a/solr/CHANGES.txt
+++ b/solr/CHANGES.txt
@@ -64,6 +64,8 @@ Bug Fixes
* SOLR-12648: Autoscaling framework based replica placement is not used unless a policy is specified or
non-empty cluster policy exists. (shalin)
+* SOLR-12750: Migrate API should lock the collection instead of shard. (shalin)
+
================== 7.5.0 ==================
Consult the LUCENE_CHANGES.txt file for additional, low level, changes in this release.
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/5d3ceed4/solr/solrj/src/java/org/apache/solr/common/params/CollectionParams.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/common/params/CollectionParams.java b/solr/solrj/src/java/org/apache/solr/common/params/CollectionParams.java
index 6fb348f..dee2f5f 100644
--- a/solr/solrj/src/java/org/apache/solr/common/params/CollectionParams.java
+++ b/solr/solrj/src/java/org/apache/solr/common/params/CollectionParams.java
@@ -86,7 +86,7 @@ public interface CollectionParams {
CREATESHARD(true, LockLevel.COLLECTION),
DELETEREPLICA(true, LockLevel.SHARD),
FORCELEADER(true, LockLevel.SHARD),
- MIGRATE(true, LockLevel.SHARD),
+ MIGRATE(true, LockLevel.COLLECTION),
ADDROLE(true, LockLevel.NONE),
REMOVEROLE(true, LockLevel.NONE),
CLUSTERPROP(true, LockLevel.NONE),