You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@accumulo.apache.org by "keith-turner (via GitHub)" <gi...@apache.org> on 2023/06/07 17:26:38 UTC
[GitHub] [accumulo] keith-turner commented on a diff in pull request #3230: Stop upgrade process if invalid table locks are found.
keith-turner commented on code in PR #3230:
URL: https://github.com/apache/accumulo/pull/3230#discussion_r1221941203
##########
server/manager/src/main/java/org/apache/accumulo/manager/upgrade/PreUpgradeValidation.java:
##########
@@ -112,9 +140,163 @@ private static String extractAuthName(ACL acl) {
@SuppressFBWarnings(value = "DM_EXIT",
justification = "Want to immediately stop all threads on upgrade error")
- private void fail(Exception e) {
+ protected void fail(Exception e) {
log.error("FATAL: Error performing pre-upgrade checks", e);
System.exit(1);
}
+ private void validateTableLocks(final ServerContext context) {
+
+ final ZooReaderWriter zrw = context.getZooReaderWriter();
+ final ZooKeeper zk = zrw.getZooKeeper();
+ final String rootPath = context.getZooKeeperRoot();
+ final String tserverLockRoot = rootPath + Constants.ZTSERVERS;
+
+ log.debug("Looking for locks that may be from previous version in path: {}", tserverLockRoot);
+
+ List<Pair<HostAndPort,ServiceLock.ServiceLockPath>> hostsWithLocks =
+ gatherLocks(zk, tserverLockRoot);
+
+ int numCheckThreads = Math.max(32, Runtime.getRuntime().availableProcessors() - 2);
+ ThreadPoolExecutor lockCheckPool = ThreadPools.getServerThreadPools().createThreadPool(8,
+ numCheckThreads, 10, MINUTES, "update-lock-check", false);
+
+ final Map<String,String> invalidLocks = new ConcurrentHashMap<>(hostsWithLocks.size());
+ // try a thrift call to the hosts - hosts running previous versions will fail the call
+ hostsWithLocks.forEach(hostLockPair -> lockCheckPool.execute(() -> {
+ tryThriftCall(context, invalidLocks, hostLockPair.getFirst(), hostLockPair.getSecond());
+ if (Thread.currentThread().isInterrupted()) {
+ throw new IllegalStateException("Interrupted try to make thrift call to check locks");
+ }
+ }));
+
+ lockCheckPool.shutdown();
+ boolean lockCheckTimeout = false;
+ final long blockTimeMinutes = 60;
+ final long lockCheckStart = System.nanoTime();
+ try {
+ // wait to all to finish
+ log.info("waiting for lock check tasks to complete. May block for up to {} minutes",
+ blockTimeMinutes);
+ if (!lockCheckPool.awaitTermination(blockTimeMinutes, MINUTES)) {
+ log.warn(
+ "Timed out waiting for lock check to finish - continuing, but tservers running prior versions may be present");
+ lockCheckTimeout = true;
+ }
+ } catch (InterruptedException ex) {
+ var remaining = lockCheckPool.shutdownNow();
+ log.warn("Interrupted waiting for lock check to finish. {} tasks were remaining.",
+ remaining.size());
+ lockCheckTimeout = true;
+ }
+ log.info("tserver lock completed in {} minutes",
+ NANOSECONDS.toMinutes(System.nanoTime() - lockCheckStart));
+ if (invalidLocks.size() == 0) {
+ log.info("All tserver locks found were processed and are valid.");
+ } else {
+ log.warn("Lock check completed with {} inlaid lock(s) found", invalidLocks.size());
+ Map<String,String> sorted = new TreeMap<>(invalidLocks);
+ sorted.forEach((p, v) -> {
+ log.warn("Invalid tserver lock: {}", p);
+ });
+ }
+
+ if (context.getConfiguration().getBoolean(Property.GENERAL_UPGRADE_VERSION_CHECK_ENABLED)) {
+ if (lockCheckTimeout || !invalidLocks.isEmpty()) {
+ throw new IllegalStateException(
+ "Failed to complete tserver lock check - cannot validate that all tservers running current version");
+ }
+ }
+ }
+
+ private static void tryThriftCall(ServerContext context, Map<String,String> invalidLocks,
+ HostAndPort host, ServiceLock.ServiceLockPath lockPath) {
+
+ Retry retry = Retry.builder().maxRetries(10).retryAfter(100, MILLISECONDS)
+ .incrementBy(250, MILLISECONDS).maxWait(10, SECONDS).backOffFactor(1.07)
+ .logInterval(1, SECONDS).createFactory().createRetry();
+
+ while (retry.canRetry()) {
+ try (TTransport transport = ThriftUtil.createTransport(host, context)) {
+ log.trace("found valid lock at: {}", lockPath);
+ return;
+ } catch (TException ex) {
+ log.trace("Could not establish a thrift connection, service lock path: {}", lockPath, ex);
+ }
+ // pause for next rty
+ try {
+ retry.useRetry();
+ retry.waitForNextAttempt(log,
+ "retrying thrift call to host: " + host + " lock path: " + lockPath);
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ invalidLocks.put(lockPath.toString(), "");
+ return;
+ } catch (Exception ex) {
+ log.debug("Failed to establish a connection for to service holding lock. {}", lockPath, ex);
+ invalidLocks.put(lockPath.toString(), "");
+ return;
+ }
+ }
+ log.debug("Failed to establish a connection for to service holding lock. {}. no retires left.",
+ lockPath);
+ invalidLocks.put(lockPath.toString(), "");
+ }
+
+ private List<Pair<HostAndPort,ServiceLock.ServiceLockPath>> gatherLocks(final ZooKeeper zk,
+ final String zkPathRoot) {
+ List<Pair<HostAndPort,ServiceLock.ServiceLockPath>> hosts = new ArrayList<>();
+ try {
+ ZKUtil.visitSubTreeDFS(zk, zkPathRoot, false, (rc, path, ctx, name) -> {
+ if (name.startsWith(ServiceLock.ZLOCK_PREFIX)) {
+ log.trace("found lock at {}", path);
+ try {
+ Stat stat = new Stat();
+ final var zLockPath = ServiceLock.path(path);
+ byte[] lockData = zk.getData(zLockPath.toString(), false, stat);
+ if (lockData == null || !validLockData(lockData, hosts, zLockPath)) {
+ log.debug("Invalid lock data - trying to delete path: {}", zLockPath);
+ zk.delete(zLockPath.toString(), stat.getVersion());
+ }
+ } catch (KeeperException.NoNodeException ex) {
+ // empty - lock no longer exists.
+ } catch (KeeperException ex) {
+ log.trace("could not read lock from ZooKeeper for {}, skipping", path, ex);
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(
+ "interrupted reading lock from ZooKeeper for path: " + path);
+ }
+ }
+ });
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(
+ "interrupted reading lock from ZooKeeper for path: " + zkPathRoot);
+ } catch (KeeperException ex) {
+ log.trace("could not read lock all locks from ZooKeeper for {}, lost list may be incomplete",
+ zkPathRoot, ex);
+ }
+ return hosts;
+ }
+
+ /**
+ * Decode the lock data and if valid add the host, port pair to the hosts map. If the data cannot
+ * be read return false to indicate invalid data.
+ */
+ private boolean validLockData(final byte[] lockData,
+ List<Pair<HostAndPort,ServiceLock.ServiceLockPath>> hosts,
+ ServiceLock.ServiceLockPath zLockPath) {
+ try {
+ Optional<ServiceLockData> sld = ServiceLockData.parse(lockData);
+ if (sld.isPresent()) {
+ HostAndPort hostAndPort = sld.orElseThrow().getAddress(ServiceLockData.ThriftService.TSERV);
+ hosts.add(new Pair<>(hostAndPort, zLockPath));
+ }
+ return true;
+ } catch (Exception ex) {
+ log.trace("Invalid lock data for path: {}", zLockPath, ex);
Review Comment:
Unless this expected for some reason, could log at a higher level.
```suggestion
log.warn("Invalid lock data for path: {}", zLockPath, ex);
```
##########
server/manager/src/main/java/org/apache/accumulo/manager/upgrade/PreUpgradeValidation.java:
##########
@@ -112,9 +140,163 @@ private static String extractAuthName(ACL acl) {
@SuppressFBWarnings(value = "DM_EXIT",
justification = "Want to immediately stop all threads on upgrade error")
- private void fail(Exception e) {
+ protected void fail(Exception e) {
log.error("FATAL: Error performing pre-upgrade checks", e);
System.exit(1);
}
+ private void validateTableLocks(final ServerContext context) {
+
+ final ZooReaderWriter zrw = context.getZooReaderWriter();
+ final ZooKeeper zk = zrw.getZooKeeper();
+ final String rootPath = context.getZooKeeperRoot();
+ final String tserverLockRoot = rootPath + Constants.ZTSERVERS;
+
+ log.debug("Looking for locks that may be from previous version in path: {}", tserverLockRoot);
+
+ List<Pair<HostAndPort,ServiceLock.ServiceLockPath>> hostsWithLocks =
+ gatherLocks(zk, tserverLockRoot);
+
+ int numCheckThreads = Math.max(32, Runtime.getRuntime().availableProcessors() - 2);
+ ThreadPoolExecutor lockCheckPool = ThreadPools.getServerThreadPools().createThreadPool(8,
+ numCheckThreads, 10, MINUTES, "update-lock-check", false);
+
+ final Map<String,String> invalidLocks = new ConcurrentHashMap<>(hostsWithLocks.size());
+ // try a thrift call to the hosts - hosts running previous versions will fail the call
+ hostsWithLocks.forEach(hostLockPair -> lockCheckPool.execute(() -> {
+ tryThriftCall(context, invalidLocks, hostLockPair.getFirst(), hostLockPair.getSecond());
+ if (Thread.currentThread().isInterrupted()) {
+ throw new IllegalStateException("Interrupted try to make thrift call to check locks");
+ }
+ }));
+
+ lockCheckPool.shutdown();
+ boolean lockCheckTimeout = false;
+ final long blockTimeMinutes = 60;
+ final long lockCheckStart = System.nanoTime();
+ try {
+ // wait to all to finish
+ log.info("waiting for lock check tasks to complete. May block for up to {} minutes",
+ blockTimeMinutes);
+ if (!lockCheckPool.awaitTermination(blockTimeMinutes, MINUTES)) {
+ log.warn(
+ "Timed out waiting for lock check to finish - continuing, but tservers running prior versions may be present");
+ lockCheckTimeout = true;
+ }
+ } catch (InterruptedException ex) {
+ var remaining = lockCheckPool.shutdownNow();
+ log.warn("Interrupted waiting for lock check to finish. {} tasks were remaining.",
+ remaining.size());
+ lockCheckTimeout = true;
+ }
+ log.info("tserver lock completed in {} minutes",
+ NANOSECONDS.toMinutes(System.nanoTime() - lockCheckStart));
+ if (invalidLocks.size() == 0) {
+ log.info("All tserver locks found were processed and are valid.");
+ } else {
+ log.warn("Lock check completed with {} inlaid lock(s) found", invalidLocks.size());
+ Map<String,String> sorted = new TreeMap<>(invalidLocks);
+ sorted.forEach((p, v) -> {
+ log.warn("Invalid tserver lock: {}", p);
+ });
+ }
+
+ if (context.getConfiguration().getBoolean(Property.GENERAL_UPGRADE_VERSION_CHECK_ENABLED)) {
+ if (lockCheckTimeout || !invalidLocks.isEmpty()) {
+ throw new IllegalStateException(
+ "Failed to complete tserver lock check - cannot validate that all tservers running current version");
+ }
+ }
+ }
+
+ private static void tryThriftCall(ServerContext context, Map<String,String> invalidLocks,
+ HostAndPort host, ServiceLock.ServiceLockPath lockPath) {
+
+ Retry retry = Retry.builder().maxRetries(10).retryAfter(100, MILLISECONDS)
+ .incrementBy(250, MILLISECONDS).maxWait(10, SECONDS).backOffFactor(1.07)
+ .logInterval(1, SECONDS).createFactory().createRetry();
+
+ while (retry.canRetry()) {
+ try (TTransport transport = ThriftUtil.createTransport(host, context)) {
+ log.trace("found valid lock at: {}", lockPath);
+ return;
+ } catch (TException ex) {
+ log.trace("Could not establish a thrift connection, service lock path: {}", lockPath, ex);
+ }
+ // pause for next rty
+ try {
+ retry.useRetry();
+ retry.waitForNextAttempt(log,
+ "retrying thrift call to host: " + host + " lock path: " + lockPath);
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ invalidLocks.put(lockPath.toString(), "");
+ return;
+ } catch (Exception ex) {
+ log.debug("Failed to establish a connection for to service holding lock. {}", lockPath, ex);
+ invalidLocks.put(lockPath.toString(), "");
+ return;
+ }
+ }
+ log.debug("Failed to establish a connection for to service holding lock. {}. no retires left.",
+ lockPath);
+ invalidLocks.put(lockPath.toString(), "");
+ }
+
+ private List<Pair<HostAndPort,ServiceLock.ServiceLockPath>> gatherLocks(final ZooKeeper zk,
+ final String zkPathRoot) {
+ List<Pair<HostAndPort,ServiceLock.ServiceLockPath>> hosts = new ArrayList<>();
+ try {
+ ZKUtil.visitSubTreeDFS(zk, zkPathRoot, false, (rc, path, ctx, name) -> {
+ if (name.startsWith(ServiceLock.ZLOCK_PREFIX)) {
+ log.trace("found lock at {}", path);
+ try {
+ Stat stat = new Stat();
+ final var zLockPath = ServiceLock.path(path);
+ byte[] lockData = zk.getData(zLockPath.toString(), false, stat);
+ if (lockData == null || !validLockData(lockData, hosts, zLockPath)) {
+ log.debug("Invalid lock data - trying to delete path: {}", zLockPath);
+ zk.delete(zLockPath.toString(), stat.getVersion());
Review Comment:
Should this throw an exception instead of delete the lock?
##########
server/manager/src/main/java/org/apache/accumulo/manager/upgrade/PreUpgradeValidation.java:
##########
@@ -112,9 +140,163 @@ private static String extractAuthName(ACL acl) {
@SuppressFBWarnings(value = "DM_EXIT",
justification = "Want to immediately stop all threads on upgrade error")
- private void fail(Exception e) {
+ protected void fail(Exception e) {
log.error("FATAL: Error performing pre-upgrade checks", e);
System.exit(1);
}
+ private void validateTableLocks(final ServerContext context) {
+
+ final ZooReaderWriter zrw = context.getZooReaderWriter();
+ final ZooKeeper zk = zrw.getZooKeeper();
+ final String rootPath = context.getZooKeeperRoot();
+ final String tserverLockRoot = rootPath + Constants.ZTSERVERS;
+
+ log.debug("Looking for locks that may be from previous version in path: {}", tserverLockRoot);
+
+ List<Pair<HostAndPort,ServiceLock.ServiceLockPath>> hostsWithLocks =
+ gatherLocks(zk, tserverLockRoot);
+
+ int numCheckThreads = Math.max(32, Runtime.getRuntime().availableProcessors() - 2);
+ ThreadPoolExecutor lockCheckPool = ThreadPools.getServerThreadPools().createThreadPool(8,
+ numCheckThreads, 10, MINUTES, "update-lock-check", false);
+
+ final Map<String,String> invalidLocks = new ConcurrentHashMap<>(hostsWithLocks.size());
+ // try a thrift call to the hosts - hosts running previous versions will fail the call
+ hostsWithLocks.forEach(hostLockPair -> lockCheckPool.execute(() -> {
+ tryThriftCall(context, invalidLocks, hostLockPair.getFirst(), hostLockPair.getSecond());
+ if (Thread.currentThread().isInterrupted()) {
+ throw new IllegalStateException("Interrupted try to make thrift call to check locks");
+ }
+ }));
+
+ lockCheckPool.shutdown();
+ boolean lockCheckTimeout = false;
+ final long blockTimeMinutes = 60;
+ final long lockCheckStart = System.nanoTime();
+ try {
+ // wait to all to finish
+ log.info("waiting for lock check tasks to complete. May block for up to {} minutes",
+ blockTimeMinutes);
+ if (!lockCheckPool.awaitTermination(blockTimeMinutes, MINUTES)) {
+ log.warn(
+ "Timed out waiting for lock check to finish - continuing, but tservers running prior versions may be present");
+ lockCheckTimeout = true;
+ }
+ } catch (InterruptedException ex) {
+ var remaining = lockCheckPool.shutdownNow();
+ log.warn("Interrupted waiting for lock check to finish. {} tasks were remaining.",
+ remaining.size());
+ lockCheckTimeout = true;
+ }
+ log.info("tserver lock completed in {} minutes",
+ NANOSECONDS.toMinutes(System.nanoTime() - lockCheckStart));
+ if (invalidLocks.size() == 0) {
+ log.info("All tserver locks found were processed and are valid.");
+ } else {
+ log.warn("Lock check completed with {} inlaid lock(s) found", invalidLocks.size());
+ Map<String,String> sorted = new TreeMap<>(invalidLocks);
+ sorted.forEach((p, v) -> {
+ log.warn("Invalid tserver lock: {}", p);
+ });
+ }
+
+ if (context.getConfiguration().getBoolean(Property.GENERAL_UPGRADE_VERSION_CHECK_ENABLED)) {
+ if (lockCheckTimeout || !invalidLocks.isEmpty()) {
+ throw new IllegalStateException(
+ "Failed to complete tserver lock check - cannot validate that all tservers running current version");
+ }
+ }
+ }
+
+ private static void tryThriftCall(ServerContext context, Map<String,String> invalidLocks,
+ HostAndPort host, ServiceLock.ServiceLockPath lockPath) {
+
+ Retry retry = Retry.builder().maxRetries(10).retryAfter(100, MILLISECONDS)
+ .incrementBy(250, MILLISECONDS).maxWait(10, SECONDS).backOffFactor(1.07)
+ .logInterval(1, SECONDS).createFactory().createRetry();
+
+ while (retry.canRetry()) {
+ try (TTransport transport = ThriftUtil.createTransport(host, context)) {
+ log.trace("found valid lock at: {}", lockPath);
+ return;
+ } catch (TException ex) {
+ log.trace("Could not establish a thrift connection, service lock path: {}", lockPath, ex);
+ }
+ // pause for next rty
+ try {
+ retry.useRetry();
+ retry.waitForNextAttempt(log,
+ "retrying thrift call to host: " + host + " lock path: " + lockPath);
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ invalidLocks.put(lockPath.toString(), "");
+ return;
+ } catch (Exception ex) {
+ log.debug("Failed to establish a connection for to service holding lock. {}", lockPath, ex);
+ invalidLocks.put(lockPath.toString(), "");
+ return;
+ }
+ }
+ log.debug("Failed to establish a connection for to service holding lock. {}. no retires left.",
+ lockPath);
+ invalidLocks.put(lockPath.toString(), "");
+ }
+
+ private List<Pair<HostAndPort,ServiceLock.ServiceLockPath>> gatherLocks(final ZooKeeper zk,
+ final String zkPathRoot) {
+ List<Pair<HostAndPort,ServiceLock.ServiceLockPath>> hosts = new ArrayList<>();
+ try {
+ ZKUtil.visitSubTreeDFS(zk, zkPathRoot, false, (rc, path, ctx, name) -> {
+ if (name.startsWith(ServiceLock.ZLOCK_PREFIX)) {
+ log.trace("found lock at {}", path);
+ try {
+ Stat stat = new Stat();
+ final var zLockPath = ServiceLock.path(path);
+ byte[] lockData = zk.getData(zLockPath.toString(), false, stat);
+ if (lockData == null || !validLockData(lockData, hosts, zLockPath)) {
+ log.debug("Invalid lock data - trying to delete path: {}", zLockPath);
+ zk.delete(zLockPath.toString(), stat.getVersion());
+ }
+ } catch (KeeperException.NoNodeException ex) {
+ // empty - lock no longer exists.
+ } catch (KeeperException ex) {
+ log.trace("could not read lock from ZooKeeper for {}, skipping", path, ex);
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(
+ "interrupted reading lock from ZooKeeper for path: " + path);
+ }
+ }
+ });
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(
+ "interrupted reading lock from ZooKeeper for path: " + zkPathRoot);
+ } catch (KeeperException ex) {
+ log.trace("could not read lock all locks from ZooKeeper for {}, lost list may be incomplete",
Review Comment:
why ignore this and why proceed if we can't read the locks from ZK?
##########
server/manager/src/main/java/org/apache/accumulo/manager/upgrade/PreUpgradeValidation.java:
##########
@@ -112,9 +140,163 @@ private static String extractAuthName(ACL acl) {
@SuppressFBWarnings(value = "DM_EXIT",
justification = "Want to immediately stop all threads on upgrade error")
- private void fail(Exception e) {
+ protected void fail(Exception e) {
log.error("FATAL: Error performing pre-upgrade checks", e);
System.exit(1);
}
+ private void validateTableLocks(final ServerContext context) {
+
+ final ZooReaderWriter zrw = context.getZooReaderWriter();
+ final ZooKeeper zk = zrw.getZooKeeper();
+ final String rootPath = context.getZooKeeperRoot();
+ final String tserverLockRoot = rootPath + Constants.ZTSERVERS;
+
+ log.debug("Looking for locks that may be from previous version in path: {}", tserverLockRoot);
+
+ List<Pair<HostAndPort,ServiceLock.ServiceLockPath>> hostsWithLocks =
+ gatherLocks(zk, tserverLockRoot);
+
+ int numCheckThreads = Math.max(32, Runtime.getRuntime().availableProcessors() - 2);
+ ThreadPoolExecutor lockCheckPool = ThreadPools.getServerThreadPools().createThreadPool(8,
+ numCheckThreads, 10, MINUTES, "update-lock-check", false);
+
+ final Map<String,String> invalidLocks = new ConcurrentHashMap<>(hostsWithLocks.size());
+ // try a thrift call to the hosts - hosts running previous versions will fail the call
+ hostsWithLocks.forEach(hostLockPair -> lockCheckPool.execute(() -> {
+ tryThriftCall(context, invalidLocks, hostLockPair.getFirst(), hostLockPair.getSecond());
+ if (Thread.currentThread().isInterrupted()) {
+ throw new IllegalStateException("Interrupted try to make thrift call to check locks");
+ }
+ }));
+
+ lockCheckPool.shutdown();
+ boolean lockCheckTimeout = false;
+ final long blockTimeMinutes = 60;
+ final long lockCheckStart = System.nanoTime();
+ try {
+ // wait to all to finish
+ log.info("waiting for lock check tasks to complete. May block for up to {} minutes",
+ blockTimeMinutes);
+ if (!lockCheckPool.awaitTermination(blockTimeMinutes, MINUTES)) {
+ log.warn(
+ "Timed out waiting for lock check to finish - continuing, but tservers running prior versions may be present");
+ lockCheckTimeout = true;
+ }
+ } catch (InterruptedException ex) {
+ var remaining = lockCheckPool.shutdownNow();
+ log.warn("Interrupted waiting for lock check to finish. {} tasks were remaining.",
+ remaining.size());
+ lockCheckTimeout = true;
+ }
+ log.info("tserver lock completed in {} minutes",
+ NANOSECONDS.toMinutes(System.nanoTime() - lockCheckStart));
+ if (invalidLocks.size() == 0) {
+ log.info("All tserver locks found were processed and are valid.");
+ } else {
+ log.warn("Lock check completed with {} inlaid lock(s) found", invalidLocks.size());
+ Map<String,String> sorted = new TreeMap<>(invalidLocks);
+ sorted.forEach((p, v) -> {
+ log.warn("Invalid tserver lock: {}", p);
+ });
+ }
+
+ if (context.getConfiguration().getBoolean(Property.GENERAL_UPGRADE_VERSION_CHECK_ENABLED)) {
+ if (lockCheckTimeout || !invalidLocks.isEmpty()) {
+ throw new IllegalStateException(
+ "Failed to complete tserver lock check - cannot validate that all tservers running current version");
+ }
+ }
+ }
+
+ private static void tryThriftCall(ServerContext context, Map<String,String> invalidLocks,
+ HostAndPort host, ServiceLock.ServiceLockPath lockPath) {
+
+ Retry retry = Retry.builder().maxRetries(10).retryAfter(100, MILLISECONDS)
+ .incrementBy(250, MILLISECONDS).maxWait(10, SECONDS).backOffFactor(1.07)
+ .logInterval(1, SECONDS).createFactory().createRetry();
+
+ while (retry.canRetry()) {
+ try (TTransport transport = ThriftUtil.createTransport(host, context)) {
+ log.trace("found valid lock at: {}", lockPath);
+ return;
+ } catch (TException ex) {
+ log.trace("Could not establish a thrift connection, service lock path: {}", lockPath, ex);
+ }
+ // pause for next rty
+ try {
+ retry.useRetry();
+ retry.waitForNextAttempt(log,
+ "retrying thrift call to host: " + host + " lock path: " + lockPath);
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ invalidLocks.put(lockPath.toString(), "");
+ return;
+ } catch (Exception ex) {
+ log.debug("Failed to establish a connection for to service holding lock. {}", lockPath, ex);
+ invalidLocks.put(lockPath.toString(), "");
+ return;
+ }
+ }
+ log.debug("Failed to establish a connection for to service holding lock. {}. no retires left.",
+ lockPath);
+ invalidLocks.put(lockPath.toString(), "");
+ }
+
+ private List<Pair<HostAndPort,ServiceLock.ServiceLockPath>> gatherLocks(final ZooKeeper zk,
+ final String zkPathRoot) {
+ List<Pair<HostAndPort,ServiceLock.ServiceLockPath>> hosts = new ArrayList<>();
+ try {
+ ZKUtil.visitSubTreeDFS(zk, zkPathRoot, false, (rc, path, ctx, name) -> {
+ if (name.startsWith(ServiceLock.ZLOCK_PREFIX)) {
+ log.trace("found lock at {}", path);
+ try {
+ Stat stat = new Stat();
+ final var zLockPath = ServiceLock.path(path);
+ byte[] lockData = zk.getData(zLockPath.toString(), false, stat);
+ if (lockData == null || !validLockData(lockData, hosts, zLockPath)) {
+ log.debug("Invalid lock data - trying to delete path: {}", zLockPath);
+ zk.delete(zLockPath.toString(), stat.getVersion());
+ }
+ } catch (KeeperException.NoNodeException ex) {
+ // empty - lock no longer exists.
+ } catch (KeeperException ex) {
+ log.trace("could not read lock from ZooKeeper for {}, skipping", path, ex);
Review Comment:
Why ignore this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org