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/10/12 21:30:03 UTC
[lucene-solr] 03/05: @973 Bring back early limited,
fast count based retry on connection loss.
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 bed65e566625108a83c5e3861efebe4f755c2533
Author: markrmiller@gmail.com <ma...@gmail.com>
AuthorDate: Mon Oct 12 15:55:45 2020 -0500
@973 Bring back early limited, fast count based retry on connection loss.
---
.../org/apache/solr/common/cloud/SolrZkClient.java | 102 ++++++++++++++++-----
.../apache/solr/common/cloud/ZkCmdExecutor.java | 34 ++-----
.../solr/common/cloud/ZkMaintenanceUtils.java | 3 +-
3 files changed, 89 insertions(+), 50 deletions(-)
diff --git a/solr/solrj/src/java/org/apache/solr/common/cloud/SolrZkClient.java b/solr/solrj/src/java/org/apache/solr/common/cloud/SolrZkClient.java
index 851e33d..d776952 100644
--- a/solr/solrj/src/java/org/apache/solr/common/cloud/SolrZkClient.java
+++ b/solr/solrj/src/java/org/apache/solr/common/cloud/SolrZkClient.java
@@ -97,6 +97,8 @@ public class SolrZkClient implements Closeable {
private final ConnectionManager connManager;
+ private ZkCmdExecutor zkCmdExecutor;
+
// what about ensuring order of state updates per collection??
final ExecutorService zkCallbackExecutor = ParWork.getParExecutorService("zkCallbackExecutor", 1, 1, 60000, new BlockingArrayQueue());
@@ -156,6 +158,14 @@ public class SolrZkClient implements Closeable {
this.zkACLProvider = zkACLProvider;
}
+ zkCmdExecutor = new ZkCmdExecutor(3, new IsClosed() {
+
+ @Override
+ public boolean isClosed() {
+ return SolrZkClient.this.higherLevelIsClosed.isClosed() || isClosed;
+ }
+ });
+
connManager = new ConnectionManager("ZooKeeperConnection Watcher:"
+ zkServerAddress, this, zkServerAddress, zkClientTimeout, onReconnect, beforeReconnect);
@@ -229,23 +239,37 @@ public class SolrZkClient implements Closeable {
return keeper != null && keeper.getState() == ZooKeeper.States.CONNECTED;
}
- public void delete(final String path, final int version)
+ public void delete(final String path, final int version) throws KeeperException, InterruptedException {
+ delete(path, version, true);
+ }
+ public void delete(final String path, final int version, boolean retryOnConnLoss)
throws InterruptedException, KeeperException {
- ZooKeeper keeper = connManager.getKeeper();
+ ZooKeeper keeper = connManager.getKeeper();
+ if (retryOnConnLoss) {
+ zkCmdExecutor.retryOperation(() -> {
+ keeper.delete(path, version);
+ return null;
+ });
+ } else {
keeper.delete(path, version);
+ }
}
/**
* Wraps the watcher so that it doesn't fire off ZK's event queue. In order to guarantee that a watch object will
* only be triggered once for a given notification, users need to wrap their watcher using this method before
- * calling {@link #exists(String, org.apache.zookeeper.Watcher)} or
- * {@link #getData(String, org.apache.zookeeper.Watcher, org.apache.zookeeper.data.Stat)}.
+ * calling {@link #exists(String, org.apache.zookeeper.Watcher, boolean)} or
+ * {@link #getData(String, org.apache.zookeeper.Watcher, org.apache.zookeeper.data.Stat, boolean)}.
*/
public Watcher wrapWatcher(final Watcher watcher) {
if (watcher == null || watcher instanceof ProcessWatchWithExecutor) return watcher;
return new ProcessWatchWithExecutor(watcher, zkCallbackExecutor);
}
+ public Stat exists(final String path, final Watcher watcher) throws KeeperException, InterruptedException {
+ return exists(path, watcher, true);
+ }
+
/**
* Return the stat of the node of the given path. Return null if no such a
* node exists.
@@ -263,39 +287,61 @@ public class SolrZkClient implements Closeable {
* @throws InterruptedException If the server transaction is interrupted.
* @throws IllegalArgumentException if an invalid path is specified
*/
- public Stat exists(final String path, final Watcher watcher)
+ public Stat exists(final String path, final Watcher watcher, boolean retryOnConnLoss)
throws KeeperException, InterruptedException {
- ZooKeeper keeper = connManager.getKeeper();
+ ZooKeeper keeper = connManager.getKeeper();
+ if (retryOnConnLoss) {
+ return zkCmdExecutor.retryOperation(() -> keeper.exists(path, wrapWatcher(watcher)));
+ } else {
return keeper.exists(path, wrapWatcher(watcher));
-
+ }
}
/**
* Returns true if path exists
*/
- public Boolean exists(final String path)
+ public Boolean exists(final String path, boolean retryOnConnLoss)
throws KeeperException, InterruptedException {
- ZooKeeper keeper = connManager.getKeeper();
+ ZooKeeper keeper = connManager.getKeeper();
+ if (retryOnConnLoss) {
+ return zkCmdExecutor.retryOperation(() -> keeper.exists(path, null) != null);
+ } else {
return keeper.exists(path, null) != null;
+ }
}
- /**
- * Returns children of the node at the path
- */
+ public Boolean exists(final String path) throws KeeperException, InterruptedException {
+ return this.exists(path, true);
+ }
+
+ /**
+ * Returns children of the node at the path
+ */
public List<String> getChildren(final String path, final Watcher watcher, boolean retryOnConnLoss)
throws KeeperException, InterruptedException {
- ZooKeeper keeper = connManager.getKeeper();
+ ZooKeeper keeper = connManager.getKeeper();
+ if (retryOnConnLoss) {
+ return zkCmdExecutor.retryOperation(() -> keeper.getChildren(path, wrapWatcher(watcher)));
+ } else {
return keeper.getChildren(path, wrapWatcher(watcher));
+ }
}
- /**
- * Returns node's data
- */
- public byte[] getData(final String path, final Watcher watcher, final Stat stat)
+ public byte[] getData(final String path, final Watcher watcher, final Stat stat) throws KeeperException, InterruptedException {
+ return getData(path, watcher, stat, true);
+ }
+
+ /**
+ * Returns node's data
+ */
+ public byte[] getData(final String path, final Watcher watcher, final Stat stat, boolean retryOnConnLoss)
throws KeeperException, InterruptedException {
- ZooKeeper keeper = connManager.getKeeper();
- if (keeper == null) return null;
+ ZooKeeper keeper = connManager.getKeeper();
+ if (retryOnConnLoss) {
+ return zkCmdExecutor.retryOperation(() -> keeper.getData(path, wrapWatcher(watcher), stat));
+ } else {
return keeper.getData(path, wrapWatcher(watcher), stat);
+ }
}
/**
@@ -453,8 +499,6 @@ public class SolrZkClient implements Closeable {
Watcher watcher, boolean failOnExists, boolean retryOnConnLoss, int skipPathParts) throws KeeperException, InterruptedException {
log.error("mkdir path={}", path);
log.debug("makePath: {}", path);
- boolean retry = false;
- retryOnConnLoss = false;
if (path.startsWith("/")) {
path = path.substring(1, path.length());
@@ -474,14 +518,22 @@ public class SolrZkClient implements Closeable {
if (i == paths.length - 1) {
mode = createMode;
bytes = data;
- if (!retryOnConnLoss) retry = false;
}
ZooKeeper keeper = connManager.getKeeper();
try {
- keeper.create(currentPath, bytes, zkACLProvider.getACLsToAdd(currentPath), mode);
+ if (retryOnConnLoss) {
+ final CreateMode finalMode = mode;
+ final byte[] finalBytes = bytes;
+ zkCmdExecutor.retryOperation(() -> {
+ keeper.create(currentPath, finalBytes, zkACLProvider.getACLsToAdd(currentPath), finalMode);
+ return null;
+ });
+ } else {
+ keeper.create(currentPath, bytes, zkACLProvider.getACLsToAdd(currentPath), mode);
+ }
} catch (NoAuthException e) {
// in auth cases, we may not have permission for an earlier part of a path, which is fine
- if (i == paths.length - 1 || !exists(currentPath)) {
+ if (i == paths.length - 1 || !exists(currentPath, retryOnConnLoss)) {
throw e;
}
@@ -491,7 +543,7 @@ public class SolrZkClient implements Closeable {
// TODO: version ? for now, don't worry about race
setData(currentPath, data, -1, retryOnConnLoss);
// set new watch
- exists(currentPath, watcher);
+ exists(currentPath, watcher, retryOnConnLoss);
return;
}
diff --git a/solr/solrj/src/java/org/apache/solr/common/cloud/ZkCmdExecutor.java b/solr/solrj/src/java/org/apache/solr/common/cloud/ZkCmdExecutor.java
index 3971048..431b058 100644
--- a/solr/solrj/src/java/org/apache/solr/common/cloud/ZkCmdExecutor.java
+++ b/solr/solrj/src/java/org/apache/solr/common/cloud/ZkCmdExecutor.java
@@ -16,6 +16,7 @@
*/
package org.apache.solr.common.cloud;
+import org.apache.solr.common.SolrException;
import org.apache.solr.common.cloud.ConnectionManager.IsClosed;
import org.apache.solr.common.util.TimeOut;
import org.apache.solr.common.util.TimeSource;
@@ -32,23 +33,21 @@ public class ZkCmdExecutor {
private long retryDelay = 50L;
private int retryCount;
- private long timeoutms;
private IsClosed isClosed;
- public ZkCmdExecutor(int timeoutms) {
- this(timeoutms, null);
+ public ZkCmdExecutor(int retryCount) {
+ this(retryCount, null);
}
/**
* TODO: At this point, this should probably take a SolrZkClient in
* its constructor.
*
- * @param timeoutms
- * the client timeout for the ZooKeeper clients that will be used
- * with this class.
+ * @param retryCount
+ * number of retries on connectionloss
*/
- public ZkCmdExecutor(long timeoutms, IsClosed isClosed) {
- this.timeoutms = timeoutms;
+ public ZkCmdExecutor(int retryCount, IsClosed isClosed) {
+ this.retryCount = retryCount;
this.isClosed = isClosed;
}
@@ -68,34 +67,21 @@ public class ZkCmdExecutor {
public <T> T retryOperation(ZkOperation operation)
throws KeeperException, InterruptedException {
KeeperException exception = null;
- TimeOut timeout = new TimeOut(timeoutms, TimeUnit.MILLISECONDS, TimeSource.NANO_TIME);
int tryCnt = 0;
- while (true) {
+ while (tryCnt < retryCount) {
try {
-
- if (timeout.hasTimedOut()) {
- throw new RuntimeException("Timed out attempting zk call");
- }
-
return (T) operation.execute();
} catch (KeeperException.ConnectionLossException e) {
log.warn("ConnectionLost to ZK");
if (exception == null) {
exception = e;
}
- if (Thread.currentThread().isInterrupted()) {
- Thread.currentThread().interrupt();
- throw new InterruptedException();
- }
-
- if (timeout.hasTimedOut()) {
- throw new RuntimeException("Timed out attempting zk call");
- }
retryDelay(tryCnt);
}
tryCnt++;
}
+ throw exception;
}
/**
@@ -106,7 +92,7 @@ public class ZkCmdExecutor {
*/
protected void retryDelay(int attemptCount) throws InterruptedException {
long sleep = retryDelay;
- log.info("delaying for retry, attempt={} retryDelay={} sleep={} timeout={}", attemptCount, retryDelay, sleep, timeoutms);
+ log.info("delaying for retry, attempt={} retryDelay={} sleep={}", attemptCount, retryDelay, sleep);
Thread.sleep(sleep);
}
diff --git a/solr/solrj/src/java/org/apache/solr/common/cloud/ZkMaintenanceUtils.java b/solr/solrj/src/java/org/apache/solr/common/cloud/ZkMaintenanceUtils.java
index 5ad21ab..44a93b3 100644
--- a/solr/solrj/src/java/org/apache/solr/common/cloud/ZkMaintenanceUtils.java
+++ b/solr/solrj/src/java/org/apache/solr/common/cloud/ZkMaintenanceUtils.java
@@ -19,6 +19,7 @@ package org.apache.solr.common.cloud;
import org.apache.solr.client.solrj.SolrServerException;
import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -339,7 +340,7 @@ public class ZkMaintenanceUtils {
}
private static boolean isEphemeral(SolrZkClient zkClient, String zkPath) throws KeeperException, InterruptedException {
- Stat znodeStat = zkClient.exists(zkPath, null);
+ Stat znodeStat = zkClient.exists(zkPath, null, true);
return znodeStat.getEphemeralOwner() != 0;
}