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/11/16 19:01:07 UTC
[lucene-solr] branch reference_impl_dev updated (0e9826b -> caea02d)
This is an automated email from the ASF dual-hosted git repository.
markrmiller pushed a change to branch reference_impl_dev
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git.
from 0e9826b @1232 Send the recovery lock into the recovery strat.
new c2d7e90 @1233 Close elector should only be in unload not solrcore#close.
new d5db80d @1234 Cut unused dist queue stuff.
new caea02d @1235 Cleanup logging, Overseer close.
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../java/org/apache/solr/cloud/LeaderElector.java | 6 +-
.../src/java/org/apache/solr/cloud/Overseer.java | 136 +++++-----
.../apache/solr/cloud/OverseerElectionContext.java | 4 +-
.../apache/solr/cloud/OverseerNodePrioritizer.java | 120 ---------
.../solr/cloud/OverseerTaskExecutorTask.java | 8 +-
.../apache/solr/cloud/OverseerTaskProcessor.java | 139 ----------
.../org/apache/solr/cloud/OverseerTaskQueue.java | 40 +--
.../solr/cloud/ShardLeaderElectionContext.java | 7 +-
.../solr/cloud/ShardLeaderElectionContextBase.java | 2 +-
.../java/org/apache/solr/cloud/StatePublisher.java | 4 +-
.../java/org/apache/solr/cloud/ZkController.java | 27 +-
.../org/apache/solr/cloud/ZkDistributedQueue.java | 297 +--------------------
.../cloud/api/collections/OverseerRoleCmd.java | 5 +-
.../java/org/apache/solr/core/CoreContainer.java | 3 +-
.../src/java/org/apache/solr/core/SolrCore.java | 8 -
.../src/java/org/apache/solr/core/SolrCores.java | 2 +-
.../solr/handler/admin/CollectionsHandler.java | 18 +-
.../handler/component/SpellCheckComponent.java | 4 +-
.../java/org/apache/solr/servlet/HttpSolrCall.java | 4 +-
.../apache/solr/cloud/DistributedQueueTest.java | 22 +-
.../solr/handler/component/StatsComponentTest.java | 2 +
.../client/solrj/impl/BaseCloudSolrClient.java | 2 +-
.../apache/solr/common/cloud/ZkStateReader.java | 16 +-
23 files changed, 148 insertions(+), 728 deletions(-)
delete mode 100644 solr/core/src/java/org/apache/solr/cloud/OverseerNodePrioritizer.java
[lucene-solr] 02/03: @1234 Cut unused dist queue stuff.
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
markrmiller pushed a commit to branch reference_impl_dev
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git
commit d5db80ddf071bbe81c2d31a422f48e4319697e7f
Author: markrmiller@gmail.com <ma...@gmail.com>
AuthorDate: Mon Nov 16 12:12:17 2020 -0600
@1234 Cut unused dist queue stuff.
---
.../org/apache/solr/cloud/ZkDistributedQueue.java | 297 +--------------------
.../apache/solr/cloud/DistributedQueueTest.java | 22 +-
2 files changed, 18 insertions(+), 301 deletions(-)
diff --git a/solr/core/src/java/org/apache/solr/cloud/ZkDistributedQueue.java b/solr/core/src/java/org/apache/solr/cloud/ZkDistributedQueue.java
index dba4487..70a61e5 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ZkDistributedQueue.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ZkDistributedQueue.java
@@ -17,8 +17,6 @@
package org.apache.solr.cloud;
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.common.cloud.ConnectionManager.IsClosed;
import org.apache.solr.common.cloud.SolrZkClient;
@@ -27,7 +25,6 @@ import org.apache.solr.common.util.Utils;
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;
@@ -44,7 +41,6 @@ import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Set;
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;
@@ -143,12 +139,7 @@ public class ZkDistributedQueue implements DistributedQueue {
*/
@Override
public byte[] peek() throws KeeperException, InterruptedException {
- Timer.Context time = stats.time(dir + "_peek");
- try {
- return firstElement();
- } finally {
- time.stop();
- }
+ throw new UnsupportedOperationException();
}
/**
@@ -160,7 +151,7 @@ public class ZkDistributedQueue implements DistributedQueue {
*/
@Override
public byte[] peek(boolean block) throws KeeperException, InterruptedException {
- return block ? peek(Long.MAX_VALUE) : peek();
+ throw new UnsupportedOperationException();
}
/**
@@ -172,28 +163,7 @@ public class ZkDistributedQueue implements DistributedQueue {
*/
@Override
public byte[] peek(long wait) throws KeeperException, InterruptedException {
- 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();
- }
+ throw new UnsupportedOperationException();
}
/**
@@ -204,12 +174,7 @@ public class ZkDistributedQueue implements DistributedQueue {
*/
@Override
public byte[] poll() throws KeeperException, InterruptedException {
- Timer.Context time = stats.time(dir + "_poll");
- try {
- return removeFirst();
- } finally {
- time.stop();
- }
+ throw new UnsupportedOperationException();
}
/**
@@ -307,21 +272,7 @@ public class ZkDistributedQueue implements DistributedQueue {
*/
@Override
public byte[] take() throws KeeperException, InterruptedException {
- // 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();
- }
+ throw new UnsupportedOperationException();
}
private static Set<String> OPERATIONS = new HashSet<>();
@@ -338,48 +289,9 @@ public class ZkDistributedQueue implements DistributedQueue {
*/
@Override
public void offer(byte[] data) throws KeeperException, InterruptedException {
- // TODO change to accept json
- Map json = (Map) Utils.fromJSON(data);
-
- Timer.Context time = stats.time(dir + "_offer");
-
- try {
- while (true) {
- try {
- if (maxQueueSize > 0) {
- if (offerPermits.get() <= 0 || offerPermits.getAndDecrement() <= 0) { // nocommit review
- // If a max queue size is set, check it before creating a new queue item.
- Stat stat = zookeeper.exists(dir, null, true);
- if (stat == null) {
- // jump to the code below, which tries to create dir if it doesn't exist
- throw new KeeperException.NoNodeException();
- }
- int remainingCapacity = maxQueueSize - stat.getNumChildren();
- 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.
- zookeeper.create(dir + "/" + PREFIX, data, CreateMode.PERSISTENT_SEQUENTIAL, true);
- isDirty = true;
- return;
- } catch (KeeperException.NoNodeException e) {
- try {
- zookeeper.create(dir, DATA, CreateMode.PERSISTENT, true);
- } catch (KeeperException.NodeExistsException ne) {
- // someone created it
- }
- }
- }
- } finally {
- time.stop();
- }
+ // TODO - if too many items on the queue, just block
+ zookeeper.create(dir + "/" + PREFIX, data, CreateMode.PERSISTENT_SEQUENTIAL, true);
+ return;
}
public Stats getZkStats() {
@@ -418,68 +330,14 @@ public class ZkDistributedQueue implements DistributedQueue {
* list is inherently stale.
*/
private String firstChild(boolean remove, boolean refetchIfDirty) throws KeeperException, InterruptedException {
- updateLock.lockInterruptibly();
- try {
- if (log.isDebugEnabled()) log.debug("firstChild isDirty={}, refetchIfDirty={}, remove={}", isDirty, refetchIfDirty, remove);
- // We always return from cache first, the cache will be cleared if the node is not exist
- if (!knownChildren.isEmpty() && !(isDirty && refetchIfDirty)) {
- if (log.isDebugEnabled()) log.debug("returning first child");
- return remove ? knownChildren.pollFirst() : knownChildren.first();
- }
-
- if (!isDirty && knownChildren.isEmpty()) {
- if (log.isDebugEnabled()) log.debug("isDirty={} knownChildren is empty, returning null", isDirty);
- 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.
-
- if (log.isDebugEnabled()) log.debug("isDirty={} fetch known children", isDirty);
- ChildWatcher newWatcher = (watcherCount == 0) ? new ChildWatcher() : null;
- knownChildren = fetchZkChildren(newWatcher);
- if (newWatcher != null) {
- watcherCount++; // watcher was successfully set
- }
- isDirty = false;
- if (knownChildren.isEmpty()) {
- if (log.isDebugEnabled()) log.debug("known children is empty return null");
- return null;
- }
- if (log.isDebugEnabled()) log.debug("signal all and get first, remove={}", remove);
- changed.signalAll();
- return remove ? knownChildren.pollFirst() : knownChildren.first();
- } finally {
- updateLock.unlock();
- }
+ throw new UnsupportedOperationException();
}
/**
* Return the current set of children from ZK; does not change internal state.
*/
TreeSet<String> fetchZkChildren(Watcher watcher) throws InterruptedException, KeeperException {
- while (true) {
- try {
- TreeSet<String> orderedChildren = new TreeSet<>();
- if (log.isDebugEnabled()) log.debug("fetchZkChildren {}", dir);
- List<String> childNames = zookeeper.getChildren(dir, watcher, true);
- if (log.isDebugEnabled()) log.debug("got {} nodes", childNames);
- 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);
- }
- if (log.isDebugEnabled()) log.debug("returning {}", orderedChildren);
- return orderedChildren;
- } catch (KeeperException.NoNodeException e) {
- zookeeper.makePath(dir, false, true);
- // go back to the loop and try again
- }
- }
+ throw new UnsupportedOperationException();
}
/**
@@ -490,98 +348,7 @@ public class ZkDistributedQueue implements DistributedQueue {
*/
@Override
public Collection<Pair<String, byte[]>> peekElements(int max, long waitMillis, Predicate<String> acceptFilter) throws KeeperException, InterruptedException {
- List<String> foundChildren = new ArrayList<>();
- long waitNanos = TimeUnit.MILLISECONDS.toNanos(waitMillis);
- if (log.isDebugEnabled()) log.debug("peekElements {} {}", max, waitMillis);
- boolean first = true;
- while (true) {
- // Trigger a refresh, but only force it if this is not the first iteration.
- if (log.isDebugEnabled()) log.debug("call first child {}", !first);
- 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 {
- byte[] data = zookeeper.getData(dir + "/" + child, null, null, true);
- result.add(new Pair<>(child, data));
- } catch (KeeperException.NoNodeException 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 KeeperException, InterruptedException {
- while (true) {
- if (log.isDebugEnabled()) log.debug("calling firstElement");
- String firstChild = firstChild(false, false);
- if (firstChild == null) {
- if (log.isDebugEnabled()) log.debug("return null");
- return null;
- }
- try {
- if (log.isDebugEnabled()) log.debug("get data {}", dir + "/" + firstChild);
- return zookeeper.getData(dir + "/" + firstChild, null, null, true);
- } catch (KeeperException.NoNodeException e) {
- if (log.isDebugEnabled()) log.debug("no node found {}, clear known children and set dirty", dir + "/" + firstChild);
- // 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();
- }
- }
- }
+ throw new UnsupportedOperationException();
}
private byte[] removeFirst() throws KeeperException, InterruptedException {
@@ -609,46 +376,4 @@ public class ZkDistributedQueue implements DistributedQueue {
}
}
}
-
- @VisibleForTesting int watcherCount() throws InterruptedException {
- updateLock.lockInterruptibly();
- try {
- return watcherCount;
- } finally {
- updateLock.unlock();
- }
- }
-
- @VisibleForTesting boolean isDirty() throws InterruptedException {
- updateLock.lockInterruptibly();
- try {
-
- if (log.isDebugEnabled()) log.debug("isDirty {}", isDirty);
- return isDirty;
- } finally {
- updateLock.unlock();
- }
- }
-
- @VisibleForTesting class ChildWatcher implements Watcher {
-
- @Override
- public void process(WatchedEvent event) {
-
- if (log.isDebugEnabled()) log.debug("ChildWatcher fired {} {}", event.getType(), event.getPath());
- // 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();
- }
- }
- }
}
diff --git a/solr/core/src/test/org/apache/solr/cloud/DistributedQueueTest.java b/solr/core/src/test/org/apache/solr/cloud/DistributedQueueTest.java
index 449ff6b..470a10f 100644
--- a/solr/core/src/test/org/apache/solr/cloud/DistributedQueueTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/DistributedQueueTest.java
@@ -143,12 +143,9 @@ public class DistributedQueueTest extends SolrTestCaseJ4 {
assertNull(dq.peek(100));
- // assertFalse(dq.isDirty());
- assertEquals(1, dq.watcherCount());
forceSessionExpire();
- assertEquals(0, dq.watcherCount());
// Rerun the earlier test make sure updates are still seen, post reconnection.
future = testExecutor.submit(() -> new String(dq.peek(true), UTF8));
@@ -171,28 +168,23 @@ public class DistributedQueueTest extends SolrTestCaseJ4 {
String dqZNode = "/distqueue/test";
ZkDistributedQueue dq = makeDistributedQueue(dqZNode);
assertTrue(dq.peekElements(1, 1, s1 -> true).isEmpty());
- assertEquals(1, dq.watcherCount());
+
assertTrue(dq.peekElements(1, 1, s1 -> true).isEmpty());
- assertEquals(1, dq.watcherCount());
+
assertNull(dq.peek());
- assertEquals(1, dq.watcherCount());
+
assertNull(dq.peek(1));
- assertEquals(1, dq.watcherCount());
+
dq.offer("hello world".getBytes(UTF8));
assertNotNull(dq.peek()); // synchronously available
// dirty and watcher state indeterminate here, race with watcher
Thread.sleep(100); // watcher should have fired now
assertNotNull(dq.peek());
- // in case of race condition, childWatcher is kicked off after peek()
- if (dq.watcherCount() == 0) {
- dq.poll();
- dq.offer("hello world".getBytes(UTF8));
- dq.peek();
- }
- assertEquals(1, dq.watcherCount());
+
+
assertFalse(dq.peekElements(1, 1, s -> true).isEmpty());
- assertEquals(1, dq.watcherCount());
+
}
@Test
[lucene-solr] 01/03: @1233 Close elector should only be in unload
not solrcore#close.
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
markrmiller pushed a commit to branch reference_impl_dev
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git
commit c2d7e906c4612597f37c67eda69d4cf530067419
Author: markrmiller@gmail.com <ma...@gmail.com>
AuthorDate: Mon Nov 16 11:28:49 2020 -0600
@1233 Close elector should only be in unload not solrcore#close.
---
solr/core/src/java/org/apache/solr/core/CoreContainer.java | 2 +-
solr/core/src/java/org/apache/solr/core/SolrCore.java | 8 --------
.../org/apache/solr/handler/component/StatsComponentTest.java | 2 ++
3 files changed, 3 insertions(+), 9 deletions(-)
diff --git a/solr/core/src/java/org/apache/solr/core/CoreContainer.java b/solr/core/src/java/org/apache/solr/core/CoreContainer.java
index 1a07678..f05dafe 100644
--- a/solr/core/src/java/org/apache/solr/core/CoreContainer.java
+++ b/solr/core/src/java/org/apache/solr/core/CoreContainer.java
@@ -1743,9 +1743,9 @@ public class CoreContainer implements Closeable {
if (isZooKeeperAware()) {
IOUtils.closeQuietly(getZkController().getShardLeaderElector(name));
+ getZkController().removeShardLeaderElector(name);
}
- // check for core-init errors first
CoreLoadFailure loadFailure = coreInitFailures.remove(name);
if (loadFailure != null) {
// getting the index directory requires opening a DirectoryFactory with a SolrConfig, etc,
diff --git a/solr/core/src/java/org/apache/solr/core/SolrCore.java b/solr/core/src/java/org/apache/solr/core/SolrCore.java
index 00d68ea..995b6fd 100644
--- a/solr/core/src/java/org/apache/solr/core/SolrCore.java
+++ b/solr/core/src/java/org/apache/solr/core/SolrCore.java
@@ -1624,14 +1624,6 @@ public final class SolrCore implements SolrInfoBean, Closeable {
this.isClosed = true;
searcherExecutor.shutdown();
- if (coreContainer.isZooKeeperAware() && !old_reloaded) {
- LeaderElector elector = coreContainer.getZkController().getShardLeaderElector(name);
- if (elector != null) {
- IOUtils.closeQuietly(elector);
- coreContainer.getZkController().removeShardLeaderElector(name);
- }
- }
-
closer.collect("snapshotsDir", () -> {
Directory snapshotsDir = snapshotMgr.getSnapshotsDir();
this.directoryFactory.doneWithDirectory(snapshotsDir);
diff --git a/solr/core/src/test/org/apache/solr/handler/component/StatsComponentTest.java b/solr/core/src/test/org/apache/solr/handler/component/StatsComponentTest.java
index 3e48176..6b49aeb 100644
--- a/solr/core/src/test/org/apache/solr/handler/component/StatsComponentTest.java
+++ b/solr/core/src/test/org/apache/solr/handler/component/StatsComponentTest.java
@@ -1622,6 +1622,7 @@ public class StatsComponentTest extends SolrTestCaseJ4 {
}
/** @see #testHllOptions */
+ @Nightly
public void testCardinality() throws Exception {
SolrCore core = h.getCore();
// insure we have the same hasher a_l would use
@@ -1887,6 +1888,7 @@ public class StatsComponentTest extends SolrTestCaseJ4 {
}
// simple percentiles test
+ @Nightly
public void testPercentiles() throws Exception {
// NOTE: deliberately not in numeric order
[lucene-solr] 03/03: @1235 Cleanup logging, Overseer close.
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
markrmiller pushed a commit to branch reference_impl_dev
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git
commit caea02dbc2c6ca345810379bc78f4ac42dd187da
Author: markrmiller@gmail.com <ma...@gmail.com>
AuthorDate: Mon Nov 16 12:57:54 2020 -0600
@1235 Cleanup logging, Overseer close.
---
.../java/org/apache/solr/cloud/LeaderElector.java | 6 +-
.../src/java/org/apache/solr/cloud/Overseer.java | 136 ++++++++++----------
.../apache/solr/cloud/OverseerElectionContext.java | 4 +-
.../apache/solr/cloud/OverseerNodePrioritizer.java | 120 ------------------
.../solr/cloud/OverseerTaskExecutorTask.java | 8 +-
.../apache/solr/cloud/OverseerTaskProcessor.java | 139 ---------------------
.../org/apache/solr/cloud/OverseerTaskQueue.java | 40 ++----
.../solr/cloud/ShardLeaderElectionContext.java | 7 +-
.../solr/cloud/ShardLeaderElectionContextBase.java | 2 +-
.../java/org/apache/solr/cloud/StatePublisher.java | 4 +-
.../java/org/apache/solr/cloud/ZkController.java | 27 ++--
.../cloud/api/collections/OverseerRoleCmd.java | 5 +-
.../java/org/apache/solr/core/CoreContainer.java | 1 -
.../src/java/org/apache/solr/core/SolrCores.java | 2 +-
.../solr/handler/admin/CollectionsHandler.java | 18 +--
.../handler/component/SpellCheckComponent.java | 4 +-
.../java/org/apache/solr/servlet/HttpSolrCall.java | 4 +-
.../client/solrj/impl/BaseCloudSolrClient.java | 2 +-
.../apache/solr/common/cloud/ZkStateReader.java | 16 +--
19 files changed, 127 insertions(+), 418 deletions(-)
diff --git a/solr/core/src/java/org/apache/solr/cloud/LeaderElector.java b/solr/core/src/java/org/apache/solr/cloud/LeaderElector.java
index c28a9af..234208e 100644
--- a/solr/core/src/java/org/apache/solr/cloud/LeaderElector.java
+++ b/solr/core/src/java/org/apache/solr/cloud/LeaderElector.java
@@ -107,7 +107,7 @@ public class LeaderElector implements Closeable {
InterruptedException, IOException {
//if (checkClosed(context)) return false;
- log.info("Check if I am leader {}", context.getClass().getSimpleName());
+ if (log.isDebugEnabled()) log.debug("Check if I am leader {}", context.getClass().getSimpleName());
if (isClosed) {
log.info("elector is closed, won't join election");
}
@@ -154,7 +154,7 @@ public class LeaderElector implements Closeable {
if (leaderSeqNodeName.equals(seqs.get(0))) {
// I am the leader
- log.info("I am the potential leader {}, running leader process", context.leaderProps);
+ if (log.isDebugEnabled()) log.debug("I am the potential leader {}, running leader process", context.leaderProps.getName());
try {
if (isClosed || (zkController != null && zkController.getCoreContainer().isShutDown())) {
@@ -362,7 +362,7 @@ public class LeaderElector implements Closeable {
int seq = getSeq(context.leaderSeqPath);
- log.info("Do checkIfIamLeader");
+ if (log.isDebugEnabled()) log.debug("Do checkIfIamLeader");
boolean tryagain = checkIfIamLeader(context, replacement);
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 4ab00e7..861083c 100644
--- a/solr/core/src/java/org/apache/solr/cloud/Overseer.java
+++ b/solr/core/src/java/org/apache/solr/cloud/Overseer.java
@@ -155,6 +155,7 @@ public class Overseer implements SolrCloseable {
public static final String OVERSEER_ELECT = "/overseer/overseer_elect";
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+ private static volatile Overseer OUR_JVM_OVERSEER = null;
private volatile boolean closeAndDone;
private volatile boolean initedHttpClient = false;
@@ -265,13 +266,23 @@ public class Overseer implements SolrCloseable {
}
public synchronized void start(String id, ElectionContext context) throws KeeperException {
- log.info("Staring Overseer");
+ log.info("Starting Overseer");
if (getCoreContainer().isShutDown() || closeAndDone) {
if (log.isDebugEnabled()) log.debug("Already closed, exiting");
return;
}
- doClose();
+ if (!closed) {
+ throw new IllegalStateException("Cannot start an Overseer that is not closed");
+ }
+
+ if (OUR_JVM_OVERSEER != null) {
+ throw new IllegalStateException("Cannot start an Overseer if another is running");
+ }
+
+ OUR_JVM_OVERSEER = this;
+
+ // doClose();
closed = false;
MDCLoggingContext.setNode(zkController == null ?
@@ -283,7 +294,7 @@ public class Overseer implements SolrCloseable {
// stateManagmentExecutor = ParWork.getParExecutorService("stateManagmentExecutor",
// 1, 1, 3000, new SynchronousQueue());
taskExecutor = ParWork.getParExecutorService("overseerTaskExecutor",
- 10, 32, 10000, new SynchronousQueue());
+ 10, 32, 1000, new SynchronousQueue());
// try {
// if (context != null) context.close();
@@ -469,47 +480,19 @@ public class Overseer implements SolrCloseable {
public void closeAndDone() {
this.closeAndDone = true;
- this.closed = true;
+
}
public boolean isCloseAndDone() {
return closeAndDone;
}
- public void close() {
- log.info("Overseer (id={}) closing closeAndDone={} frp}", id, closeAndDone);
+ public synchronized void close() {
+ log.info("Overseer (id={}) closing closeAndDone={}}", id, closeAndDone);
- if (closeAndDone) {
- if (overseerOnlyClient != null) {
- overseerOnlyClient.disableCloseLock();
- }
-
- if (overseerLbClient != null) {
- overseerLbClient.close();
- overseerLbClient = null;
- }
-
- if (overseerOnlyClient != null) {
- overseerOnlyClient.close();
- overseerOnlyClient = null;
- }
- } else {
- if (!zkController.getCoreContainer().isShutDown() && !zkController.isShudownCalled() && !zkController.isClosed()) {
- log.info("rejoining the overseer election after closing");
- zkController.rejoinOverseerElection( false);
- }
- }
- }
-
- @Override
- public boolean isClosed() {
- return closed;
- }
-
- void doClose() {
closed = true;
- log.info("doClose() - start closeAndDone={}", closeAndDone);
+ log.info("doClose - closeAndDone={}", closeAndDone);
this.zkStateWriter = null;
if (queueWatcher != null) {
@@ -521,36 +504,65 @@ public class Overseer implements SolrCloseable {
}
-// if (stateManagmentExecutor != null) {
-// log.info("shutdown stateManagmentExecutor");
-// stateManagmentExecutor.shutdown();
-// }
-//
+ // if (stateManagmentExecutor != null) {
+ // log.info("shutdown stateManagmentExecutor");
+ // stateManagmentExecutor.shutdown();
+ // }
+ //
-// if (stateManagmentExecutor != null) {
-// stateManagmentExecutor.shutdownNow();
-// }
+ // if (stateManagmentExecutor != null) {
+ // stateManagmentExecutor.shutdownNow();
+ // }
- // ExecutorUtil.shutdownAndAwaitTermination(stateManagmentExecutor );
+ // ExecutorUtil.shutdownAndAwaitTermination(stateManagmentExecutor );
if (taskExecutor != null) {
try {
- taskExecutor.shutdownNow();
+ taskExecutor.shutdown();
} catch (NullPointerException okay) {
// okay
}
}
+ if (closeAndDone) {
+ if (overseerOnlyClient != null) {
+ overseerOnlyClient.disableCloseLock();
+ }
+
+ if (overseerLbClient != null) {
+ overseerLbClient.close();
+ overseerLbClient = null;
+ }
+
+ if (overseerOnlyClient != null) {
+ overseerOnlyClient.close();
+ overseerOnlyClient = null;
+ }
+ }
ExecutorUtil.shutdownAndAwaitTermination(taskExecutor );
+ if (!closeAndDone) {
+ if (!zkController.getCoreContainer().isShutDown() && !zkController.isShudownCalled() && !zkController.isClosed()) {
+ log.info("rejoining the overseer election after closing");
+ zkController.rejoinOverseerElection( false);
+ }
+ }
+
if (log.isDebugEnabled()) {
- log.debug("doClose() - end");
+ log.debug("doClose - end");
}
+ OUR_JVM_OVERSEER = null;
assert ObjectReleaseTracker.release(this);
}
+ @Override
+ public boolean isClosed() {
+ return closed;
+ }
+
+
/**
* Get queue that can be used to send messages to Overseer.
* <p>
@@ -716,17 +728,10 @@ public class Overseer implements SolrCloseable {
return true;
}
- public void writePendingUpdates() throws InterruptedException {
- log.info("writePendingUpdates ");
+ public void writePendingUpdates() {
+ if (log.isDebugEnabled()) log.debug("writePendingUpdates");
new OverseerTaskExecutorTask.WriteTask(getCoreContainer(), zkStateWriter).run();
-// try {
-// future.get();
-// } catch (ExecutionException e) {
-// log.error("", e);
-// throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, e);
-// }
-
}
private static abstract class QueueWatcher implements Watcher, Closeable {
@@ -751,22 +756,22 @@ public class Overseer implements SolrCloseable {
private List<String> setWatch() {
try {
- log.info("set watch on Overseer work queue {}", path);
+ if (log.isDebugEnabled()) log.debug("set watch on Overseer work queue {}", path);
List<String> children = zkController.getZkClient().getChildren(path, this, true);
Collections.sort(children);
return children;
} catch (KeeperException.SessionExpiredException e) {
log.warn("ZooKeeper session expired");
- overseer.doClose();
- return null;
+ overseer.close();
+ throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, e);
} catch (InterruptedException | AlreadyClosedException e) {
log.info("Already closed");
- overseer.doClose();
- return null;
+ overseer.close();
+ throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, e);
} catch (Exception e) {
log.error("Unexpected error in Overseer state update loop", e);
- overseer.doClose();
+ overseer.close();
throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, e);
}
}
@@ -844,12 +849,7 @@ public class Overseer implements SolrCloseable {
}
}
- try {
- overseer.writePendingUpdates();
- } catch (InterruptedException e) {
- log.error("Overseer state update queue processing interrupted");
- return;
- }
+ overseer.writePendingUpdates();
zkController.getZkClient().delete(fullPaths, true);
@@ -949,7 +949,7 @@ public class Overseer implements SolrCloseable {
// return;
// }
- log.info("response {}", response);
+ if (log.isDebugEnabled()) log.debug("response {}", response);
if (asyncId != null) {
if (response != null && (response.getResponse().get("failure") != null || response.getResponse().get("exception") != null)) {
diff --git a/solr/core/src/java/org/apache/solr/cloud/OverseerElectionContext.java b/solr/core/src/java/org/apache/solr/cloud/OverseerElectionContext.java
index b174208..8c2a8bf 100644
--- a/solr/core/src/java/org/apache/solr/cloud/OverseerElectionContext.java
+++ b/solr/core/src/java/org/apache/solr/cloud/OverseerElectionContext.java
@@ -80,8 +80,6 @@ final class OverseerElectionContext extends ShardLeaderElectionContextBase {
super.runLeaderProcess(context, weAreReplacement, pauseBeforeStartMs);
- log.info("Registered as Overseer leader, starting Overseer ...");
-
if (!overseer.getZkController().getCoreContainer().isShutDown() && !overseer.getZkController().isShudownCalled()
&& !overseer.isDone()) {
log.info("Starting overseer after winnning Overseer election {}", id);
@@ -112,7 +110,7 @@ final class OverseerElectionContext extends ShardLeaderElectionContextBase {
}
closer.collect("overseer", () -> {
try {
- overseer.doClose();
+ overseer.close();
} catch (Exception e) {
ParWork.propagateInterrupt(e);
log.error("Exception closing Overseer", e);
diff --git a/solr/core/src/java/org/apache/solr/cloud/OverseerNodePrioritizer.java b/solr/core/src/java/org/apache/solr/cloud/OverseerNodePrioritizer.java
deleted file mode 100644
index 6a384b23..0000000
--- a/solr/core/src/java/org/apache/solr/cloud/OverseerNodePrioritizer.java
+++ /dev/null
@@ -1,120 +0,0 @@
-/*
- * 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;
-
-import org.apache.solr.client.solrj.impl.Http2SolrClient;
-import org.apache.solr.cloud.overseer.OverseerAction;
-import org.apache.solr.common.cloud.SolrZkClient;
-import org.apache.solr.common.cloud.ZkNodeProps;
-import org.apache.solr.common.cloud.ZkStateReader;
-import org.apache.solr.common.params.CoreAdminParams;
-import org.apache.solr.common.params.CoreAdminParams.CoreAdminAction;
-import org.apache.solr.common.params.ModifiableSolrParams;
-import org.apache.solr.common.util.Utils;
-import org.apache.solr.handler.component.ShardHandler;
-import org.apache.solr.handler.component.ShardHandlerFactory;
-import org.apache.solr.handler.component.ShardRequest;
-import org.apache.zookeeper.data.Stat;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import static org.apache.solr.common.params.CommonParams.ID;
-import java.lang.invoke.MethodHandles;
-import java.util.List;
-import java.util.Map;
-
-/**
- * Responsible for prioritization of Overseer nodes, for example with the
- * ADDROLE collection command.
- */
-public class OverseerNodePrioritizer {
-
- private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
-
- private final ZkStateReader zkStateReader;
- private final String adminPath;
- private final ShardHandlerFactory shardHandlerFactory;
-
- private ZkDistributedQueue stateUpdateQueue;
-
- private Http2SolrClient httpClient;
-
- public OverseerNodePrioritizer(ZkStateReader zkStateReader, ZkDistributedQueue stateUpdateQueue, String adminPath, ShardHandlerFactory shardHandlerFactory, Http2SolrClient httpClient) {
- this.zkStateReader = zkStateReader;
- this.adminPath = adminPath;
- this.shardHandlerFactory = shardHandlerFactory;
- this.stateUpdateQueue = stateUpdateQueue;
- }
-
- public synchronized void prioritizeOverseerNodes(String overseerId) throws Exception {
- SolrZkClient zk = zkStateReader.getZkClient();
- if(!zk.exists(ZkStateReader.ROLES))return;
- Map m = (Map) Utils.fromJSON(zk.getData(ZkStateReader.ROLES, null, new Stat()));
-
- List overseerDesignates = (List) m.get("overseer");
- if(overseerDesignates==null || overseerDesignates.isEmpty()) return;
- String ldr = OverseerTaskProcessor.getLeaderNode(zk);
- if(overseerDesignates.contains(ldr)) return;
- log.info("prioritizing overseer nodes at {} overseer designates are {}", overseerId, overseerDesignates);
- List<String> electionNodes = OverseerTaskProcessor.getSortedElectionNodes(zk, Overseer.OVERSEER_ELECT + LeaderElector.ELECTION_NODE);
- if(electionNodes.size()<2) return;
- log.info("sorted nodes {}", electionNodes);
-
- String designateNodeId = null;
- for (String electionNode : electionNodes) {
- if(overseerDesignates.contains( LeaderElector.getNodeName(electionNode))){
- designateNodeId = electionNode;
- break;
- }
- }
-
- if(designateNodeId == null){
- log.warn("No live overseer designate ");
- return;
- }
- if(!designateNodeId.equals( electionNodes.get(1))) { //checking if it is already at no:1
- log.info("asking node {} to come join election at head", designateNodeId);
- invokeOverseerOp(designateNodeId, "rejoinAtHead"); //ask designate to come first
- if (log.isInfoEnabled()) {
- log.info("asking the old first in line {} to rejoin election ", electionNodes.get(1));
- }
- invokeOverseerOp(electionNodes.get(1), "rejoin");//ask second inline to go behind
- }
- //now ask the current leader to QUIT , so that the designate can takeover
- stateUpdateQueue.offer(
- Utils.toJSON(new ZkNodeProps(Overseer.QUEUE_OPERATION, OverseerAction.QUIT.toLower(),
- ID, OverseerTaskProcessor.getLeaderId(zkStateReader.getZkClient()))));
-
- }
-
- private void invokeOverseerOp(String electionNode, String op) {
- ModifiableSolrParams params = new ModifiableSolrParams();
- ShardHandler shardHandler = shardHandlerFactory.getShardHandler();
- params.set(CoreAdminParams.ACTION, CoreAdminAction.OVERSEEROP.toString());
- params.set("op", op);
- params.set("qt", adminPath);
- params.set("electionNode", electionNode);
- ShardRequest sreq = new ShardRequest();
- sreq.purpose = 1;
- String replica = zkStateReader.getBaseUrlForNodeName(LeaderElector.getNodeName(electionNode));
- sreq.shards = new String[]{replica};
- sreq.actualShards = sreq.shards;
- sreq.params = params;
- shardHandler.submit(sreq, replica, sreq.params);
- shardHandler.takeCompletedOrError();
- }
-}
diff --git a/solr/core/src/java/org/apache/solr/cloud/OverseerTaskExecutorTask.java b/solr/core/src/java/org/apache/solr/cloud/OverseerTaskExecutorTask.java
index 00ddddf..5495d01 100644
--- a/solr/core/src/java/org/apache/solr/cloud/OverseerTaskExecutorTask.java
+++ b/solr/core/src/java/org/apache/solr/cloud/OverseerTaskExecutorTask.java
@@ -47,11 +47,11 @@ public class OverseerTaskExecutorTask implements Runnable {
return;
}
- log.info("Queue operation is {}", operation);
+ if (log.isDebugEnabled()) log.debug("Queue operation is {}", operation);
- log.info("Process message {} {}", message, operation);
+ if (log.isDebugEnabled()) log.debug("Process message {} {}", message, operation);
- log.info("Enqueue message {}", operation);
+ if (log.isDebugEnabled()) log.debug("Enqueue message {}", operation);
zkStateWriter.enqueueUpdate(null, message, true);
@@ -101,7 +101,7 @@ public class OverseerTaskExecutorTask implements Runnable {
@Override
public void run() {
- log.info("OverseerTaskExecutorTask, going to process message {}", message);
+ if (log.isDebugEnabled()) log.debug("OverseerTaskExecutorTask, going to process message {}", message);
try {
processQueueItem(message);
diff --git a/solr/core/src/java/org/apache/solr/cloud/OverseerTaskProcessor.java b/solr/core/src/java/org/apache/solr/cloud/OverseerTaskProcessor.java
index 0d4720f..ccef59e 100644
--- a/solr/core/src/java/org/apache/solr/cloud/OverseerTaskProcessor.java
+++ b/solr/core/src/java/org/apache/solr/cloud/OverseerTaskProcessor.java
@@ -28,7 +28,6 @@ import org.apache.solr.common.util.IOUtils;
import org.apache.solr.common.util.StrUtils;
import org.apache.solr.common.util.Utils;
import org.apache.solr.core.CoreContainer;
-import org.apache.solr.logging.MDCLoggingContext;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
@@ -144,145 +143,7 @@ public class OverseerTaskProcessor implements Runnable, Closeable {
@Override
public void run() {
- MDCLoggingContext.setNode(thisNode);
- log.debug("Process current queue of overseer operations");
- String oldestItemInWorkQueue = null;
- // hasLeftOverItems - used for avoiding re-execution of async tasks that were processed by a previous Overseer.
- // This variable is set in case there's any task found on the workQueue when the OCP starts up and
- // the id for the queue tail is used as a marker to check for the task in completed/failed map in zk.
- // Beyond the marker, all tasks can safely be assumed to have never been executed.
- boolean hasLeftOverItems = true;
-
- try {
- oldestItemInWorkQueue = workQueue.getTailId();
- } catch (KeeperException e) {
- // We don't need to handle this. This is just a fail-safe which comes in handy in skipping already processed
- // async calls.
- SolrException.log(log, "", e);
- } catch (Exception e) {
- ParWork.propagateInterrupt(e);
- if (e instanceof KeeperException.SessionExpiredException) {
- return;
- }
- if (e instanceof InterruptedException
- || e instanceof AlreadyClosedException) {
- return;
- }
- throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, e);
- }
-
- if (oldestItemInWorkQueue == null) hasLeftOverItems = false;
- else {
- if (log.isDebugEnabled()) log.debug("Found already existing elements in the work-queue. Last element: {}", oldestItemInWorkQueue);
- }
-
- while (!this.isClosed()) {
- try {
-
- if (log.isDebugEnabled()) log.debug("Cleaning up work-queue. #Running tasks: {} #Completed tasks: {}", runningTasksSize(), completedTasks.size());
- cleanUpWorkQueue();
-
- printTrackingMaps();
-
- boolean waited = false;
-
- while (runningTasksSize() > MAX_PARALLEL_TASKS) {
- synchronized (waitLock) {
- waitLock.wait(250);
- }
- waited = true;
- }
-
- if (waited) cleanUpWorkQueue();
-
- ArrayList<QueueEvent> heads = new ArrayList<>(blockedTasks.size() + MAX_PARALLEL_TASKS);
- if (log.isDebugEnabled()) log.debug("Add {} blocked tasks to process", blockedTasks.size());
- heads.addAll(blockedTasks.values());
- blockedTasks.clear(); // clear it now; may get refilled below.
-
- if (heads.size() < MAX_BLOCKED_TASKS) {
- //instead of reading MAX_PARALLEL_TASKS items always, we should only fetch as much as we can execute
- int toFetch = Math.min(MAX_BLOCKED_TASKS - heads.size(), MAX_PARALLEL_TASKS - runningTasksSize());
- if (log.isDebugEnabled()) log.debug("PeekTopN for {} items", toFetch);
- List<QueueEvent> newTasks = workQueue.peekTopN(toFetch, excludedTasks, 10);
- if (log.isDebugEnabled()) log.debug("Got {} tasks from work-queue : [{}]", newTasks.size(), newTasks);
- heads.addAll(newTasks);
- }
-
- taskBatch.batchId++;
-
- for (QueueEvent head : heads) {
-
- if (runningZKTasks.contains(head.getId())) {
- log.warn("Task found in running ZKTasks already, continuing");
- continue;
- }
-
- if (head.getBytes() == null) {
- log.info("Found empty entry id={} event={}", head.getId(), head.getWatchedEvent());
- continue;
- }
-
- final ZkNodeProps message = ZkNodeProps.load(head.getBytes());
- final String asyncId = message.getStr(ASYNC);
- if (hasLeftOverItems) {
- if (head.getId().equals(oldestItemInWorkQueue)) hasLeftOverItems = false;
- if (asyncId != null && (completedMap.contains(asyncId) || failureMap.contains(asyncId))) {
- if (log.isDebugEnabled()) log.debug("Found already processed task in workQueue, cleaning up. AsyncId [{}]", asyncId);
- workQueue.remove(head);
- continue;
- }
- }
- String operation = message.getStr(Overseer.QUEUE_OPERATION);
- if (operation == null) {
- log.error("Msg does not have required " + Overseer.QUEUE_OPERATION + ": {}", message);
- workQueue.remove(head);
- continue;
- }
- OverseerMessageHandler messageHandler = selector.selectOverseerMessageHandler(message);
- String op = message.getStr("operation");
- OverseerMessageHandler.Lock lock = null;
- lock = messageHandler.lockTask(message, taskBatch);
- if (lock == null) {
- if (log.isDebugEnabled()) log.debug("Exclusivity check failed for [{}]", message.toString());
- // we may end crossing the size of the MAX_BLOCKED_TASKS. They are fine
- if (blockedTasks.size() < MAX_BLOCKED_TASKS) blockedTasks.put(head.getId(), head);
- continue;
- }
- try {
- markTaskAsRunning(head, asyncId);
- if (log.isDebugEnabled()) {
- log.debug("Marked task [{}] as running", head.getId());
- }
- } catch (Exception e) {
- if (e instanceof KeeperException.SessionExpiredException || e instanceof InterruptedException) {
- ParWork.propagateInterrupt(e);
- log.error("ZooKeeper session has expired");
- return;
- }
-
- throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, e);
- }
- if (log.isDebugEnabled()) log.debug(messageHandler.getName() + ": Get the message id:" + head.getId() + " message:" + message.toString());
- Runner runner = new Runner(messageHandler, message, operation, head, lock);
- taskFutures.put(runner, ParWork.getRootSharedExecutor().submit(runner));
- }
-
- } catch (InterruptedException | AlreadyClosedException e) {
- ParWork.propagateInterrupt(e, true);
- return;
- } catch (KeeperException.SessionExpiredException e) {
- log.warn("Zookeeper expiration");
- return;
- } catch (Exception e) {
- log.error("Unexpected exception", e);
- }
- }
-
- if (log.isDebugEnabled()) {
- log.debug("run() - end");
- }
}
private int runningTasksSize() {
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 88f494b..a4d0b61 100644
--- a/solr/core/src/java/org/apache/solr/cloud/OverseerTaskQueue.java
+++ b/solr/core/src/java/org/apache/solr/cloud/OverseerTaskQueue.java
@@ -19,7 +19,6 @@ package org.apache.solr.cloud;
import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
import java.util.List;
-import java.util.TreeSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
@@ -160,7 +159,7 @@ public class OverseerTaskQueue extends ZkDistributedQueue {
}
// If latchEventType is not null, only fire if the type matches
- log.info("{} fired on path {} state {} latchEventType {}", event.getType(), event.getPath(), event.getState(), latchEventType);
+ if (log.isDebugEnabled()) log.debug("{} fired on path {} state {} latchEventType {}", event.getType(), event.getPath(), event.getState(), latchEventType);
if (latchEventType == null || event.getType() == latchEventType) {
lock.lock();
@@ -220,7 +219,7 @@ public class OverseerTaskQueue extends ZkDistributedQueue {
*/
public QueueEvent offer(byte[] data, long timeout) throws KeeperException,
InterruptedException {
- log.info("offer operation to the Overseeer queue {}", Utils.fromJSON(data));
+ if (log.isDebugEnabled()) log.debug("offer operation to the Overseeer queue {}", Utils.fromJSON(data));
if (shuttingDown.get()) {
throw new SolrException(SolrException.ErrorCode.CONFLICT,"Solr is shutting down, no more overseer tasks may be offered");
}
@@ -230,25 +229,25 @@ public class OverseerTaskQueue extends ZkDistributedQueue {
// otherwise we may miss the response.
String watchID = createResponseNode();
- log.info("watchId for response node {}, setting a watch ... ", watchID);
+ if (log.isDebugEnabled()) log.debug("watchId for response node {}, setting a watch ... ", watchID);
LatchWatcher watcher = new LatchWatcher(Watcher.Event.EventType.NodeDataChanged);
Stat stat = zookeeper.exists(watchID, watcher, true);
// create the request node
String path = createRequestNode(data, watchID);
- log.info("created request node at {}", path);
+ if (log.isDebugEnabled()) log.debug("created request node at {}", path);
if (stat != null) {
pendingResponses.incrementAndGet();
- log.info("wait on latch {}", timeout);
+ if (log.isDebugEnabled()) log.debug("wait on latch {}", timeout);
watcher.await(timeout);
}
byte[] bytes = zookeeper.getData(watchID, null, null, true);
- log.info("get data from response node {} {} {}", watchID, bytes == null ? null : bytes.length, watcher.getWatchedEvent());
+ if (log.isDebugEnabled()) log.debug("get data from response node {} {} {}", watchID, bytes == null ? null : bytes.length, watcher.getWatchedEvent());
// create the event before deleting the node, otherwise we can get the deleted
// event from the watcher.
QueueEvent event = new QueueEvent(watchID, bytes, watcher.getWatchedEvent());
- log.info("delete response node... {}", watchID);
+ if (log.isDebugEnabled()) log.debug("delete response node... {}", watchID);
// zookeeper.delete(watchID, -1, true);
return event;
} finally {
@@ -302,31 +301,6 @@ public class OverseerTaskQueue extends ZkDistributedQueue {
}
}
-
- /**
- *
- * Gets last element of the Queue without removing it.
- */
- public String getTailId() throws KeeperException, InterruptedException {
- log.info("getTailId");
- // TODO: could we use getChildren here? Unsure what freshness guarantee the caller needs.
- TreeSet<String> orderedChildren = fetchZkChildren(null);
- log.info("getTailId found {} children", orderedChildren);
- for (String headNode : orderedChildren.descendingSet())
- if (headNode != null) {
- try {
- QueueEvent queueEvent = new QueueEvent(dir + "/" + headNode, zookeeper.getData(dir + "/" + headNode,
- null, null, true), null);
- log.info("return {}", queueEvent.getId());
- return queueEvent.getId();
- } catch (KeeperException.NoNodeException e) {
- // Another client removed the node first, try next
- log.info("no node found {}", dir + "/" + headNode);
- }
- }
- return null;
- }
-
public static class QueueEvent {
@Override
public int hashCode() {
diff --git a/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContext.java b/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContext.java
index cde67b0..e3e52bc 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContext.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContext.java
@@ -123,7 +123,7 @@ final class ShardLeaderElectionContext extends ShardLeaderElectionContextBase {
String coreName = leaderProps.getName();
- log.info("Run leader process for shard election {}", coreName);
+ log.info("Run leader process for shard [{}] election, first step is to try and sync with the shard core={}", context.leaderProps.getSlice(), coreName);
try (SolrCore core = cc.getCore(coreName)) {
if (core == null) {
log.error("No SolrCore found, cannot become leader {}", coreName);
@@ -267,10 +267,9 @@ final class ShardLeaderElectionContext extends ShardLeaderElectionContextBase {
assert zkController != null;
assert zkController.getOverseer() != null;
- log.info("Publish leader state");
- zkController.publish(zkNodes);
+ log.info("I am the new leader, publishing as active: " + leaderProps.getCoreUrl() + " " + shardId);
- log.info("I am the new leader: " + leaderProps.getCoreUrl() + " " + shardId);
+ zkController.publish(zkNodes);
} catch (AlreadyClosedException | InterruptedException e) {
ParWork.propagateInterrupt("Already closed or interrupted, bailing..", e);
diff --git a/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContextBase.java b/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContextBase.java
index 402371e..3f10620 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContextBase.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContextBase.java
@@ -179,7 +179,7 @@ class ShardLeaderElectionContextBase extends ElectionContext {
SetDataResult dresult = (SetDataResult) result;
Stat stat = dresult.getStat();
leaderZkNodeParentVersion = stat.getVersion();
- log.info("Got leaderZkNodeParentVersion {}", leaderZkNodeParentVersion);
+ if (log.isDebugEnabled()) log.debug("Got leaderZkNodeParentVersion {}", leaderZkNodeParentVersion);
}
}
//assert leaderZkNodeParentVersion != null;
diff --git a/solr/core/src/java/org/apache/solr/cloud/StatePublisher.java b/solr/core/src/java/org/apache/solr/cloud/StatePublisher.java
index 9574f8d..6f98090 100644
--- a/solr/core/src/java/org/apache/solr/cloud/StatePublisher.java
+++ b/solr/core/src/java/org/apache/solr/cloud/StatePublisher.java
@@ -64,7 +64,7 @@ public class StatePublisher implements Closeable {
try {
message = workQueue.poll(5, TimeUnit.SECONDS);
if (message != null) {
- log.info("Got state message " + message);
+ if (log.isDebugEnabled()) log.debug("Got state message " + message);
if (message == TERMINATE_OP) {
return;
}
@@ -73,7 +73,7 @@ public class StatePublisher implements Closeable {
while (message != null) {
message = workQueue.poll(0, TimeUnit.SECONDS);
- log.info("Got state message " + message);
+ if (log.isDebugEnabled()) log.debug("Got state message " + message);
if (message != null) {
if (message == TERMINATE_OP) {
return;
diff --git a/solr/core/src/java/org/apache/solr/cloud/ZkController.java b/solr/core/src/java/org/apache/solr/cloud/ZkController.java
index cdda8ef..acebbf1 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ZkController.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ZkController.java
@@ -667,7 +667,7 @@ public class ZkController implements Closeable, Runnable {
}
public void disconnect(boolean publishDown) {
- log.info("disconnect");
+ if (log.isDebugEnabled()) log.debug("disconnect");
this.dcCalled = true;
try (ParWork closer = new ParWork(this, true, true)) {
closer.collect( "replicateFromLeaders", replicateFromLeaders);
@@ -703,7 +703,7 @@ public class ZkController implements Closeable, Runnable {
* Closes the underlying ZooKeeper client.
*/
public void close() {
- log.info("Closing ZkController");
+ if (log.isDebugEnabled()) log.debug("Closing ZkController");
//assert closeTracker.close();
IOUtils.closeQuietly(overseerElector);
@@ -718,16 +718,6 @@ public class ZkController implements Closeable, Runnable {
this.isClosed = true;
- try (ParWork closer = new ParWork(this, true, true)) {
- closer.collect(overseer);
- closer.collect(replicateFromLeaders);
- closer.collect(electionContexts);
- closer.collect(collectionToTerms);
- closer.collect(sysPropsCacher);
- closer.collect(cloudManager);
- closer.collect(cloudSolrClient);
- closer.collect(overseerContexts);
- }
try {
if (statePublisher != null) {
@@ -738,6 +728,17 @@ public class ZkController implements Closeable, Runnable {
log.error("Exception closing state publisher");
}
+
+ try (ParWork closer = new ParWork(this, true, true)) {
+ closer.collect(replicateFromLeaders);
+ closer.collect(electionContexts);
+ closer.collect(collectionToTerms);
+ closer.collect(sysPropsCacher);
+ closer.collect(cloudManager);
+ closer.collect(cloudSolrClient);
+ closer.collect(overseerContexts);
+ }
+
IOUtils.closeQuietly(zkStateReader);
if (closeZkClient) {
@@ -1466,7 +1467,7 @@ public class ZkController implements Closeable, Runnable {
throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR, "", e);
}
- log.info("Wait to see leader for {}, {}", collection, shardId);
+ if (log.isDebugEnabled()) log.debug("Wait to see leader for {}, {}", collection, shardId);
Replica leader = null;
for (int i = 0; i < 15; i++) {
try {
diff --git a/solr/core/src/java/org/apache/solr/cloud/api/collections/OverseerRoleCmd.java b/solr/core/src/java/org/apache/solr/cloud/api/collections/OverseerRoleCmd.java
index ba55cd3..8527e7d 100644
--- a/solr/core/src/java/org/apache/solr/cloud/api/collections/OverseerRoleCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/api/collections/OverseerRoleCmd.java
@@ -24,7 +24,6 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
-import org.apache.solr.cloud.OverseerNodePrioritizer;
import org.apache.solr.common.cloud.ClusterState;
import org.apache.solr.common.cloud.SolrZkClient;
import org.apache.solr.common.cloud.ZkNodeProps;
@@ -45,14 +44,12 @@ public class OverseerRoleCmd implements OverseerCollectionMessageHandler.Cmd {
private final OverseerCollectionMessageHandler ocmh;
private final CollectionAction operation;
- private final OverseerNodePrioritizer overseerPrioritizer;
- public OverseerRoleCmd(OverseerCollectionMessageHandler ocmh, CollectionAction operation, OverseerNodePrioritizer prioritizer) {
+ public OverseerRoleCmd(OverseerCollectionMessageHandler ocmh, CollectionAction operation) {
this.ocmh = ocmh;
this.operation = operation;
- this.overseerPrioritizer = prioritizer;
}
@Override
diff --git a/solr/core/src/java/org/apache/solr/core/CoreContainer.java b/solr/core/src/java/org/apache/solr/core/CoreContainer.java
index f05dafe..6983ab5 100644
--- a/solr/core/src/java/org/apache/solr/core/CoreContainer.java
+++ b/solr/core/src/java/org/apache/solr/core/CoreContainer.java
@@ -990,7 +990,6 @@ public class CoreContainer implements Closeable {
@Override
public void close() throws IOException {
assert closeTracker.close();
- log.info("Closing CoreContainer");
isShutDown = true;
if (solrCores != null) {
diff --git a/solr/core/src/java/org/apache/solr/core/SolrCores.java b/solr/core/src/java/org/apache/solr/core/SolrCores.java
index 68a2827..1290637 100644
--- a/solr/core/src/java/org/apache/solr/core/SolrCores.java
+++ b/solr/core/src/java/org/apache/solr/core/SolrCores.java
@@ -90,7 +90,7 @@ class SolrCores implements Closeable {
// We are shutting down. You can't hold the lock on the various lists of cores while they shut down, so we need to
// make a temporary copy of the names and shut them down outside the lock.
public void close() {
- log.info("Closing SolrCores");
+ if (log.isDebugEnabled()) log.debug("Closing SolrCores");
this.closed = true;
waitForLoadingCoresToFinish(15000);
diff --git a/solr/core/src/java/org/apache/solr/handler/admin/CollectionsHandler.java b/solr/core/src/java/org/apache/solr/handler/admin/CollectionsHandler.java
index 06b54ce..1728898 100644
--- a/solr/core/src/java/org/apache/solr/handler/admin/CollectionsHandler.java
+++ b/solr/core/src/java/org/apache/solr/handler/admin/CollectionsHandler.java
@@ -318,7 +318,7 @@ public class CollectionsHandler extends RequestHandlerBase implements Permission
if (operation.sendToOCPQueue) {
- log.info("send request to Overseer queue and wait for response ... " + props);
+ if (log.isDebugEnabled()) log.debug("send request to Overseer queue and wait for response ... " + props);
ZkNodeProps zkProps = new ZkNodeProps(props);
SolrResponse overseerResponse = sendToOCPQueue(zkProps, operation.timeOut);
rsp.getValues().addAll(overseerResponse.getResponse());
@@ -327,10 +327,10 @@ public class CollectionsHandler extends RequestHandlerBase implements Permission
log.error("Exception", exp);
rsp.setException(exp);
}
- log.info("Overseer is done, response={}", rsp.getValues());
+ if (log.isDebugEnabled()) log.debug("Overseer is done, response={}", rsp.getValues());
} else {
// submits and doesn't wait for anything (no response)
- log.info("send request to Overseer queue and don't wait for anything (no response) ... " + props);
+ if (log.isDebugEnabled()) log.debug("send request to Overseer queue and don't wait for anything (no response) ... " + props);
coreContainer.getZkController().getOverseer().offerStateUpdate(Utils.toJSON(props));
}
@@ -346,13 +346,13 @@ public class CollectionsHandler extends RequestHandlerBase implements Permission
}
public SolrResponse sendToOCPQueue(ZkNodeProps m, long timeout) throws KeeperException, InterruptedException {
- log.info("send message to OCP {}", m);
+ if (log.isDebugEnabled()) log.debug("send message to OCP {}", m);
String operation = m.getStr(QUEUE_OPERATION);
if (operation == null) {
throw new SolrException(ErrorCode.BAD_REQUEST, "missing key " + QUEUE_OPERATION);
}
if (m.get(ASYNC) != null) {
- log.info("request is async");
+ if (log.isDebugEnabled()) log.debug("request is async");
String asyncId = m.getStr(ASYNC);
if (asyncId.equals("-1")) {
@@ -387,17 +387,17 @@ public class CollectionsHandler extends RequestHandlerBase implements Permission
return new OverseerSolrResponse(r);
}
- log.info("request is not async");
+ if (log.isDebugEnabled()) log.debug("request is not async");
long time = System.nanoTime();
QueueEvent event = coreContainer.getZkController()
.getOverseerCollectionQueue()
.offer(Utils.toJSON(m), timeout);
if (event.getBytes() != null) {
- log.info("got a response, lets deserialize {}", Utils.toJSON(m));
+ if (log.isDebugEnabled()) log.debug("got a response, lets deserialize {}", Utils.toJSON(m));
return OverseerSolrResponseSerializer.deserialize(event.getBytes());
} else {
- log.info("no data in response, checking for timeout");
+ if (log.isDebugEnabled()) log.debug("no data in response, checking for timeout");
if (System.nanoTime() - time >= TimeUnit.NANOSECONDS.convert(timeout, TimeUnit.MILLISECONDS)) {
throw new SolrException(ErrorCode.SERVER_ERROR, operation
+ " the collection time out:" + timeout / 1000 + "s");
@@ -412,7 +412,7 @@ public class CollectionsHandler extends RequestHandlerBase implements Permission
// nocommit - look into we may still need this
// we have to assume success - it was too quick for us to catch the response
- log.info("We did not find the response, there was also no timeout and we did not get a watched event ...");
+ if (log.isDebugEnabled()) log.debug("We did not find the response, there was also no timeout and we did not get a watched event ...");
NamedList<Object> resp = new NamedList<>();
resp.add("success", "true");
diff --git a/solr/core/src/java/org/apache/solr/handler/component/SpellCheckComponent.java b/solr/core/src/java/org/apache/solr/handler/component/SpellCheckComponent.java
index c301826..367febb 100644
--- a/solr/core/src/java/org/apache/solr/handler/component/SpellCheckComponent.java
+++ b/solr/core/src/java/org/apache/solr/handler/component/SpellCheckComponent.java
@@ -833,8 +833,8 @@ public class SpellCheckComponent extends SearchComponent implements SolrCoreAwar
if (currentSearcher == null) {
// firstSearcher event
try {
- if (log.isInfoEnabled()) {
- log.info("Loading spell index for spellchecker: {}", checker.getDictionaryName());
+ if (log.isDebugEnabled()) {
+ log.debug("Loading spell index for spellchecker: {}", checker.getDictionaryName());
}
checker.reload(core, newSearcher);
} catch (IOException e) {
diff --git a/solr/core/src/java/org/apache/solr/servlet/HttpSolrCall.java b/solr/core/src/java/org/apache/solr/servlet/HttpSolrCall.java
index 87b2922..ca72919 100644
--- a/solr/core/src/java/org/apache/solr/servlet/HttpSolrCall.java
+++ b/solr/core/src/java/org/apache/solr/servlet/HttpSolrCall.java
@@ -361,7 +361,7 @@ public class HttpSolrCall {
for (Map.Entry<String,Integer> entry : entries) {
String collection = entry.getKey();
Integer version = entry.getValue();
- log.info("ensure states are at at least client version {} for collection {}", version, collection);
+ if (log.isDebugEnabled()) log.debug("ensure states are at at least client version {} for collection {}", version, collection);
DocCollection docCollection = cores.getZkController().getZkStateReader().getClusterState().getCollectionOrNull(collection);
if (docCollection != null && docCollection.getZNodeVersion() < version) {
cores.getZkController().getZkStateReader().waitForState(collection, 5, TimeUnit.SECONDS, (liveNodes, collectionState) -> {
@@ -1034,7 +1034,7 @@ public class HttpSolrCall {
for (String pair : pairs) {
String[] pcs = StringUtils.split(pair, ':');
if (pcs.length == 2 && !pcs[0].isEmpty() && !pcs[1].isEmpty()) {
- log.info("compare version states {} {}", pcs[0], Integer.parseInt(pcs[1]));
+ if (log.isDebugEnabled()) log.debug("compare version states {} {}", pcs[0], Integer.parseInt(pcs[1]));
if (result == null) result = new HashMap<>();
result.put(pcs[0], Integer.parseInt(pcs[1]));
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/BaseCloudSolrClient.java b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/BaseCloudSolrClient.java
index 786653c..bcd0e94 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/BaseCloudSolrClient.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/BaseCloudSolrClient.java
@@ -882,7 +882,7 @@ public abstract class BaseCloudSolrClient extends SolrClient {
}
StringBuilder stateVerParamBuilder = null;
- log.info("build version params for collections {}", requestedCollectionNames);
+ if (log.isDebugEnabled()) log.debug("build version params for collections {}", requestedCollectionNames);
for (String requestedCollection : requestedCollectionNames) {
// track the version of state we're using on the client side using the _stateVer_ param
DocCollection coll = getZkStateReader().getClusterState().getCollectionOrNull(requestedCollection);
diff --git a/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java b/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java
index 407833a..75d57b9 100644
--- a/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java
+++ b/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java
@@ -411,7 +411,7 @@ public class ZkStateReader implements SolrCloseable, Replica.NodeNameToBaseUrl {
throw new AlreadyClosedException();
}
- log.info("createClusterStateWatchersAndUpdate");
+ if (log.isDebugEnabled()) log.debug("createClusterStateWatchersAndUpdate");
CountDownLatch latch = new CountDownLatch(1);
Watcher watcher = new Watcher() {
@@ -421,7 +421,7 @@ public class ZkStateReader implements SolrCloseable, Replica.NodeNameToBaseUrl {
if (EventType.None.equals(event.getType())) {
return;
}
- log.info("Got event on live node watcher {}", event.toString());
+ if (log.isDebugEnabled()) log.debug("Got event on live node watcher {}", event.toString());
if (event.getType() == EventType.NodeCreated) {
latch.countDown();
} else {
@@ -444,13 +444,13 @@ public class ZkStateReader implements SolrCloseable, Replica.NodeNameToBaseUrl {
try {
Stat stat = zkClient.exists("/cluster/init", watcher);
if (stat == null) {
- log.info("Collections znode not found, waiting on latch");
+ if (log.isDebugEnabled()) log.debug("Collections znode not found, waiting on latch");
try {
boolean success = latch.await(10000, TimeUnit.MILLISECONDS);
if (!success) {
log.warn("Timed waiting to see {} node in zk \n{}", "/cluster/init", clusterState);
}
- log.info("Done waiting on latch");
+ if (log.isDebugEnabled()) log.debug("Done waiting on latch");
} catch (InterruptedException e) {
log.warn("", e);
return;
@@ -820,7 +820,7 @@ public class ZkStateReader implements SolrCloseable, Replica.NodeNameToBaseUrl {
}
public void close() {
- log.info("Closing ZkStateReader");
+ if (log.isDebugEnabled()) log.debug("Closing ZkStateReader");
assert closeTracker.close();
this.closed = true;
if (notifications != null) {
@@ -1323,13 +1323,13 @@ public class ZkStateReader implements SolrCloseable, Replica.NodeNameToBaseUrl {
}
public void watchStateUpdates() {
- log.info("watch for additional state updates {}", coll);
+ if (log.isDebugEnabled()) log.debug("watch for additional state updates {}", coll);
String stateUpdatesPath = ZkStateReader.getCollectionStateUpdatesPath(coll);
watcher = new Watcher() {
@Override
public void process(WatchedEvent event) {
- log.info("_statupdates event {}", event);
+ if (log.isDebugEnabled()) log.debug("_statupdates event {}", event);
try {
// if (event.getType() == EventType.NodeDataChanged ||
@@ -1420,7 +1420,7 @@ public class ZkStateReader implements SolrCloseable, Replica.NodeNameToBaseUrl {
} else {
state = Replica.State.ACTIVE;
}
- log.info("Got additional state update {} {}", core, state == null ? "leader" : state);
+ if (log.isDebugEnabled()) log.debug("Got additional state update {} {}", core, state == null ? "leader" : state);
Replica replica = docCollection.getReplica(core);
if (replica != null) {