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/10 19:22:10 UTC
[accumulo] branch revert-3231-bugfix/change-wait-logic created (now 783f0794f5)
This is an automated email from the ASF dual-hosted git repository.
edcoleman pushed a change to branch revert-3231-bugfix/change-wait-logic
in repository https://gitbox.apache.org/repos/asf/accumulo.git
at 783f0794f5 Revert "Fix wait timeout logic for available tservers (#3231)"
This branch includes the following new commits:
new 783f0794f5 Revert "Fix wait timeout logic for available tservers (#3231)"
The 1 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.
[accumulo] 01/01: Revert "Fix wait timeout logic for available tservers (#3231)"
Posted by ed...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
edcoleman pushed a commit to branch revert-3231-bugfix/change-wait-logic
in repository https://gitbox.apache.org/repos/asf/accumulo.git
commit 783f0794f5215114ff3666378a3a46ff57ddbc1f
Author: EdColeman <de...@etcoleman.com>
AuthorDate: Fri Mar 10 19:22:04 2023 +0000
Revert "Fix wait timeout logic for available tservers (#3231)"
This reverts commit e9842290a50233262604e38cd41570221185b6bc.
---
.../java/org/apache/accumulo/master/Master.java | 55 ++++++++++++----------
1 file changed, 30 insertions(+), 25 deletions(-)
diff --git a/server/master/src/main/java/org/apache/accumulo/master/Master.java b/server/master/src/main/java/org/apache/accumulo/master/Master.java
index 9617032d82..51c1b87740 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/Master.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/Master.java
@@ -17,8 +17,6 @@
package org.apache.accumulo.master;
import static java.nio.charset.StandardCharsets.UTF_8;
-import static java.util.concurrent.TimeUnit.MILLISECONDS;
-import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static org.apache.accumulo.fate.util.UtilWaitThread.sleepUninterruptibly;
import java.io.IOException;
@@ -38,6 +36,7 @@ import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
@@ -789,7 +788,7 @@ public class Master extends AccumuloServerContext
return MasterGoalState.valueOf(new String(data));
} catch (Exception e) {
log.error("Problem getting real goal state from zookeeper: " + e);
- sleepUninterruptibly(1_000, MILLISECONDS);
+ sleepUninterruptibly(1, TimeUnit.SECONDS);
}
}
@@ -940,7 +939,7 @@ public class Master extends AccumuloServerContext
log.error("Error cleaning up migrations", ex);
}
}
- sleepUninterruptibly(TIME_BETWEEN_MIGRATION_CLEANUPS, MILLISECONDS);
+ sleepUninterruptibly(TIME_BETWEEN_MIGRATION_CLEANUPS, TimeUnit.MILLISECONDS);
}
}
@@ -1085,7 +1084,7 @@ public class Master extends AccumuloServerContext
} catch (Throwable t) {
log.error("Error balancing tablets, will wait for " + WAIT_BETWEEN_ERRORS / ONE_SECOND
+ " (seconds) and then retry", t);
- sleepUninterruptibly(WAIT_BETWEEN_ERRORS, MILLISECONDS);
+ sleepUninterruptibly(WAIT_BETWEEN_ERRORS, TimeUnit.MILLISECONDS);
}
}
}
@@ -1187,7 +1186,7 @@ public class Master extends AccumuloServerContext
// 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), MILLISECONDS);
+ sleepUninterruptibly(Math.max(1, rpcTimeout / 120_000), TimeUnit.MILLISECONDS);
}
tp.submit(new Runnable() {
@Override
@@ -1236,7 +1235,7 @@ public class Master extends AccumuloServerContext
}
tp.shutdown();
try {
- tp.awaitTermination(Math.max(10000, rpcTimeout / 3), MILLISECONDS);
+ tp.awaitTermination(Math.max(10000, rpcTimeout / 3), TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
log.debug("Interrupted while fetching status");
}
@@ -1364,7 +1363,7 @@ public class Master extends AccumuloServerContext
log.info("Waiting for AuthenticationTokenKeyManager to be initialized");
logged = true;
}
- sleepUninterruptibly(200, MILLISECONDS);
+ sleepUninterruptibly(200, TimeUnit.MILLISECONDS);
}
// And log when we are initialized
log.info("AuthenticationTokenSecretManager is initialized");
@@ -1389,7 +1388,7 @@ public class Master extends AccumuloServerContext
masterLock.replaceLockData(address.getBytes());
while (!clientService.isServing()) {
- sleepUninterruptibly(100, MILLISECONDS);
+ sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
}
// Start the daemon to scan the replication table and make units of work
@@ -1435,7 +1434,7 @@ public class Master extends AccumuloServerContext
}
while (clientService.isServing()) {
- sleepUninterruptibly(500, MILLISECONDS);
+ sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
}
log.info("Shutting down fate.");
fate.shutdown();
@@ -1479,7 +1478,8 @@ public class Master extends AccumuloServerContext
* if interrupted while blocking, propagated for caller to handle.
*/
private void blockForTservers() throws InterruptedException {
- long waitStart = System.nanoTime();
+
+ long waitStart = System.currentTimeMillis();
AccumuloConfiguration accConfig = serverConfig.getConfiguration();
long minTserverCount = accConfig.getCount(Property.MASTER_STARTUP_TSERVER_AVAIL_MIN_COUNT);
@@ -1490,6 +1490,7 @@ public class Master extends AccumuloServerContext
tserverSet.size(), Property.MASTER_STARTUP_TSERVER_AVAIL_MIN_COUNT.getKey());
return;
}
+
long maxWait = accConfig.getTimeInMillis(Property.MASTER_STARTUP_TSERVER_AVAIL_MAX_WAIT);
if (maxWait <= 0) {
@@ -1498,12 +1499,14 @@ public class Master extends AccumuloServerContext
maxWait = Long.MAX_VALUE;
}
- long retries = 10;
- long waitPeriod = maxWait / retries;
+ // 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);
- Retry tserverRetry = Retry.builder().maxRetries(retries).retryAfter(waitPeriod, MILLISECONDS)
- .incrementBy(0, MILLISECONDS).maxWait(waitPeriod, MILLISECONDS)
- .logInterval(30_000, MILLISECONDS).createRetry();
+ Retry tserverRetry =
+ Retry.builder().infiniteRetries().retryAfter(initialWait, TimeUnit.MILLISECONDS)
+ .incrementBy(15_000, TimeUnit.MILLISECONDS).maxWait(maxWait, TimeUnit.MILLISECONDS)
+ .logInterval(30_000, TimeUnit.MILLISECONDS).createRetry();
log.info("Checking for tserver availability - need to reach {} servers. Have {}",
minTserverCount, tserverSet.size());
@@ -1518,25 +1521,27 @@ public class Master extends AccumuloServerContext
// suppress last message once threshold reached.
if (needTservers) {
- tserverRetry.logRetry(log, String.format(
- "Blocking for tserver availability - need to reach %s servers. Have %s Time spent blocking %s seconds.",
+ log.info(
+ "Blocking for tserver availability - need to reach {} servers. Have {}"
+ + " Time spent blocking {} sec.",
minTserverCount, tserverSet.size(),
- NANOSECONDS.toSeconds(System.nanoTime() - waitStart)));
+ TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis() - waitStart));
}
- tserverRetry.useRetry();
}
if (tserverSet.size() < minTserverCount) {
log.warn(
"tserver availability check time expired - continuing. Requested {}, have {} tservers on line. "
- + " Time waiting {} sec",
- tserverSet.size(), minTserverCount, NANOSECONDS.toSeconds(System.nanoTime() - waitStart));
+ + " Time waiting {} ms",
+ tserverSet.size(), minTserverCount,
+ TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis() - waitStart));
} else {
log.info(
"tserver availability check completed. Requested {}, have {} tservers on line. "
- + " Time waiting {} sec",
- tserverSet.size(), minTserverCount, NANOSECONDS.toSeconds(System.nanoTime() - waitStart));
+ + " Time waiting {} ms",
+ tserverSet.size(), minTserverCount,
+ TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis() - waitStart));
}
}
@@ -1634,7 +1639,7 @@ public class Master extends AccumuloServerContext
masterLock.tryToCancelAsyncLockOrUnlock();
- sleepUninterruptibly(TIME_TO_WAIT_BETWEEN_LOCK_CHECKS, MILLISECONDS);
+ sleepUninterruptibly(TIME_TO_WAIT_BETWEEN_LOCK_CHECKS, TimeUnit.MILLISECONDS);
}
setMasterState(MasterState.HAVE_LOCK);