You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by ed...@apache.org on 2023/03/13 16:41:23 UTC
[accumulo] branch 2.1 updated: Merge branch 1.10 into 2.1
This is an automated email from the ASF dual-hosted git repository.
edcoleman pushed a commit to branch 2.1
in repository https://gitbox.apache.org/repos/asf/accumulo.git
The following commit(s) were added to refs/heads/2.1 by this push:
new 0c00ab519a Merge branch 1.10 into 2.1
0c00ab519a is described below
commit 0c00ab519ab8d010f9a5eadab332c697d7ea7461
Author: Ed Coleman <ed...@apache.org>
AuthorDate: Mon Mar 13 15:50:03 2023 +0000
Merge branch 1.10 into 2.1
---
.../java/org/apache/accumulo/manager/Manager.java | 102 +++++++++++----------
1 file changed, 54 insertions(+), 48 deletions(-)
diff --git a/server/manager/src/main/java/org/apache/accumulo/manager/Manager.java b/server/manager/src/main/java/org/apache/accumulo/manager/Manager.java
index 752c9cfd44..60931b9c2b 100644
--- a/server/manager/src/main/java/org/apache/accumulo/manager/Manager.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/Manager.java
@@ -20,6 +20,11 @@ package org.apache.accumulo.manager;
import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Collections.emptySortedMap;
+import static java.util.concurrent.TimeUnit.HOURS;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static java.util.concurrent.TimeUnit.MINUTES;
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
+import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.accumulo.core.util.UtilWaitThread.sleepUninterruptibly;
import java.io.IOException;
@@ -44,7 +49,6 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
@@ -243,7 +247,7 @@ public class Manager extends AbstractServer
Fate<Manager> fate() {
try {
// block up to 30 seconds until it's ready; if it's still not ready, introduce some logging
- if (!fateReadyLatch.await(30, TimeUnit.SECONDS)) {
+ if (!fateReadyLatch.await(30, SECONDS)) {
String msgPrefix = "Unexpected use of fate in thread " + Thread.currentThread().getName()
+ " at time " + System.currentTimeMillis();
// include stack trace so we know where it's coming from, in case we need to troubleshoot it
@@ -251,7 +255,7 @@ public class Manager extends AbstractServer
new IllegalStateException("Attempted fate action before manager finished starting up; "
+ "if this doesn't make progress, please report it as a bug to the developers"));
int minutes = 0;
- while (!fateReadyLatch.await(5, TimeUnit.MINUTES)) {
+ while (!fateReadyLatch.await(5, MINUTES)) {
minutes += 5;
log.warn("{} still blocked after {} minutes; this is getting weird", msgPrefix, minutes);
}
@@ -294,7 +298,7 @@ public class Manager extends AbstractServer
// This frees the main thread and will cause the manager to exit
clientService.stop();
Manager.this.nextEvent.event("stopped event loop");
- }, 100L, 1000L, TimeUnit.MILLISECONDS);
+ }, 100L, 1000L, MILLISECONDS);
ThreadPools.watchNonCriticalScheduledTask(future);
}
@@ -534,7 +538,7 @@ public class Manager extends AbstractServer
return ManagerGoalState.valueOf(new String(data));
} catch (Exception e) {
log.error("Problem getting real goal state from zookeeper: ", e);
- sleepUninterruptibly(1, TimeUnit.SECONDS);
+ sleepUninterruptibly(1, SECONDS);
}
}
}
@@ -687,7 +691,7 @@ public class Manager extends AbstractServer
log.error("Error cleaning up migrations", ex);
}
}
- sleepUninterruptibly(TIME_BETWEEN_MIGRATION_CLEANUPS, TimeUnit.MILLISECONDS);
+ sleepUninterruptibly(TIME_BETWEEN_MIGRATION_CLEANUPS, MILLISECONDS);
}
}
@@ -837,7 +841,7 @@ public class Manager extends AbstractServer
TraceUtil.setException(span, t, false);
log.error("Error balancing tablets, will wait for {} (seconds) and then retry ",
WAIT_BETWEEN_ERRORS / ONE_SECOND, t);
- sleepUninterruptibly(WAIT_BETWEEN_ERRORS, TimeUnit.MILLISECONDS);
+ sleepUninterruptibly(WAIT_BETWEEN_ERRORS, MILLISECONDS);
} finally {
span.end();
}
@@ -965,7 +969,7 @@ public class Manager extends AbstractServer
// Since an unbounded thread pool is being used, rate limit how fast task are added to the
// executor. This prevents the threads from growing large unless there are lots of
// unresponsive tservers.
- sleepUninterruptibly(Math.max(1, rpcTimeout / 120_000), TimeUnit.MILLISECONDS);
+ sleepUninterruptibly(Math.max(1, rpcTimeout / 120_000), MILLISECONDS);
}
tp.execute(() -> {
try {
@@ -1020,7 +1024,7 @@ public class Manager extends AbstractServer
}
tp.shutdown();
try {
- tp.awaitTermination(Math.max(10000, rpcTimeout / 3), TimeUnit.MILLISECONDS);
+ tp.awaitTermination(Math.max(10000, rpcTimeout / 3), MILLISECONDS);
} catch (InterruptedException e) {
log.debug("Interrupted while fetching status");
}
@@ -1178,11 +1182,10 @@ public class Manager extends AbstractServer
}
try {
- final AgeOffStore<Manager> store =
- new AgeOffStore<>(
- new org.apache.accumulo.core.fate.ZooStore<>(getZooKeeperRoot() + Constants.ZFATE,
- context.getZooReaderWriter()),
- TimeUnit.HOURS.toMillis(8), System::currentTimeMillis);
+ final AgeOffStore<Manager> store = new AgeOffStore<>(
+ new org.apache.accumulo.core.fate.ZooStore<>(getZooKeeperRoot() + Constants.ZFATE,
+ context.getZooReaderWriter()),
+ HOURS.toMillis(8), System::currentTimeMillis);
Fate<Manager> f = new Fate<>(this, store, TraceRepo::toLogString);
f.startTransactionRunners(getConfiguration());
@@ -1190,13 +1193,13 @@ public class Manager extends AbstractServer
fateReadyLatch.countDown();
ThreadPools.watchCriticalScheduledTask(context.getScheduledExecutor()
- .scheduleWithFixedDelay(store::ageOff, 63000, 63000, TimeUnit.MILLISECONDS));
+ .scheduleWithFixedDelay(store::ageOff, 63000, 63000, MILLISECONDS));
} catch (KeeperException | InterruptedException e) {
throw new IllegalStateException("Exception setting up FaTE cleanup thread", e);
}
- ThreadPools.watchCriticalScheduledTask(context.getScheduledExecutor().scheduleWithFixedDelay(
- () -> ScanServerMetadataEntries.clean(context), 10, 10, TimeUnit.MINUTES));
+ ThreadPools.watchCriticalScheduledTask(context.getScheduledExecutor()
+ .scheduleWithFixedDelay(() -> ScanServerMetadataEntries.clean(context), 10, 10, MINUTES));
initializeZkForReplication(zReaderWriter, zroot);
@@ -1220,7 +1223,7 @@ public class Manager extends AbstractServer
log.info("Waiting for AuthenticationTokenKeyManager to be initialized");
logged = true;
}
- sleepUninterruptibly(200, TimeUnit.MILLISECONDS);
+ sleepUninterruptibly(200, MILLISECONDS);
}
// And log when we are initialized
log.info("AuthenticationTokenSecretManager is initialized");
@@ -1235,7 +1238,7 @@ public class Manager extends AbstractServer
}
while (!clientService.isServing()) {
- sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+ sleepUninterruptibly(100, MILLISECONDS);
}
// if the replication name is ever set, then start replication services
@@ -1251,7 +1254,7 @@ public class Manager extends AbstractServer
} catch (UnknownHostException | KeeperException | InterruptedException e) {
log.error("Error occurred starting replication services. ", e);
}
- }, 0, 5000, TimeUnit.MILLISECONDS);
+ }, 0, 5000, MILLISECONDS);
ThreadPools.watchNonCriticalScheduledTask(future);
// checking stored user hashes if any of them uses an outdated algorithm
@@ -1261,7 +1264,7 @@ public class Manager extends AbstractServer
managerInitialized.set(true);
while (clientService.isServing()) {
- sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
+ sleepUninterruptibly(500, MILLISECONDS);
}
log.info("Shutting down fate.");
fate().shutdown();
@@ -1333,8 +1336,7 @@ public class Manager extends AbstractServer
* @throws InterruptedException if interrupted while blocking, propagated for caller to handle.
*/
private void blockForTservers() throws InterruptedException {
-
- long waitStart = System.currentTimeMillis();
+ long waitStart = System.nanoTime();
long minTserverCount =
getConfiguration().getCount(Property.MANAGER_STARTUP_TSERVER_AVAIL_MIN_COUNT);
@@ -1344,24 +1346,30 @@ public class Manager extends AbstractServer
tserverSet.size(), Property.MANAGER_STARTUP_TSERVER_AVAIL_MIN_COUNT.getKey());
return;
}
+ long userWait = MILLISECONDS.toSeconds(
+ getConfiguration().getTimeInMillis(Property.MANAGER_STARTUP_TSERVER_AVAIL_MAX_WAIT));
+ // Setting retry values for defined wait timeouts
+ long retries = 10;
+ // Set these to the same value so the max possible wait time always matches the provided maxWait
+ long initialWait = userWait / retries;
+ long maxWaitPeriod = initialWait;
+ long waitIncrement = 0;
- long maxWait =
- getConfiguration().getTimeInMillis(Property.MANAGER_STARTUP_TSERVER_AVAIL_MAX_WAIT);
-
- if (maxWait <= 0) {
+ if (userWait <= 0) {
log.info("tserver availability check set to block indefinitely, To change, set {} > 0.",
- Property.MANAGER_STARTUP_TSERVER_AVAIL_MAX_WAIT.getKey());
- maxWait = Long.MAX_VALUE;
- }
+ Property.MANAGER_STARTUP_TSERVER_AVAIL_MIN_COUNT.getKey());
+ userWait = Long.MAX_VALUE;
- // honor Retry condition that initial wait < max wait, otherwise use small value to allow thread
- // yield to happen
- long initialWait = Math.min(50, maxWait / 2);
+ // If indefinitely blocking, change retry values to support incremental backoff and logging.
+ retries = userWait;
+ initialWait = 1;
+ maxWaitPeriod = 30;
+ waitIncrement = 5;
+ }
- Retry tserverRetry =
- Retry.builder().infiniteRetries().retryAfter(initialWait, TimeUnit.MILLISECONDS)
- .incrementBy(15_000, TimeUnit.MILLISECONDS).maxWait(maxWait, TimeUnit.MILLISECONDS)
- .backOffFactor(1).logInterval(30_000, TimeUnit.MILLISECONDS).createRetry();
+ Retry tserverRetry = Retry.builder().maxRetries(retries).retryAfter(initialWait, SECONDS)
+ .incrementBy(waitIncrement, SECONDS).maxWait(maxWaitPeriod, SECONDS).backOffFactor(1)
+ .logInterval(30, SECONDS).createRetry();
log.info("Checking for tserver availability - need to reach {} servers. Have {}",
minTserverCount, tserverSet.size());
@@ -1376,27 +1384,25 @@ public class Manager extends AbstractServer
// suppress last message once threshold reached.
if (needTservers) {
- log.info(
- "Blocking for tserver availability - need to reach {} servers. Have {}"
- + " Time spent blocking {} sec.",
+ tserverRetry.logRetry(log, String.format(
+ "Blocking for tserver availability - need to reach %s servers. Have %s Time spent blocking %s seconds.",
minTserverCount, tserverSet.size(),
- TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis() - waitStart));
+ NANOSECONDS.toSeconds(System.nanoTime() - waitStart)));
}
+ tserverRetry.useRetry();
}
if (tserverSet.size() < minTserverCount) {
log.warn(
"tserver availability check time expired - continuing. Requested {}, have {} tservers on line. "
- + " Time waiting {} ms",
- tserverSet.size(), minTserverCount,
- TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis() - waitStart));
+ + " Time waiting {} sec",
+ tserverSet.size(), minTserverCount, NANOSECONDS.toSeconds(System.nanoTime() - waitStart));
} else {
log.info(
"tserver availability check completed. Requested {}, have {} tservers on line. "
- + " Time waiting {} ms",
- tserverSet.size(), minTserverCount,
- TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis() - waitStart));
+ + " Time waiting {} sec",
+ tserverSet.size(), minTserverCount, NANOSECONDS.toSeconds(System.nanoTime() - waitStart));
}
}
@@ -1527,7 +1533,7 @@ public class Manager extends AbstractServer
managerLock.tryToCancelAsyncLockOrUnlock();
- sleepUninterruptibly(TIME_TO_WAIT_BETWEEN_LOCK_CHECKS, TimeUnit.MILLISECONDS);
+ sleepUninterruptibly(TIME_TO_WAIT_BETWEEN_LOCK_CHECKS, MILLISECONDS);
}
setManagerState(ManagerState.HAVE_LOCK);