You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by ab...@apache.org on 2017/10/04 14:43:26 UTC
[30/45] lucene-solr:jira/solr-11285-sim: SOLR-11285: Move
getAutoScalingConfig to DistribStateManager. Further refactoring and mocking
fixes.
SOLR-11285: Move getAutoScalingConfig to DistribStateManager. Further refactoring
and mocking fixes.
Project: http://git-wip-us.apache.org/repos/asf/lucene-solr/repo
Commit: http://git-wip-us.apache.org/repos/asf/lucene-solr/commit/dddc6723
Tree: http://git-wip-us.apache.org/repos/asf/lucene-solr/tree/dddc6723
Diff: http://git-wip-us.apache.org/repos/asf/lucene-solr/diff/dddc6723
Branch: refs/heads/jira/solr-11285-sim
Commit: dddc672366f17b085595bee1ac421396217f1615
Parents: 91d76f0
Author: Andrzej Bialecki <ab...@apache.org>
Authored: Thu Sep 28 20:03:22 2017 +0200
Committer: Andrzej Bialecki <ab...@apache.org>
Committed: Thu Sep 28 20:03:22 2017 +0200
----------------------------------------------------------------------
.../org/apache/solr/cloud/AddReplicaCmd.java | 2 +-
.../src/java/org/apache/solr/cloud/Assign.java | 27 +++---
.../org/apache/solr/cloud/CreateShardCmd.java | 4 +-
.../cloud/autoscaling/ComputePlanAction.java | 2 +-
.../autoscaling/OverseerTriggerThread.java | 2 +-
...verseerCollectionConfigSetProcessorTest.java | 43 +++++++++-
.../cloud/autoscaling/ClusterDataProvider.java | 6 --
.../DelegatingClusterDataProvider.java | 23 +++--
.../cloud/autoscaling/DistribStateManager.java | 6 ++
.../solrj/cloud/autoscaling/PolicyHelper.java | 6 +-
.../impl/SolrClientClusterDataProvider.java | 11 ---
.../solrj/impl/ZkDistribStateManager.java | 25 ++++++
.../solrj/cloud/autoscaling/TestPolicy.java | 90 ++++++++++++++++++--
13 files changed, 198 insertions(+), 49 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/dddc6723/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 98c792e..10a188d 100644
--- a/solr/core/src/java/org/apache/solr/cloud/AddReplicaCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/AddReplicaCmd.java
@@ -119,7 +119,7 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
}
} else {
node = getNodesForNewReplicas(clusterState, collection, shard, 1, node,
- ocmh.overseer.getSolrCloudDataProvider().getClusterDataProvider(), ocmh.overseer.getCoreContainer()).get(0).nodeName;// TODO: use replica type in this logic too
+ ocmh.overseer.getSolrCloudDataProvider(), ocmh.overseer.getCoreContainer()).get(0).nodeName;// TODO: use replica type in this logic too
}
}
log.info("Node Identified {} for creating new replica", node);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/dddc6723/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 7e2c073..a3bfa8e 100644
--- a/solr/core/src/java/org/apache/solr/cloud/Assign.java
+++ b/solr/core/src/java/org/apache/solr/cloud/Assign.java
@@ -33,9 +33,9 @@ import java.util.stream.Collectors;
import com.google.common.collect.ImmutableMap;
import org.apache.solr.client.solrj.cloud.autoscaling.AlreadyExistsException;
import org.apache.solr.client.solrj.cloud.autoscaling.BadVersionException;
-import org.apache.solr.client.solrj.cloud.autoscaling.ClusterDataProvider;
import org.apache.solr.client.solrj.cloud.autoscaling.DistribStateManager;
import org.apache.solr.client.solrj.cloud.autoscaling.PolicyHelper;
+import org.apache.solr.client.solrj.cloud.autoscaling.SolrCloudDataProvider;
import org.apache.solr.client.solrj.cloud.autoscaling.VersionedData;
import org.apache.solr.client.solrj.impl.SolrClientCloudDataProvider;
import org.apache.solr.client.solrj.cloud.autoscaling.AutoScalingConfig;
@@ -90,10 +90,15 @@ public class Assign {
while (true) {
try {
+ int version = -1;
+ int currentId = 0;
VersionedData data = stateManager.getData(path, null);
- int currentId = NumberUtils.bytesToInt(data.getData());
+ if (data != null) {
+ currentId = NumberUtils.bytesToInt(data.getData());
+ version = data.getVersion();
+ }
byte[] bytes = NumberUtils.intToBytes(++currentId);
- stateManager.setData(path, bytes, data.getVersion());
+ stateManager.setData(path, bytes, version);
return currentId;
} catch (BadVersionException e) {
continue;
@@ -240,7 +245,7 @@ public class Assign {
int numPullReplicas) throws IOException, InterruptedException {
List<Map> rulesMap = (List) message.get("rule");
String policyName = message.getStr(POLICY);
- AutoScalingConfig autoScalingConfig = ocmh.overseer.getSolrCloudDataProvider().getClusterDataProvider().getAutoScalingConfig();
+ AutoScalingConfig autoScalingConfig = ocmh.overseer.getSolrCloudDataProvider().getDistribStateManager().getAutoScalingConfig();
if (rulesMap == null && policyName == null && autoScalingConfig.isEmpty()) {
log.debug("Identify nodes using default");
@@ -289,7 +294,7 @@ public class Assign {
PolicyHelper.SESSION_REF.set(ocmh.policySessionRef);
try {
return getPositionsUsingPolicy(collectionName,
- shardNames, numNrtReplicas, numTlogReplicas, numPullReplicas, policyName, ocmh.overseer.getSolrCloudDataProvider().getClusterDataProvider(), nodeList);
+ shardNames, numNrtReplicas, numTlogReplicas, numPullReplicas, policyName, ocmh.overseer.getSolrCloudDataProvider(), nodeList);
} finally {
PolicyHelper.SESSION_REF.remove();
}
@@ -318,7 +323,7 @@ public class Assign {
// 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,
- Object createNodeSet, ClusterDataProvider cdp, CoreContainer cc) throws IOException, InterruptedException {
+ Object createNodeSet, SolrCloudDataProvider dataProvider, CoreContainer cc) throws IOException, InterruptedException {
log.debug("getNodesForNewReplicas() shard: {} , replicas : {} , createNodeSet {}", shard, nrtReplicas, createNodeSet );
DocCollection coll = clusterState.getCollection(collectionName);
Integer maxShardsPerNode = coll.getMaxShardsPerNode();
@@ -354,10 +359,10 @@ public class Assign {
replicaPositions = getNodesViaRules(clusterState, shard, nrtReplicas, cc, coll, createNodeList, l);
}
String policyName = coll.getStr(POLICY);
- AutoScalingConfig autoScalingConfig = cdp.getAutoScalingConfig();
+ AutoScalingConfig autoScalingConfig = dataProvider.getDistribStateManager().getAutoScalingConfig();
if (policyName != null || !autoScalingConfig.getPolicy().getClusterPolicy().isEmpty()) {
replicaPositions = Assign.getPositionsUsingPolicy(collectionName, Collections.singletonList(shard), nrtReplicas, 0, 0,
- policyName, cdp, createNodeList);
+ policyName, dataProvider, createNodeList);
}
if(replicaPositions != null){
@@ -378,18 +383,18 @@ public class Assign {
int nrtReplicas,
int tlogReplicas,
int pullReplicas,
- String policyName, ClusterDataProvider cdp,
+ String policyName, SolrCloudDataProvider dataProvider,
List<String> nodesList) throws IOException, InterruptedException {
log.debug("shardnames {} NRT {} TLOG {} PULL {} , policy {}, nodeList {}", shardNames, nrtReplicas, tlogReplicas, pullReplicas, policyName, nodesList);
SolrClientCloudDataProvider clientDataProvider = null;
List<ReplicaPosition> replicaPositions = null;
- AutoScalingConfig autoScalingConfig = cdp.getAutoScalingConfig();
+ AutoScalingConfig autoScalingConfig = dataProvider.getDistribStateManager().getAutoScalingConfig();
try {
Map<String, String> kvMap = Collections.singletonMap(collName, policyName);
replicaPositions = PolicyHelper.getReplicaLocations(
collName,
autoScalingConfig,
- cdp,
+ dataProvider,
kvMap,
shardNames,
nrtReplicas,
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/dddc6723/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 b6cb382..98f8217 100644
--- a/solr/core/src/java/org/apache/solr/cloud/CreateShardCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/CreateShardCmd.java
@@ -103,7 +103,7 @@ public class CreateShardCmd implements Cmd {
numPullReplicas);
} else {
List<Assign.ReplicaCount> sortedNodeList = getNodesForNewReplicas(clusterState, collectionName, sliceName, totalReplicas,
- createNodeSetStr, ocmh.overseer.getSolrCloudDataProvider().getClusterDataProvider(), ocmh.overseer.getCoreContainer());
+ createNodeSetStr, ocmh.overseer.getSolrCloudDataProvider(), ocmh.overseer.getCoreContainer());
int i = 0;
positions = new ArrayList<>();
for (Map.Entry<Replica.Type, Integer> e : ImmutableMap.of(Replica.Type.NRT, numNrtReplicas,
@@ -174,7 +174,7 @@ public class CreateShardCmd implements Cmd {
static boolean usePolicyFramework(DocCollection collection, OverseerCollectionMessageHandler ocmh)
throws IOException, InterruptedException {
- AutoScalingConfig autoScalingConfig = ocmh.overseer.getSolrCloudDataProvider().getClusterDataProvider().getAutoScalingConfig();
+ AutoScalingConfig autoScalingConfig = ocmh.overseer.getSolrCloudDataProvider().getDistribStateManager().getAutoScalingConfig();
return !autoScalingConfig.isEmpty() || collection.getPolicyName() != null;
}
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/dddc6723/solr/core/src/java/org/apache/solr/cloud/autoscaling/ComputePlanAction.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/ComputePlanAction.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/ComputePlanAction.java
index b8b73e7..42cdf18 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/ComputePlanAction.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/ComputePlanAction.java
@@ -54,7 +54,7 @@ public class ComputePlanAction extends TriggerActionBase {
log.debug("-- processing event: {} with context properties: {}", event, context.getProperties());
SolrCloudDataProvider dataProvider = context.getDataProvider();
try {
- AutoScalingConfig autoScalingConf = dataProvider.getClusterDataProvider().getAutoScalingConfig();
+ AutoScalingConfig autoScalingConf = dataProvider.getDistribStateManager().getAutoScalingConfig();
if (autoScalingConf.isEmpty()) {
log.error("Action: " + getName() + " executed but no policy is configured");
return;
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/dddc6723/solr/core/src/java/org/apache/solr/cloud/autoscaling/OverseerTriggerThread.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/OverseerTriggerThread.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/OverseerTriggerThread.java
index 8682443..dbe2ea4 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/OverseerTriggerThread.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/OverseerTriggerThread.java
@@ -260,7 +260,7 @@ public class OverseerTriggerThread implements Runnable, Closeable {
if (isClosed) {
return;
}
- AutoScalingConfig currentConfig = dataProvider.getClusterDataProvider().getAutoScalingConfig(watcher);
+ AutoScalingConfig currentConfig = dataProvider.getDistribStateManager().getAutoScalingConfig(watcher);
log.debug("Refreshing {} with znode version {}", ZkStateReader.SOLR_AUTOSCALING_CONF_PATH, currentConfig.getZkVersion());
if (znodeVersion >= currentConfig.getZkVersion()) {
// protect against reordered watcher fires by ensuring that we only move forward
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/dddc6723/solr/core/src/test/org/apache/solr/cloud/OverseerCollectionConfigSetProcessorTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/OverseerCollectionConfigSetProcessorTest.java b/solr/core/src/test/org/apache/solr/cloud/OverseerCollectionConfigSetProcessorTest.java
index 14817b0..ca8e757 100644
--- a/solr/core/src/test/org/apache/solr/cloud/OverseerCollectionConfigSetProcessorTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/OverseerCollectionConfigSetProcessorTest.java
@@ -25,8 +25,11 @@ import java.util.concurrent.TimeUnit;
import org.apache.solr.SolrTestCaseJ4;
import org.apache.solr.client.solrj.SolrResponse;
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.ClusterDataProvider;
+import org.apache.solr.client.solrj.cloud.autoscaling.DistribStateManager;
import org.apache.solr.client.solrj.cloud.autoscaling.SolrCloudDataProvider;
+import org.apache.solr.client.solrj.cloud.autoscaling.VersionedData;
import org.apache.solr.cloud.Overseer.LeaderStatus;
import org.apache.solr.cloud.OverseerTaskQueue.QueueEvent;
import org.apache.solr.common.cloud.ClusterState;
@@ -47,12 +50,15 @@ import org.apache.solr.handler.component.ShardHandler;
import org.apache.solr.handler.component.ShardHandlerFactory;
import org.apache.solr.handler.component.ShardRequest;
import org.apache.solr.util.TimeOut;
+import org.apache.zookeeper.CreateMode;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -79,6 +85,7 @@ public class OverseerCollectionConfigSetProcessorTest extends SolrTestCaseJ4 {
private static ZkStateReader zkStateReaderMock;
private static ClusterState clusterStateMock;
private static SolrZkClient solrZkClientMock;
+ private static DistribStateManager stateManagerMock;
private static AutoScalingConfig autoScalingConfig = new AutoScalingConfig(Collections.emptyMap());
private final Map zkMap = new HashMap();
private final Map<String, ClusterState.CollectionRef> collectionsSet = new HashMap<>();
@@ -127,6 +134,7 @@ public class OverseerCollectionConfigSetProcessorTest extends SolrTestCaseJ4 {
zkControllerMock = mock(ZkController.class);
cloudDataProviderMock = mock(SolrCloudDataProvider.class);
clusterDataProviderMock = mock(ClusterDataProvider.class);
+ stateManagerMock = mock(DistribStateManager.class);
}
@AfterClass
@@ -163,6 +171,7 @@ public class OverseerCollectionConfigSetProcessorTest extends SolrTestCaseJ4 {
reset(zkControllerMock);
reset(cloudDataProviderMock);
reset(clusterDataProviderMock);
+ reset(stateManagerMock);
zkMap.clear();
collectionsSet.clear();
@@ -284,7 +293,39 @@ public class OverseerCollectionConfigSetProcessorTest extends SolrTestCaseJ4 {
when(zkControllerMock.getSolrCloudDataProvider()).thenReturn(cloudDataProviderMock);
when(cloudDataProviderMock.getClusterDataProvider()).thenReturn(clusterDataProviderMock);
when(clusterDataProviderMock.getClusterState()).thenReturn(clusterStateMock);
- when(clusterDataProviderMock.getAutoScalingConfig()).thenReturn(autoScalingConfig);
+ when(cloudDataProviderMock.getDistribStateManager()).thenReturn(stateManagerMock);
+ when(stateManagerMock.hasData(anyString())).thenAnswer(invocation -> zkMap.containsKey(invocation.getArgument(0)));
+ when(stateManagerMock.getAutoScalingConfig()).thenReturn(autoScalingConfig);
+ doAnswer(new Answer<Void>() {
+ @Override
+ public Void answer(InvocationOnMock invocation) throws Throwable {
+ if (!zkMap.containsKey(invocation.getArgument(0))) {
+ zkMap.put(invocation.getArgument(0), "");
+ }
+ return null;
+ }
+ }).when(stateManagerMock).makePath(anyString());
+ doAnswer(new Answer<Void>() {
+ @Override
+ public Void answer(InvocationOnMock invocation) throws Throwable {
+ VersionedData d = new VersionedData(0, invocation.getArgument(1), "test");
+ zkMap.put(invocation.getArgument(0), d);
+ return null;
+ }
+ }).when(stateManagerMock).createData(anyString(), any(byte[].class), any(CreateMode.class));
+ doAnswer(new Answer<Void>() {
+ @Override
+ public Void answer(InvocationOnMock invocation) throws Throwable {
+ VersionedData d = (VersionedData)zkMap.get(invocation.getArgument(0));
+ if (d != null && d.getVersion() != (Integer)invocation.getArgument(2)) {
+ throw new BadVersionException(invocation.getArgument(0));
+ }
+ int version = (Integer)invocation.getArgument(2) + 1;
+ zkMap.put(invocation.getArgument(0), new VersionedData(version, invocation.getArgument(1), "test"));
+ return null;
+ }
+ }).when(stateManagerMock).setData(anyString(), any(byte[].class), anyInt());
+ when(stateManagerMock.getData(anyString(), any())).thenAnswer(invocation -> zkMap.get(invocation.getArgument(0)));
zkMap.put("/configs/myconfig", null);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/dddc6723/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/ClusterDataProvider.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/ClusterDataProvider.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/ClusterDataProvider.java
index 35346b2..1ffb00b 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/ClusterDataProvider.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/ClusterDataProvider.java
@@ -63,12 +63,6 @@ public interface ClusterDataProvider extends Closeable {
return value;
}
- AutoScalingConfig getAutoScalingConfig(Watcher watcher) throws InterruptedException, IOException;
-
- default AutoScalingConfig getAutoScalingConfig() throws InterruptedException, IOException {
- return getAutoScalingConfig(null);
- }
-
/**
* Get the collection-specific policy
*/
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/dddc6723/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/DelegatingClusterDataProvider.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/DelegatingClusterDataProvider.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/DelegatingClusterDataProvider.java
index 149fad2..57348d3 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/DelegatingClusterDataProvider.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/DelegatingClusterDataProvider.java
@@ -1,13 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
package org.apache.solr.client.solrj.cloud.autoscaling;
import java.io.IOException;
-import java.net.ConnectException;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import org.apache.solr.common.cloud.ClusterState;
-import org.apache.zookeeper.Watcher;
/**
*
@@ -45,11 +59,6 @@ public class DelegatingClusterDataProvider implements ClusterDataProvider {
}
@Override
- public AutoScalingConfig getAutoScalingConfig(Watcher watcher) throws ConnectException, InterruptedException, IOException {
- return delegate.getAutoScalingConfig(watcher);
- }
-
- @Override
public String getPolicyNameByCollection(String coll) {
return delegate.getPolicyNameByCollection(coll);
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/dddc6723/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/DistribStateManager.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/DistribStateManager.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/DistribStateManager.java
index eb28070..b9d9655 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/DistribStateManager.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/DistribStateManager.java
@@ -54,4 +54,10 @@ public interface DistribStateManager {
List<OpResult> multi(final Iterable<Op> ops) throws BadVersionException, NoSuchElementException, AlreadyExistsException, IOException, InterruptedException;
+ AutoScalingConfig getAutoScalingConfig(Watcher watcher) throws InterruptedException, IOException;
+
+ default AutoScalingConfig getAutoScalingConfig() throws InterruptedException, IOException {
+ return getAutoScalingConfig(null);
+ }
+
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/dddc6723/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 5e56891..5dd8a33 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
@@ -43,7 +43,7 @@ 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,
+ SolrCloudDataProvider dataProvider,
Map<String, String> optionalPolicyMapping,
List<String> shardNames,
int nrtReplicas,
@@ -51,8 +51,8 @@ public class PolicyHelper {
int pullReplicas,
List<String> nodesList) {
List<ReplicaPosition> positions = new ArrayList<>();
- final ClusterDataProvider delegate = cdp;
- cdp = new DelegatingClusterDataProvider(delegate) {
+ final ClusterDataProvider delegate = dataProvider.getClusterDataProvider();
+ ClusterDataProvider cdp = new DelegatingClusterDataProvider(delegate) {
@Override
public ClusterState getClusterState() throws IOException {
return delegate.getClusterState();
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/dddc6723/solr/solrj/src/java/org/apache/solr/client/solrj/impl/SolrClientClusterDataProvider.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/SolrClientClusterDataProvider.java b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/SolrClientClusterDataProvider.java
index e08eda5..fd0722a 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/SolrClientClusterDataProvider.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/SolrClientClusterDataProvider.java
@@ -31,7 +31,6 @@ import java.util.Set;
import org.apache.solr.client.solrj.SolrRequest;
import org.apache.solr.client.solrj.SolrServerException;
-import org.apache.solr.client.solrj.cloud.autoscaling.AutoScalingConfig;
import org.apache.solr.client.solrj.cloud.autoscaling.ClusterDataProvider;
import org.apache.solr.client.solrj.cloud.autoscaling.ReplicaInfo;
import org.apache.solr.client.solrj.request.GenericSolrRequest;
@@ -51,7 +50,6 @@ import org.apache.solr.common.util.NamedList;
import org.apache.solr.common.util.StrUtils;
import org.apache.solr.common.util.Utils;
import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.Watcher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -126,15 +124,6 @@ public class SolrClientClusterDataProvider implements ClusterDataProvider, MapWr
ew.put("nodeValues", nodeVsTags);
}
- @Override
- public AutoScalingConfig getAutoScalingConfig(Watcher watcher) throws IOException {
- try {
- return zkStateReader.getAutoScalingConfig(watcher);
- } catch (KeeperException | InterruptedException e) {
- throw new IOException(e);
- }
- }
-
static class ClientSnitchCtx
extends SnitchContext {
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/dddc6723/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ZkDistribStateManager.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ZkDistribStateManager.java b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ZkDistribStateManager.java
index 9fa6e22..f0efb9c 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ZkDistribStateManager.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ZkDistribStateManager.java
@@ -18,14 +18,19 @@
package org.apache.solr.client.solrj.impl;
import java.io.IOException;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.NoSuchElementException;
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.DistribStateManager;
import org.apache.solr.client.solrj.cloud.autoscaling.VersionedData;
import org.apache.solr.common.cloud.SolrZkClient;
+import org.apache.solr.common.cloud.ZkStateReader;
+import org.apache.solr.common.params.AutoScalingParams;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.Op;
@@ -33,6 +38,8 @@ import org.apache.zookeeper.OpResult;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.data.Stat;
+import static org.apache.solr.common.util.Utils.fromJSON;
+
/**
* Implementation of {@link DistribStateManager} that uses Zookeeper.
*/
@@ -152,4 +159,22 @@ public class ZkDistribStateManager implements DistribStateManager {
}
}
+ @Override
+ public AutoScalingConfig getAutoScalingConfig(Watcher watcher) throws InterruptedException, IOException {
+ Map<String, Object> map = new HashMap<>();
+ Stat stat = new Stat();
+ try {
+ byte[] bytes = zkClient.getData(ZkStateReader.SOLR_AUTOSCALING_CONF_PATH, watcher, stat, true);
+ if (bytes != null && bytes.length > 0) {
+ map = (Map<String, Object>) fromJSON(bytes);
+ }
+ } catch (KeeperException.NoNodeException e) {
+ // ignore
+ } catch (KeeperException e) {
+ throw new IOException(e);
+ }
+ map.put(AutoScalingParams.ZK_VERSION, stat.getVersion());
+ return new AutoScalingConfig(map);
+ }
+
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/dddc6723/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 1e67d59..5516902 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
@@ -31,6 +31,7 @@ import java.util.Map;
import com.google.common.collect.ImmutableList;
import org.apache.solr.SolrTestCaseJ4;
import org.apache.solr.client.solrj.SolrRequest;
+import org.apache.solr.client.solrj.SolrResponse;
import org.apache.solr.client.solrj.cloud.autoscaling.Clause.Violation;
import org.apache.solr.client.solrj.cloud.autoscaling.Policy.Suggester.Hint;
import org.apache.solr.client.solrj.request.CollectionAdminRequest;
@@ -870,7 +871,34 @@ public class TestPolicy extends SolrTestCaseJ4 {
"}");
AutoScalingConfig config = new AutoScalingConfig(policies);
- List<ReplicaPosition> locations = PolicyHelper.getReplicaLocations("c", config, provider, null,
+ SolrCloudDataProvider dataProvider = new SolrCloudDataProvider() {
+ @Override
+ public ClusterDataProvider getClusterDataProvider() {
+ return provider;
+ }
+
+ @Override
+ public DistribStateManager getDistribStateManager() {
+ return null;
+ }
+
+ @Override
+ public DistributedQueueFactory getDistributedQueueFactory() {
+ return null;
+ }
+
+ @Override
+ public SolrResponse request(SolrRequest req) throws IOException {
+ return null;
+ }
+
+ @Override
+ public byte[] httpRequest(String url, SolrRequest.METHOD method, Map<String, String> headers, String payload, int timeout, boolean followRedirects) throws IOException {
+ return new byte[0];
+ }
+ };
+
+ List<ReplicaPosition> locations = PolicyHelper.getReplicaLocations("c", config, dataProvider, null,
Arrays.asList("s1", "s2"), 1, 0, 0,
null);
@@ -884,7 +912,7 @@ public class TestPolicy extends SolrTestCaseJ4 {
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,
+ locations = PolicyHelper.getReplicaLocations("c2", config2, dataProvider, null, Arrays.asList("s1", "s2"), 1, 0, 0,
null);
sessionRefVersion = PolicyHelper.REF_VERSION.get();
ref1Copy = PolicyHelper.SESSION_REF.get();
@@ -898,7 +926,7 @@ public class TestPolicy extends SolrTestCaseJ4 {
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,
+ locations = PolicyHelper.getReplicaLocations("c3", config2, dataProvider, null, Arrays.asList("s1", "s2"), 1, 0, 0,
null);
sessionRefVersion = PolicyHelper.REF_VERSION.get();
ref1Copy = PolicyHelper.SESSION_REF.get();
@@ -1179,7 +1207,7 @@ public class TestPolicy extends SolrTestCaseJ4 {
" '127.0.0.1:50096_solr':{" +
" 'cores':0," +
" 'port':'50096'}}");
- ClusterDataProvider dataProvider = new DelegatingClusterDataProvider(null) {
+ ClusterDataProvider provider = new DelegatingClusterDataProvider(null) {
@Override
public Map<String, Object> getNodeValues(String node, Collection<String> keys) {
Map<String, Object> result = new LinkedHashMap<>();
@@ -1208,6 +1236,32 @@ public class TestPolicy extends SolrTestCaseJ4 {
}
};
+ SolrCloudDataProvider dataProvider = new SolrCloudDataProvider() {
+ @Override
+ public ClusterDataProvider getClusterDataProvider() {
+ return provider;
+ }
+
+ @Override
+ public DistribStateManager getDistribStateManager() {
+ return null;
+ }
+
+ @Override
+ public DistributedQueueFactory getDistributedQueueFactory() {
+ return null;
+ }
+
+ @Override
+ public SolrResponse request(SolrRequest req) throws IOException {
+ return null;
+ }
+
+ @Override
+ public byte[] httpRequest(String url, SolrRequest.METHOD method, Map<String, String> headers, String payload, int timeout, boolean followRedirects) throws IOException {
+ return new byte[0];
+ }
+ };
List<ReplicaPosition> locations = PolicyHelper.getReplicaLocations(
"newColl", new AutoScalingConfig((Map<String, Object>)Utils.fromJSONString(autoScaleJson)),
dataProvider, Collections.singletonMap("newColl", "c1"), Arrays.asList("shard1", "shard2"), 1, 0, 0, null);
@@ -1242,7 +1296,7 @@ public class TestPolicy extends SolrTestCaseJ4 {
"node4:{cores:0, freedisk: 900, heap:16900, nodeRole:overseer, sysprop.rack:rack2}" +
"}");
- ClusterDataProvider dataProvider = new DelegatingClusterDataProvider(null) {
+ ClusterDataProvider provider = new DelegatingClusterDataProvider(null) {
@Override
public Map<String, Object> getNodeValues(String node, Collection<String> keys) {
Map<String, Object> result = new LinkedHashMap<>();
@@ -1271,6 +1325,32 @@ public class TestPolicy extends SolrTestCaseJ4 {
}
};
+ SolrCloudDataProvider dataProvider = new SolrCloudDataProvider() {
+ @Override
+ public ClusterDataProvider getClusterDataProvider() {
+ return provider;
+ }
+
+ @Override
+ public DistribStateManager getDistribStateManager() {
+ return null;
+ }
+
+ @Override
+ public DistributedQueueFactory getDistributedQueueFactory() {
+ return null;
+ }
+
+ @Override
+ public SolrResponse request(SolrRequest req) throws IOException {
+ return null;
+ }
+
+ @Override
+ public byte[] httpRequest(String url, SolrRequest.METHOD method, Map<String, String> headers, String payload, int timeout, boolean followRedirects) throws IOException {
+ return new byte[0];
+ }
+ };
List<ReplicaPosition> locations = PolicyHelper.getReplicaLocations(
"newColl", new AutoScalingConfig((Map<String, Object>) Utils.fromJSONString(autoScaleJson)),
dataProvider, Collections.singletonMap("newColl", "policy1"), Arrays.asList("shard1", "shard2"), 3,0,0, null);