You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by ma...@apache.org on 2020/07/20 03:50:56 UTC
[lucene-solr] branch reference_impl updated: @238 - Tweak.
This is an automated email from the ASF dual-hosted git repository.
markrmiller pushed a commit to branch reference_impl
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git
The following commit(s) were added to refs/heads/reference_impl by this push:
new 37da95b @238 - Tweak.
37da95b is described below
commit 37da95bc893d0037e679abc6a6373fedb03101a7
Author: markrmiller@gmail.com <ma...@gmail.com>
AuthorDate: Sun Jul 19 22:50:34 2020 -0500
@238 - Tweak.
---
.../src/java/org/apache/solr/cloud/Overseer.java | 33 ++++++-------
.../org/apache/solr/cloud/OverseerTaskQueue.java | 9 +++-
.../autoscaling/sim/GenericDistributedQueue.java | 18 +++----
.../autoscaling/sim/SimClusterStateProvider.java | 56 +++++++++++-----------
.../sim/SimDistributedQueueFactory.java | 12 ++---
.../autoscaling/sim/SimNodeStateProvider.java | 12 ++---
solr/reference_branch/start-solr.sh | 2 +-
.../src/resources/logconf/log4j2-startup-debug.xml | 1 +
8 files changed, 73 insertions(+), 70 deletions(-)
diff --git a/solr/core/src/java/org/apache/solr/cloud/Overseer.java b/solr/core/src/java/org/apache/solr/cloud/Overseer.java
index 84ca239..fcf09ed 100644
--- a/solr/core/src/java/org/apache/solr/cloud/Overseer.java
+++ b/solr/core/src/java/org/apache/solr/cloud/Overseer.java
@@ -227,7 +227,7 @@ public class Overseer implements SolrCloseable {
// we do not sure which message is bad message, therefore we will re-process node one by one
int fallbackQueueSize = Integer.MAX_VALUE;
ZkDistributedQueue fallbackQueue = workQueue;
- while (!this.isClosed && !Thread.currentThread().isInterrupted()) {
+ while (!isClosed() && !Thread.currentThread().isInterrupted()) {
if (zkStateWriter == null) {
try {
zkStateWriter = new ZkStateWriter(reader, stats);
@@ -283,7 +283,7 @@ public class Overseer implements SolrCloseable {
log.warn("Solr cannot talk to ZK, exiting Overseer work queue loop", e);
return;
} catch (InterruptedException | AlreadyClosedException e) {
- ParWork.propegateInterrupt(e);
+ ParWork.propegateInterrupt(e, true);
return;
} catch (Exception e) {
log.error("Unexpected error in Overseer state update loop", e);
@@ -298,7 +298,7 @@ public class Overseer implements SolrCloseable {
// We do not need to filter any nodes here cause all processed nodes are removed once we flush clusterstate
queue = new LinkedList<>(stateUpdateQueue.peekElements(1000, 2000L, (x) -> true));
} catch (InterruptedException | AlreadyClosedException e) {
- ParWork.propegateInterrupt(e);
+ ParWork.propegateInterrupt(e, true);
return;
} catch (KeeperException.SessionExpiredException e) {
log.error("run()", e);
@@ -309,6 +309,8 @@ public class Overseer implements SolrCloseable {
log.error("Unexpected error in Overseer state update loop", e);
if (!isClosed()) {
continue;
+ } else {
+ return;
}
}
try {
@@ -327,7 +329,7 @@ public class Overseer implements SolrCloseable {
processedNodes.clear();
});
}
- if (isClosed) break;
+ if (isClosed()) return;
// if an event comes in the next 100ms batch it together
queue = new LinkedList<>(stateUpdateQueue.peekElements(1000, 100, node -> !processedNodes.contains(node)));
}
@@ -354,7 +356,7 @@ public class Overseer implements SolrCloseable {
} finally {
log.info("Overseer Loop exiting : {}", LeaderElector.getNodeName(myId));
- if (!isClosed) {
+ if (!isClosed()) {
Overseer.this.close();
}
}
@@ -527,17 +529,9 @@ public class Overseer implements SolrCloseable {
@Override
public void close() throws IOException {
- thread.close();
- while (isAlive()) {
- try {
- join(100);
- Thread.currentThread().interrupt();
- } catch (InterruptedException e) {
- ParWork.propegateInterrupt(e);
- throw new RuntimeException("Interrupted waiting to close");
- }
- }
this.isClosed = true;
+ thread.close();
+ Thread.currentThread().interrupt();
}
public Closeable getThread() {
@@ -849,18 +843,21 @@ public class Overseer implements SolrCloseable {
try (ParWork closer = new ParWork(this, true)) {
closer.collect(() -> {
- ccThread.interrupt();
+
IOUtils.closeQuietly(ccThread);
+ ccThread.interrupt();
});
closer.collect(() -> {
- updaterThread.interrupt();
+
IOUtils.closeQuietly(updaterThread);
+ updaterThread.interrupt();
});
closer.collect(() -> {
- triggerThread.interrupt();
+
IOUtils.closeQuietly(triggerThread);
+ triggerThread.interrupt();
});
closer.addCollect("OverseerInternals");
diff --git a/solr/core/src/java/org/apache/solr/cloud/OverseerTaskQueue.java b/solr/core/src/java/org/apache/solr/cloud/OverseerTaskQueue.java
index 4feb936..f91e596 100644
--- a/solr/core/src/java/org/apache/solr/cloud/OverseerTaskQueue.java
+++ b/solr/core/src/java/org/apache/solr/cloud/OverseerTaskQueue.java
@@ -34,6 +34,8 @@ import org.apache.solr.common.SolrException;
import org.apache.solr.common.cloud.SolrZkClient;
import org.apache.solr.common.cloud.ZkNodeProps;
import org.apache.solr.common.util.Pair;
+import org.apache.solr.common.util.TimeOut;
+import org.apache.solr.common.util.TimeSource;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
@@ -137,7 +139,7 @@ public class OverseerTaskQueue extends ZkDistributedQueue {
private final Lock lock;
private final Condition eventReceived;
- private WatchedEvent event;
+ private volatile WatchedEvent event;
private Event.EventType latchEventType;
LatchWatcher() {
@@ -184,7 +186,10 @@ public class OverseerTaskQueue extends ZkDistributedQueue {
if (this.event != null) {
return;
}
- eventReceived.await(timeoutMs, TimeUnit.MILLISECONDS);
+ TimeOut timeout = new TimeOut(timeoutMs, TimeUnit.MILLISECONDS, TimeSource.NANO_TIME);
+ while (this.event == null && !timeout.hasTimedOut() && !Thread.currentThread().isInterrupted()) {
+ eventReceived.await(500, TimeUnit.MILLISECONDS);
+ }
} finally {
lock.unlock();
}
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/GenericDistributedQueue.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/GenericDistributedQueue.java
index 109c516..37cebe1 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/GenericDistributedQueue.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/GenericDistributedQueue.java
@@ -183,7 +183,7 @@ public class GenericDistributedQueue implements DistributedQueue {
} else {
time = stats.time(dir + "_peek_wait" + wait);
}
- updateLock.lockInterruptibly();
+ updateLock.lock();
try {
long waitNanos = TimeUnit.MILLISECONDS.toNanos(wait);
while (waitNanos > 0) {
@@ -282,7 +282,7 @@ public class GenericDistributedQueue implements DistributedQueue {
public byte[] take() throws Exception {
// Same as for element. Should refactor this.
Timer.Context timer = stats.time(dir + "_take");
- updateLock.lockInterruptibly();
+ updateLock.lock();
try {
while (true) {
byte[] result = removeFirst();
@@ -379,7 +379,7 @@ public class GenericDistributedQueue implements DistributedQueue {
* list is inherently stale.
*/
private String firstChild(boolean remove, boolean refetchIfDirty) throws Exception {
- updateLock.lockInterruptibly();
+ updateLock.lock();
try {
// We always return from cache first, the cache will be cleared if the node is not exist
if (!knownChildren.isEmpty() && !(isDirty && refetchIfDirty)) {
@@ -453,7 +453,7 @@ public class GenericDistributedQueue implements DistributedQueue {
// Trigger a refresh, but only force it if this is not the first iteration.
firstChild(false, !first);
- updateLock.lockInterruptibly();
+ updateLock.lock();
try {
for (String child : knownChildren) {
if (acceptFilter.test(child)) {
@@ -496,7 +496,7 @@ public class GenericDistributedQueue implements DistributedQueue {
result.add(new Pair<>(child, data.getData()));
} catch (NoSuchElementException e) {
// Another client deleted the node first, remove the in-memory and continue.
- updateLock.lockInterruptibly();
+ updateLock.lock();
try {
knownChildren.remove(child);
} finally {
@@ -523,7 +523,7 @@ public class GenericDistributedQueue implements DistributedQueue {
return data != null ? data.getData() : null;
} catch (NoSuchElementException e) {
// Another client deleted the node first, remove the in-memory and retry.
- updateLock.lockInterruptibly();
+ updateLock.lock();
try {
// Efficient only for single-consumer
knownChildren.clear();
@@ -549,7 +549,7 @@ public class GenericDistributedQueue implements DistributedQueue {
return result.getData();
} catch (NoSuchElementException e) {
// Another client deleted the node first, remove the in-memory and retry.
- updateLock.lockInterruptibly();
+ updateLock.lock();
try {
// Efficient only for single-consumer
knownChildren.clear();
@@ -562,7 +562,7 @@ public class GenericDistributedQueue implements DistributedQueue {
}
@VisibleForTesting int watcherCount() throws InterruptedException {
- updateLock.lockInterruptibly();
+ updateLock.lock();
try {
return watcherCount;
} finally {
@@ -571,7 +571,7 @@ public class GenericDistributedQueue implements DistributedQueue {
}
@VisibleForTesting boolean isDirty() throws InterruptedException {
- updateLock.lockInterruptibly();
+ updateLock.lock();
try {
return isDirty;
} finally {
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java
index 232599a..716e689 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java
@@ -326,7 +326,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
*/
@SuppressWarnings({"unchecked"})
public void simSetClusterState(ClusterState initialState) throws Exception {
- lock.lockInterruptibly();
+ lock.lock();
try {
collProperties.clear();
colShardReplicaMap.clear();
@@ -453,7 +453,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
*/
public boolean simRemoveNode(String nodeId) throws Exception {
ensureNotClosed();
- lock.lockInterruptibly();
+ lock.lock();
try {
Set<String> collections = new HashSet<>();
// mark every replica on that node as down
@@ -487,7 +487,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
* Remove all replica information related to dead nodes.
*/
public void simRemoveDeadNodes() throws Exception {
- lock.lockInterruptibly();
+ lock.lock();
try {
Set<String> myNodes = new HashSet<>(nodeReplicaMap.keySet());
myNodes.removeAll(liveNodes.get());
@@ -568,7 +568,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
createEphemeralLiveNode(nodeId);
Set<String> collections = new HashSet<>();
- lock.lockInterruptibly();
+ lock.lock();
try {
setReplicaStates(nodeId, Replica.State.RECOVERING, collections);
} finally {
@@ -577,7 +577,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
cloudManager.getTimeSource().sleep(1000);
- lock.lockInterruptibly();
+ lock.lock();
try {
setReplicaStates(nodeId, Replica.State.ACTIVE, collections);
if (!collections.isEmpty()) {
@@ -659,7 +659,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
@SuppressWarnings({"unchecked"})
public void simAddReplica(String nodeId, ReplicaInfo replicaInfo, boolean runLeaderElection) throws Exception {
ensureNotClosed();
- lock.lockInterruptibly();
+ lock.lock();
try {
// make sure SolrCore name is unique across cluster and coreNodeName within collection
@@ -753,7 +753,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
public void simRemoveReplica(String nodeId, String collection, String coreNodeName) throws Exception {
ensureNotClosed();
- lock.lockInterruptibly();
+ lock.lock();
try {
@SuppressWarnings({"unchecked"})
final List<ReplicaInfo> replicas = nodeReplicaMap.computeIfAbsent
@@ -835,7 +835,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
private void simRunLeaderElection(Collection<String> collections, boolean saveClusterState) throws Exception {
ensureNotClosed();
if (saveClusterState) {
- lock.lockInterruptibly();
+ lock.lock();
try {
collections.forEach(c -> collectionsStatesRef.get(c).invalidate());
} finally {
@@ -865,7 +865,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
log.trace("Attempting leader election ({} / {})", collection, slice);
final AtomicBoolean stateChanged = new AtomicBoolean(Boolean.FALSE);
- lock.lockInterruptibly();
+ lock.lock();
try {
final ClusterState state = getClusterState();
final DocCollection col = state.getCollectionOrNull(collection);
@@ -1036,7 +1036,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
ZkWriteCommand cmd = ZkWriteCommand.noop();
- lock.lockInterruptibly();
+ lock.lock();
try {
cmd = new ClusterStateMutator(cloudManager).createCollection(clusterState, props);
if (cmd.noop) {
@@ -1151,7 +1151,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
});
// force recreation of collection states
- lock.lockInterruptibly();
+ lock.lock();
try {
collectionsStatesRef.get(collectionName).invalidate();
} finally {
@@ -1183,7 +1183,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
results.add(CoreAdminParams.REQUESTID, async);
}
- lock.lockInterruptibly();
+ lock.lock();
try {
collProperties.remove(collection);
sliceProperties.remove(collection);
@@ -1232,7 +1232,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
* Remove all collections.
*/
public void simDeleteAllCollections() throws Exception {
- lock.lockInterruptibly();
+ lock.lock();
try {
collectionsStatesRef.keySet().forEach(name -> {
try {
@@ -1349,7 +1349,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
String collectionName = message.getStr(COLLECTION_PROP);
String sliceName = message.getStr(SHARD_ID_PROP);
ClusterState clusterState = getClusterState();
- lock.lockInterruptibly();
+ lock.lock();
try {
ZkWriteCommand cmd = new CollectionMutator(cloudManager).createShard(clusterState, message);
if (cmd.noop) {
@@ -1558,7 +1558,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
log.trace("-- switching slice states after split shard: collection={}, parent={}, subSlices={}", collectionName,
sliceName.get(), subSlices);
}
- lock.lockInterruptibly();
+ lock.lock();
try {
Map<String, Object> sProps = sliceProperties.computeIfAbsent(collectionName, c -> new ConcurrentHashMap<>())
.computeIfAbsent(sliceName.get(), s -> new ConcurrentHashMap<>());
@@ -1627,7 +1627,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
opDelay(collectionName, CollectionParams.CollectionAction.DELETESHARD.name());
- lock.lockInterruptibly();
+ lock.lock();
try {
sliceProperties.computeIfAbsent(collectionName, c -> new ConcurrentHashMap<>()).remove(sliceName);
colShardReplicaMap.computeIfAbsent(collectionName, c -> new ConcurrentHashMap<>()).remove(sliceName);
@@ -1748,7 +1748,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
}
}
if (!deletesPerShard.isEmpty()) {
- lock.lockInterruptibly();
+ lock.lock();
try {
for (Map.Entry<String, AtomicLong> entry : deletesPerShard.entrySet()) {
String shard = entry.getKey();
@@ -1796,7 +1796,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
if (numDocs == null || numDocs.intValue() == 0) {
continue;
}
- lock.lockInterruptibly();
+ lock.lock();
try {
Number indexSize = (Number)ri.getVariable(Type.CORE_IDX.metricsAttribute);
if (indexSize != null) {
@@ -1847,7 +1847,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
// XXX don't add more than 2bln docs in one request
boolean modified = false;
- lock.lockInterruptibly();
+ lock.lock();
try {
coll = getClusterState().getCollection(collection);
Slice[] slices = coll.getActiveSlicesArr();
@@ -1989,7 +1989,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
}
SolrParams params = req.getParams();
if (params != null && (params.getBool(UpdateParams.OPTIMIZE, false) || params.getBool(UpdateParams.EXPUNGE_DELETES, false))) {
- lock.lockInterruptibly();
+ lock.lock();
try {
coll.getSlices().forEach(s -> {
Replica leader = s.getLeader();
@@ -2092,7 +2092,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
* @param properties properties to set
*/
public void simSetClusterProperties(Map<String, Object> properties) throws Exception {
- lock.lockInterruptibly();
+ lock.lock();
try {
clusterProperties.clear();
if (properties != null) {
@@ -2111,7 +2111,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
* @param value property value
*/
public void simSetClusterProperty(String key, Object value) throws Exception {
- lock.lockInterruptibly();
+ lock.lock();
try {
if (value != null) {
clusterProperties.put(key, value);
@@ -2130,7 +2130,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
* @param properties properties
*/
public void simSetCollectionProperties(String coll, Map<String, Object> properties) throws Exception {
- lock.lockInterruptibly();
+ lock.lock();
try {
if (properties == null) {
collProperties.remove(coll);
@@ -2152,7 +2152,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
* @param value property value
*/
public void simSetCollectionProperty(String coll, String key, String value) throws Exception {
- lock.lockInterruptibly();
+ lock.lock();
try {
final Map<String, Object> props = collProperties.computeIfAbsent(coll, c -> new HashMap<>());
if (value == null) {
@@ -2173,7 +2173,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
* @param properties slice properties
*/
public void simSetSliceProperties(String coll, String slice, Map<String, Object> properties) throws Exception {
- lock.lockInterruptibly();
+ lock.lock();
try {
final Map<String, Object> sliceProps = sliceProperties.computeIfAbsent
(coll, c -> new HashMap<>()).computeIfAbsent(slice, s -> new HashMap<>());
@@ -2381,7 +2381,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
}
public Map<String, Map<String, Object>> simGetCollectionStats() throws IOException, InterruptedException {
- lock.lockInterruptibly();
+ lock.lock();
try {
final Map<String, Map<String, Object>> stats = new TreeMap<>();
ClusterState state = getClusterState();
@@ -2523,7 +2523,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
@Override
public ClusterState getClusterState() throws IOException {
try {
- lock.lockInterruptibly();
+ lock.lock();
try {
Map<String, DocCollection> states = getCollectionStates();
ClusterState state = new ClusterState(0, liveNodes.get(), states);
@@ -2537,7 +2537,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
}
private Map<String, DocCollection> getCollectionStates() throws IOException, InterruptedException {
- lock.lockInterruptibly();
+ lock.lock();
try {
Map<String, DocCollection> collectionStates = new HashMap<>();
collectionsStatesRef.forEach((name, cached) -> {
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SimDistributedQueueFactory.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SimDistributedQueueFactory.java
index fb17881..7ecef55 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SimDistributedQueueFactory.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SimDistributedQueueFactory.java
@@ -115,7 +115,7 @@ public class SimDistributedQueueFactory implements DistributedQueueFactory {
private Pair<String, byte[]> peekInternal(long wait) throws Exception {
Preconditions.checkArgument(wait > 0);
long waitNanos = TimeUnit.MILLISECONDS.toNanos(wait);
- updateLock.lockInterruptibly();
+ updateLock.lock();
try {
while (waitNanos > 0) {
Pair<String, byte[]> pair = queue.peek();
@@ -136,7 +136,7 @@ public class SimDistributedQueueFactory implements DistributedQueueFactory {
@Override
public byte[] poll() throws Exception {
Timer.Context time = stats.time(dir + "_poll");
- updateLock.lockInterruptibly();
+ updateLock.lock();
try {
Pair<String, byte[]> pair = queue.poll();
if (pair != null) {
@@ -154,7 +154,7 @@ public class SimDistributedQueueFactory implements DistributedQueueFactory {
@Override
public byte[] remove() throws Exception {
Timer.Context time = stats.time(dir + "_remove");
- updateLock.lockInterruptibly();
+ updateLock.lock();
try {
byte[] res = queue.remove().second();
changed.signalAll();
@@ -168,7 +168,7 @@ public class SimDistributedQueueFactory implements DistributedQueueFactory {
@Override
public byte[] take() throws Exception {
Timer.Context timer = stats.time(dir + "_take");
- updateLock.lockInterruptibly();
+ updateLock.lock();
try {
while (true) {
byte[] result = poll();
@@ -187,7 +187,7 @@ public class SimDistributedQueueFactory implements DistributedQueueFactory {
@SuppressWarnings({"unchecked", "rawtypes"})
public void offer(byte[] data) throws Exception {
Timer.Context time = stats.time(dir + "_offer");
- updateLock.lockInterruptibly();
+ updateLock.lock();
try {
queue.offer(new Pair(String.format(Locale.ROOT, "qn-%010d", seq), data));
seq++;
@@ -203,7 +203,7 @@ public class SimDistributedQueueFactory implements DistributedQueueFactory {
@Override
public Collection<Pair<String, byte[]>> peekElements(int max, long waitMillis, Predicate<String> acceptFilter) throws Exception {
- updateLock.lockInterruptibly();
+ updateLock.lock();
try {
List<Pair<String, byte[]>> res = new LinkedList<>();
final int maximum = max < 0 ? Integer.MAX_VALUE : max;
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SimNodeStateProvider.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SimNodeStateProvider.java
index 6effce0..29e9c37 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SimNodeStateProvider.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SimNodeStateProvider.java
@@ -92,7 +92,7 @@ public class SimNodeStateProvider implements NodeStateProvider, Closeable {
* @return previous property value or null if property or node didn't exist.
*/
public Object simUpdateNodeValue(String node, String key, Function<Object, Object> updater) throws InterruptedException {
- lock.lockInterruptibly();
+ lock.lock();
try {
Map<String, Object> values = nodeValues.computeIfAbsent(node, n -> new ConcurrentHashMap<>());
return values.put(key, updater.apply(values.get(key)));
@@ -108,7 +108,7 @@ public class SimNodeStateProvider implements NodeStateProvider, Closeable {
* @param values values.
*/
public void simSetNodeValues(String node, Map<String, Object> values) throws InterruptedException {
- lock.lockInterruptibly();
+ lock.lock();
try {
Map<String, Object> existing = nodeValues.computeIfAbsent(node, n -> new ConcurrentHashMap<>());
existing.clear();
@@ -131,7 +131,7 @@ public class SimNodeStateProvider implements NodeStateProvider, Closeable {
* @param value property value
*/
public void simSetNodeValue(String node, String key, Object value) throws InterruptedException {
- lock.lockInterruptibly();
+ lock.lock();
try {
Map<String, Object> existing = nodeValues.computeIfAbsent(node, n -> new ConcurrentHashMap<>());
if (value == null) {
@@ -156,7 +156,7 @@ public class SimNodeStateProvider implements NodeStateProvider, Closeable {
*/
@SuppressWarnings({"unchecked"})
public void simAddNodeValue(String node, String key, Object value) throws InterruptedException {
- lock.lockInterruptibly();
+ lock.lock();
try {
Map<String, Object> values = nodeValues.computeIfAbsent(node, n -> new ConcurrentHashMap<>());
Object existing = values.get(key);
@@ -185,7 +185,7 @@ public class SimNodeStateProvider implements NodeStateProvider, Closeable {
*/
public void simRemoveNodeValues(String node) throws InterruptedException {
log.debug("--removing value for {}", node);
- lock.lockInterruptibly();
+ lock.lock();
try {
Map<String, Object> values = nodeValues.remove(node);
if (values != null && values.containsKey("nodeRole")) {
@@ -203,7 +203,7 @@ public class SimNodeStateProvider implements NodeStateProvider, Closeable {
public void simRemoveDeadNodes() throws InterruptedException {
Set<String> myNodes = new HashSet<>(nodeValues.keySet());
myNodes.removeAll(liveNodesSet.get());
- lock.lockInterruptibly();
+ lock.lock();
try {
AtomicBoolean updateRoles = new AtomicBoolean(false);
myNodes.forEach(n -> {
diff --git a/solr/reference_branch/start-solr.sh b/solr/reference_branch/start-solr.sh
index 88d2084..388c9de 100644
--- a/solr/reference_branch/start-solr.sh
+++ b/solr/reference_branch/start-solr.sh
@@ -22,4 +22,4 @@ if [ ! -d "lucene-solr" ]; then
fi
-bash /opt/solr/reference_impl/solr/bin/solr start -c -m 1g -z "localhost:2181" -p ${SOLR_PORT:-8983} -force -f
+bash /opt/solr/reference_impl/solr/bin/solr start -c -m 1g -z "localhost:2181" -p ${SOLR_PORT:-9998} -force -f
diff --git a/solr/test-framework/src/resources/logconf/log4j2-startup-debug.xml b/solr/test-framework/src/resources/logconf/log4j2-startup-debug.xml
index e0eb1b3..dba5066 100644
--- a/solr/test-framework/src/resources/logconf/log4j2-startup-debug.xml
+++ b/solr/test-framework/src/resources/logconf/log4j2-startup-debug.xml
@@ -42,6 +42,7 @@
<AsyncLogger name="org.apache.solr.cloud.Overseer" level="DEBUG"/>
<AsyncLogger name="org.apache.solr.cloud.OverseerTaskProcessor" level="DEBUG"/>
<AsyncLogger name="org.apache.solr.cloud.ZkDistributedQueue" level="DEBUG"/>
+ <AsyncLogger name="org.apache.solr.cloud.OverseerTaskQueue" level="DEBUG"/>
<!-- <AsyncLogger name="org.apache.solr.common.cloud.SolrZkClient" level="DEBUG"/> -->
<AsyncLogger name="org.apache.solr.cloud.overseer.SliceMutator" level="DEBUG"/>
<AsyncLogger name="org.apache.solr.client.solrj.impl.LBSolrClient" level="DEBUG"/>