You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by no...@apache.org on 2017/10/04 08:24:55 UTC
[03/50] [abbrv] lucene-solr:master: SOLR:10822: Share a
Policy.session object between multiple collection admin calls to get the
correct computations
SOLR:10822: Share a Policy.session object between multiple collection admin calls to get the correct computations
Project: http://git-wip-us.apache.org/repos/asf/lucene-solr/repo
Commit: http://git-wip-us.apache.org/repos/asf/lucene-solr/commit/91446f7f
Tree: http://git-wip-us.apache.org/repos/asf/lucene-solr/tree/91446f7f
Diff: http://git-wip-us.apache.org/repos/asf/lucene-solr/diff/91446f7f
Branch: refs/heads/master
Commit: 91446f7f773319ef9cfd97e45af5620cb45f220c
Parents: 64c480f
Author: Noble Paul <no...@apache.org>
Authored: Tue Aug 15 16:06:35 2017 +0930
Committer: Noble Paul <no...@apache.org>
Committed: Tue Aug 15 16:06:35 2017 +0930
----------------------------------------------------------------------
solr/CHANGES.txt | 2 +
.../org/apache/solr/cloud/AddReplicaCmd.java | 16 +-
.../src/java/org/apache/solr/cloud/Assign.java | 20 +-
.../apache/solr/cloud/CreateCollectionCmd.java | 6 +-
.../org/apache/solr/cloud/CreateShardCmd.java | 139 ++++++------
.../cloud/OverseerCollectionMessageHandler.java | 3 +
.../java/org/apache/solr/cloud/RestoreCmd.java | 219 ++++++++++---------
.../org/apache/solr/cloud/SplitShardCmd.java | 7 +-
.../solrj/cloud/autoscaling/PolicyHelper.java | 143 +++++++++---
.../solrj/cloud/autoscaling/TestPolicy.java | 79 ++++++-
10 files changed, 417 insertions(+), 217 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/91446f7f/solr/CHANGES.txt
----------------------------------------------------------------------
diff --git a/solr/CHANGES.txt b/solr/CHANGES.txt
index d48581e..daad2ff 100644
--- a/solr/CHANGES.txt
+++ b/solr/CHANGES.txt
@@ -179,6 +179,8 @@ Other Changes
* SOLR-11195: Require class attribute for shard and cluster metric reporter configuration. (Christine Poerschke)
+* SOLR:10822: Share a Policy.session object between multiple collection admin calls to get the correct computations (noble)
+
================== 7.0.0 ==================
Versions of Major Components
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/91446f7f/solr/core/src/java/org/apache/solr/cloud/AddReplicaCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/AddReplicaCmd.java b/solr/core/src/java/org/apache/solr/cloud/AddReplicaCmd.java
index c7889c9..6535d34 100644
--- a/solr/core/src/java/org/apache/solr/cloud/AddReplicaCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/AddReplicaCmd.java
@@ -24,9 +24,12 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.Locale;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.commons.lang.StringUtils;
import org.apache.solr.client.solrj.cloud.autoscaling.Policy;
+import org.apache.solr.client.solrj.cloud.autoscaling.PolicyHelper;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.cloud.ClusterState;
import org.apache.solr.common.cloud.DocCollection;
@@ -46,7 +49,6 @@ import org.slf4j.LoggerFactory;
import static org.apache.solr.cloud.Assign.getNodesForNewReplicas;
import static org.apache.solr.cloud.OverseerCollectionMessageHandler.COLL_CONF;
-import static org.apache.solr.cloud.OverseerCollectionMessageHandler.RANDOM;
import static org.apache.solr.cloud.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;
@@ -95,13 +97,14 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
ShardHandler shardHandler = ocmh.shardHandlerFactory.getShardHandler();
boolean skipCreateReplicaInClusterState = message.getBool(SKIP_CREATE_REPLICA_IN_CLUSTER_STATE, false);
+ final Long policyVersionBefore = PolicyHelper.REF_VERSION.get();
+ AtomicLong policyVersionAfter = new AtomicLong(-1);
// Kind of unnecessary, but it does put the logic of whether to override maxShardsPerNode in one place.
if (!skipCreateReplicaInClusterState) {
if (CreateShardCmd.usePolicyFramework(coll, ocmh)) {
if (node == null) {
if(coll.getPolicyName() != null) message.getProperties().put(Policy.POLICY, coll.getPolicyName());
- node = Assign.identifyNodes(() -> ocmh.overseer.getZkController().getCoreContainer(),
- ocmh.zkStateReader,
+ node = Assign.identifyNodes(ocmh,
clusterState,
Collections.emptyList(),
collection,
@@ -111,6 +114,9 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
replicaType == Replica.Type.TLOG ? 0 : 1,
replicaType == Replica.Type.PULL ? 0 : 1
).get(0).node;
+ if (policyVersionBefore == null && PolicyHelper.REF_VERSION.get() != null) {
+ policyVersionAfter.set(PolicyHelper.REF_VERSION.get());
+ }
}
} else {
node = getNodesForNewReplicas(clusterState, collection, shard, 1, node,
@@ -206,6 +212,10 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
Runnable runnable = () -> {
ocmh.processResponses(results, shardHandler, true, "ADDREPLICA failed to create replica", asyncId, requestMap);
ocmh.waitForCoreNodeName(collection, fnode, fcoreName);
+ if (policyVersionAfter.get() > -1) {
+ PolicyHelper.REF_VERSION.remove();
+ ocmh.policySessionRef.decref(policyVersionAfter.get());
+ }
if (onComplete != null) onComplete.run();
};
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/91446f7f/solr/core/src/java/org/apache/solr/cloud/Assign.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/Assign.java b/solr/core/src/java/org/apache/solr/cloud/Assign.java
index 35019c7..c6897cd 100644
--- a/solr/core/src/java/org/apache/solr/cloud/Assign.java
+++ b/solr/core/src/java/org/apache/solr/cloud/Assign.java
@@ -28,7 +28,6 @@ import java.util.Locale;
import java.util.Map;
import java.util.Random;
import java.util.Set;
-import java.util.function.Supplier;
import java.util.stream.Collectors;
import com.google.common.collect.ImmutableMap;
@@ -236,8 +235,7 @@ public class Assign {
return nodeList;
}
- public static List<ReplicaPosition> identifyNodes(Supplier<CoreContainer> coreContainer,
- ZkStateReader zkStateReader,
+ public static List<ReplicaPosition> identifyNodes(OverseerCollectionMessageHandler ocmh,
ClusterState clusterState,
List<String> nodeList,
String collectionName,
@@ -248,7 +246,7 @@ public class Assign {
int numPullReplicas) throws KeeperException, InterruptedException {
List<Map> rulesMap = (List) message.get("rule");
String policyName = message.getStr(POLICY);
- AutoScalingConfig autoScalingConfig = zkStateReader.getAutoScalingConfig();
+ AutoScalingConfig autoScalingConfig = ocmh.zkStateReader.getAutoScalingConfig();
if (rulesMap == null && policyName == null && autoScalingConfig.getPolicy().getClusterPolicy().isEmpty()) {
log.debug("Identify nodes using default");
@@ -275,8 +273,16 @@ public class Assign {
if (policyName != null || !autoScalingConfig.getPolicy().getClusterPolicy().isEmpty()) {
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, zkStateReader, nodeList);
+ synchronized (ocmh) {
+ PolicyHelper.SESSION_REF.set(ocmh.policySessionRef);
+ try {
+ return getPositionsUsingPolicy(collectionName,
+ shardNames, numNrtReplicas, numTlogReplicas, numPullReplicas, policyName, ocmh.zkStateReader, nodeList);
+ } finally {
+ PolicyHelper.SESSION_REF.remove();
+ }
+
+ }
} else {
log.debug("Identify nodes using rules framework");
List<Rule> rules = new ArrayList<>();
@@ -289,7 +295,7 @@ public class Assign {
(List<Map>) message.get(SNITCH),
new HashMap<>(),//this is a new collection. So, there are no nodes in any shard
nodeList,
- coreContainer.get(),
+ ocmh.overseer.getZkController().getCoreContainer(),
clusterState);
Map<ReplicaPosition, String> nodeMappings = replicaAssigner.getNodeMappings();
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/91446f7f/solr/core/src/java/org/apache/solr/cloud/CreateCollectionCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/CreateCollectionCmd.java b/solr/core/src/java/org/apache/solr/cloud/CreateCollectionCmd.java
index 978cfdf2..87f9bca 100644
--- a/solr/core/src/java/org/apache/solr/cloud/CreateCollectionCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/CreateCollectionCmd.java
@@ -30,6 +30,7 @@ import java.util.Properties;
import java.util.concurrent.TimeUnit;
import org.apache.solr.client.solrj.cloud.autoscaling.Policy;
+import org.apache.solr.client.solrj.cloud.autoscaling.PolicyHelper;
import org.apache.solr.cloud.OverseerCollectionMessageHandler.Cmd;
import org.apache.solr.cloud.autoscaling.AutoScalingHandler;
import org.apache.solr.cloud.overseer.ClusterStateMutator;
@@ -196,8 +197,7 @@ public class CreateCollectionCmd implements Cmd {
}
}
- replicaPositions = Assign.identifyNodes(() -> ocmh.overseer.getZkController().getCoreContainer(),
- ocmh.zkStateReader
+ replicaPositions = Assign.identifyNodes(ocmh
, clusterState, nodeList, collectionName, message, shardNames, numNrtReplicas, numTlogReplicas, numPullReplicas);
}
@@ -331,6 +331,8 @@ public class CreateCollectionCmd implements Cmd {
throw ex;
} catch (Exception ex) {
throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, null, ex);
+ } finally {
+ PolicyHelper.clearFlagAndDecref(ocmh.policySessionRef);
}
}
String getConfigName(String coll, ZkNodeProps message) throws KeeperException, InterruptedException {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/91446f7f/solr/core/src/java/org/apache/solr/cloud/CreateShardCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/CreateShardCmd.java b/solr/core/src/java/org/apache/solr/cloud/CreateShardCmd.java
index e277690..21258e4 100644
--- a/solr/core/src/java/org/apache/solr/cloud/CreateShardCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/CreateShardCmd.java
@@ -28,6 +28,7 @@ import java.util.concurrent.TimeUnit;
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.cloud.OverseerCollectionMessageHandler.Cmd;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.cloud.ClusterState;
@@ -87,76 +88,80 @@ public class CreateShardCmd implements Cmd {
ZkStateReader zkStateReader = ocmh.zkStateReader;
boolean usePolicyFramework = usePolicyFramework(collection,ocmh);
List<ReplicaPosition> positions = null;
- if (usePolicyFramework) {
- if (collection.getPolicyName() != null) message.getProperties().put(Policy.POLICY, collection.getPolicyName());
- positions = Assign.identifyNodes(() -> ocmh.overseer.getZkController().getCoreContainer(),
- zkStateReader,
- clusterState,
- Assign.getLiveOrLiveAndCreateNodeSetList(clusterState.getLiveNodes(), message, RANDOM),
- collectionName,
- message,
- Collections.singletonList(sliceName),
- numNrtReplicas,
- numTlogReplicas,
- numPullReplicas);
- } else {
- List<Assign.ReplicaCount> sortedNodeList = getNodesForNewReplicas(clusterState, collectionName, sliceName, totalReplicas,
- createNodeSetStr, ocmh.overseer.getZkController().getCoreContainer());
- 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++;
+ CountDownLatch countDownLatch;
+ try {
+ if (usePolicyFramework) {
+ if (collection.getPolicyName() != null) message.getProperties().put(Policy.POLICY, collection.getPolicyName());
+ positions = Assign.identifyNodes(ocmh,
+ clusterState,
+ Assign.getLiveOrLiveAndCreateNodeSetList(clusterState.getLiveNodes(), message, RANDOM),
+ collectionName,
+ message,
+ Collections.singletonList(sliceName),
+ numNrtReplicas,
+ numTlogReplicas,
+ numPullReplicas);
+ } else {
+ List<Assign.ReplicaCount> sortedNodeList = getNodesForNewReplicas(clusterState, collectionName, sliceName, totalReplicas,
+ createNodeSetStr, ocmh.overseer.getZkController().getCoreContainer());
+ 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++;
+ }
}
}
- }
- 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 countDownLatch = new CountDownLatch(totalReplicas);
- for (ReplicaPosition position : positions) {
- String nodeName = position.node;
- String coreName = Assign.buildCoreName(ocmh.zkStateReader.getZkClient(), 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);
- 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);
+ 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 CountDownLatch(totalReplicas);
+ for (ReplicaPosition position : positions) {
+ String nodeName = position.node;
+ String coreName = Assign.buildCoreName(ocmh.zkStateReader.getZkClient(), 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);
+ 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"));
}
- success.addAll((NamedList) addResult.get("success"));
- }
- });
+ });
+ }
+ } finally {
+ PolicyHelper.clearFlagAndDecref(ocmh.policySessionRef);
}
log.debug("Waiting for create shard action to complete");
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/91446f7f/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionMessageHandler.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionMessageHandler.java b/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionMessageHandler.java
index 7a08030..577cf64 100644
--- a/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionMessageHandler.java
+++ b/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionMessageHandler.java
@@ -36,6 +36,7 @@ import com.google.common.collect.ImmutableMap;
import org.apache.commons.lang.StringUtils;
import org.apache.solr.client.solrj.SolrResponse;
import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.cloud.autoscaling.PolicyHelper;
import org.apache.solr.client.solrj.impl.HttpSolrClient;
import org.apache.solr.client.solrj.impl.HttpSolrClient.RemoteSolrException;
import org.apache.solr.client.solrj.request.AbstractUpdateRequest;
@@ -973,6 +974,8 @@ public class OverseerCollectionMessageHandler implements OverseerMessageHandler
);
}
+ public final PolicyHelper.SessionRef policySessionRef = new PolicyHelper.SessionRef();
+
@Override
public void close() throws IOException {
if (tpe != null) {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/91446f7f/solr/core/src/java/org/apache/solr/cloud/RestoreCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/RestoreCmd.java b/solr/core/src/java/org/apache/solr/cloud/RestoreCmd.java
index fa493c7..9edf729 100644
--- a/solr/core/src/java/org/apache/solr/cloud/RestoreCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/RestoreCmd.java
@@ -32,7 +32,9 @@ import java.util.Objects;
import java.util.Optional;
import java.util.Properties;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.solr.client.solrj.cloud.autoscaling.PolicyHelper;
import org.apache.solr.cloud.overseer.OverseerAction;
import org.apache.solr.common.cloud.ReplicaPosition;
import org.apache.solr.common.SolrException;
@@ -214,135 +216,138 @@ public class RestoreCmd implements OverseerCollectionMessageHandler.Cmd {
List<String> sliceNames = new ArrayList<>();
restoreCollection.getSlices().forEach(x -> sliceNames.add(x.getName()));
- List<ReplicaPosition> replicaPositions = Assign.identifyNodes(() -> ocmh.overseer.getZkController().getCoreContainer(),
- ocmh.zkStateReader, clusterState,
- nodeList, restoreCollectionName,
- message, sliceNames,
- numNrtReplicas, numTlogReplicas, numPullReplicas);
-
- //Create one replica per shard and copy backed up data to it
- for (Slice slice : restoreCollection.getSlices()) {
- log.debug("Adding replica for shard={} collection={} ", slice.getName(), restoreCollection);
- HashMap<String, Object> propMap = new HashMap<>();
- propMap.put(Overseer.QUEUE_OPERATION, CREATESHARD);
- propMap.put(COLLECTION_PROP, restoreCollectionName);
- propMap.put(SHARD_ID_PROP, slice.getName());
-
- if (numNrtReplicas >= 1) {
- propMap.put(REPLICA_TYPE, Replica.Type.NRT.name());
- } else if (numTlogReplicas >= 1) {
- propMap.put(REPLICA_TYPE, Replica.Type.TLOG.name());
- } else {
- throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "Unexpected number of replicas, replicationFactor, " +
- Replica.Type.NRT + " or " + Replica.Type.TLOG + " must be greater than 0");
- }
-
- // Get the first node matching the shard to restore in
- String node;
- for (ReplicaPosition replicaPosition : replicaPositions) {
- if (Objects.equals(replicaPosition.shard, slice.getName())) {
- node = replicaPosition.node;
- propMap.put(CoreAdminParams.NODE, node);
- replicaPositions.remove(replicaPosition);
- break;
+ try {
+ List<ReplicaPosition> replicaPositions = Assign.identifyNodes(
+ ocmh, clusterState,
+ nodeList, restoreCollectionName,
+ message, sliceNames,
+ numNrtReplicas, numTlogReplicas, numPullReplicas);
+ //Create one replica per shard and copy backed up data to it
+ for (Slice slice : restoreCollection.getSlices()) {
+ log.debug("Adding replica for shard={} collection={} ", slice.getName(), restoreCollection);
+ HashMap<String, Object> propMap = new HashMap<>();
+ propMap.put(Overseer.QUEUE_OPERATION, CREATESHARD);
+ propMap.put(COLLECTION_PROP, restoreCollectionName);
+ propMap.put(SHARD_ID_PROP, slice.getName());
+
+ if (numNrtReplicas >= 1) {
+ propMap.put(REPLICA_TYPE, Replica.Type.NRT.name());
+ } else if (numTlogReplicas >= 1) {
+ propMap.put(REPLICA_TYPE, Replica.Type.TLOG.name());
+ } else {
+ throw new SolrException(ErrorCode.BAD_REQUEST, "Unexpected number of replicas, replicationFactor, " +
+ Replica.Type.NRT + " or " + Replica.Type.TLOG + " must be greater than 0");
}
- }
- // add async param
- if (asyncId != null) {
- propMap.put(ASYNC, asyncId);
- }
- ocmh.addPropertyParams(message, propMap);
+ // Get the first node matching the shard to restore in
+ String node;
+ for (ReplicaPosition replicaPosition : replicaPositions) {
+ if (Objects.equals(replicaPosition.shard, slice.getName())) {
+ node = replicaPosition.node;
+ propMap.put(CoreAdminParams.NODE, node);
+ replicaPositions.remove(replicaPosition);
+ break;
+ }
+ }
- ocmh.addReplica(clusterState, new ZkNodeProps(propMap), new NamedList(), null);
- }
+ // add async param
+ if (asyncId != null) {
+ propMap.put(ASYNC, asyncId);
+ }
+ ocmh.addPropertyParams(message, propMap);
- //refresh the location copy of collection state
- restoreCollection = zkStateReader.getClusterState().getCollection(restoreCollectionName);
+ ocmh.addReplica(clusterState, new ZkNodeProps(propMap), new NamedList(), null);
+ }
- //Copy data from backed up index to each replica
- for (Slice slice : restoreCollection.getSlices()) {
- ModifiableSolrParams params = new ModifiableSolrParams();
- params.set(CoreAdminParams.ACTION, CoreAdminParams.CoreAdminAction.RESTORECORE.toString());
- params.set(NAME, "snapshot." + slice.getName());
- params.set(CoreAdminParams.BACKUP_LOCATION, backupPath.toASCIIString());
- params.set(CoreAdminParams.BACKUP_REPOSITORY, repo);
+ //refresh the location copy of collection state
+ restoreCollection = zkStateReader.getClusterState().getCollection(restoreCollectionName);
- ocmh.sliceCmd(clusterState, params, null, slice, shardHandler, asyncId, requestMap);
- }
- ocmh.processResponses(new NamedList(), shardHandler, true, "Could not restore core", asyncId, requestMap);
+ //Copy data from backed up index to each replica
+ for (Slice slice : restoreCollection.getSlices()) {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+ params.set(CoreAdminParams.ACTION, CoreAdminParams.CoreAdminAction.RESTORECORE.toString());
+ params.set(NAME, "snapshot." + slice.getName());
+ params.set(CoreAdminParams.BACKUP_LOCATION, backupPath.toASCIIString());
+ params.set(CoreAdminParams.BACKUP_REPOSITORY, repo);
- //Mark all shards in ACTIVE STATE
- {
- HashMap<String, Object> propMap = new HashMap<>();
- propMap.put(Overseer.QUEUE_OPERATION, OverseerAction.UPDATESHARDSTATE.toLower());
- propMap.put(ZkStateReader.COLLECTION_PROP, restoreCollectionName);
- for (Slice shard : restoreCollection.getSlices()) {
- propMap.put(shard.getName(), Slice.State.ACTIVE.toString());
+ ocmh.sliceCmd(clusterState, params, null, slice, shardHandler, asyncId, requestMap);
+ }
+ ocmh.processResponses(new NamedList(), shardHandler, true, "Could not restore core", asyncId, requestMap);
+
+ //Mark all shards in ACTIVE STATE
+ {
+ HashMap<String, Object> propMap = new HashMap<>();
+ propMap.put(Overseer.QUEUE_OPERATION, OverseerAction.UPDATESHARDSTATE.toLower());
+ propMap.put(ZkStateReader.COLLECTION_PROP, restoreCollectionName);
+ for (Slice shard : restoreCollection.getSlices()) {
+ propMap.put(shard.getName(), Slice.State.ACTIVE.toString());
+ }
+ inQueue.offer(Utils.toJSON(new ZkNodeProps(propMap)));
}
- inQueue.offer(Utils.toJSON(new ZkNodeProps(propMap)));
- }
- //refresh the location copy of collection state
- restoreCollection = zkStateReader.getClusterState().getCollection(restoreCollectionName);
+ //refresh the location copy of collection state
+ restoreCollection = zkStateReader.getClusterState().getCollection(restoreCollectionName);
- if (totalReplicasPerShard > 1) {
- log.info("Adding replicas to restored collection={}", restoreCollection);
- for (Slice slice : restoreCollection.getSlices()) {
+ if (totalReplicasPerShard > 1) {
+ log.info("Adding replicas to restored collection={}", restoreCollection);
+ for (Slice slice : restoreCollection.getSlices()) {
- //Add the remaining replicas for each shard, considering it's type
- int createdNrtReplicas = 0, createdTlogReplicas = 0, createdPullReplicas = 0;
+ //Add the remaining replicas for each shard, considering it's type
+ int createdNrtReplicas = 0, createdTlogReplicas = 0, createdPullReplicas = 0;
- // We already created either a NRT or an TLOG replica as leader
- if (numNrtReplicas > 0) {
- createdNrtReplicas++;
- } else if (createdTlogReplicas > 0) {
- createdTlogReplicas++;
- }
-
- for (int i = 1; i < totalReplicasPerShard; i++) {
- Replica.Type typeToCreate;
- if (createdNrtReplicas < numNrtReplicas) {
+ // We already created either a NRT or an TLOG replica as leader
+ if (numNrtReplicas > 0) {
createdNrtReplicas++;
- typeToCreate = Replica.Type.NRT;
- } else if (createdTlogReplicas < numTlogReplicas) {
+ } else if (createdTlogReplicas > 0) {
createdTlogReplicas++;
- typeToCreate = Replica.Type.TLOG;
- } else {
- createdPullReplicas++;
- typeToCreate = Replica.Type.PULL;
- assert createdPullReplicas <= numPullReplicas: "Unexpected number of replicas";
}
- log.debug("Adding replica for shard={} collection={} of type {} ", slice.getName(), restoreCollection, typeToCreate);
- HashMap<String, Object> propMap = new HashMap<>();
- propMap.put(COLLECTION_PROP, restoreCollectionName);
- propMap.put(SHARD_ID_PROP, slice.getName());
- propMap.put(REPLICA_TYPE, typeToCreate.name());
-
- // Get the first node matching the shard to restore in
- String node;
- for (ReplicaPosition replicaPosition : replicaPositions) {
- if (Objects.equals(replicaPosition.shard, slice.getName())) {
- node = replicaPosition.node;
- propMap.put(CoreAdminParams.NODE, node);
- replicaPositions.remove(replicaPosition);
- break;
+ for (int i = 1; i < totalReplicasPerShard; i++) {
+ Replica.Type typeToCreate;
+ if (createdNrtReplicas < numNrtReplicas) {
+ createdNrtReplicas++;
+ typeToCreate = Replica.Type.NRT;
+ } else if (createdTlogReplicas < numTlogReplicas) {
+ createdTlogReplicas++;
+ typeToCreate = Replica.Type.TLOG;
+ } else {
+ createdPullReplicas++;
+ typeToCreate = Replica.Type.PULL;
+ assert createdPullReplicas <= numPullReplicas: "Unexpected number of replicas";
}
- }
- // add async param
- if (asyncId != null) {
- propMap.put(ASYNC, asyncId);
- }
- ocmh.addPropertyParams(message, propMap);
+ log.debug("Adding replica for shard={} collection={} of type {} ", slice.getName(), restoreCollection, typeToCreate);
+ HashMap<String, Object> propMap = new HashMap<>();
+ propMap.put(COLLECTION_PROP, restoreCollectionName);
+ propMap.put(SHARD_ID_PROP, slice.getName());
+ propMap.put(REPLICA_TYPE, typeToCreate.name());
+
+ // Get the first node matching the shard to restore in
+ String node;
+ for (ReplicaPosition replicaPosition : replicaPositions) {
+ if (Objects.equals(replicaPosition.shard, slice.getName())) {
+ node = replicaPosition.node;
+ propMap.put(CoreAdminParams.NODE, node);
+ replicaPositions.remove(replicaPosition);
+ break;
+ }
+ }
+
+ // add async param
+ if (asyncId != null) {
+ propMap.put(ASYNC, asyncId);
+ }
+ ocmh.addPropertyParams(message, propMap);
- ocmh.addReplica(zkStateReader.getClusterState(), new ZkNodeProps(propMap), results, null);
+ ocmh.addReplica(zkStateReader.getClusterState(), new ZkNodeProps(propMap), results, null);
+ }
}
}
- }
- log.info("Completed restoring collection={} backupName={}", restoreCollection, backupName);
+ log.info("Completed restoring collection={} backupName={}", restoreCollection, backupName);
+ } finally {
+ PolicyHelper.clearFlagAndDecref(ocmh.policySessionRef);
+ }
}
private int getInt(ZkNodeProps message, String propertyName, Integer default1, int default2) {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/91446f7f/solr/core/src/java/org/apache/solr/cloud/SplitShardCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/SplitShardCmd.java b/solr/core/src/java/org/apache/solr/cloud/SplitShardCmd.java
index 8c69f54..3bd6d89 100644
--- a/solr/core/src/java/org/apache/solr/cloud/SplitShardCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/SplitShardCmd.java
@@ -27,6 +27,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
+import org.apache.solr.client.solrj.cloud.autoscaling.PolicyHelper;
import org.apache.solr.client.solrj.request.CoreAdminRequest;
import org.apache.solr.cloud.OverseerCollectionMessageHandler.Cmd;
import org.apache.solr.cloud.overseer.OverseerAction;
@@ -381,8 +382,8 @@ public class SplitShardCmd implements Cmd {
// TODO: change this to handle sharding a slice into > 2 sub-shards.
- List<ReplicaPosition> replicaPositions = Assign.identifyNodes(() -> ocmh.overseer.getZkController().getCoreContainer(),
- ocmh.zkStateReader, clusterState,
+ List<ReplicaPosition> replicaPositions = Assign.identifyNodes(ocmh,
+ clusterState,
new ArrayList<>(clusterState.getLiveNodes()),
collectionName,
new ZkNodeProps(collection.getProperties()),
@@ -504,6 +505,8 @@ public class SplitShardCmd implements Cmd {
} catch (Exception e) {
log.error("Error executing split operation for collection: " + collectionName + " parent shard: " + slice, e);
throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, null, e);
+ } finally {
+ PolicyHelper.clearFlagAndDecref(ocmh.policySessionRef);
}
}
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/91446f7f/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/PolicyHelper.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/PolicyHelper.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/PolicyHelper.java
index 1a6f33e..a8c99ec 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/PolicyHelper.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/PolicyHelper.java
@@ -21,22 +21,24 @@ package org.apache.solr.client.solrj.cloud.autoscaling;
import java.util.ArrayList;
import java.util.Collection;
import java.util.EnumMap;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.solr.client.solrj.SolrRequest;
import org.apache.solr.client.solrj.cloud.autoscaling.Policy.Suggester.Hint;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.cloud.ReplicaPosition;
-import org.apache.solr.common.params.CoreAdminParams;
import org.apache.solr.common.util.Utils;
import static org.apache.solr.common.params.CollectionParams.CollectionAction.ADDREPLICA;
import static org.apache.solr.common.params.CoreAdminParams.NODE;
public class PolicyHelper {
+ private static ThreadLocal<Map<String, String>> policyMapping = new ThreadLocal<>();
public static List<ReplicaPosition> getReplicaLocations(String collName, AutoScalingConfig autoScalingConfig,
ClusterDataProvider cdp,
Map<String, String> optionalPolicyMapping,
@@ -46,7 +48,6 @@ public class PolicyHelper {
int pullReplicas,
List<String> nodesList) {
List<ReplicaPosition> positions = new ArrayList<>();
- if (optionalPolicyMapping != null) {
final ClusterDataProvider delegate = cdp;
cdp = new ClusterDataProvider() {
@Override
@@ -66,41 +67,127 @@ public class PolicyHelper {
@Override
public String getPolicyNameByCollection(String coll) {
- return optionalPolicyMapping.containsKey(coll) ?
+ return policyMapping.get() != null && policyMapping.get().containsKey(coll) ?
optionalPolicyMapping.get(coll) :
delegate.getPolicyNameByCollection(coll);
}
};
- }
- Policy.Session session = autoScalingConfig.getPolicy().createSession(cdp);
- Map<Replica.Type, Integer> typeVsCount = new EnumMap<>(Replica.Type.class);
- typeVsCount.put(Replica.Type.NRT, nrtReplicas);
- typeVsCount.put(Replica.Type.TLOG, tlogReplicas);
- typeVsCount.put(Replica.Type.PULL, pullReplicas);
- for (String shardName : shardNames) {
- int idx = 0;
- for (Map.Entry<Replica.Type, Integer> e : typeVsCount.entrySet()) {
- for (int i = 0; i < e.getValue(); i++) {
- Policy.Suggester suggester = session.getSuggester(ADDREPLICA)
- .hint(Hint.COLL, collName)
- .hint(Hint.REPLICATYPE, e.getKey())
- .hint(Hint.SHARD, shardName);
- if (nodesList != null) {
- for (String nodeName : nodesList) {
- suggester = suggester.hint(Hint.TARGET_NODE, nodeName);
+ policyMapping.set(optionalPolicyMapping);
+ Policy.Session session = null;
+ try {
+ session = SESSION_REF.get() != null ?
+ SESSION_REF.get().initOrGet(cdp, autoScalingConfig.getPolicy()) :
+ autoScalingConfig.getPolicy().createSession(cdp);
+
+ Map<Replica.Type, Integer> typeVsCount = new EnumMap<>(Replica.Type.class);
+ typeVsCount.put(Replica.Type.NRT, nrtReplicas);
+ typeVsCount.put(Replica.Type.TLOG, tlogReplicas);
+ typeVsCount.put(Replica.Type.PULL, pullReplicas);
+ for (String shardName : shardNames) {
+ int idx = 0;
+ for (Map.Entry<Replica.Type, Integer> e : typeVsCount.entrySet()) {
+ for (int i = 0; i < e.getValue(); i++) {
+ Policy.Suggester suggester = session.getSuggester(ADDREPLICA)
+ .hint(Hint.COLL, collName)
+ .hint(Hint.REPLICATYPE, e.getKey())
+ .hint(Hint.SHARD, shardName);
+ if (nodesList != null) {
+ for (String nodeName : nodesList) {
+ suggester = suggester.hint(Hint.TARGET_NODE, nodeName);
+ }
}
+ SolrRequest op = suggester.getOperation();
+ if (op == null) {
+ throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "No node can satisfy the rules " +
+ Utils.toJSONString(Utils.getDeepCopy(session.expandedClauses, 4, true)));
+ }
+ session = suggester.getSession();
+ positions.add(new ReplicaPosition(shardName, ++idx, e.getKey(), op.getParams().get(NODE)));
}
- SolrRequest op = suggester.getOperation();
- if (op == null) {
- throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "No node can satisfy the rules " + Utils.toJSONString(Utils.getDeepCopy(session.expandedClauses, 4, true)));
- }
- session = suggester.getSession();
- positions.add(new ReplicaPosition(shardName, ++idx, e.getKey(), op.getParams().get(NODE)));
}
}
+ } finally {
+ if (session != null && SESSION_REF.get() != null) SESSION_REF.get().updateSession(session);
+ policyMapping.remove();
}
-
return positions;
}
+
+
+ public static final int SESSION_EXPIRY = 180;//3 seconds
+ public static ThreadLocal<Long> REF_VERSION = new ThreadLocal<>();
+
+ public static class SessionRef {
+ private final AtomicLong myVersion = new AtomicLong(0);
+ AtomicInteger refCount = new AtomicInteger();
+ private Policy.Session session;
+ long lastUsedTime;
+
+ public SessionRef() {
+ }
+
+ public long getRefVersion(){
+ return myVersion.get();
+ }
+
+
+ public void decref(long version) {
+ synchronized (SessionRef.class) {
+ if (session == null) return;
+ if(myVersion.get() != version) return;
+ if (refCount.decrementAndGet() <= 0) {
+ session = null;
+ lastUsedTime = 0;
+ }
+ }
+ }
+
+ public int getRefCount() {
+ return refCount.get();
+ }
+
+ public Policy.Session get() {
+ synchronized (SessionRef.class) {
+ if (session == null) return null;
+ if (TimeUnit.SECONDS.convert(System.nanoTime() - lastUsedTime, TimeUnit.NANOSECONDS) > SESSION_EXPIRY) {
+ session = null;
+ return null;
+ } else {
+ REF_VERSION.set(myVersion.get());
+ refCount.incrementAndGet();
+ return session;
+ }
+ }
+ }
+
+ public Policy.Session initOrGet(ClusterDataProvider cdp, Policy policy) {
+ synchronized (SessionRef.class) {
+ Policy.Session session = get();
+ if (session != null) return session;
+ this.session = policy.createSession(cdp);
+ myVersion.incrementAndGet();
+ lastUsedTime = System.nanoTime();
+ REF_VERSION.set(myVersion.get());
+ refCount.set(1);
+ return this.session;
+ }
+ }
+
+
+ private void updateSession(Policy.Session session) {
+ this.session = session;
+ lastUsedTime = System.nanoTime();
+ }
+ }
+
+ public static void clearFlagAndDecref(SessionRef policySessionRef) {
+ Long refVersion = REF_VERSION.get();
+ if (refVersion != null) policySessionRef.decref(refVersion);
+ REF_VERSION.remove();
+ }
+
+ public static ThreadLocal<SessionRef> SESSION_REF = new ThreadLocal<>();
+
+
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/91446f7f/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 483489a..a11eab4 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
@@ -27,6 +27,7 @@ import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import com.google.common.collect.ImmutableList;
import org.apache.solr.SolrTestCaseJ4;
@@ -40,6 +41,7 @@ import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.params.SolrParams;
import org.apache.solr.common.util.Utils;
import org.apache.solr.common.util.ValidatingJsonMap;
+import org.junit.Test;
import static org.apache.solr.common.params.CollectionParams.CollectionAction.ADDREPLICA;
import static org.apache.solr.common.params.CollectionParams.CollectionAction.MOVEREPLICA;
@@ -832,7 +834,82 @@ public class TestPolicy extends SolrTestCaseJ4 {
.getOperation();
assertNotNull(opReq);
assertEquals("node5", opReq.getParams().get("targetNode"));
+ }
+
+ @Test
+ public void testSessionCaching() {
+ PolicyHelper.SessionRef ref1 = new PolicyHelper.SessionRef();
+ String autoScalingjson = " '{cluster-policy':[" +
+ " { 'cores':'<10', 'node':'#ANY'}," +
+ " { 'replica':'<2', 'shard':'#EACH', 'node':'#ANY'}," +
+ " { 'nodeRole':'overseer','replica':0}]," +
+ " 'cluster-preferences':[{'minimize':'cores'}]}";
+ Policy policy = new Policy((Map<String, Object>) Utils.fromJSONString(autoScalingjson));
+ PolicyHelper.SESSION_REF.set(ref1);
+ String nodeValues = " {" +
+ " 'node4':{" +
+ " 'node':'10.0.0.4:8987_solr'," +
+ " 'cores':1," +
+ " 'freedisk':884.7097854614258}," +
+ " 'node3':{" +
+ " 'node':'10.0.0.4:8989_solr'," +
+ " 'cores':1," +
+ " 'freedisk':884.7097854614258}," +
+ " 'node2':{" +
+ " 'node':'10.0.0.4:7574_solr'," +
+ " 'cores':1," +
+ " 'freedisk':884.7097854614258}," +
+ "}";
+
+ ClusterDataProvider provider = getClusterDataProvider((Map<String, Map>) Utils.fromJSONString(nodeValues), clusterState);
+ Map policies = (Map) Utils.fromJSONString("{" +
+ " 'cluster-preferences': [" +
+ " { 'maximize': 'freedisk', 'precision': 50}," +
+ " { 'minimize': 'cores', 'precision': 50}" +
+ " ]," +
+ " 'cluster-policy': [" +
+ " { 'replica': 0, 'nodeRole': 'overseer'}" +
+ " { 'replica': '<2', 'shard': '#EACH', 'node': '#ANY'}," +
+ " ]" +
+ "}");
+ AutoScalingConfig config = new AutoScalingConfig(policies);
+ List<ReplicaPosition> locations = PolicyHelper.getReplicaLocations("c", config, provider, null,
+ Arrays.asList("s1", "s2"), 1, 0, 0,
+ null);
+
+ long sessionRefVersion = PolicyHelper.REF_VERSION.get();
+ PolicyHelper.SessionRef ref1Copy = PolicyHelper.SESSION_REF.get();
+ PolicyHelper.SESSION_REF.remove();
+ Policy.Session session = ref1Copy.get();
+ assertNotNull(session);
+ assertEquals(ref1, ref1Copy);
+ assertTrue(session.getPolicy() == config.getPolicy());
+ ref1Copy.decref(sessionRefVersion);
+ PolicyHelper.SESSION_REF.set(ref1);
+ AutoScalingConfig config2 = new AutoScalingConfig(policies);
+ locations = PolicyHelper.getReplicaLocations("c2", config2, provider, null, Arrays.asList("s1", "s2"), 1, 0, 0,
+ null);
+ sessionRefVersion = PolicyHelper.REF_VERSION.get();
+ ref1Copy = PolicyHelper.SESSION_REF.get();
+ PolicyHelper.SESSION_REF.remove();
+ session = ref1Copy.get();
+ ref1Copy.decref(sessionRefVersion);
+ assertEquals(ref1, ref1Copy);
+ assertFalse(session.getPolicy() == config2.getPolicy());
+ assertTrue(session.getPolicy() == config.getPolicy());
+ assertEquals(2, ref1Copy.getRefCount());
+ ref1.decref(sessionRefVersion);//decref 1
+ ref1.decref(sessionRefVersion);//decref 2
+ PolicyHelper.SESSION_REF.set(ref1);
+ locations = PolicyHelper.getReplicaLocations("c3", config2, provider, null, Arrays.asList("s1", "s2"), 1, 0, 0,
+ null);
+ sessionRefVersion = PolicyHelper.REF_VERSION.get();
+ ref1Copy = PolicyHelper.SESSION_REF.get();
+ PolicyHelper.SESSION_REF.remove();
+ session = ref1Copy.get();
+ ref1Copy.decref(sessionRefVersion);
+ assertTrue(session.getPolicy() == config2.getPolicy());
}
@@ -1176,7 +1253,7 @@ public class TestPolicy extends SolrTestCaseJ4 {
dataProvider, Collections.singletonMap("newColl", "policy1"), Arrays.asList("shard1", "shard2"), 3,0,0, null);
assertTrue(locations.stream().allMatch(it -> ImmutableList.of("node2", "node1", "node3").contains(it.node)) );
}
-
+
public void testMoveReplicaSuggester(){
String dataproviderdata = "{" +
" 'liveNodes':[" +