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/08/29 12:26:34 UTC
[02/10] lucene-solr:jira/solr-11285: SOLR-11285: Initial refactoring.
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/aee54ff7/solr/core/src/java/org/apache/solr/cloud/autoscaling/ExecutePlanAction.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/ExecutePlanAction.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/ExecutePlanAction.java
index 2cd9824..01c25c0 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/ExecutePlanAction.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/ExecutePlanAction.java
@@ -25,6 +25,7 @@ import java.util.List;
import org.apache.solr.client.solrj.SolrRequest;
import org.apache.solr.client.solrj.SolrResponse;
import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.cloud.autoscaling.ClusterDataProvider;
import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.apache.solr.client.solrj.impl.HttpSolrClient;
import org.apache.solr.common.SolrException;
@@ -43,32 +44,29 @@ public class ExecutePlanAction extends TriggerActionBase {
@Override
public void process(TriggerEvent event, ActionContext context) {
log.debug("-- processing event: {} with context properties: {}", event, context.getProperties());
- CoreContainer container = context.getCoreContainer();
+ ClusterDataProvider clusterDataProvider = context.getClusterDataProvider();
List<SolrRequest> operations = (List<SolrRequest>) context.getProperty("operations");
if (operations == null || operations.isEmpty()) {
log.info("No operations to execute for event: {}", event);
return;
}
- try (CloudSolrClient cloudSolrClient = new CloudSolrClient.Builder()
- .withZkHost(container.getZkController().getZkServerAddress())
- .withHttpClient(container.getUpdateShardHandler().getHttpClient())
- .build()) {
+ try {
for (SolrRequest operation : operations) {
log.info("Executing operation: {}", operation.getParams());
try {
- SolrResponse response = operation.process(cloudSolrClient);
+ SolrResponse response = clusterDataProvider.request(operation);
context.getProperties().compute("responses", (s, o) -> {
List<NamedList<Object>> responses = (List<NamedList<Object>>) o;
if (responses == null) responses = new ArrayList<>(operations.size());
responses.add(response.getResponse());
return responses;
});
- } catch (SolrServerException | HttpSolrClient.RemoteSolrException e) {
+ } catch (Exception e) {
throw new SolrException(SolrException.ErrorCode.SERVER_ERROR,
"Unexpected exception executing operation: " + operation.getParams(), e);
}
}
- } catch (IOException e) {
+ } catch (Exception e) {
throw new SolrException(SolrException.ErrorCode.SERVER_ERROR,
"Unexpected IOException while processing event: " + event, e);
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/aee54ff7/solr/core/src/java/org/apache/solr/cloud/autoscaling/HttpTriggerListener.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/HttpTriggerListener.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/HttpTriggerListener.java
index 2af4f30..dadc3af 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/HttpTriggerListener.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/HttpTriggerListener.java
@@ -30,6 +30,7 @@ import org.apache.http.client.methods.HttpPost;
import org.apache.http.client.protocol.HttpClientContext;
import org.apache.http.entity.StringEntity;
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.TriggerEventProcessorStage;
import org.apache.solr.client.solrj.impl.HttpClientUtil;
import org.apache.solr.common.util.Utils;
@@ -72,9 +73,9 @@ public class HttpTriggerListener extends TriggerListenerBase {
private boolean followRedirects;
@Override
- public void init(CoreContainer coreContainer, AutoScalingConfig.TriggerListenerConfig config) {
- super.init(coreContainer, config);
- httpClient = coreContainer.getUpdateShardHandler().getHttpClient();
+ public void init(ClusterDataProvider clusterDataProvider, AutoScalingConfig.TriggerListenerConfig config) {
+ super.init(clusterDataProvider, config);
+ httpClient = clusterDataProvider.getHttpClient();
urlTemplate = (String)config.properties.get("url");
payloadTemplate = (String)config.properties.get("payload");
contentType = (String)config.properties.get("contentType");
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/aee54ff7/solr/core/src/java/org/apache/solr/cloud/autoscaling/NodeAddedTrigger.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/NodeAddedTrigger.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/NodeAddedTrigger.java
index 0c0278b..387458c 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/NodeAddedTrigger.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/NodeAddedTrigger.java
@@ -17,29 +17,24 @@
package org.apache.solr.cloud.autoscaling;
-import java.io.IOException;
import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
import java.util.Collection;
-import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
-import java.util.Locale;
import java.util.Map;
-import java.util.Objects;
+import java.util.NoSuchElementException;
import java.util.Set;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
-import org.apache.lucene.util.IOUtils;
+import org.apache.solr.client.solrj.cloud.autoscaling.ClusterDataProvider;
import org.apache.solr.client.solrj.cloud.autoscaling.TriggerEventType;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.cloud.ZkStateReader;
-import org.apache.solr.core.CoreContainer;
+import org.apache.solr.core.SolrResourceLoader;
import org.apache.solr.util.TimeSource;
-import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -49,60 +44,28 @@ import org.slf4j.LoggerFactory;
public class NodeAddedTrigger extends TriggerBase {
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- private final String name;
- private final Map<String, Object> properties;
- private final CoreContainer container;
- private final List<TriggerAction> actions;
- private final AtomicReference<AutoScaling.TriggerEventProcessor> processorRef;
- private final boolean enabled;
- private final int waitForSecond;
- private final TriggerEventType eventType;
private final TimeSource timeSource;
- private boolean isClosed = false;
-
private Set<String> lastLiveNodes;
private Map<String, Long> nodeNameVsTimeAdded = new HashMap<>();
public NodeAddedTrigger(String name, Map<String, Object> properties,
- CoreContainer container) {
- super(container.getZkController().getZkClient());
- this.name = name;
- this.properties = properties;
- this.container = container;
+ SolrResourceLoader loader,
+ ClusterDataProvider clusterDataProvider) {
+ super(name, properties, loader, clusterDataProvider);
this.timeSource = TimeSource.CURRENT_TIME;
- this.processorRef = new AtomicReference<>();
- List<Map<String, String>> o = (List<Map<String, String>>) properties.get("actions");
- if (o != null && !o.isEmpty()) {
- actions = new ArrayList<>(3);
- for (Map<String, String> map : o) {
- TriggerAction action = container.getResourceLoader().newInstance(map.get("class"), TriggerAction.class);
- actions.add(action);
- }
- } else {
- actions = Collections.emptyList();
- }
- lastLiveNodes = new HashSet<>(container.getZkController().getZkStateReader().getClusterState().getLiveNodes());
+ lastLiveNodes = new HashSet<>(clusterDataProvider.getLiveNodes());
log.debug("Initial livenodes: {}", lastLiveNodes);
- this.enabled = Boolean.parseBoolean(String.valueOf(properties.getOrDefault("enabled", "true")));
- this.waitForSecond = ((Long) properties.getOrDefault("waitFor", -1L)).intValue();
- this.eventType = TriggerEventType.valueOf(properties.get("event").toString().toUpperCase(Locale.ROOT));
log.debug("NodeAddedTrigger {} instantiated with properties: {}", name, properties);
}
@Override
public void init() {
- List<Map<String, String>> o = (List<Map<String, String>>) properties.get("actions");
- if (o != null && !o.isEmpty()) {
- for (int i = 0; i < o.size(); i++) {
- Map<String, String> map = o.get(i);
- actions.get(i).init(map);
- }
- }
+ super.init();
// pick up added nodes for which marker paths were created
try {
- List<String> added = container.getZkController().getZkClient().getChildren(ZkStateReader.SOLR_AUTOSCALING_NODE_ADDED_PATH, null, true);
+ List<String> added = clusterDataProvider.listData(ZkStateReader.SOLR_AUTOSCALING_NODE_ADDED_PATH);
added.forEach(n -> {
// don't add nodes that have since gone away
if (lastLiveNodes.contains(n)) {
@@ -111,78 +74,15 @@ public class NodeAddedTrigger extends TriggerBase {
}
removeMarker(n);
});
- } catch (KeeperException.NoNodeException e) {
+ } catch (NoSuchElementException e) {
// ignore
- } catch (KeeperException | InterruptedException e) {
+ } catch (Exception e) {
log.warn("Exception retrieving nodeLost markers", e);
}
}
@Override
- public void setProcessor(AutoScaling.TriggerEventProcessor processor) {
- processorRef.set(processor);
- }
-
- @Override
- public AutoScaling.TriggerEventProcessor getProcessor() {
- return processorRef.get();
- }
-
- @Override
- public String getName() {
- return name;
- }
-
- @Override
- public TriggerEventType getEventType() {
- return eventType;
- }
-
- @Override
- public boolean isEnabled() {
- return enabled;
- }
-
- @Override
- public int getWaitForSecond() {
- return waitForSecond;
- }
-
- @Override
- public Map<String, Object> getProperties() {
- return properties;
- }
-
- @Override
- public List<TriggerAction> getActions() {
- return actions;
- }
-
- @Override
- public boolean equals(Object obj) {
- if (obj instanceof NodeAddedTrigger) {
- NodeAddedTrigger that = (NodeAddedTrigger) obj;
- return this.name.equals(that.name)
- && this.properties.equals(that.properties);
- }
- return false;
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(name, properties);
- }
-
- @Override
- public void close() throws IOException {
- synchronized (this) {
- isClosed = true;
- IOUtils.closeWhileHandlingException(actions);
- }
- }
-
- @Override
public void restoreState(AutoScaling.Trigger old) {
assert old.isClosed();
if (old instanceof NodeAddedTrigger) {
@@ -229,8 +129,7 @@ public class NodeAddedTrigger extends TriggerBase {
}
log.debug("Running NodeAddedTrigger {}", name);
- ZkStateReader reader = container.getZkController().getZkStateReader();
- Set<String> newLiveNodes = reader.getClusterState().getLiveNodes();
+ Set<String> newLiveNodes = new HashSet<>(clusterDataProvider.getLiveNodes());
log.debug("Found livenodes: {}", newLiveNodes);
// have any nodes that we were tracking been removed from the cluster?
@@ -287,24 +186,17 @@ public class NodeAddedTrigger extends TriggerBase {
private void removeMarker(String nodeName) {
String path = ZkStateReader.SOLR_AUTOSCALING_NODE_ADDED_PATH + "/" + nodeName;
try {
- if (container.getZkController().getZkClient().exists(path, true)) {
- container.getZkController().getZkClient().delete(path, -1, true);
+ if (clusterDataProvider.hasData(path)) {
+ clusterDataProvider.removeData(path, -1);
}
- } catch (KeeperException.NoNodeException e) {
+ } catch (NoSuchElementException e) {
// ignore
- } catch (KeeperException | InterruptedException e) {
+ } catch (Exception e) {
log.debug("Exception removing nodeAdded marker " + nodeName, e);
}
}
- @Override
- public boolean isClosed() {
- synchronized (this) {
- return isClosed;
- }
- }
-
public static class NodeAddedEvent extends TriggerEvent {
public NodeAddedEvent(TriggerEventType eventType, String source, List<Long> times, List<String> nodeNames) {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/aee54ff7/solr/core/src/java/org/apache/solr/cloud/autoscaling/NodeLostTrigger.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/NodeLostTrigger.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/NodeLostTrigger.java
index 18dafcb..d1e15f7 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/NodeLostTrigger.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/NodeLostTrigger.java
@@ -17,29 +17,24 @@
package org.apache.solr.cloud.autoscaling;
-import java.io.IOException;
import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
import java.util.Collection;
-import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
-import java.util.Locale;
import java.util.Map;
-import java.util.Objects;
+import java.util.NoSuchElementException;
import java.util.Set;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
-import org.apache.lucene.util.IOUtils;
+import org.apache.solr.client.solrj.cloud.autoscaling.ClusterDataProvider;
import org.apache.solr.client.solrj.cloud.autoscaling.TriggerEventType;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.cloud.ZkStateReader;
-import org.apache.solr.core.CoreContainer;
+import org.apache.solr.core.SolrResourceLoader;
import org.apache.solr.util.TimeSource;
-import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -49,59 +44,27 @@ import org.slf4j.LoggerFactory;
public class NodeLostTrigger extends TriggerBase {
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- private final String name;
- private final Map<String, Object> properties;
- private final CoreContainer container;
- private final List<TriggerAction> actions;
- private final AtomicReference<AutoScaling.TriggerEventProcessor> processorRef;
- private final boolean enabled;
- private final int waitForSecond;
- private final TriggerEventType eventType;
private final TimeSource timeSource;
- private boolean isClosed = false;
-
private Set<String> lastLiveNodes;
private Map<String, Long> nodeNameVsTimeRemoved = new HashMap<>();
public NodeLostTrigger(String name, Map<String, Object> properties,
- CoreContainer container) {
- super(container.getZkController().getZkClient());
- this.name = name;
- this.properties = properties;
- this.container = container;
+ SolrResourceLoader loader,
+ ClusterDataProvider clusterDataProvider) {
+ super(name, properties, loader, clusterDataProvider);
this.timeSource = TimeSource.CURRENT_TIME;
- this.processorRef = new AtomicReference<>();
- List<Map<String, String>> o = (List<Map<String, String>>) properties.get("actions");
- if (o != null && !o.isEmpty()) {
- actions = new ArrayList<>(3);
- for (Map<String, String> map : o) {
- TriggerAction action = container.getResourceLoader().newInstance(map.get("class"), TriggerAction.class);
- actions.add(action);
- }
- } else {
- actions = Collections.emptyList();
- }
- lastLiveNodes = new HashSet<>(container.getZkController().getZkStateReader().getClusterState().getLiveNodes());
+ lastLiveNodes = new HashSet<>(clusterDataProvider.getLiveNodes());
log.debug("Initial livenodes: {}", lastLiveNodes);
- this.enabled = Boolean.parseBoolean(String.valueOf(properties.getOrDefault("enabled", "true")));
- this.waitForSecond = ((Long) properties.getOrDefault("waitFor", -1L)).intValue();
- this.eventType = TriggerEventType.valueOf(properties.get("event").toString().toUpperCase(Locale.ROOT));
}
@Override
public void init() {
- List<Map<String, String>> o = (List<Map<String, String>>) properties.get("actions");
- if (o != null && !o.isEmpty()) {
- for (int i = 0; i < o.size(); i++) {
- Map<String, String> map = o.get(i);
- actions.get(i).init(map);
- }
- }
+ super.init();
// pick up lost nodes for which marker paths were created
try {
- List<String> lost = container.getZkController().getZkClient().getChildren(ZkStateReader.SOLR_AUTOSCALING_NODE_LOST_PATH, null, true);
+ List<String> lost = clusterDataProvider.listData(ZkStateReader.SOLR_AUTOSCALING_NODE_LOST_PATH);
lost.forEach(n -> {
// don't add nodes that have since came back
if (!lastLiveNodes.contains(n)) {
@@ -110,77 +73,19 @@ public class NodeLostTrigger extends TriggerBase {
}
removeMarker(n);
});
- } catch (KeeperException.NoNodeException e) {
+ } catch (NoSuchElementException e) {
// ignore
- } catch (KeeperException | InterruptedException e) {
+ } catch (Exception e) {
log.warn("Exception retrieving nodeLost markers", e);
}
}
@Override
- public void setProcessor(AutoScaling.TriggerEventProcessor processor) {
- processorRef.set(processor);
- }
-
- @Override
- public AutoScaling.TriggerEventProcessor getProcessor() {
- return processorRef.get();
- }
-
- @Override
- public String getName() {
- return name;
- }
-
- @Override
- public TriggerEventType getEventType() {
- return eventType;
- }
-
- @Override
- public boolean isEnabled() {
- return enabled;
- }
-
- @Override
- public int getWaitForSecond() {
- return waitForSecond;
- }
-
- @Override
- public Map<String, Object> getProperties() {
- return properties;
- }
-
- @Override
public List<TriggerAction> getActions() {
return actions;
}
@Override
- public boolean equals(Object obj) {
- if (obj instanceof NodeLostTrigger) {
- NodeLostTrigger that = (NodeLostTrigger) obj;
- return this.name.equals(that.name)
- && this.properties.equals(that.properties);
- }
- return false;
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(name, properties);
- }
-
- @Override
- public void close() throws IOException {
- synchronized (this) {
- isClosed = true;
- IOUtils.closeWhileHandlingException(actions);
- }
- }
-
- @Override
public void restoreState(AutoScaling.Trigger old) {
assert old.isClosed();
if (old instanceof NodeLostTrigger) {
@@ -226,8 +131,7 @@ public class NodeLostTrigger extends TriggerBase {
}
}
- ZkStateReader reader = container.getZkController().getZkStateReader();
- Set<String> newLiveNodes = reader.getClusterState().getLiveNodes();
+ Set<String> newLiveNodes = new HashSet<>(clusterDataProvider.getLiveNodes());
log.debug("Running NodeLostTrigger: {} with currently live nodes: {}", name, newLiveNodes);
// have any nodes that we were tracking been added to the cluster?
@@ -286,23 +190,16 @@ public class NodeLostTrigger extends TriggerBase {
private void removeMarker(String nodeName) {
String path = ZkStateReader.SOLR_AUTOSCALING_NODE_LOST_PATH + "/" + nodeName;
try {
- if (container.getZkController().getZkClient().exists(path, true)) {
- container.getZkController().getZkClient().delete(path, -1, true);
+ if (clusterDataProvider.hasData(path)) {
+ clusterDataProvider.removeData(path, -1);
}
- } catch (KeeperException.NoNodeException e) {
+ } catch (NoSuchElementException e) {
// ignore
- } catch (KeeperException | InterruptedException e) {
+ } catch (Exception e) {
log.warn("Exception removing nodeLost marker " + nodeName, e);
}
}
- @Override
- public boolean isClosed() {
- synchronized (this) {
- return isClosed;
- }
- }
-
public static class NodeLostEvent extends TriggerEvent {
public NodeLostEvent(TriggerEventType eventType, String source, List<Long> times, List<String> nodeNames) {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/aee54ff7/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 6b7aa91..b2bc3c3 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
@@ -20,16 +20,19 @@ package org.apache.solr.cloud.autoscaling;
import java.io.Closeable;
import java.io.IOException;
import java.lang.invoke.MethodHandles;
+import java.net.ConnectException;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.NoSuchElementException;
import java.util.Set;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.lucene.store.AlreadyClosedException;
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.TriggerEventType;
import org.apache.solr.cloud.ZkController;
import org.apache.solr.common.SolrException;
@@ -37,6 +40,7 @@ import org.apache.solr.common.cloud.SolrZkClient;
import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.cloud.ZooKeeperException;
import org.apache.solr.common.util.IOUtils;
+import org.apache.solr.core.SolrResourceLoader;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
@@ -52,11 +56,7 @@ public class OverseerTriggerThread implements Runnable, Closeable {
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- private final ZkController zkController;
-
- private final ZkStateReader zkStateReader;
-
- private final SolrZkClient zkClient;
+ private final ClusterDataProvider clusterDataProvider;
private final ScheduledTriggers scheduledTriggers;
@@ -77,12 +77,10 @@ public class OverseerTriggerThread implements Runnable, Closeable {
private AutoScalingConfig autoScalingConfig;
- public OverseerTriggerThread(ZkController zkController) {
- this.zkController = zkController;
- zkStateReader = zkController.getZkStateReader();
- zkClient = zkController.getZkClient();
- scheduledTriggers = new ScheduledTriggers(zkController);
- triggerFactory = new AutoScaling.TriggerFactory(zkController.getCoreContainer());
+ public OverseerTriggerThread(SolrResourceLoader loader, ClusterDataProvider clusterDataProvider) {
+ this.clusterDataProvider = clusterDataProvider;
+ scheduledTriggers = new ScheduledTriggers(loader, clusterDataProvider);
+ triggerFactory = new AutoScaling.TriggerFactoryImpl(loader, clusterDataProvider);
}
@Override
@@ -106,11 +104,8 @@ public class OverseerTriggerThread implements Runnable, Closeable {
try {
refreshAutoScalingConf(new AutoScalingWatcher());
- } catch (KeeperException.ConnectionLossException | KeeperException.SessionExpiredException e) {
+ } catch (ConnectException e) {
log.warn("ZooKeeper watch triggered for autoscaling conf, but Solr cannot talk to ZK: [{}]", e.getMessage());
- } catch (KeeperException e) {
- log.error("A ZK error has occurred", e);
- throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR, "A ZK error has occurred", e);
} catch (InterruptedException e) {
// Restore the interrupted status
Thread.currentThread().interrupt();
@@ -201,26 +196,26 @@ public class OverseerTriggerThread implements Runnable, Closeable {
if (cleanOldNodeLostMarkers) {
log.debug("-- clean old nodeLost markers");
try {
- List<String> markers = zkClient.getChildren(ZkStateReader.SOLR_AUTOSCALING_NODE_LOST_PATH, null, true);
+ List<String> markers = clusterDataProvider.listData(ZkStateReader.SOLR_AUTOSCALING_NODE_LOST_PATH);
markers.forEach(n -> {
removeNodeMarker(ZkStateReader.SOLR_AUTOSCALING_NODE_LOST_PATH, n);
});
- } catch (KeeperException.NoNodeException e) {
+ } catch (NoSuchElementException e) {
// ignore
- } catch (KeeperException | InterruptedException e) {
+ } catch (Exception e) {
log.warn("Error removing old nodeLost markers", e);
}
}
if (cleanOldNodeAddedMarkers) {
log.debug("-- clean old nodeAdded markers");
try {
- List<String> markers = zkClient.getChildren(ZkStateReader.SOLR_AUTOSCALING_NODE_ADDED_PATH, null, true);
+ List<String> markers = clusterDataProvider.listData(ZkStateReader.SOLR_AUTOSCALING_NODE_ADDED_PATH);
markers.forEach(n -> {
removeNodeMarker(ZkStateReader.SOLR_AUTOSCALING_NODE_ADDED_PATH, n);
});
- } catch (KeeperException.NoNodeException e) {
+ } catch (NoSuchElementException e) {
// ignore
- } catch (KeeperException | InterruptedException e) {
+ } catch (Exception e) {
log.warn("Error removing old nodeAdded markers", e);
}
@@ -231,11 +226,11 @@ public class OverseerTriggerThread implements Runnable, Closeable {
private void removeNodeMarker(String path, String nodeName) {
path = path + "/" + nodeName;
try {
- zkClient.delete(path, -1, true);
+ clusterDataProvider.removeData(path, -1);
log.debug(" -- deleted " + path);
- } catch (KeeperException.NoNodeException e) {
+ } catch (NoSuchElementException e) {
// ignore
- } catch (KeeperException | InterruptedException e) {
+ } catch (Exception e) {
log.warn("Error removing old marker " + path, e);
}
}
@@ -250,11 +245,8 @@ public class OverseerTriggerThread implements Runnable, Closeable {
try {
refreshAutoScalingConf(this);
- } catch (KeeperException.ConnectionLossException | KeeperException.SessionExpiredException e) {
- log.warn("ZooKeeper watch triggered for autoscaling conf, but Solr cannot talk to ZK: [{}]", e.getMessage());
- } catch (KeeperException e) {
- log.error("A ZK error has occurred", e);
- throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR, "A ZK error has occurred", e);
+ } catch (ConnectException e) {
+ log.warn("ZooKeeper watch triggered for autoscaling conf, but we cannot talk to ZK: [{}]", e.getMessage());
} catch (InterruptedException e) {
// Restore the interrupted status
Thread.currentThread().interrupt();
@@ -266,13 +258,13 @@ public class OverseerTriggerThread implements Runnable, Closeable {
}
- private void refreshAutoScalingConf(Watcher watcher) throws KeeperException, InterruptedException {
+ private void refreshAutoScalingConf(Watcher watcher) throws ConnectException, InterruptedException, IOException {
updateLock.lock();
try {
if (isClosed) {
return;
}
- AutoScalingConfig currentConfig = zkStateReader.getAutoScalingConfig(watcher);
+ AutoScalingConfig currentConfig = clusterDataProvider.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/aee54ff7/solr/core/src/java/org/apache/solr/cloud/autoscaling/ScheduledTriggers.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/ScheduledTriggers.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/ScheduledTriggers.java
index ec9ffc1..1d0f5d6 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/ScheduledTriggers.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/ScheduledTriggers.java
@@ -41,15 +41,18 @@ import java.util.stream.Collectors;
import org.apache.lucene.store.AlreadyClosedException;
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.TriggerEventProcessorStage;
import org.apache.solr.cloud.ActionThrottle;
import org.apache.solr.cloud.Overseer;
import org.apache.solr.cloud.ZkController;
+import org.apache.solr.common.SolrException;
import org.apache.solr.common.cloud.SolrZkClient;
import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.util.ExecutorUtil;
import org.apache.solr.common.util.IOUtils;
import org.apache.solr.core.CoreContainer;
+import org.apache.solr.core.SolrResourceLoader;
import org.apache.solr.util.DefaultSolrThreadFactory;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.Op;
@@ -85,17 +88,17 @@ public class ScheduledTriggers implements Closeable {
private final ActionThrottle actionThrottle;
- private final SolrZkClient zkClient;
+ private final ClusterDataProvider clusterDataProvider;
- private final Overseer.Stats queueStats;
+ private final SolrResourceLoader loader;
- private final CoreContainer coreContainer;
+ private final Overseer.Stats queueStats;
private final TriggerListeners listeners;
private AutoScalingConfig autoScalingConfig;
- public ScheduledTriggers(ZkController zkController) {
+ public ScheduledTriggers(SolrResourceLoader loader, ClusterDataProvider clusterDataProvider) {
// todo make the core pool size configurable
// it is important to use more than one because a time taking trigger can starve other scheduled triggers
// ideally we should have as many core threads as the number of triggers but firstly, we don't know beforehand
@@ -108,8 +111,8 @@ public class ScheduledTriggers implements Closeable {
actionExecutor = ExecutorUtil.newMDCAwareSingleThreadExecutor(new DefaultSolrThreadFactory("AutoscalingActionExecutor"));
// todo make the wait time configurable
actionThrottle = new ActionThrottle("action", DEFAULT_MIN_MS_BETWEEN_ACTIONS);
- coreContainer = zkController.getCoreContainer();
- zkClient = zkController.getZkClient();
+ this.clusterDataProvider = clusterDataProvider;
+ this.loader = loader;
queueStats = new Overseer.Stats();
listeners = new TriggerListeners();
}
@@ -136,7 +139,12 @@ public class ScheduledTriggers implements Closeable {
if (isClosed) {
throw new AlreadyClosedException("ScheduledTriggers has been closed and cannot be used anymore");
}
- ScheduledTrigger scheduledTrigger = new ScheduledTrigger(newTrigger, zkClient, queueStats);
+ ScheduledTrigger scheduledTrigger;
+ try {
+ scheduledTrigger = new ScheduledTrigger(newTrigger, clusterDataProvider, queueStats);
+ } catch (IOException e) {
+ throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "exception creating scheduled trigger", e);
+ }
ScheduledTrigger old = scheduledTriggers.putIfAbsent(newTrigger.getName(), scheduledTrigger);
if (old != null) {
if (old.trigger.equals(newTrigger)) {
@@ -182,7 +190,7 @@ public class ScheduledTriggers implements Closeable {
// let the action executor thread wait instead of the trigger thread so we use the throttle here
actionThrottle.minimumWaitBetweenActions();
actionThrottle.markAttemptingAction();
- ActionContext actionContext = new ActionContext(coreContainer, newTrigger, new HashMap<>());
+ ActionContext actionContext = new ActionContext(clusterDataProvider, newTrigger, new HashMap<>());
for (TriggerAction action : actions) {
List<String> beforeActions = (List<String>)actionContext.getProperties().computeIfAbsent(TriggerEventProcessorStage.BEFORE_ACTION.toString(), k -> new ArrayList<String>());
beforeActions.add(action.getName());
@@ -244,23 +252,23 @@ public class ScheduledTriggers implements Closeable {
String statePath = ZkStateReader.SOLR_AUTOSCALING_TRIGGER_STATE_PATH + "/" + triggerName;
String eventsPath = ZkStateReader.SOLR_AUTOSCALING_EVENTS_PATH + "/" + triggerName;
try {
- if (zkClient.exists(statePath, true)) {
- zkClient.delete(statePath, -1, true);
+ if (clusterDataProvider.hasData(statePath)) {
+ clusterDataProvider.removeData(statePath, -1);
}
- } catch (KeeperException | InterruptedException e) {
+ } catch (Exception e) {
log.warn("Failed to remove state for removed trigger " + statePath, e);
}
try {
- if (zkClient.exists(eventsPath, true)) {
- List<String> events = zkClient.getChildren(eventsPath, null, true);
+ if (clusterDataProvider.hasData(eventsPath)) {
+ List<String> events = clusterDataProvider.listData(eventsPath);
List<Op> ops = new ArrayList<>(events.size() + 1);
events.forEach(ev -> {
ops.add(Op.delete(eventsPath + "/" + ev, -1));
});
ops.add(Op.delete(eventsPath, -1));
- zkClient.multi(ops, true);
+ clusterDataProvider.multi(ops);
}
- } catch (KeeperException | InterruptedException e) {
+ } catch (Exception e) {
log.warn("Failed to remove events for removed trigger " + eventsPath, e);
}
}
@@ -297,9 +305,9 @@ public class ScheduledTriggers implements Closeable {
boolean replay;
volatile boolean isClosed;
- ScheduledTrigger(AutoScaling.Trigger trigger, SolrZkClient zkClient, Overseer.Stats stats) {
+ ScheduledTrigger(AutoScaling.Trigger trigger, ClusterDataProvider clusterDataProvider, Overseer.Stats stats) throws IOException {
this.trigger = trigger;
- this.queue = new TriggerEventQueue(zkClient, trigger.getName(), stats);
+ this.queue = new TriggerEventQueue(clusterDataProvider, trigger.getName(), stats);
this.replay = true;
this.isClosed = false;
}
@@ -426,13 +434,13 @@ public class ScheduledTriggers implements Closeable {
if (listener == null) { // create new instance
String clazz = config.listenerClass;
try {
- listener = coreContainer.getResourceLoader().newInstance(clazz, TriggerListener.class);
+ listener = loader.newInstance(clazz, TriggerListener.class);
} catch (Exception e) {
log.warn("Invalid TriggerListener class name '" + clazz + "', skipping...", e);
}
if (listener != null) {
try {
- listener.init(coreContainer, config);
+ listener.init(clusterDataProvider, config);
listenersPerName.put(config.name, listener);
} catch (Exception e) {
log.warn("Error initializing TriggerListener " + config, e);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/aee54ff7/solr/core/src/java/org/apache/solr/cloud/autoscaling/SystemLogListener.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/SystemLogListener.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/SystemLogListener.java
index a72b174..b18d5b7 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/SystemLogListener.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/SystemLogListener.java
@@ -31,6 +31,7 @@ import java.util.StringJoiner;
import org.apache.solr.client.solrj.SolrRequest;
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.TriggerEventProcessorStage;
import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.apache.solr.client.solrj.request.UpdateRequest;
@@ -73,8 +74,8 @@ public class SystemLogListener extends TriggerListenerBase {
private boolean enabled = true;
@Override
- public void init(CoreContainer coreContainer, AutoScalingConfig.TriggerListenerConfig config) {
- super.init(coreContainer, config);
+ public void init(ClusterDataProvider clusterDataProvider, AutoScalingConfig.TriggerListenerConfig config) {
+ super.init(clusterDataProvider, config);
collection = (String)config.properties.getOrDefault(CollectionAdminParams.COLLECTION, CollectionAdminParams.SYSTEM_COLL);
enabled = Boolean.parseBoolean(String.valueOf(config.properties.getOrDefault("enabled", true)));
}
@@ -85,10 +86,7 @@ public class SystemLogListener extends TriggerListenerBase {
if (!enabled) {
return;
}
- try (CloudSolrClient cloudSolrClient = new CloudSolrClient.Builder()
- .withZkHost(coreContainer.getZkController().getZkServerAddress())
- .withHttpClient(coreContainer.getUpdateShardHandler().getHttpClient())
- .build()) {
+ try {
SolrInputDocument doc = new SolrInputDocument();
doc.addField(CommonParams.TYPE, DOC_TYPE);
doc.addField(SOURCE_FIELD, SOURCE);
@@ -122,7 +120,8 @@ public class SystemLogListener extends TriggerListenerBase {
}
UpdateRequest req = new UpdateRequest();
req.add(doc);
- cloudSolrClient.request(req, collection);
+ req.setParam(CollectionAdminParams.COLLECTION, collection);
+ clusterDataProvider.request(req);
} catch (Exception e) {
if ((e instanceof SolrException) && e.getMessage().contains("Collection not found")) {
// relatively benign
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/aee54ff7/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerBase.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerBase.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerBase.java
index 7aff846..fd03d38 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerBase.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerBase.java
@@ -16,14 +16,26 @@
*/
package org.apache.solr.cloud.autoscaling;
+import java.io.IOException;
import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
import java.util.Map;
+import java.util.Objects;
import java.util.TreeMap;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.lucene.util.IOUtils;
+import org.apache.solr.client.solrj.cloud.autoscaling.ClusterDataProvider;
+import org.apache.solr.client.solrj.cloud.autoscaling.TriggerEventType;
import org.apache.solr.common.cloud.SolrZkClient;
import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.util.Utils;
+import org.apache.solr.core.SolrResourceLoader;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.data.Stat;
@@ -37,19 +49,130 @@ import org.slf4j.LoggerFactory;
public abstract class TriggerBase implements AutoScaling.Trigger {
private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- protected SolrZkClient zkClient;
+ protected final String name;
+ protected final ClusterDataProvider clusterDataProvider;
+ protected final Map<String, Object> properties = new HashMap<>();
+ protected final TriggerEventType eventType;
+ protected final int waitForSecond;
protected Map<String,Object> lastState;
+ protected final AtomicReference<AutoScaling.TriggerEventProcessor> processorRef = new AtomicReference<>();
+ protected final List<TriggerAction> actions;
+ protected final boolean enabled;
+ protected boolean isClosed;
- protected TriggerBase(SolrZkClient zkClient) {
- this.zkClient = zkClient;
+ protected TriggerBase(String name, Map<String, Object> properties, SolrResourceLoader loader, ClusterDataProvider clusterDataProvider) {
+ this.name = name;
+ if (properties != null) {
+ this.properties.putAll(properties);
+ }
+ this.clusterDataProvider = clusterDataProvider;
+ this.enabled = Boolean.parseBoolean(String.valueOf(this.properties.getOrDefault("enabled", "true")));
+ this.eventType = TriggerEventType.valueOf(this.properties.getOrDefault("event", TriggerEventType.INVALID.toString()).toString().toUpperCase(Locale.ROOT));
+ this.waitForSecond = ((Long) this.properties.getOrDefault("waitFor", -1L)).intValue();
+ List<Map<String, String>> o = (List<Map<String, String>>) properties.get("actions");
+ if (o != null && !o.isEmpty()) {
+ actions = new ArrayList<>(3);
+ for (Map<String, String> map : o) {
+ TriggerAction action = loader.newInstance(map.get("class"), TriggerAction.class);
+ actions.add(action);
+ }
+ } else {
+ actions = Collections.emptyList();
+ }
+
try {
- zkClient.makePath(ZkStateReader.SOLR_AUTOSCALING_TRIGGER_STATE_PATH, false, true);
- } catch (KeeperException | InterruptedException e) {
+ if (!clusterDataProvider.hasData(ZkStateReader.SOLR_AUTOSCALING_TRIGGER_STATE_PATH)) {
+ clusterDataProvider.makePath(ZkStateReader.SOLR_AUTOSCALING_TRIGGER_STATE_PATH);
+ }
+ } catch (IOException e) {
LOG.warn("Exception checking ZK path " + ZkStateReader.SOLR_AUTOSCALING_TRIGGER_STATE_PATH, e);
}
}
+ @Override
+ public void init() {
+ List<Map<String, String>> o = (List<Map<String, String>>) properties.get("actions");
+ if (o != null && !o.isEmpty()) {
+ for (int i = 0; i < o.size(); i++) {
+ Map<String, String> map = o.get(i);
+ actions.get(i).init(map);
+ }
+ }
+ }
+
+ @Override
+ public void setProcessor(AutoScaling.TriggerEventProcessor processor) {
+ processorRef.set(processor);
+ }
+
+ @Override
+ public AutoScaling.TriggerEventProcessor getProcessor() {
+ return processorRef.get();
+ }
+
+ @Override
+ public String getName() {
+ return name;
+ }
+
+ @Override
+ public TriggerEventType getEventType() {
+ return eventType;
+ }
+
+ @Override
+ public boolean isEnabled() {
+ return enabled;
+ }
+
+ @Override
+ public int getWaitForSecond() {
+ return waitForSecond;
+ }
+
+ @Override
+ public Map<String, Object> getProperties() {
+ return properties;
+ }
+
+ @Override
+ public List<TriggerAction> getActions() {
+ return actions;
+ }
+
+ @Override
+ public boolean isClosed() {
+ synchronized (this) {
+ return isClosed;
+ }
+ }
+ @Override
+ public void close() throws IOException {
+ synchronized (this) {
+ isClosed = true;
+ IOUtils.closeWhileHandlingException(actions);
+ }
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(name, properties);
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj == null) {
+ return false;
+ }
+ if (obj.getClass().equals(this.getClass())) {
+ TriggerBase that = (TriggerBase) obj;
+ return this.name.equals(that.name)
+ && this.properties.equals(that.properties);
+ }
+ return false;
+ }
+
/**
* Prepare and return internal state of this trigger in a format suitable for persisting in ZK.
* @return map of internal state properties. Note: values must be supported by {@link Utils#toJSON(Object)}.
@@ -72,15 +195,15 @@ public abstract class TriggerBase implements AutoScaling.Trigger {
byte[] data = Utils.toJSON(state);
String path = ZkStateReader.SOLR_AUTOSCALING_TRIGGER_STATE_PATH + "/" + getName();
try {
- if (zkClient.exists(path, true)) {
+ if (clusterDataProvider.hasData(path)) {
// update
- zkClient.setData(path, data, -1, true);
+ clusterDataProvider.setData(path, data, -1);
} else {
// create
- zkClient.create(path, data, CreateMode.PERSISTENT, true);
+ clusterDataProvider.createData(path, data, CreateMode.PERSISTENT);
}
lastState = state;
- } catch (KeeperException | InterruptedException e) {
+ } catch (IOException e) {
LOG.warn("Exception updating trigger state '" + path + "'", e);
}
}
@@ -90,10 +213,10 @@ public abstract class TriggerBase implements AutoScaling.Trigger {
byte[] data = null;
String path = ZkStateReader.SOLR_AUTOSCALING_TRIGGER_STATE_PATH + "/" + getName();
try {
- if (zkClient.exists(path, true)) {
- data = zkClient.getData(path, null, new Stat(), true);
+ if (clusterDataProvider.hasData(path)) {
+ ClusterDataProvider.VersionedData versionedDat = clusterDataProvider.getData(path);
}
- } catch (KeeperException | InterruptedException e) {
+ } catch (Exception e) {
LOG.warn("Exception getting trigger state '" + path + "'", e);
}
if (data != null) {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/aee54ff7/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerEventQueue.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerEventQueue.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerEventQueue.java
index 99f641c..c0b248f 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerEventQueue.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerEventQueue.java
@@ -1,23 +1,24 @@
package org.apache.solr.cloud.autoscaling;
+import java.io.IOException;
import java.lang.invoke.MethodHandles;
import java.util.Map;
+import java.util.Queue;
+import org.apache.solr.client.solrj.cloud.DistributedQueue;
+import org.apache.solr.client.solrj.cloud.autoscaling.ClusterDataProvider;
import org.apache.solr.client.solrj.cloud.autoscaling.TriggerEventType;
-import org.apache.solr.cloud.DistributedQueue;
import org.apache.solr.cloud.Overseer;
-import org.apache.solr.common.cloud.SolrZkClient;
import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.util.Utils;
import org.apache.solr.util.TimeSource;
-import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
*
*/
-public class TriggerEventQueue extends DistributedQueue {
+public class TriggerEventQueue {
private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
public static final String ENQUEUE_TIME = "_enqueue_time_";
@@ -25,9 +26,11 @@ public class TriggerEventQueue extends DistributedQueue {
private final String triggerName;
private final TimeSource timeSource;
+ private final DistributedQueue delegate;
- public TriggerEventQueue(SolrZkClient zookeeper, String triggerName, Overseer.Stats stats) {
- super(zookeeper, ZkStateReader.SOLR_AUTOSCALING_EVENTS_PATH + "/" + triggerName, stats);
+ public TriggerEventQueue(ClusterDataProvider clusterDataProvider, String triggerName, Overseer.Stats stats) throws IOException {
+ // TODO: collect stats
+ this.delegate = clusterDataProvider.getDistributedQueueFactory().makeQueue(ZkStateReader.SOLR_AUTOSCALING_EVENTS_PATH + "/" + triggerName);
this.triggerName = triggerName;
this.timeSource = TimeSource.CURRENT_TIME;
}
@@ -36,9 +39,9 @@ public class TriggerEventQueue extends DistributedQueue {
event.getProperties().put(ENQUEUE_TIME, timeSource.getTime());
try {
byte[] data = Utils.toJSON(event);
- offer(data);
+ delegate.offer(data);
return true;
- } catch (KeeperException | InterruptedException e) {
+ } catch (Exception e) {
LOG.warn("Exception adding event " + event + " to queue " + triggerName, e);
return false;
}
@@ -47,7 +50,7 @@ public class TriggerEventQueue extends DistributedQueue {
public TriggerEvent peekEvent() {
byte[] data;
try {
- while ((data = peek()) != null) {
+ while ((data = delegate.peek()) != null) {
if (data.length == 0) {
LOG.warn("ignoring empty data...");
continue;
@@ -60,7 +63,7 @@ public class TriggerEventQueue extends DistributedQueue {
continue;
}
}
- } catch (KeeperException | InterruptedException e) {
+ } catch (Exception e) {
LOG.warn("Exception peeking queue of trigger " + triggerName, e);
}
return null;
@@ -69,7 +72,7 @@ public class TriggerEventQueue extends DistributedQueue {
public TriggerEvent pollEvent() {
byte[] data;
try {
- while ((data = poll()) != null) {
+ while ((data = delegate.poll()) != null) {
if (data.length == 0) {
LOG.warn("ignoring empty data...");
continue;
@@ -82,7 +85,7 @@ public class TriggerEventQueue extends DistributedQueue {
continue;
}
}
- } catch (KeeperException | InterruptedException e) {
+ } catch (Exception e) {
LOG.warn("Exception polling queue of trigger " + triggerName, e);
}
return null;
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/aee54ff7/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerListener.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerListener.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerListener.java
index 3688bfc..4236522 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerListener.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerListener.java
@@ -19,6 +19,7 @@ package org.apache.solr.cloud.autoscaling;
import java.io.Closeable;
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.TriggerEventProcessorStage;
import org.apache.solr.core.CoreContainer;
@@ -28,7 +29,7 @@ import org.apache.solr.core.CoreContainer;
*/
public interface TriggerListener extends Closeable {
- void init(CoreContainer coreContainer, AutoScalingConfig.TriggerListenerConfig config) throws Exception;
+ void init(ClusterDataProvider clusterDataProvider, AutoScalingConfig.TriggerListenerConfig config) throws Exception;
AutoScalingConfig.TriggerListenerConfig getConfig();
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/aee54ff7/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerListenerBase.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerListenerBase.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerListenerBase.java
index 01a4413..d8b8fa3 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerListenerBase.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerListenerBase.java
@@ -19,7 +19,7 @@ package org.apache.solr.cloud.autoscaling;
import java.io.IOException;
import org.apache.solr.client.solrj.cloud.autoscaling.AutoScalingConfig;
-import org.apache.solr.core.CoreContainer;
+import org.apache.solr.client.solrj.cloud.autoscaling.ClusterDataProvider;
/**
* Base class for implementations of {@link TriggerListener}.
@@ -27,11 +27,11 @@ import org.apache.solr.core.CoreContainer;
public abstract class TriggerListenerBase implements TriggerListener {
protected AutoScalingConfig.TriggerListenerConfig config;
- protected CoreContainer coreContainer;
+ protected ClusterDataProvider clusterDataProvider;
@Override
- public void init(CoreContainer coreContainer, AutoScalingConfig.TriggerListenerConfig config) {
- this.coreContainer = coreContainer;
+ public void init(ClusterDataProvider clusterDataProvider, AutoScalingConfig.TriggerListenerConfig config) {
+ this.clusterDataProvider = clusterDataProvider;
this.config = config;
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/aee54ff7/solr/core/src/java/org/apache/solr/cloud/autoscaling/ZkDistributedQueueFactory.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/ZkDistributedQueueFactory.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/ZkDistributedQueueFactory.java
new file mode 100644
index 0000000..4cf57f9
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/ZkDistributedQueueFactory.java
@@ -0,0 +1,28 @@
+package org.apache.solr.cloud.autoscaling;
+
+import java.io.IOException;
+
+import org.apache.solr.client.solrj.cloud.DistributedQueue;
+import org.apache.solr.client.solrj.cloud.autoscaling.ClusterDataProvider;
+import org.apache.solr.cloud.ZkDistributedQueue;
+import org.apache.solr.common.cloud.SolrZkClient;
+
+/**
+ *
+ */
+public class ZkDistributedQueueFactory implements ClusterDataProvider.DistributedQueueFactory {
+ private final SolrZkClient zkClient;
+
+ public ZkDistributedQueueFactory(SolrZkClient zkClient) {
+ this.zkClient = zkClient;
+ }
+ @Override
+ public DistributedQueue makeQueue(String path) throws IOException {
+ return new ZkDistributedQueue(zkClient, path);
+ }
+
+ @Override
+ public void removeQueue(String path) throws IOException {
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/aee54ff7/solr/core/src/java/org/apache/solr/core/CoreContainer.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/core/CoreContainer.java b/solr/core/src/java/org/apache/solr/core/CoreContainer.java
index 0198f9d..57847f0 100644
--- a/solr/core/src/java/org/apache/solr/core/CoreContainer.java
+++ b/solr/core/src/java/org/apache/solr/core/CoreContainer.java
@@ -956,6 +956,8 @@ public class CoreContainer {
SolrException.log(log, null, e);
} catch (KeeperException e) {
SolrException.log(log, null, e);
+ } catch (Exception e) {
+ SolrException.log(log, null, e);
}
}
@@ -1377,6 +1379,8 @@ public class CoreContainer {
throw new SolrException(ErrorCode.SERVER_ERROR, "Interrupted while unregistering core [" + name + "] from cloud state");
} catch (KeeperException e) {
throw new SolrException(ErrorCode.SERVER_ERROR, "Error unregistering core [" + name + "] from cloud state", e);
+ } catch (Exception e) {
+ throw new SolrException(ErrorCode.SERVER_ERROR, "Error unregistering core [" + name + "] from cloud state", e);
}
}
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/aee54ff7/solr/core/src/java/org/apache/solr/core/ZkContainer.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/core/ZkContainer.java b/solr/core/src/java/org/apache/solr/core/ZkContainer.java
index 16ba4d8..37155ca 100644
--- a/solr/core/src/java/org/apache/solr/core/ZkContainer.java
+++ b/solr/core/src/java/org/apache/solr/core/ZkContainer.java
@@ -227,6 +227,8 @@ public class ZkContainer {
} catch (InterruptedException e) {
Thread.interrupted();
ZkContainer.log.error("", e);
+ } catch (Exception e) {
+ ZkContainer.log.error("", e);
}
}
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/aee54ff7/solr/core/src/java/org/apache/solr/update/processor/DistributedUpdateProcessor.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/update/processor/DistributedUpdateProcessor.java b/solr/core/src/java/org/apache/solr/update/processor/DistributedUpdateProcessor.java
index 45f6ea2..849fc49 100644
--- a/solr/core/src/java/org/apache/solr/update/processor/DistributedUpdateProcessor.java
+++ b/solr/core/src/java/org/apache/solr/update/processor/DistributedUpdateProcessor.java
@@ -46,7 +46,7 @@ import org.apache.solr.client.solrj.request.GenericSolrRequest;
import org.apache.solr.client.solrj.request.UpdateRequest;
import org.apache.solr.client.solrj.response.SimpleSolrResponse;
import org.apache.solr.cloud.CloudDescriptor;
-import org.apache.solr.cloud.DistributedQueue;
+import org.apache.solr.client.solrj.cloud.DistributedQueue;
import org.apache.solr.cloud.Overseer;
import org.apache.solr.cloud.ZkController;
import org.apache.solr.cloud.overseer.OverseerAction;
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/aee54ff7/solr/core/src/test/org/apache/solr/cloud/DeleteShardTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/DeleteShardTest.java b/solr/core/src/test/org/apache/solr/cloud/DeleteShardTest.java
index ed3d03b..028525e 100644
--- a/solr/core/src/test/org/apache/solr/cloud/DeleteShardTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/DeleteShardTest.java
@@ -21,6 +21,7 @@ import java.util.HashMap;
import java.util.Map;
import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.cloud.DistributedQueue;
import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.apache.solr.client.solrj.request.CollectionAdminRequest;
import org.apache.solr.client.solrj.request.CoreStatus;
@@ -82,8 +83,7 @@ public class DeleteShardTest extends SolrCloudTestCase {
}
- protected void setSliceState(String collection, String slice, State state) throws SolrServerException, IOException,
- KeeperException, InterruptedException {
+ protected void setSliceState(String collection, String slice, State state) throws Exception {
CloudSolrClient client = cluster.getSolrClient();
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/aee54ff7/solr/core/src/test/org/apache/solr/cloud/DistributedQueueTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/DistributedQueueTest.java b/solr/core/src/test/org/apache/solr/cloud/DistributedQueueTest.java
index ed33dc1..7d7e5f5 100644
--- a/solr/core/src/test/org/apache/solr/cloud/DistributedQueueTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/DistributedQueueTest.java
@@ -25,6 +25,7 @@ import java.util.concurrent.TimeoutException;
import java.util.function.Predicate;
import org.apache.solr.SolrTestCaseJ4;
+import org.apache.solr.client.solrj.cloud.DistributedQueue;
import org.apache.solr.common.cloud.SolrZkClient;
import org.apache.solr.common.util.ExecutorUtil;
import org.apache.solr.common.util.SolrjNamedThreadFactory;
@@ -95,7 +96,7 @@ public class DistributedQueueTest extends SolrTestCaseJ4 {
String dqZNode = "/distqueue/test";
byte[] data = "hello world".getBytes(UTF8);
- DistributedQueue consumer = makeDistributedQueue(dqZNode);
+ ZkDistributedQueue consumer = makeDistributedQueue(dqZNode);
DistributedQueue producer = makeDistributedQueue(dqZNode);
DistributedQueue producer2 = makeDistributedQueue(dqZNode);
@@ -124,7 +125,7 @@ public class DistributedQueueTest extends SolrTestCaseJ4 {
String dqZNode = "/distqueue/test";
String testData = "hello world";
- DistributedQueue dq = makeDistributedQueue(dqZNode);
+ ZkDistributedQueue dq = makeDistributedQueue(dqZNode);
assertNull(dq.peek());
Future<String> future = executor.submit(() -> new String(dq.peek(true), UTF8));
@@ -171,7 +172,7 @@ public class DistributedQueueTest extends SolrTestCaseJ4 {
@Test
public void testLeakChildWatcher() throws Exception {
String dqZNode = "/distqueue/test";
- DistributedQueue dq = makeDistributedQueue(dqZNode);
+ ZkDistributedQueue dq = makeDistributedQueue(dqZNode);
assertTrue(dq.peekElements(1, 1, s1 -> true).isEmpty());
assertEquals(1, dq.watcherCount());
assertFalse(dq.isDirty());
@@ -280,8 +281,8 @@ public class DistributedQueueTest extends SolrTestCaseJ4 {
assertFalse(sessionId == zkClient.getSolrZooKeeper().getSessionId());
}
- protected DistributedQueue makeDistributedQueue(String dqZNode) throws Exception {
- return new DistributedQueue(zkClient, setupNewDistributedQueueZNode(dqZNode));
+ protected ZkDistributedQueue makeDistributedQueue(String dqZNode) throws Exception {
+ return new ZkDistributedQueue(zkClient, setupNewDistributedQueueZNode(dqZNode));
}
private static class QueueChangerThread extends Thread {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/aee54ff7/solr/core/src/test/org/apache/solr/cloud/ForceLeaderTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/ForceLeaderTest.java b/solr/core/src/test/org/apache/solr/cloud/ForceLeaderTest.java
index 749abdf..69658a5 100644
--- a/solr/core/src/test/org/apache/solr/cloud/ForceLeaderTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/ForceLeaderTest.java
@@ -26,6 +26,7 @@ import org.apache.solr.client.solrj.SolrClient;
import org.apache.solr.client.solrj.SolrRequest;
import org.apache.solr.client.solrj.SolrRequest.METHOD;
import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.cloud.DistributedQueue;
import org.apache.solr.client.solrj.embedded.JettySolrRunner;
import org.apache.solr.client.solrj.impl.HttpSolrClient;
import org.apache.solr.client.solrj.request.CollectionAdminRequest;
@@ -230,8 +231,7 @@ public class ForceLeaderTest extends HttpPartitionTest {
}
}
- protected void setReplicaState(String collection, String slice, Replica replica, Replica.State state) throws SolrServerException, IOException,
- KeeperException, InterruptedException {
+ protected void setReplicaState(String collection, String slice, Replica replica, Replica.State state) throws Exception {
DistributedQueue inQueue = Overseer.getStateUpdateQueue(cloudClient.getZkStateReader().getZkClient());
ZkStateReader zkStateReader = cloudClient.getZkStateReader();
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/aee54ff7/solr/core/src/test/org/apache/solr/cloud/MultiThreadedOCPTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/MultiThreadedOCPTest.java b/solr/core/src/test/org/apache/solr/cloud/MultiThreadedOCPTest.java
index 537dbba..6ab8593 100644
--- a/solr/core/src/test/org/apache/solr/cloud/MultiThreadedOCPTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/MultiThreadedOCPTest.java
@@ -23,6 +23,7 @@ import java.util.Random;
import org.apache.solr.client.solrj.SolrClient;
import org.apache.solr.client.solrj.SolrRequest;
import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.cloud.DistributedQueue;
import org.apache.solr.client.solrj.impl.HttpSolrClient;
import org.apache.solr.client.solrj.request.CollectionAdminRequest;
import org.apache.solr.client.solrj.request.CollectionAdminRequest.Create;
@@ -68,7 +69,7 @@ public class MultiThreadedOCPTest extends AbstractFullDistribZkTestBase {
private void testFillWorkQueue() throws Exception {
try (SolrClient client = createNewSolrClient("", getBaseUrl((HttpSolrClient) clients.get(0)))) {
- DistributedQueue distributedQueue = new DistributedQueue(cloudClient.getZkStateReader().getZkClient(),
+ DistributedQueue distributedQueue = new ZkDistributedQueue(cloudClient.getZkStateReader().getZkClient(),
"/overseer/collection-queue-work", new Overseer.Stats());
//fill the work queue with blocked tasks by adding more than the no:of parallel tasks
for (int i = 0; i < MAX_PARALLEL_TASKS+5; i++) {
@@ -149,7 +150,7 @@ public class MultiThreadedOCPTest extends AbstractFullDistribZkTestBase {
private void testTaskExclusivity() throws Exception, SolrServerException {
- DistributedQueue distributedQueue = new DistributedQueue(cloudClient.getZkStateReader().getZkClient(),
+ DistributedQueue distributedQueue = new ZkDistributedQueue(cloudClient.getZkStateReader().getZkClient(),
"/overseer/collection-queue-work", new Overseer.Stats());
try (SolrClient client = createNewSolrClient("", getBaseUrl((HttpSolrClient) clients.get(0)))) {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/aee54ff7/solr/core/src/test/org/apache/solr/cloud/OverseerTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/OverseerTest.java b/solr/core/src/test/org/apache/solr/cloud/OverseerTest.java
index 378e371..8e95974 100644
--- a/solr/core/src/test/org/apache/solr/cloud/OverseerTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/OverseerTest.java
@@ -37,6 +37,7 @@ import com.codahale.metrics.Snapshot;
import com.codahale.metrics.Timer;
import org.apache.lucene.util.LuceneTestCase.Slow;
import org.apache.solr.SolrTestCaseJ4;
+import org.apache.solr.client.solrj.cloud.DistributedQueue;
import org.apache.solr.cloud.overseer.OverseerAction;
import org.apache.solr.common.cloud.ClusterState;
import org.apache.solr.common.cloud.DocCollection;
@@ -133,7 +134,7 @@ public class OverseerTest extends SolrTestCaseJ4 {
zkClient.close();
}
- public void createCollection(String collection, int numShards) throws KeeperException, InterruptedException {
+ public void createCollection(String collection, int numShards) throws Exception {
ZkNodeProps m = new ZkNodeProps(Overseer.QUEUE_OPERATION, CollectionParams.CollectionAction.CREATE.toLower(),
"name", collection,
@@ -146,7 +147,7 @@ public class OverseerTest extends SolrTestCaseJ4 {
}
public String publishState(String collection, String coreName, String coreNodeName, String shard, Replica.State stateName, int numShards)
- throws KeeperException, InterruptedException, IOException {
+ throws Exception {
if (stateName == null) {
ElectionContext ec = electionContext.remove(coreName);
if (ec != null) {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/aee54ff7/solr/core/src/test/org/apache/solr/cloud/TestRandomRequestDistribution.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/TestRandomRequestDistribution.java b/solr/core/src/test/org/apache/solr/cloud/TestRandomRequestDistribution.java
index 415f80f..ea1f3e5 100644
--- a/solr/core/src/test/org/apache/solr/cloud/TestRandomRequestDistribution.java
+++ b/solr/core/src/test/org/apache/solr/cloud/TestRandomRequestDistribution.java
@@ -31,6 +31,7 @@ import org.apache.lucene.util.TestUtil;
import org.apache.solr.BaseDistributedSearchTestCase;
import org.apache.solr.SolrTestCaseJ4;
import org.apache.solr.client.solrj.SolrQuery;
+import org.apache.solr.client.solrj.cloud.DistributedQueue;
import org.apache.solr.client.solrj.embedded.JettySolrRunner;
import org.apache.solr.client.solrj.impl.HttpSolrClient;
import org.apache.solr.client.solrj.request.CollectionAdminRequest;
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/aee54ff7/solr/core/src/test/org/apache/solr/cloud/autoscaling/AutoAddReplicasPlanActionTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/AutoAddReplicasPlanActionTest.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/AutoAddReplicasPlanActionTest.java
index 4a0495d..a8f470c 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/AutoAddReplicasPlanActionTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/AutoAddReplicasPlanActionTest.java
@@ -150,7 +150,7 @@ public class AutoAddReplicasPlanActionTest extends SolrCloudTestCase{
private List<SolrRequest> getOperations(JettySolrRunner actionJetty, String lostNodeName) {
AutoAddReplicasPlanAction action = new AutoAddReplicasPlanAction();
TriggerEvent lostNode = new NodeLostTrigger.NodeLostEvent(TriggerEventType.NODELOST, ".auto_add_replicas", Collections.singletonList(System.currentTimeMillis()), Collections.singletonList(lostNodeName));
- ActionContext context = new ActionContext(actionJetty.getCoreContainer(), null, new HashMap<>());
+ ActionContext context = new ActionContext(actionJetty.getCoreContainer().getZkController().getClusterDataProvider(), null, new HashMap<>());
action.process(lostNode, context);
List<SolrRequest> operations = (List) context.getProperty("operations");
return operations;
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/aee54ff7/solr/core/src/test/org/apache/solr/cloud/autoscaling/ExecutePlanActionTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/ExecutePlanActionTest.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/ExecutePlanActionTest.java
index 1cb692e..23c880a 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/ExecutePlanActionTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/ExecutePlanActionTest.java
@@ -110,7 +110,7 @@ public class ExecutePlanActionTest extends SolrCloudTestCase {
NodeLostTrigger.NodeLostEvent nodeLostEvent = new NodeLostTrigger.NodeLostEvent(TriggerEventType.NODELOST,
"mock_trigger_name", Collections.singletonList(TimeSource.CURRENT_TIME.getTime()),
Collections.singletonList(sourceNodeName));
- ActionContext actionContext = new ActionContext(survivor.getCoreContainer(), null,
+ ActionContext actionContext = new ActionContext(survivor.getCoreContainer().getZkController().getClusterDataProvider(), null,
new HashMap<>(Collections.singletonMap("operations", operations)));
action.process(nodeLostEvent, actionContext);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/aee54ff7/solr/core/src/test/org/apache/solr/cloud/autoscaling/NodeAddedTriggerTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/NodeAddedTriggerTest.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/NodeAddedTriggerTest.java
index 687aec7..8dd7340 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/NodeAddedTriggerTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/NodeAddedTriggerTest.java
@@ -30,6 +30,7 @@ import java.util.concurrent.atomic.AtomicReference;
import org.apache.solr.client.solrj.embedded.JettySolrRunner;
import org.apache.solr.cloud.SolrCloudTestCase;
import org.apache.solr.core.CoreContainer;
+import org.apache.solr.core.SolrResourceLoader;
import org.apache.solr.util.TimeSource;
import org.junit.Before;
import org.junit.BeforeClass;
@@ -72,7 +73,8 @@ public class NodeAddedTriggerTest extends SolrCloudTestCase {
long waitForSeconds = 1 + random().nextInt(5);
Map<String, Object> props = createTriggerProps(waitForSeconds);
- try (NodeAddedTrigger trigger = new NodeAddedTrigger("node_added_trigger", props, container)) {
+ try (NodeAddedTrigger trigger = new NodeAddedTrigger("node_added_trigger", props, container.getResourceLoader(),
+ container.getZkController().getClusterDataProvider())) {
trigger.setProcessor(noFirstRunProcessor);
trigger.run();
@@ -112,7 +114,8 @@ public class NodeAddedTriggerTest extends SolrCloudTestCase {
// add a new node but remove it before the waitFor period expires
// and assert that the trigger doesn't fire at all
- try (NodeAddedTrigger trigger = new NodeAddedTrigger("node_added_trigger", props, container)) {
+ try (NodeAddedTrigger trigger = new NodeAddedTrigger("node_added_trigger", props, container.getResourceLoader(),
+ container.getZkController().getClusterDataProvider())) {
final long waitTime = 2;
props.put("waitFor", waitTime);
trigger.setProcessor(noFirstRunProcessor);
@@ -157,7 +160,8 @@ public class NodeAddedTriggerTest extends SolrCloudTestCase {
action.put("name", "testActionInit");
action.put("class", NodeAddedTriggerTest.AssertInitTriggerAction.class.getName());
actions.add(action);
- try (NodeAddedTrigger trigger = new NodeAddedTrigger("node_added_trigger", props, container)) {
+ try (NodeAddedTrigger trigger = new NodeAddedTrigger("node_added_trigger", props, container.getResourceLoader(),
+ container.getZkController().getClusterDataProvider())) {
assertEquals(true, actionConstructorCalled.get());
assertEquals(false, actionInitCalled.get());
assertEquals(false, actionCloseCalled.get());
@@ -198,7 +202,8 @@ public class NodeAddedTriggerTest extends SolrCloudTestCase {
public void testListenerAcceptance() throws Exception {
CoreContainer container = cluster.getJettySolrRunners().get(0).getCoreContainer();
Map<String, Object> props = createTriggerProps(0);
- try (NodeAddedTrigger trigger = new NodeAddedTrigger("node_added_trigger", props, container)) {
+ try (NodeAddedTrigger trigger = new NodeAddedTrigger("node_added_trigger", props, container.getResourceLoader(),
+ container.getZkController().getClusterDataProvider())) {
trigger.setProcessor(noFirstRunProcessor);
trigger.run(); // starts tracking live nodes
@@ -234,7 +239,8 @@ public class NodeAddedTriggerTest extends SolrCloudTestCase {
// add a new node but update the trigger before the waitFor period expires
// and assert that the new trigger still fires
- NodeAddedTrigger trigger = new NodeAddedTrigger("node_added_trigger", props, container);
+ NodeAddedTrigger trigger = new NodeAddedTrigger("node_added_trigger", props, container.getResourceLoader(),
+ container.getZkController().getClusterDataProvider());
trigger.setProcessor(noFirstRunProcessor);
trigger.run();
@@ -242,7 +248,8 @@ public class NodeAddedTriggerTest extends SolrCloudTestCase {
trigger.run(); // this run should detect the new node
trigger.close(); // close the old trigger
- try (NodeAddedTrigger newTrigger = new NodeAddedTrigger("some_different_name", props, container)) {
+ try (NodeAddedTrigger newTrigger = new NodeAddedTrigger("some_different_name", props, container.getResourceLoader(),
+ container.getZkController().getClusterDataProvider())) {
try {
newTrigger.restoreState(trigger);
fail("Trigger should only be able to restore state from an old trigger of the same name");
@@ -251,7 +258,8 @@ public class NodeAddedTriggerTest extends SolrCloudTestCase {
}
}
- try (NodeAddedTrigger newTrigger = new NodeAddedTrigger("node_added_trigger", props, container)) {
+ try (NodeAddedTrigger newTrigger = new NodeAddedTrigger("node_added_trigger", props, container.getResourceLoader(),
+ container.getZkController().getClusterDataProvider())) {
AtomicBoolean fired = new AtomicBoolean(false);
AtomicReference<TriggerEvent> eventRef = new AtomicReference<>();
newTrigger.setProcessor(event -> {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/aee54ff7/solr/core/src/test/org/apache/solr/cloud/autoscaling/NodeLostTriggerTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/NodeLostTriggerTest.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/NodeLostTriggerTest.java
index 9833c5d..0198098 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/NodeLostTriggerTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/NodeLostTriggerTest.java
@@ -73,7 +73,8 @@ public class NodeLostTriggerTest extends SolrCloudTestCase {
long waitForSeconds = 1 + random().nextInt(5);
Map<String, Object> props = createTriggerProps(waitForSeconds);
- try (NodeLostTrigger trigger = new NodeLostTrigger("node_lost_trigger", props, container)) {
+ try (NodeLostTrigger trigger = new NodeLostTrigger("node_lost_trigger", props, container.getResourceLoader(),
+ container.getZkController().getClusterDataProvider())) {
trigger.setProcessor(noFirstRunProcessor);
trigger.run();
String lostNodeName1 = cluster.getJettySolrRunner(1).getNodeName();
@@ -117,7 +118,8 @@ public class NodeLostTriggerTest extends SolrCloudTestCase {
// remove a node but add it back before the waitFor period expires
// and assert that the trigger doesn't fire at all
- try (NodeLostTrigger trigger = new NodeLostTrigger("node_lost_trigger", props, container)) {
+ try (NodeLostTrigger trigger = new NodeLostTrigger("node_lost_trigger", props, container.getResourceLoader(),
+ container.getZkController().getClusterDataProvider())) {
final long waitTime = 2;
props.put("waitFor", waitTime);
trigger.setProcessor(noFirstRunProcessor);
@@ -173,7 +175,8 @@ public class NodeLostTriggerTest extends SolrCloudTestCase {
action.put("name", "testActionInit");
action.put("class", AssertInitTriggerAction.class.getName());
actions.add(action);
- try (NodeLostTrigger trigger = new NodeLostTrigger("node_added_trigger", props, container)) {
+ try (NodeLostTrigger trigger = new NodeLostTrigger("node_added_trigger", props, container.getResourceLoader(),
+ container.getZkController().getClusterDataProvider())) {
assertEquals(true, actionConstructorCalled.get());
assertEquals(false, actionInitCalled.get());
assertEquals(false, actionCloseCalled.get());
@@ -214,7 +217,8 @@ public class NodeLostTriggerTest extends SolrCloudTestCase {
public void testListenerAcceptance() throws Exception {
CoreContainer container = cluster.getJettySolrRunners().get(0).getCoreContainer();
Map<String, Object> props = createTriggerProps(0);
- try (NodeLostTrigger trigger = new NodeLostTrigger("node_added_trigger", props, container)) {
+ try (NodeLostTrigger trigger = new NodeLostTrigger("node_added_trigger", props, container.getResourceLoader(),
+ container.getZkController().getClusterDataProvider())) {
trigger.setProcessor(noFirstRunProcessor);
JettySolrRunner newNode = cluster.startJettySolrRunner();
@@ -267,7 +271,8 @@ public class NodeLostTriggerTest extends SolrCloudTestCase {
// remove a node but update the trigger before the waitFor period expires
// and assert that the new trigger still fires
- NodeLostTrigger trigger = new NodeLostTrigger("node_lost_trigger", props, container);
+ NodeLostTrigger trigger = new NodeLostTrigger("node_lost_trigger", props, container.getResourceLoader(),
+ container.getZkController().getClusterDataProvider());
trigger.setProcessor(noFirstRunProcessor);
trigger.run();
@@ -284,7 +289,8 @@ public class NodeLostTriggerTest extends SolrCloudTestCase {
trigger.run(); // this run should detect the lost node
trigger.close(); // close the old trigger
- try (NodeLostTrigger newTrigger = new NodeLostTrigger("some_different_name", props, container)) {
+ try (NodeLostTrigger newTrigger = new NodeLostTrigger("some_different_name", props, container.getResourceLoader(),
+ container.getZkController().getClusterDataProvider())) {
try {
newTrigger.restoreState(trigger);
fail("Trigger should only be able to restore state from an old trigger of the same name");
@@ -293,7 +299,8 @@ public class NodeLostTriggerTest extends SolrCloudTestCase {
}
}
- try (NodeLostTrigger newTrigger = new NodeLostTrigger("node_lost_trigger", props, container)) {
+ try (NodeLostTrigger newTrigger = new NodeLostTrigger("node_lost_trigger", props, container.getResourceLoader(),
+ container.getZkController().getClusterDataProvider())) {
AtomicBoolean fired = new AtomicBoolean(false);
AtomicReference<TriggerEvent> eventRef = new AtomicReference<>();
newTrigger.setProcessor(event -> {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/aee54ff7/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 bd3d5fd..6bf1082 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
@@ -168,7 +168,7 @@ public class TestPolicyCloud extends SolrCloudTestCase {
CollectionAdminRequest.createCollection("metricsTest", "conf", 1, 1)
.process(cluster.getSolrClient());
DocCollection collection = getCollectionState("metricsTest");
- SolrClientDataProvider provider = new SolrClientDataProvider(solrClient);
+ SolrClientDataProvider provider = new SolrClientDataProvider(new ZkDistributedQueueFactory(cluster.getZkClient()), solrClient);
List<String> tags = Arrays.asList("metrics:solr.node:ADMIN./admin/authorization.clientErrors:count",
"metrics:solr.jvm:buffers.direct.Count");
Map<String, Object> val = provider.getNodeValues(collection .getReplicas().get(0).getNodeName(), tags);
@@ -268,7 +268,7 @@ public class TestPolicyCloud extends SolrCloudTestCase {
CollectionAdminRequest.createCollectionWithImplicitRouter("policiesTest", "conf", "shard1", 2)
.process(cluster.getSolrClient());
DocCollection rulesCollection = getCollectionState("policiesTest");
- SolrClientDataProvider provider = new SolrClientDataProvider(cluster.getSolrClient());
+ SolrClientDataProvider provider = new SolrClientDataProvider(new ZkDistributedQueueFactory(cluster.getZkClient()), cluster.getSolrClient());
Map<String, Object> val = provider.getNodeValues(rulesCollection.getReplicas().get(0).getNodeName(), Arrays.asList(
"freedisk",
"cores",
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/aee54ff7/solr/core/src/test/org/apache/solr/cloud/autoscaling/TriggerIntegrationTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/TriggerIntegrationTest.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/TriggerIntegrationTest.java
index a0eab35..24dafc8 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/TriggerIntegrationTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/TriggerIntegrationTest.java
@@ -34,6 +34,7 @@ import java.util.concurrent.locks.ReentrantLock;
import org.apache.solr.client.solrj.SolrRequest;
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.TriggerEventProcessorStage;
import org.apache.solr.client.solrj.cloud.autoscaling.TriggerEventType;
import org.apache.solr.client.solrj.embedded.JettySolrRunner;
@@ -965,8 +966,8 @@ public class TriggerIntegrationTest extends SolrCloudTestCase {
public static class TestTriggerListener extends TriggerListenerBase {
@Override
- public void init(CoreContainer coreContainer, AutoScalingConfig.TriggerListenerConfig config) {
- super.init(coreContainer, config);
+ public void init(ClusterDataProvider clusterDataProvider, AutoScalingConfig.TriggerListenerConfig config) {
+ super.init(clusterDataProvider, config);
listenerCreated.countDown();
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/aee54ff7/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/DistributedQueue.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/DistributedQueue.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/DistributedQueue.java
new file mode 100644
index 0000000..44684f0
--- /dev/null
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/DistributedQueue.java
@@ -0,0 +1,44 @@
+/*
+ * 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;
+
+import java.util.Collection;
+import java.util.function.Predicate;
+
+import org.apache.solr.common.util.Pair;
+
+/**
+ *
+ */
+public interface DistributedQueue {
+ byte[] peek() throws Exception;
+
+ byte[] peek(boolean block) throws Exception;
+
+ byte[] peek(long wait) throws Exception;
+
+ byte[] poll() throws Exception;
+
+ byte[] remove() throws Exception;
+
+ byte[] take() throws Exception;
+
+ void offer(byte[] data) throws Exception;
+
+ Collection<Pair<String, byte[]>> peekElements(int max, long waitMillis, Predicate<String> acceptFilter) throws Exception;
+
+}