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/11/29 17:13:38 UTC
[2/2] lucene-solr:jira/solr-11285-sim: SOLR-11403: Implement and use
GenericDistributedQueue to make sure the simulator creates actual event data
in DistribStateManager. Implement a large cluster test.
SOLR-11403: Implement and use GenericDistributedQueue to make sure the simulator
creates actual event data in DistribStateManager.
Implement a large cluster test.
Project: http://git-wip-us.apache.org/repos/asf/lucene-solr/repo
Commit: http://git-wip-us.apache.org/repos/asf/lucene-solr/commit/23df975b
Tree: http://git-wip-us.apache.org/repos/asf/lucene-solr/tree/23df975b
Diff: http://git-wip-us.apache.org/repos/asf/lucene-solr/diff/23df975b
Branch: refs/heads/jira/solr-11285-sim
Commit: 23df975b2a0155daa69af72e7eecaf630c27033e
Parents: 4e63be7
Author: Andrzej Bialecki <ab...@apache.org>
Authored: Wed Nov 29 18:12:29 2017 +0100
Committer: Andrzej Bialecki <ab...@apache.org>
Committed: Wed Nov 29 18:12:29 2017 +0100
----------------------------------------------------------------------
.../cloud/autoscaling/ComputePlanAction.java | 5 +-
.../cloud/autoscaling/NodeAddedTrigger.java | 2 +-
.../solr/cloud/autoscaling/NodeLostTrigger.java | 2 +-
.../cloud/autoscaling/ScheduledTriggers.java | 3 +
.../sim/GenericDistributedQueue.java | 600 +++++++++++++++++++
.../sim/GenericDistributedQueueFactory.java | 45 ++
.../cloud/autoscaling/sim/SimCloudManager.java | 29 +-
.../sim/SimClusterStateProvider.java | 51 +-
.../autoscaling/sim/SimDistribStateManager.java | 112 ++--
.../sim/SimDistributedQueueFactory.java | 4 +-
.../autoscaling/sim/SimNodeStateProvider.java | 6 +-
.../autoscaling/sim/SimSolrCloudTestCase.java | 18 +-
.../sim/TestClusterStateProvider.java | 2 +-
.../sim/TestDistribStateManager.java | 2 +-
.../autoscaling/sim/TestExecutePlanAction.java | 7 +-
.../sim/TestGenericDistributedQueue.java | 16 +
.../cloud/autoscaling/sim/TestLargeCluster.java | 20 +-
.../cloud/autoscaling/sim/TestPolicyCloud.java | 1 -
.../sim/TestSimDistributedQueue.java | 8 +-
.../autoscaling/sim/TestTriggerIntegration.java | 1 +
.../cloud/autoscaling/DistribStateManager.java | 2 +
.../solrj/impl/ZkDistribStateManager.java | 9 +-
22 files changed, 829 insertions(+), 116 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/23df975b/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 b36ee59..9eed1f2 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
@@ -23,6 +23,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.TimeUnit;
import org.apache.solr.client.solrj.SolrRequest;
import org.apache.solr.client.solrj.cloud.autoscaling.AutoScalingConfig;
@@ -61,9 +62,11 @@ public class ComputePlanAction extends TriggerActionBase {
Suggester suggester = getSuggester(session, event, cloudManager);
log.debug("{} Created suggester with node(s): {}, event {}", event.eventType, event.getProperty(TriggerEvent.NODE_NAMES), event.id);
while (true) {
+ long start = cloudManager.getTimeSource().getTime();
SolrRequest operation = suggester.getSuggestion();
if (operation == null) break;
- log.info("Computed Plan: {}", operation.getParams());
+ log.info("Computed Plan ({} ms): {}",
+ TimeUnit.MILLISECONDS.convert(cloudManager.getTimeSource().getTime() - start, TimeUnit.NANOSECONDS), operation.getParams());
Map<String, Object> props = context.getProperties();
props.compute("operations", (k, v) -> {
List<SolrRequest> operations = (List<SolrRequest>) v;
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/23df975b/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 cc89698..a1e19e9 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
@@ -127,7 +127,7 @@ public class NodeAddedTrigger extends TriggerBase {
log.debug("Running NodeAddedTrigger {}", name);
Set<String> newLiveNodes = new HashSet<>(cloudManager.getClusterStateProvider().getLiveNodes());
- log.debug("Found livenodes: {}", newLiveNodes);
+ log.debug("Found livenodes: {}", newLiveNodes.size());
// have any nodes that we were tracking been removed from the cluster?
// if so, remove them from the tracking map
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/23df975b/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 d5ba80a..57c76c0 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
@@ -123,7 +123,7 @@ public class NodeLostTrigger extends TriggerBase {
}
Set<String> newLiveNodes = new HashSet<>(cloudManager.getClusterStateProvider().getLiveNodes());
- log.debug("Running NodeLostTrigger: {} with currently live nodes: {}", name, newLiveNodes);
+ log.debug("Running NodeLostTrigger: {} with currently live nodes: {}", name, newLiveNodes.size());
// have any nodes that we were tracking been added to the cluster?
// if so, remove them from the tracking map
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/23df975b/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 0471201..6f1c8d5 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
@@ -236,6 +236,7 @@ public class ScheduledTriggers implements Closeable {
}
actionExecutor.submit(() -> {
assert hasPendingActions.get();
+ long eventProcessingStart = cloudManager.getTimeSource().getTime();
log.debug("-- processing actions for " + event);
try {
// let the action executor thread wait instead of the trigger thread so we use the throttle here
@@ -272,6 +273,8 @@ public class ScheduledTriggers implements Closeable {
cooldownStart.set(cloudManager.getTimeSource().getTime());
hasPendingActions.set(false);
}
+ log.debug("-- processing took {} ms for event id={}",
+ TimeUnit.MILLISECONDS.convert(cloudManager.getTimeSource().getTime() - eventProcessingStart, TimeUnit.NANOSECONDS), event.id);
});
} else {
if (enqueued) {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/23df975b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/GenericDistributedQueue.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/GenericDistributedQueue.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/GenericDistributedQueue.java
new file mode 100644
index 0000000..0a61709
--- /dev/null
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/GenericDistributedQueue.java
@@ -0,0 +1,600 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.solr.cloud.autoscaling.sim;
+
+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.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.TreeSet;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Predicate;
+
+import com.codahale.metrics.Timer;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import org.apache.solr.client.solrj.cloud.DistributedQueue;
+import org.apache.solr.client.solrj.cloud.autoscaling.AlreadyExistsException;
+import org.apache.solr.client.solrj.cloud.autoscaling.DistribStateManager;
+import org.apache.solr.client.solrj.cloud.autoscaling.VersionedData;
+import org.apache.solr.cloud.OverseerTaskQueue;
+import org.apache.solr.cloud.Stats;
+import org.apache.solr.common.SolrException;
+import org.apache.solr.common.SolrException.ErrorCode;
+import org.apache.solr.common.util.Pair;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.Op;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.data.Stat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A distributed queue that uses {@link org.apache.solr.client.solrj.cloud.autoscaling.DistribStateManager}.
+ * Implementation based on {@link org.apache.solr.cloud.ZkDistributedQueue}
+ */
+public class GenericDistributedQueue implements DistributedQueue {
+ private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ static final String PREFIX = "qn-";
+
+ /**
+ * Theory of operation:
+ * <p>
+ * Under ordinary circumstances we neither watch nor poll for children in ZK.
+ * Instead we keep an in-memory list of known child names. When the in-memory
+ * list is exhausted, we then fetch from ZK.
+ * <p>
+ * We only bother setting a child watcher when the queue has no children in ZK.
+ */
+ private static final Object _IMPLEMENTATION_NOTES = null;
+
+ final String dir;
+
+ final DistribStateManager stateManager;
+
+ final Stats stats;
+
+ /**
+ * A lock that guards all of the mutable state that follows.
+ */
+ private final ReentrantLock updateLock = new ReentrantLock();
+
+ /**
+ * Contains the last set of children fetched from ZK. Elements are removed from the head of
+ * this in-memory set as they are consumed from the queue. Due to the distributed nature
+ * of the queue, elements may appear in this set whose underlying nodes have been consumed in ZK.
+ * Therefore, methods like {@link #peek()} have to double-check actual node existence, and methods
+ * like {@link #poll()} must resolve any races by attempting to delete the underlying node.
+ */
+ private TreeSet<String> knownChildren = new TreeSet<>();
+
+ /**
+ * Used to wait on ZK changes to the child list; you must hold {@link #updateLock} before waiting on this condition.
+ */
+ private final Condition changed = updateLock.newCondition();
+
+ private boolean isDirty = true;
+
+ private int watcherCount = 0;
+
+ private final int maxQueueSize;
+
+ /**
+ * If {@link #maxQueueSize} is set, the number of items we can queue without rechecking the server.
+ */
+ private final AtomicInteger offerPermits = new AtomicInteger(0);
+
+ public GenericDistributedQueue(DistribStateManager stateManager, String dir) {
+ this(stateManager, dir, new Stats());
+ }
+
+ public GenericDistributedQueue(DistribStateManager stateManager, String dir, Stats stats) {
+ this(stateManager, dir, stats, 0);
+ }
+
+ public GenericDistributedQueue(DistribStateManager stateManager, String dir, Stats stats, int maxQueueSize) {
+ this.dir = dir;
+
+ try {
+ if (!stateManager.hasData(dir)) {
+ try {
+ stateManager.makePath(dir);
+ } catch (AlreadyExistsException e) {
+ // ignore
+ }
+ }
+ } catch (IOException | KeeperException e) {
+ throw new SolrException(ErrorCode.SERVER_ERROR, e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new SolrException(ErrorCode.SERVER_ERROR, e);
+ }
+
+ this.stateManager = stateManager;
+ this.stats = stats;
+ this.maxQueueSize = maxQueueSize;
+ }
+
+ /**
+ * Returns the data at the first element of the queue, or null if the queue is
+ * empty.
+ *
+ * @return data at the first element of the queue, or null.
+ */
+ @Override
+ public byte[] peek() throws Exception {
+ Timer.Context time = stats.time(dir + "_peek");
+ try {
+ return firstElement();
+ } finally {
+ time.stop();
+ }
+ }
+
+ /**
+ * Returns the data at the first element of the queue, or null if the queue is
+ * empty and block is false.
+ *
+ * @param block if true, blocks until an element enters the queue
+ * @return data at the first element of the queue, or null.
+ */
+ @Override
+ public byte[] peek(boolean block) throws Exception {
+ return block ? peek(Long.MAX_VALUE) : peek();
+ }
+
+ /**
+ * Returns the data at the first element of the queue, or null if the queue is
+ * empty after wait ms.
+ *
+ * @param wait max wait time in ms.
+ * @return data at the first element of the queue, or null.
+ */
+ @Override
+ public byte[] peek(long wait) throws Exception {
+ Preconditions.checkArgument(wait > 0);
+ Timer.Context time;
+ if (wait == Long.MAX_VALUE) {
+ time = stats.time(dir + "_peek_wait_forever");
+ } else {
+ time = stats.time(dir + "_peek_wait" + wait);
+ }
+ updateLock.lockInterruptibly();
+ try {
+ long waitNanos = TimeUnit.MILLISECONDS.toNanos(wait);
+ while (waitNanos > 0) {
+ byte[] result = firstElement();
+ if (result != null) {
+ return result;
+ }
+ waitNanos = changed.awaitNanos(waitNanos);
+ }
+ return null;
+ } finally {
+ updateLock.unlock();
+ time.stop();
+ }
+ }
+
+ /**
+ * Attempts to remove the head of the queue and return it. Returns null if the
+ * queue is empty.
+ *
+ * @return Head of the queue or null.
+ */
+ @Override
+ public byte[] poll() throws Exception {
+ Timer.Context time = stats.time(dir + "_poll");
+ try {
+ return removeFirst();
+ } finally {
+ time.stop();
+ }
+ }
+
+ /**
+ * Attempts to remove the head of the queue and return it.
+ *
+ * @return The former head of the queue
+ */
+ @Override
+ public byte[] remove() throws Exception {
+ Timer.Context time = stats.time(dir + "_remove");
+ try {
+ byte[] result = removeFirst();
+ if (result == null) {
+ throw new NoSuchElementException();
+ }
+ return result;
+ } finally {
+ time.stop();
+ }
+ }
+
+ public void remove(Collection<String> paths) throws Exception {
+ if (paths.isEmpty()) return;
+ List<Op> ops = new ArrayList<>();
+ for (String path : paths) {
+ ops.add(Op.delete(dir + "/" + path, -1));
+ }
+ for (int from = 0; from < ops.size(); from += 1000) {
+ int to = Math.min(from + 1000, ops.size());
+ if (from < to) {
+ try {
+ stateManager.multi(ops.subList(from, to));
+ } catch (NoSuchElementException e) {
+ // don't know which nodes are not exist, so try to delete one by one node
+ for (int j = from; j < to; j++) {
+ try {
+ stateManager.removeData(ops.get(j).getPath(), -1);
+ } catch (NoSuchElementException e2) {
+ LOG.debug("Can not remove node which is not exist : " + ops.get(j).getPath());
+ }
+ }
+ }
+ }
+ }
+
+ int cacheSizeBefore = knownChildren.size();
+ knownChildren.removeAll(paths);
+ if (cacheSizeBefore - paths.size() == knownChildren.size() && knownChildren.size() != 0) {
+ stats.setQueueLength(knownChildren.size());
+ } else {
+ // There are elements get deleted but not present in the cache,
+ // the cache seems not valid anymore
+ knownChildren.clear();
+ isDirty = true;
+ }
+ }
+
+ /**
+ * Removes the head of the queue and returns it, blocks until it succeeds.
+ *
+ * @return The former head of the queue
+ */
+ @Override
+ public byte[] take() throws Exception {
+ // Same as for element. Should refactor this.
+ Timer.Context timer = stats.time(dir + "_take");
+ updateLock.lockInterruptibly();
+ try {
+ while (true) {
+ byte[] result = removeFirst();
+ if (result != null) {
+ return result;
+ }
+ changed.await();
+ }
+ } finally {
+ updateLock.unlock();
+ timer.stop();
+ }
+ }
+
+ /**
+ * Inserts data into queue. If there are no other queue consumers, the offered element
+ * will be immediately visible when this method returns.
+ */
+ @Override
+ public void offer(byte[] data) throws Exception {
+ Timer.Context time = stats.time(dir + "_offer");
+ try {
+ while (true) {
+ try {
+ if (maxQueueSize > 0) {
+ if (offerPermits.get() <= 0 || offerPermits.getAndDecrement() <= 0) {
+ // If a max queue size is set, check it before creating a new queue item.
+ if (!stateManager.hasData(dir)) {
+ // jump to the code below, which tries to create dir if it doesn't exist
+ throw new NoSuchElementException();
+ }
+ List<String> children = stateManager.listData(dir);
+ int remainingCapacity = maxQueueSize - children.size();
+ if (remainingCapacity <= 0) {
+ throw new IllegalStateException("queue is full");
+ }
+
+ // Allow this client to push up to 1% of the remaining queue capacity without rechecking.
+ offerPermits.set(remainingCapacity / 100);
+ }
+ }
+
+ // Explicitly set isDirty here so that synchronous same-thread calls behave as expected.
+ // This will get set again when the watcher actually fires, but that's ok.
+ stateManager.createData(dir + "/" + PREFIX, data, CreateMode.PERSISTENT_SEQUENTIAL);
+ isDirty = true;
+ return;
+ } catch (NoSuchElementException e) {
+ try {
+ stateManager.createData(dir, new byte[0], CreateMode.PERSISTENT);
+ } catch (NoSuchElementException ne) {
+ // someone created it
+ }
+ }
+ }
+ } finally {
+ time.stop();
+ }
+ }
+
+ public Stats getZkStats() {
+ return stats;
+ }
+
+ @Override
+ public Map<String, Object> getStats() {
+ if (stats == null) {
+ return Collections.emptyMap();
+ }
+ Map<String, Object> res = new HashMap<>();
+ res.put("queueLength", stats.getQueueLength());
+ final Map<String, Object> statsMap = new HashMap<>();
+ res.put("stats", statsMap);
+ stats.getStats().forEach((op, stat) -> {
+ final Map<String, Object> statMap = new HashMap<>();
+ statMap.put("success", stat.success.get());
+ statMap.put("errors", stat.errors.get());
+ final List<Map<String, Object>> failed = new ArrayList<>(stat.failureDetails.size());
+ statMap.put("failureDetails", failed);
+ stat.failureDetails.forEach(failedOp -> {
+ Map<String, Object> fo = new HashMap<>();
+ fo.put("req", failedOp.req);
+ fo.put("resp", failedOp.resp);
+ });
+ statsMap.put(op, statMap);
+ });
+ return res;
+ }
+
+ /**
+ * Returns the name if the first known child node, or {@code null} if the queue is empty.
+ * This is the only place {@link #knownChildren} is ever updated!
+ * The caller must double check that the actual node still exists, since the in-memory
+ * list is inherently stale.
+ */
+ private String firstChild(boolean remove, boolean refetchIfDirty) throws Exception {
+ updateLock.lockInterruptibly();
+ try {
+ // We always return from cache first, the cache will be cleared if the node is not exist
+ if (!knownChildren.isEmpty() && !(isDirty && refetchIfDirty)) {
+ return remove ? knownChildren.pollFirst() : knownChildren.first();
+ }
+
+ if (!isDirty && knownChildren.isEmpty()) {
+ return null;
+ }
+
+ // Dirty, try to fetch an updated list of children from ZK.
+ // Only set a new watcher if there isn't already a watcher.
+ ChildWatcher newWatcher = (watcherCount == 0) ? new ChildWatcher() : null;
+ knownChildren = fetchZkChildren(newWatcher);
+ if (newWatcher != null) {
+ watcherCount++; // watcher was successfully set
+ }
+ isDirty = false;
+ if (knownChildren.isEmpty()) {
+ return null;
+ }
+ changed.signalAll();
+ return remove ? knownChildren.pollFirst() : knownChildren.first();
+ } finally {
+ updateLock.unlock();
+ }
+ }
+
+ /**
+ * Return the current set of children from ZK; does not change internal state.
+ */
+ TreeSet<String> fetchZkChildren(Watcher watcher) throws Exception {
+ while (true) {
+ try {
+ TreeSet<String> orderedChildren = new TreeSet<>();
+
+ List<String> childNames = stateManager.listData(dir, watcher);
+ stats.setQueueLength(childNames.size());
+ for (String childName : childNames) {
+ // Check format
+ if (!childName.regionMatches(0, PREFIX, 0, PREFIX.length())) {
+ LOG.debug("Found child node with improper name: " + childName);
+ continue;
+ }
+ orderedChildren.add(childName);
+ }
+ return orderedChildren;
+ } catch (NoSuchElementException e) {
+ try {
+ stateManager.makePath(dir);
+ } catch (AlreadyExistsException e2) {
+ // ignore
+ }
+ // go back to the loop and try again
+ }
+ }
+ }
+
+ /**
+ * Return the currently-known set of elements, using child names from memory. If no children are found, or no
+ * children pass {@code acceptFilter}, waits up to {@code waitMillis} for at least one child to become available.
+ * <p>
+ * Package-private to support {@link OverseerTaskQueue} specifically.</p>
+ */
+ @Override
+ public Collection<Pair<String, byte[]>> peekElements(int max, long waitMillis, Predicate<String> acceptFilter) throws Exception {
+ List<String> foundChildren = new ArrayList<>();
+ long waitNanos = TimeUnit.MILLISECONDS.toNanos(waitMillis);
+ boolean first = true;
+ while (true) {
+ // Trigger a refresh, but only force it if this is not the first iteration.
+ firstChild(false, !first);
+
+ updateLock.lockInterruptibly();
+ try {
+ for (String child : knownChildren) {
+ if (acceptFilter.test(child)) {
+ foundChildren.add(child);
+ }
+ }
+ if (!foundChildren.isEmpty()) {
+ break;
+ }
+ if (waitNanos <= 0) {
+ break;
+ }
+
+ // If this is our first time through, force a refresh before waiting.
+ if (first) {
+ first = false;
+ continue;
+ }
+
+ waitNanos = changed.awaitNanos(waitNanos);
+ } finally {
+ updateLock.unlock();
+ }
+
+ if (!foundChildren.isEmpty()) {
+ break;
+ }
+ }
+
+ // Technically we could restart the method if we fail to actually obtain any valid children
+ // from ZK, but this is a super rare case, and the latency of the ZK fetches would require
+ // much more sophisticated waitNanos tracking.
+ List<Pair<String, byte[]>> result = new ArrayList<>();
+ for (String child : foundChildren) {
+ if (result.size() >= max) {
+ break;
+ }
+ try {
+ VersionedData data = stateManager.getData(dir + "/" + child);
+ result.add(new Pair<>(child, data.getData()));
+ } catch (NoSuchElementException e) {
+ // Another client deleted the node first, remove the in-memory and continue.
+ updateLock.lockInterruptibly();
+ try {
+ knownChildren.remove(child);
+ } finally {
+ updateLock.unlock();
+ }
+ }
+ }
+ return result;
+ }
+
+ /**
+ * Return the head of the queue without modifying the queue.
+ *
+ * @return the data at the head of the queue.
+ */
+ private byte[] firstElement() throws Exception {
+ while (true) {
+ String firstChild = firstChild(false, false);
+ if (firstChild == null) {
+ return null;
+ }
+ try {
+ VersionedData data = stateManager.getData(dir + "/" + firstChild);
+ return data != null ? data.getData() : null;
+ } catch (NoSuchElementException e) {
+ // Another client deleted the node first, remove the in-memory and retry.
+ updateLock.lockInterruptibly();
+ try {
+ // Efficient only for single-consumer
+ knownChildren.clear();
+ isDirty = true;
+ } finally {
+ updateLock.unlock();
+ }
+ }
+ }
+ }
+
+ private byte[] removeFirst() throws Exception {
+ while (true) {
+ String firstChild = firstChild(true, false);
+ if (firstChild == null) {
+ return null;
+ }
+ try {
+ String path = dir + "/" + firstChild;
+ VersionedData result = stateManager.getData(path);
+ stateManager.removeData(path, -1);
+ stats.setQueueLength(knownChildren.size());
+ return result.getData();
+ } catch (NoSuchElementException e) {
+ // Another client deleted the node first, remove the in-memory and retry.
+ updateLock.lockInterruptibly();
+ try {
+ // Efficient only for single-consumer
+ knownChildren.clear();
+ isDirty = true;
+ } finally {
+ updateLock.unlock();
+ }
+ }
+ }
+ }
+
+ @VisibleForTesting int watcherCount() throws InterruptedException {
+ updateLock.lockInterruptibly();
+ try {
+ return watcherCount;
+ } finally {
+ updateLock.unlock();
+ }
+ }
+
+ @VisibleForTesting boolean isDirty() throws InterruptedException {
+ updateLock.lockInterruptibly();
+ try {
+ return isDirty;
+ } finally {
+ updateLock.unlock();
+ }
+ }
+
+ @VisibleForTesting class ChildWatcher implements Watcher {
+
+ @Override
+ public void process(WatchedEvent event) {
+ // session events are not change events, and do not remove the watcher; except for Expired
+ if (Event.EventType.None.equals(event.getType()) && !Event.KeeperState.Expired.equals(event.getState())) {
+ return;
+ }
+ updateLock.lock();
+ try {
+ isDirty = true;
+ watcherCount--;
+ // optimistically signal any waiters that the queue may not be empty now, so they can wake up and retry
+ changed.signalAll();
+ } finally {
+ updateLock.unlock();
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/23df975b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/GenericDistributedQueueFactory.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/GenericDistributedQueueFactory.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/GenericDistributedQueueFactory.java
new file mode 100644
index 0000000..d06698c
--- /dev/null
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/GenericDistributedQueueFactory.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.solr.cloud.autoscaling.sim;
+
+import java.io.IOException;
+
+import org.apache.solr.client.solrj.cloud.DistributedQueue;
+import org.apache.solr.client.solrj.cloud.DistributedQueueFactory;
+import org.apache.solr.client.solrj.cloud.autoscaling.DistribStateManager;
+
+/**
+ *
+ */
+public class GenericDistributedQueueFactory implements DistributedQueueFactory {
+
+ private final DistribStateManager stateManager;
+
+ public GenericDistributedQueueFactory(DistribStateManager stateManager) {
+ this.stateManager = stateManager;
+ }
+
+ @Override
+ public DistributedQueue makeQueue(String path) throws IOException {
+ return new GenericDistributedQueue(stateManager, path);
+ }
+
+ @Override
+ public void removeQueue(String path) throws IOException {
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/23df975b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimCloudManager.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimCloudManager.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimCloudManager.java
index 1da4a5f..4a37e35 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimCloudManager.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimCloudManager.java
@@ -91,7 +91,7 @@ public class SimCloudManager implements SolrCloudManager {
private final SimNodeStateProvider nodeStateProvider;
private final AutoScalingHandler autoScalingHandler;
private final Set<String> liveNodes = ConcurrentHashMap.newKeySet();
- private final SimDistributedQueueFactory queueFactory;
+ private final DistributedQueueFactory queueFactory;
private final ObjectCache objectCache = new ObjectCache();
private TimeSource timeSource;
private SolrClient solrClient;
@@ -120,8 +120,8 @@ public class SimCloudManager implements SolrCloudManager {
this.timeSource = timeSource != null ? timeSource : TimeSource.NANO_TIME;
this.clusterStateProvider = new SimClusterStateProvider(liveNodes, this);
this.nodeStateProvider = new SimNodeStateProvider(liveNodes, this.stateManager, this.clusterStateProvider, null);
- this.queueFactory = new SimDistributedQueueFactory();
- this.simCloudManagerPool = ExecutorUtil.newMDCAwareFixedThreadPool(50, new DefaultSolrThreadFactory("simCloudManagerPool"));
+ this.queueFactory = new GenericDistributedQueueFactory(stateManager);
+ this.simCloudManagerPool = ExecutorUtil.newMDCAwareFixedThreadPool(200, new DefaultSolrThreadFactory("simCloudManagerPool"));
this.autoScalingHandler = new AutoScalingHandler(this, new SolrResourceLoader());
ThreadGroup triggerThreadGroup = new ThreadGroup("Simulated Overseer autoscaling triggers");
OverseerTriggerThread trigger = new OverseerTriggerThread(new SolrResourceLoader(), this,
@@ -140,9 +140,9 @@ public class SimCloudManager implements SolrCloudManager {
SimCloudManager cloudManager = new SimCloudManager(timeSource);
for (int i = 1; i <= numNodes; i++) {
Map<String, Object> values = createNodeValues(null);
- if (i == 1) { // designated Overseer
+// if (i == 1) { // designated Overseer ?
//values.put(ImplicitSnitch.NODEROLE, "overseer");
- }
+// }
String nodeId = (String)values.get(ImplicitSnitch.NODE);
cloudManager.getSimClusterStateProvider().simAddNode(nodeId);
cloudManager.getSimNodeStateProvider().simSetNodeValues(nodeId, values);
@@ -211,14 +211,14 @@ public class SimCloudManager implements SolrCloudManager {
String nodeId = (String)values.get(ImplicitSnitch.NODE);
clusterStateProvider.simAddNode(nodeId);
nodeStateProvider.simSetNodeValues(nodeId, values);
- LOG.debug("-- added node " + nodeId);
+ LOG.trace("-- added node " + nodeId);
return nodeId;
}
public void simRemoveNode(String nodeId) throws Exception {
clusterStateProvider.simRemoveNode(nodeId);
nodeStateProvider.simRemoveNodeValues(nodeId);
- LOG.debug("-- removed node " + nodeId);
+ LOG.trace("-- removed node " + nodeId);
}
public void simRemoveRandomNodes(int number, Random random) throws Exception {
@@ -274,10 +274,6 @@ public class SimCloudManager implements SolrCloudManager {
return stateManager;
}
- public SimDistributedQueueFactory getSimDistributedQueueFactory() {
- return queueFactory;
- }
-
// --------- interface methods -----------
@@ -330,7 +326,10 @@ public class SimCloudManager implements SolrCloudManager {
}
public SolrResponse simHandleSolrRequest(SolrRequest req) throws IOException, InterruptedException {
- LOG.debug("--- got SolrRequest: " + req.getMethod() + " " + req.getPath() +
+ // pay the penalty for remote request, at least 5 ms
+ timeSource.sleep(5);
+
+ LOG.trace("--- got SolrRequest: " + req.getMethod() + " " + req.getPath() +
(req.getParams() != null ? " " + req.getParams().toQueryString() : ""));
if (req.getPath() != null && req.getPath().startsWith("/admin/autoscaling") ||
req.getPath().startsWith("/cluster/autoscaling")) {
@@ -341,7 +340,7 @@ public class SimCloudManager implements SolrCloudManager {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
cw.write(baos);
String payload = baos.toString("UTF-8");
- LOG.debug("-- payload: " + payload);
+ LOG.trace("-- payload: {}", payload);
queryRequest.setContentStreams(Collections.singletonList(new ContentStreamBase.StringStream(payload)));
queryRequest.getContext().put("httpMethod", req.getMethod().toString());
SolrQueryResponse queryResponse = new SolrQueryResponse();
@@ -351,7 +350,7 @@ public class SimCloudManager implements SolrCloudManager {
}
SolrResponse rsp = new SolrResponseBase();
rsp.setResponse(queryResponse.getValues());
- LOG.debug("-- response: " + rsp);
+ LOG.trace("-- response: {}", rsp);
return rsp;
}
if (req instanceof UpdateRequest) {
@@ -379,7 +378,7 @@ public class SimCloudManager implements SolrCloudManager {
if (action == null) {
throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "Unknown action: " + a);
}
- LOG.info("Invoking Collection Action :{} with params {}", action.toLower(), req.getParams().toQueryString());
+ LOG.debug("Invoking Collection Action :{} with params {}", action.toLower(), req.getParams().toQueryString());
NamedList results = new NamedList();
rsp.setResponse(results);
switch (action) {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/23df975b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java
index 7906b01..b5b91a2 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java
@@ -32,6 +32,7 @@ import java.util.Random;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
@@ -74,7 +75,7 @@ import static org.apache.solr.common.cloud.ZkStateReader.SHARD_ID_PROP;
import static org.apache.solr.common.params.CommonParams.NAME;
/**
- *
+ * Simulated {@link ClusterStateProvider}.
*/
public class SimClusterStateProvider implements ClusterStateProvider {
private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
@@ -92,15 +93,17 @@ public class SimClusterStateProvider implements ClusterStateProvider {
private final ActionThrottle leaderThrottle;
- // op -> delay
+ // default map of: operation -> delay
private final Map<String, Long> defaultOpDelays = new HashMap<>();
- // collection -> op -> delay
+ // per-collection map of: collection -> op -> delay
private final Map<String, Map<String, Long>> opDelays = new ConcurrentHashMap<>();
private ClusterState lastSavedState = null;
private Map<String, Object> lastSavedProperties = null;
+ private AtomicReference<Map<String, DocCollection>> collectionsStatesRef = new AtomicReference<>();
+
/**
* Zero-arg constructor. The instance needs to be initialized using the <code>sim*</code> methods in order
* to ensure proper behavior, otherwise it will behave as a cluster with zero live nodes and zero replicas.
@@ -110,13 +113,14 @@ public class SimClusterStateProvider implements ClusterStateProvider {
this.cloudManager = cloudManager;
this.stateManager = cloudManager.getDistribStateManager();
this.leaderThrottle = new ActionThrottle("leader", 5000, cloudManager.getTimeSource());
+ // names are CollectionAction names, delays are in ms (simulated time)
defaultOpDelays.put(CollectionParams.CollectionAction.MOVEREPLICA.name(), 5000L);
defaultOpDelays.put(CollectionParams.CollectionAction.DELETEREPLICA.name(), 5000L);
- defaultOpDelays.put(CollectionParams.CollectionAction.ADDREPLICA.name(), 5000L);
+ defaultOpDelays.put(CollectionParams.CollectionAction.ADDREPLICA.name(), 500L);
defaultOpDelays.put(CollectionParams.CollectionAction.SPLITSHARD.name(), 5000L);
defaultOpDelays.put(CollectionParams.CollectionAction.CREATESHARD.name(), 5000L);
defaultOpDelays.put(CollectionParams.CollectionAction.DELETESHARD.name(), 5000L);
- defaultOpDelays.put(CollectionParams.CollectionAction.CREATE.name(), 5000L);
+ defaultOpDelays.put(CollectionParams.CollectionAction.CREATE.name(), 500L);
defaultOpDelays.put(CollectionParams.CollectionAction.DELETE.name(), 5000L);
}
@@ -168,6 +172,8 @@ public class SimClusterStateProvider implements ClusterStateProvider {
return nodeReplicaMap.putIfAbsent(nodeId, new ArrayList<>()) == null;
}
+ // utility to run leader election in a separate thread and with throttling
+ // Note: leader election is a no-op if a shard leader already exists for each shard
private class LeaderElection implements Callable<Boolean> {
Collection<String> collections;
boolean saveClusterState;
@@ -205,7 +211,9 @@ public class SimClusterStateProvider implements ClusterStateProvider {
});
}
boolean res = liveNodes.remove(nodeId);
- cloudManager.submit(new LeaderElection(collections, true));
+ if (!collections.isEmpty()) {
+ cloudManager.submit(new LeaderElection(collections, true));
+ }
return res;
} finally {
lock.unlock();
@@ -250,6 +258,9 @@ public class SimClusterStateProvider implements ClusterStateProvider {
replicaInfo.getVariables().put(ZkStateReader.STATE_PROP, Replica.State.ACTIVE.toString());
// add property expected in tests
replicaInfo.getVariables().put(Suggestion.coreidxsize, 123450000);
+ // at this point nuke our cached DocCollection state
+ collectionsStatesRef.set(null);
+
replicas.add(replicaInfo);
Map<String, Object> values = cloudManager.getSimNodeStateProvider().simGetAllNodeValues()
.computeIfAbsent(nodeId, id -> new ConcurrentHashMap<>(SimCloudManager.createNodeValues(id)));
@@ -297,6 +308,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
}
private ClusterState saveClusterState() throws IOException {
+ collectionsStatesRef.set(null);
ClusterState currentState = getClusterState();
if (lastSavedState != null && lastSavedState.equals(currentState)) {
return lastSavedState;
@@ -322,10 +334,9 @@ public class SimClusterStateProvider implements ClusterStateProvider {
}
private synchronized void simRunLeaderElection(Collection<String> collections, boolean saveClusterState) throws Exception {
- if (saveClusterState) {
- saveClusterState();
- }
ClusterState state = getClusterState();
+ AtomicBoolean stateChanged = new AtomicBoolean(Boolean.FALSE);
+
state.forEachCollection(dc -> {
if (!collections.contains(dc.getName())) {
return;
@@ -333,7 +344,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
dc.getSlices().forEach(s -> {
Replica leader = s.getLeader();
if (leader == null || !liveNodes.contains(leader.getNodeName())) {
- LOG.info("Running leader election for " + dc.getName() + " / " + s.getName());
+ LOG.trace("Running leader election for " + dc.getName() + " / " + s.getName());
if (s.getReplicas().isEmpty()) { // no replicas - punt
return;
}
@@ -351,7 +362,9 @@ public class SimClusterStateProvider implements ClusterStateProvider {
if (ri == null) {
throw new IllegalStateException("-- could not find ReplicaInfo for replica " + r);
}
- ri.getVariables().remove(ZkStateReader.LEADER_PROP);
+ if (ri.getVariables().remove(ZkStateReader.LEADER_PROP) != null) {
+ stateChanged.set(true);
+ }
if (r.isActive(liveNodes)) {
active.add(ri);
} else { // if it's on a node that is not live mark it down
@@ -377,12 +390,16 @@ public class SimClusterStateProvider implements ClusterStateProvider {
return;
}
ri.getVariables().put(ZkStateReader.LEADER_PROP, "true");
+ stateChanged.set(true);
LOG.info("-- elected new leader for " + dc.getName() + " / " + s.getName() + ": " + ri);
} else {
-
+ LOG.trace("-- already has leader for {} / {}", dc.getName(), s.getName());
}
});
});
+ if (saveClusterState || stateChanged.get()) {
+ saveClusterState();
+ }
}
public void simCreateCollection(ZkNodeProps props, NamedList results) throws Exception {
@@ -422,6 +439,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
throw new RuntimeException(e);
}
});
+ collectionsStatesRef.set(null);
// add collection props
DocCollection coll = cmd.collection;
collProperties.computeIfAbsent(collectionName, c -> new ConcurrentHashMap<>()).putAll(coll.getProperties());
@@ -833,6 +851,10 @@ public class SimClusterStateProvider implements ClusterStateProvider {
}
private Map<String, DocCollection> getCollectionStates() {
+ Map<String, DocCollection> collectionStates = collectionsStatesRef.get();
+ if (collectionStates != null) {
+ return collectionStates;
+ }
lock.lock();
try {
Map<String, Map<String, Map<String, Replica>>> collMap = new HashMap<>();
@@ -853,6 +875,10 @@ public class SimClusterStateProvider implements ClusterStateProvider {
collMap.computeIfAbsent(c, co -> new ConcurrentHashMap<>()).computeIfAbsent(slice, s -> new ConcurrentHashMap<>());
});
});
+ // add empty collections
+ collProperties.keySet().forEach(c -> {
+ collMap.computeIfAbsent(c, co -> new ConcurrentHashMap<>());
+ });
Map<String, DocCollection> res = new HashMap<>();
collMap.forEach((coll, shards) -> {
@@ -866,6 +892,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
DocCollection dc = new DocCollection(coll, slices, collProps, DocRouter.DEFAULT, 0, ZkStateReader.CLUSTER_STATE);
res.put(coll, dc);
});
+ collectionsStatesRef.set(res);
return res;
} finally {
lock.unlock();
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/23df975b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimDistribStateManager.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimDistribStateManager.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimDistribStateManager.java
index 2e45312..a570e6d 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimDistribStateManager.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimDistribStateManager.java
@@ -78,15 +78,14 @@ public class SimDistribStateManager implements DistribStateManager {
private Map<String, Node> children = new ConcurrentHashMap<>();
Set<Watcher> dataWatches = ConcurrentHashMap.newKeySet();
Set<Watcher> childrenWatches = ConcurrentHashMap.newKeySet();
- private ExecutorService watcherPool;
- Node(Node parent, String name, String path, CreateMode mode, String clientId, ExecutorService watcherPool) {
+ Node(Node parent, String name, String path, CreateMode mode, String clientId) {
this.parent = parent;
this.name = name;
this.path = path;
this.mode = mode;
this.clientId = clientId;
- this.watcherPool = watcherPool;
+
}
public void clear() {
@@ -103,14 +102,6 @@ public class SimDistribStateManager implements DistribStateManager {
}
}
- private void runTask(Runnable r) {
- if (watcherPool != null) {
- watcherPool.submit(r);
- } else {
- r.run();
- }
- }
-
public void setData(byte[] data, int version) throws BadVersionException, IOException {
Set<Watcher> currentWatchers = new HashSet<>(dataWatches);
dataLock.lock();
@@ -128,11 +119,9 @@ public class SimDistribStateManager implements DistribStateManager {
} finally {
dataLock.unlock();
}
- runTask(() -> {
- for (Watcher w : currentWatchers) {
- w.process(new WatchedEvent(Watcher.Event.EventType.NodeDataChanged, Watcher.Event.KeeperState.SyncConnected, path));
- }
- });
+ for (Watcher w : currentWatchers) {
+ w.process(new WatchedEvent(Watcher.Event.EventType.NodeDataChanged, Watcher.Event.KeeperState.SyncConnected, path));
+ }
}
public VersionedData getData(Watcher w) {
@@ -158,11 +147,9 @@ public class SimDistribStateManager implements DistribStateManager {
} finally {
dataLock.unlock();
}
- runTask(() -> {
- for (Watcher w : currentWatchers) {
- w.process(new WatchedEvent(Watcher.Event.EventType.NodeChildrenChanged, Watcher.Event.KeeperState.SyncConnected, path));
- }
- });
+ for (Watcher w : currentWatchers) {
+ w.process(new WatchedEvent(Watcher.Event.EventType.NodeChildrenChanged, Watcher.Event.KeeperState.SyncConnected, path));
+ }
}
public void removeChild(String name, int version) throws NoSuchElementException, BadVersionException, IOException {
@@ -174,20 +161,16 @@ public class SimDistribStateManager implements DistribStateManager {
throw new BadVersionException(version, path);
}
children.remove(name);
- Set<Watcher> currentChildrenWatchers = new HashSet<>(childrenWatches);
+ Set<Watcher> currentWatchers = new HashSet<>(childrenWatches);
childrenWatches.clear();
- runTask(() -> {
- for (Watcher w : currentChildrenWatchers) {
- w.process(new WatchedEvent(Watcher.Event.EventType.NodeChildrenChanged, Watcher.Event.KeeperState.SyncConnected, path));
- }
- });
- Set<Watcher> currentDataWatchers = new HashSet<>(n.dataWatches);
+ for (Watcher w : currentWatchers) {
+ w.process(new WatchedEvent(Watcher.Event.EventType.NodeChildrenChanged, Watcher.Event.KeeperState.SyncConnected, path));
+ }
+ currentWatchers = new HashSet<>(n.dataWatches);
n.dataWatches.clear();
- runTask(() -> {
- for (Watcher w : currentDataWatchers) {
- w.process(new WatchedEvent(Watcher.Event.EventType.NodeDeleted, Watcher.Event.KeeperState.SyncConnected, n.path));
- }
- });
+ for (Watcher w : currentWatchers) {
+ w.process(new WatchedEvent(Watcher.Event.EventType.NodeDeleted, Watcher.Event.KeeperState.SyncConnected, n.path));
+ }
// TODO: not sure if it's correct to recurse and fire watches???
Set<String> kids = new HashSet<>(n.children.keySet());
for (String kid : kids) {
@@ -218,7 +201,7 @@ public class SimDistribStateManager implements DistribStateManager {
private static final ReentrantLock multiLock = new ReentrantLock();
public static Node createNewRootNode() {
- return new Node(null, "", "/", CreateMode.PERSISTENT, "__root__", null);
+ return new Node(null, "", "/", CreateMode.PERSISTENT, "__root__");
}
private final ExecutorService watchersPool;
@@ -299,7 +282,7 @@ public class SimDistribStateManager implements DistribStateManager {
parentNode.seq++;
}
currentPath.append(currentName);
- n = new Node(parentNode, currentName, currentPath.toString(), mode, id, watchersPool);
+ n = new Node(parentNode, currentName, currentPath.toString(), mode, id);
parentNode.setChild(currentName, n);
} else {
break;
@@ -323,9 +306,7 @@ public class SimDistribStateManager implements DistribStateManager {
} finally {
multiLock.unlock();
}
- if (watchersPool != null) {
- watchersPool.shutdownNow();
- }
+
}
@Override
@@ -355,17 +336,40 @@ public class SimDistribStateManager implements DistribStateManager {
}
@Override
+ public List<String> listData(String path, Watcher watcher) throws NoSuchElementException, IOException {
+ Node n;
+ List<String> res;
+ multiLock.lock();
+ try {
+ n = traverse(path, false, CreateMode.PERSISTENT);
+ if (n == null) {
+ throw new NoSuchElementException(path);
+ }
+ res = new ArrayList<>(n.children.keySet());
+ Collections.sort(res);
+ } finally {
+ multiLock.unlock();
+ }
+ if (watcher != null) {
+ n.dataWatches.add(watcher);
+ n.childrenWatches.add(watcher);
+ }
+ return res;
+ }
+
+ @Override
public VersionedData getData(String path, Watcher watcher) throws NoSuchElementException, IOException {
+ Node n = null;
multiLock.lock();
try {
- Node n = traverse(path, false, CreateMode.PERSISTENT);
+ n = traverse(path, false, CreateMode.PERSISTENT);
if (n == null) {
throw new NoSuchElementException(path);
}
- return n.getData(watcher);
} finally {
multiLock.unlock();
}
+ return n.getData(watcher);
}
@Override
@@ -380,20 +384,21 @@ public class SimDistribStateManager implements DistribStateManager {
@Override
public void makePath(String path, byte[] data, CreateMode createMode, boolean failOnExists) throws AlreadyExistsException, IOException, KeeperException, InterruptedException {
+ Node n = null;
multiLock.lock();
try {
if (failOnExists && hasData(path)) {
throw new AlreadyExistsException(path);
}
- Node n = traverse(path, true, createMode);
- try {
- n.setData(data, -1);
- } catch (BadVersionException e) {
- throw new IOException("should not happen!", e);
- }
+ n = traverse(path, true, createMode);
} finally {
multiLock.unlock();
}
+ try {
+ n.setData(data, -1);
+ } catch (BadVersionException e) {
+ throw new IOException("should not happen!", e);
+ }
}
@Override
@@ -414,16 +419,19 @@ public class SimDistribStateManager implements DistribStateManager {
throw new NoSuchElementException(sb.toString());
}
}
+ Node n = null;
multiLock.lock();
try {
- Node n = traverse(path, true, mode);
+ n = traverse(path, true, mode);
+ } finally {
+ multiLock.unlock();
+ }
+ try {
n.setData(data, -1);
return n.path;
} catch (BadVersionException e) {
// not happening
return null;
- } finally {
- multiLock.unlock();
}
}
@@ -449,16 +457,16 @@ public class SimDistribStateManager implements DistribStateManager {
@Override
public void setData(String path, byte[] data, int version) throws NoSuchElementException, BadVersionException, IOException {
multiLock.lock();
+ Node n = null;
try {
- Node n = traverse(path, false, CreateMode.PERSISTENT);
+ n = traverse(path, false, CreateMode.PERSISTENT);
if (n == null) {
throw new NoSuchElementException(path);
}
- n.setData(data, version);
} finally {
multiLock.unlock();
}
-
+ n.setData(data, version);
}
@Override
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/23df975b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimDistributedQueueFactory.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimDistributedQueueFactory.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimDistributedQueueFactory.java
index d105e20..ea00e6c 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimDistributedQueueFactory.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimDistributedQueueFactory.java
@@ -46,7 +46,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- *
+ * Simulated {@link DistributedQueueFactory} that keeps all data in memory.
*/
public class SimDistributedQueueFactory implements DistributedQueueFactory {
private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
@@ -188,7 +188,7 @@ public class SimDistributedQueueFactory implements DistributedQueueFactory {
try {
queue.offer(new Pair(String.format(Locale.ROOT, "qn-%010d", seq), data));
seq++;
- LOG.debug("=== offer " + System.nanoTime());
+ LOG.trace("=== offer " + System.nanoTime());
changed.signalAll();
} finally {
updateLock.unlock();
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/23df975b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimNodeStateProvider.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimNodeStateProvider.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimNodeStateProvider.java
index a4943db..577c976 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimNodeStateProvider.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimNodeStateProvider.java
@@ -38,8 +38,8 @@ import org.slf4j.LoggerFactory;
/**
* Simulated {@link NodeStateProvider}.
- * Note: in order to set-up node-level metrics use {@link #simSetNodeValues(String, Map)}. However, in order
- * to set-up core-level metrics use {@link SimClusterStateProvider#simSetCollectionValue(String, String, Object, boolean)}.
+ * Note: in order to setup node-level metrics use {@link #simSetNodeValues(String, Map)}. However, in order
+ * to setup core-level metrics use {@link SimClusterStateProvider#simSetCollectionValue(String, String, Object, boolean)}.
*/
public class SimNodeStateProvider implements NodeStateProvider {
private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
@@ -182,7 +182,7 @@ public class SimNodeStateProvider implements NodeStateProvider {
@Override
public Map<String, Object> getNodeValues(String node, Collection<String> tags) {
- LOG.trace("-- requested values for " + node + ": " + tags);
+ LOG.debug("-- requested values for " + node + ": " + tags);
if (!liveNodes.contains(node)) {
nodeValues.remove(node);
return Collections.emptyMap();
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/23df975b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimSolrCloudTestCase.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimSolrCloudTestCase.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimSolrCloudTestCase.java
index c428b35..d7e075b 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimSolrCloudTestCase.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimSolrCloudTestCase.java
@@ -100,18 +100,19 @@ public class SimSolrCloudTestCase extends SolrTestCaseJ4 {
* @param collection the collection to watch
* @param predicate a predicate to match against the collection state
*/
- protected void waitForState(String message, String collection, CollectionStatePredicate predicate) {
+ protected long waitForState(String message, String collection, CollectionStatePredicate predicate) {
AtomicReference<DocCollection> state = new AtomicReference<>();
AtomicReference<Set<String>> liveNodesLastSeen = new AtomicReference<>();
try {
- waitForState(collection, DEFAULT_TIMEOUT, TimeUnit.SECONDS, (n, c) -> {
+ return waitForState(collection, DEFAULT_TIMEOUT, TimeUnit.SECONDS, (n, c) -> {
state.set(c);
liveNodesLastSeen.set(n);
return predicate.matches(n, c);
});
} catch (Exception e) {
- fail(message + "\n" + e.getMessage() + "\nLive Nodes: " + liveNodesLastSeen.get() + "\nLast available state: " + state.get());
+ fail(message + "\n" + e.toString() + "\nLive Nodes: " + liveNodesLastSeen.get() + "\nLast available state: " + state.get());
}
+ return 0;
}
/**
@@ -124,21 +125,24 @@ public class SimSolrCloudTestCase extends SolrTestCaseJ4 {
* @param wait how long to wait
* @param unit the units of the wait parameter
* @param predicate the predicate to call on state changes
+ * @return number of milliseconds elapsed
* @throws InterruptedException on interrupt
* @throws TimeoutException on timeout
* @throws IOException on watcher register / unregister error
*/
- public void waitForState(final String collection, long wait, TimeUnit unit, CollectionStatePredicate predicate)
+ public long waitForState(final String collection, long wait, TimeUnit unit, CollectionStatePredicate predicate)
throws InterruptedException, TimeoutException, IOException {
TimeOut timeout = new TimeOut(wait, unit, cluster.getTimeSource());
while (!timeout.hasTimedOut()) {
ClusterState state = cluster.getClusterStateProvider().getClusterState();
DocCollection coll = state.getCollectionOrNull(collection);
- if (coll == null) { // already removed or does not exist
- return;
+ // due to the way we manage collections in SimClusterStateProvider a null here
+ // can mean that a collection is still being created but has no replicas
+ if (coll == null) { // does not yet exist?
+ continue;
}
if (predicate.matches(state.getLiveNodes(), coll)) {
- return;
+ return timeout.timeElapsed(TimeUnit.MILLISECONDS);
}
timeout.sleep(50);
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/23df975b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestClusterStateProvider.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestClusterStateProvider.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestClusterStateProvider.java
index 0ee3a10..38dd7dc 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestClusterStateProvider.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestClusterStateProvider.java
@@ -51,7 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- *
+ * This test compares the cluster state of a real cluster and a simulated one.
*/
public class TestClusterStateProvider extends SolrCloudTestCase {
private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/23df975b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestDistribStateManager.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestDistribStateManager.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestDistribStateManager.java
index 82d3444..a9c5140 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestDistribStateManager.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestDistribStateManager.java
@@ -43,7 +43,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- *
+ * This test compares a ZK-based {@link DistribStateManager} to the simulated one.
*/
public class TestDistribStateManager extends SolrTestCaseJ4 {
private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/23df975b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestExecutePlanAction.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestExecutePlanAction.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestExecutePlanAction.java
index 7045e6e..637f2d6 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestExecutePlanAction.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestExecutePlanAction.java
@@ -22,6 +22,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
@@ -104,8 +105,7 @@ public class TestExecutePlanAction extends SimSolrCloudTestCase {
create.setMaxShardsPerNode(1);
create.process(solrClient);
- waitForState("Timed out waiting for replicas of new collection to be active",
- collectionName, clusterShape(1, 2));
+ log.info("Collection ready after " + waitForState(collectionName, 120, TimeUnit.SECONDS, clusterShape(1, 2)) + "ms");
String sourceNodeName = cluster.getSimClusterStateProvider().simGetRandomNode(random());
ClusterState clusterState = cluster.getClusterStateProvider().getClusterState();
@@ -166,8 +166,7 @@ public class TestExecutePlanAction extends SimSolrCloudTestCase {
assertNotNull(response.get("success"));
}
- waitForState("Timed out waiting for replicas of new collection to be active",
- collectionName, clusterShape(1, 2));
+ log.info("Collection ready after " + waitForState(collectionName, 120, TimeUnit.SECONDS, clusterShape(1, 2)) + "ms");
}
@Test
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/23df975b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestGenericDistributedQueue.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestGenericDistributedQueue.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestGenericDistributedQueue.java
new file mode 100644
index 0000000..df2dacd
--- /dev/null
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestGenericDistributedQueue.java
@@ -0,0 +1,16 @@
+package org.apache.solr.cloud.autoscaling.sim;
+
+import org.apache.solr.client.solrj.cloud.DistributedQueue;
+import org.apache.solr.client.solrj.cloud.autoscaling.DistribStateManager;
+
+/**
+ *
+ */
+public class TestGenericDistributedQueue extends TestSimDistributedQueue {
+ DistribStateManager stateManager = new SimDistribStateManager();
+
+ @Override
+ protected DistributedQueue makeDistributedQueue(String dqZNode) throws Exception {
+ return new GenericDistributedQueue(stateManager, dqZNode);
+ }
+}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/23df975b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestLargeCluster.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestLargeCluster.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestLargeCluster.java
index 7769a5f..726f839 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestLargeCluster.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestLargeCluster.java
@@ -24,6 +24,7 @@ import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.solr.client.solrj.SolrClient;
@@ -60,7 +61,7 @@ public class TestLargeCluster extends SimSolrCloudTestCase {
public static final int SPEED = 50;
- public static final int NUM_NODES = 500;
+ public static final int NUM_NODES = 100;
static Map<String, List<CapturedEvent>> listenerEvents = new ConcurrentHashMap<>();
static AtomicInteger triggerFiredCount = new AtomicInteger();
@@ -83,7 +84,6 @@ public class TestLargeCluster extends SimSolrCloudTestCase {
triggerFiredCount.set(0);
listenerEvents.clear();
// clear any events or markers
- // todo: consider the impact of such cleanup on regular cluster restarts
removeChildren(ZkStateReader.SOLR_AUTOSCALING_EVENTS_PATH);
removeChildren(ZkStateReader.SOLR_AUTOSCALING_TRIGGER_STATE_PATH);
removeChildren(ZkStateReader.SOLR_AUTOSCALING_NODE_LOST_PATH);
@@ -155,28 +155,30 @@ public class TestLargeCluster extends SimSolrCloudTestCase {
// pick a few random nodes
List<String> nodes = new ArrayList<>();
- int limit = 100;
+ int limit = 30;
for (String node : cluster.getClusterStateProvider().getLiveNodes()) {
nodes.add(node);
+ if (nodes.size() > limit) {
+ break;
+ }
}
Collections.shuffle(nodes, random());
String collectionName = "testBasic";
CollectionAdminRequest.Create create = CollectionAdminRequest.createCollection(collectionName,
- "conf", 10, 5, 5, 5);
+ "conf", 2, 5, 5, 5);
create.setMaxShardsPerNode(1);
create.setCreateNodeSet(String.join(",", nodes));
create.process(solrClient);
- waitForState("Timed out waiting for replicas of new collection to be active",
- collectionName, clusterShape(10, 15));
+ log.info("Ready after " + waitForState(collectionName, 30 * nodes.size(), TimeUnit.SECONDS, clusterShape(2, 15)) + "ms");
+ int KILL_NODES = 8;
// kill off a number of nodes
- for (int i = 0; i < 20; i++) {
+ for (int i = 0; i < KILL_NODES; i++) {
cluster.simRemoveNode(nodes.get(i));
}
- waitForState("Timed out waiting for replicas of new collection to be active",
- collectionName, clusterShape(10, 15));
+ log.info("Ready after " + waitForState(collectionName, 90 * KILL_NODES, TimeUnit.SECONDS, clusterShape(2, 15)) + "ms");
log.info(cluster.simGetSystemCollection().toString());
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/23df975b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestPolicyCloud.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestPolicyCloud.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestPolicyCloud.java
index d6255e9..feb7567 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestPolicyCloud.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestPolicyCloud.java
@@ -54,7 +54,6 @@ import org.slf4j.LoggerFactory;
import static org.apache.solr.cloud.autoscaling.AutoScalingHandlerTest.createAutoScalingRequest;
-@LuceneTestCase.Slow
public class TestPolicyCloud extends SimSolrCloudTestCase {
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/23df975b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestSimDistributedQueue.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestSimDistributedQueue.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestSimDistributedQueue.java
index e5d4e47..1c41795 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestSimDistributedQueue.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestSimDistributedQueue.java
@@ -40,7 +40,7 @@ public class TestSimDistributedQueue extends SolrTestCaseJ4 {
@Test
public void testDistributedQueue() throws Exception {
- String dqZNode = "/distqueue/test";
+ String dqZNode = "/distqueue/test1";
byte[] data = "hello world".getBytes(UTF8);
DistributedQueue dq = makeDistributedQueue(dqZNode);
@@ -83,7 +83,7 @@ public class TestSimDistributedQueue extends SolrTestCaseJ4 {
@Test
public void testDistributedQueueBlocking() throws Exception {
- String dqZNode = "/distqueue/test";
+ String dqZNode = "/distqueue/test2";
String testData = "hello world";
DistributedQueue dq = makeDistributedQueue(dqZNode);
@@ -120,7 +120,7 @@ public class TestSimDistributedQueue extends SolrTestCaseJ4 {
@Test
public void testLocallyOffer() throws Exception {
- String dqZNode = "/distqueue/test";
+ String dqZNode = "/distqueue/test3";
DistributedQueue dq = makeDistributedQueue(dqZNode);
dq.peekElements(1, 1, s -> true);
for (int i = 0; i < 100; i++) {
@@ -135,7 +135,7 @@ public class TestSimDistributedQueue extends SolrTestCaseJ4 {
@Test
public void testPeekElements() throws Exception {
- String dqZNode = "/distqueue/test";
+ String dqZNode = "/distqueue/test4";
byte[] data = "hello world".getBytes(UTF8);
DistributedQueue dq = makeDistributedQueue(dqZNode);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/23df975b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestTriggerIntegration.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestTriggerIntegration.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestTriggerIntegration.java
index fb3b0bc..3f6a656 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestTriggerIntegration.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestTriggerIntegration.java
@@ -601,6 +601,7 @@ public class TestTriggerIntegration extends SimSolrCloudTestCase {
public static long eventQueueActionWait = 5000;
+ // simulation framework doesn't support overseer
/*
@Test
public void testEventQueue() throws Exception {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/23df975b/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 fb5790c..26aaead 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
@@ -38,6 +38,8 @@ public interface DistribStateManager extends SolrCloseable {
List<String> listData(String path) throws NoSuchElementException, IOException, KeeperException, InterruptedException;
+ List<String> listData(String path, Watcher watcher) throws NoSuchElementException, IOException, KeeperException, InterruptedException;
+
VersionedData getData(String path, Watcher watcher) throws NoSuchElementException, IOException, KeeperException, InterruptedException;
default VersionedData getData(String path) throws NoSuchElementException, IOException, KeeperException, InterruptedException {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/23df975b/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 94c0a5a..b0cc710 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
@@ -61,9 +61,9 @@ public class ZkDistribStateManager implements DistribStateManager {
}
@Override
- public List<String> listData(String path) throws NoSuchElementException, IOException, KeeperException, InterruptedException {
+ public List<String> listData(String path, Watcher watcher) throws NoSuchElementException, IOException, KeeperException, InterruptedException {
try {
- return zkClient.getChildren(path, null, true);
+ return zkClient.getChildren(path, watcher, true);
} catch (KeeperException.NoNodeException e) {
throw new NoSuchElementException(path);
} catch (InterruptedException e) {
@@ -72,6 +72,11 @@ public class ZkDistribStateManager implements DistribStateManager {
}
@Override
+ public List<String> listData(String path) throws NoSuchElementException, IOException, KeeperException, InterruptedException {
+ return listData(path, null);
+ }
+
+ @Override
public VersionedData getData(String path, Watcher watcher) throws NoSuchElementException, IOException, KeeperException, InterruptedException {
Stat stat = new Stat();
try {