This is an automated email from the ASF dual-hosted git repository.
DomGarguilo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/accumulo.git
The following commit(s) were added to refs/heads/main by this push:
new 3dab7b0a6d Use Duration for Fate.shutdown and EventQueue.poll (#6642)
3dab7b0a6d is described below
commit 3dab7b0a6d9147774c550de2893298dd471ea062
Author: Dom G. <[email protected]>
AuthorDate: Thu Nov 12 15:01:35 2026 -0500
Use Duration for Fate.shutdown and EventQueue.poll (#6642)
Simplify and improve readbility of logic by replacing a `long` time value +
`TimeUnit` with a Duration object
Co-Authored-By: Claude Opus 5.5
---
.../java/org/apache/accumulo/core/fate/Fate.java | 32 +++++++---------
.../apache/accumulo/core/fate/FateExecutor.java | 44 +++++++++++-----------
.../apache/accumulo/core/util/CountDownTimer.java | 5 ++-
.../java/org/apache/accumulo/core/util/Timer.java | 2 +-
.../accumulo/core/util/CountDownTimerTest.java | 7 ++++
.../org/apache/accumulo/manager/EventQueue.java | 8 ++--
.../accumulo/manager/TabletGroupWatcher.java | 3 +-
.../accumulo/manager/fate/FateWorkerEnv.java | 5 ++-
.../apache/accumulo/manager/EventQueueTest.java | 4 +-
.../test/fate/FateExecutionOrderITBase.java | 5 +--
.../org/apache/accumulo/test/fate/FateITBase.java | 20 +++++-----
.../accumulo/test/fate/FateOpsCommandsITBase.java | 17 ++++-----
.../accumulo/test/fate/FatePoolsWatcherITBase.java | 13 ++++---
.../accumulo/test/fate/MultipleStoresITBase.java | 9 ++---
14 files changed, 88 insertions(+), 86 deletions(-)
diff --git a/core/src/main/java/org/apache/accumulo/core/fate/Fate.java
b/core/src/main/java/org/apache/accumulo/core/fate/Fate.java
index 77e07076b5..8446da66c8 100644
--- a/core/src/main/java/org/apache/accumulo/core/fate/Fate.java
+++ b/core/src/main/java/org/apache/accumulo/core/fate/Fate.java
@@ -37,7 +37,6 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
@@ -47,6 +46,7 @@ import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.logging.FateLogger;
import org.apache.accumulo.core.manager.thrift.TFateOperation;
+import org.apache.accumulo.core.util.CountDownTimer;
import org.apache.accumulo.core.util.threads.ThreadPools;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -395,9 +395,8 @@ public class Fate<T> extends FateClient<T> {
* optionally waits on them. Leaves the fate object in a state where it can
still update and read
* fate data, like add a new fate operation or get the status of an existing
fate operation.
*/
- public void shutdown(long timeout, TimeUnit timeUnit) {
- log.info("Shutting down {} FATE, waiting: {} seconds", store.type(),
- TimeUnit.SECONDS.convert(timeout, timeUnit));
+ public void shutdown(Duration timeout) {
+ log.info("Shutting down {} FATE, waiting: {} seconds", store.type(),
timeout.toSeconds());
// important this is set before shutdownNow is called as the background
// threads will check this to see if shutdown related errors should be
ignored.
if (keepRunning.compareAndSet(true, false)) {
@@ -412,18 +411,17 @@ public class Fate<T> extends FateClient<T> {
fatePoolsWatcherFuture.cancel(false);
}
- if (timeout > 0) {
- long start = System.nanoTime();
+ if (timeout.compareTo(Duration.ZERO) > 0) {
+ final CountDownTimer countdown = CountDownTimer.startNew(timeout);
try {
- waitForAllFateExecShutdown(start, timeout, timeUnit);
- waitForDeadResCleanerShutdown(start, timeout, timeUnit);
+ waitForAllFateExecShutdown(countdown);
+ waitForDeadResCleanerShutdown(countdown);
if (anyFateExecutorIsAlive() || deadResCleanerIsAlive()) {
log.warn(
- "Waited for {}ms for all fate {} background threads to stop, but
some are still running. "
+ "Waited {}ms for all fate {} background threads to stop, but
some are still running. "
+ "fate executor threads:{} dead reservation cleaner
thread:{} ",
- TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start),
store.type(),
- anyFateExecutorIsAlive(), deadResCleanerIsAlive());
+ timeout.toMillis(), store.type(), anyFateExecutorIsAlive(),
deadResCleanerIsAlive());
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
@@ -450,7 +448,7 @@ public class Fate<T> extends FateClient<T> {
* Initiates shutdown of all fate threads and prevents reads and updates of
fates persisted data.
*/
public void close() {
- shutdown(0, SECONDS);
+ shutdown(Duration.ZERO);
store.close();
}
@@ -485,18 +483,16 @@ public class Fate<T> extends FateClient<T> {
return deadResCleanerExecutor != null &&
!deadResCleanerExecutor.isTerminated();
}
- private void waitForAllFateExecShutdown(long start, long timeout, TimeUnit
timeUnit)
- throws InterruptedException {
+ private void waitForAllFateExecShutdown(CountDownTimer countdown) throws
InterruptedException {
synchronized (fateExecutors) {
for (var fateExecutor : fateExecutors) {
- fateExecutor.waitForShutdown(start, timeout, timeUnit);
+ fateExecutor.waitForShutdown(countdown);
}
}
}
- private void waitForDeadResCleanerShutdown(long start, long timeout,
TimeUnit timeUnit)
- throws InterruptedException {
- while (((System.nanoTime() - start) < timeUnit.toNanos(timeout)) &&
deadResCleanerIsAlive()) {
+ private void waitForDeadResCleanerShutdown(CountDownTimer countdown) throws
InterruptedException {
+ while (!countdown.isExpired() && deadResCleanerIsAlive()) {
if (deadResCleanerExecutor != null &&
!deadResCleanerExecutor.awaitTermination(1, SECONDS)) {
log.debug("Fate {} is waiting for dead reservation cleaner thread to
terminate",
store.type());
diff --git a/core/src/main/java/org/apache/accumulo/core/fate/FateExecutor.java
b/core/src/main/java/org/apache/accumulo/core/fate/FateExecutor.java
index 22fcaf67c8..9cd7837d05 100644
--- a/core/src/main/java/org/apache/accumulo/core/fate/FateExecutor.java
+++ b/core/src/main/java/org/apache/accumulo/core/fate/FateExecutor.java
@@ -52,6 +52,7 @@ import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.fate.Fate.TxInfo;
import org.apache.accumulo.core.fate.FateStore.FateTxStore;
import org.apache.accumulo.core.fate.ReadOnlyFateStore.TStatus;
+import org.apache.accumulo.core.util.CountDownTimer;
import org.apache.accumulo.core.util.ShutdownUtil;
import org.apache.accumulo.core.util.Timer;
import org.apache.accumulo.core.util.threads.ThreadPoolNames;
@@ -268,33 +269,30 @@ public class FateExecutor<T> {
transactionExecutor.shutdownNow();
}
- protected void waitForShutdown(long start, long timeout, TimeUnit timeUnit)
- throws InterruptedException {
- if (timeout > 0) {
- while (((System.nanoTime() - start) < timeUnit.toNanos(timeout)) &&
isAlive()) {
- if (!transactionExecutor.awaitTermination(1, SECONDS)) {
- log.debug("Fate {} is waiting for {} worker threads for fate ops {}
to terminate",
- fate.getStore().type(), runningTxRunners.size(), fateOps);
- continue;
- }
-
- workFinder.join(1_000);
- if (workFinder.isAlive()) {
- log.debug("Fate {} is waiting for work finder thread for fate ops {}
to terminate",
- fate.getStore().type(), fateOps);
- workFinder.interrupt();
- }
+ protected void waitForShutdown(CountDownTimer countdown) throws
InterruptedException {
+ while (!countdown.isExpired() && isAlive()) {
+ if (!transactionExecutor.awaitTermination(1, SECONDS)) {
+ log.debug("Fate {} is waiting for {} worker threads for fate ops {} to
terminate",
+ fate.getStore().type(), runningTxRunners.size(), fateOps);
+ continue;
}
- if (isAlive()) {
- log.warn(
- "Waited for {}ms for the {} fate executor operating on fate ops {}
to stop, but it"
- + " is still running. Summary of run state of its threads:
work finder:{}"
- + " transaction executor:{}",
- TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start),
fate.getStore().type(),
- fateOps, workFinder.isAlive(),
!transactionExecutor.isTerminated());
+ workFinder.join(1_000);
+ if (workFinder.isAlive()) {
+ log.debug("Fate {} is waiting for work finder thread for fate ops {}
to terminate",
+ fate.getStore().type(), fateOps);
+ workFinder.interrupt();
}
}
+
+ if (isAlive()) {
+ log.warn(
+ "The {} fate executor operating on fate ops {} did not stop within
the shutdown"
+ + " timeout and is still running. Summary of run state of its
threads: work finder:{}"
+ + " transaction executor:{}",
+ fate.getStore().type(), fateOps, workFinder.isAlive(),
+ !transactionExecutor.isTerminated());
+ }
}
/**
diff --git
a/core/src/main/java/org/apache/accumulo/core/util/CountDownTimer.java
b/core/src/main/java/org/apache/accumulo/core/util/CountDownTimer.java
index 5497e06a27..1643bcdce3 100644
--- a/core/src/main/java/org/apache/accumulo/core/util/CountDownTimer.java
+++ b/core/src/main/java/org/apache/accumulo/core/util/CountDownTimer.java
@@ -50,11 +50,12 @@ public class CountDownTimer {
/**
* Starts a new countdown timer with the specified duration.
*
- * @param duration the countdown duration, must be non-negative.
+ * @param duration the countdown duration, must be non-negative. Durations
too large to represent
+ * in nanoseconds (~292 years) are capped, so the timer effectively
never expires.
*/
public static CountDownTimer startNew(Duration duration) {
Preconditions.checkArgument(!duration.isNegative());
- return new CountDownTimer(duration.toNanos());
+ return new CountDownTimer(Timer.toNanos(duration));
}
/**
diff --git a/core/src/main/java/org/apache/accumulo/core/util/Timer.java
b/core/src/main/java/org/apache/accumulo/core/util/Timer.java
index 88cfbeee76..c9937dc253 100644
--- a/core/src/main/java/org/apache/accumulo/core/util/Timer.java
+++ b/core/src/main/java/org/apache/accumulo/core/util/Timer.java
@@ -96,7 +96,7 @@ public final class Timer {
return (startNanos - otherTimer.startNanos) > 0;
}
- private static long toNanos(Duration duration) {
+ static long toNanos(Duration duration) {
try {
// This can overflow when very large, such as when the
// duration is created using Long.MAX_VALUE millis
diff --git
a/core/src/test/java/org/apache/accumulo/core/util/CountDownTimerTest.java
b/core/src/test/java/org/apache/accumulo/core/util/CountDownTimerTest.java
index b0ce9953a6..b727e2903e 100644
--- a/core/src/test/java/org/apache/accumulo/core/util/CountDownTimerTest.java
+++ b/core/src/test/java/org/apache/accumulo/core/util/CountDownTimerTest.java
@@ -94,4 +94,11 @@ public class CountDownTimerTest {
assertThrows(IllegalArgumentException.class,
() -> CountDownTimer.startNew(Duration.ofMillis(-1)));
}
+
+ @Test
+ public void testHugeDurationSaturates() {
+ var cdTimer = CountDownTimer.startNew(Duration.ofMillis(Long.MAX_VALUE));
+ assertFalse(cdTimer.isExpired());
+ assertTrue(cdTimer.timeLeft(TimeUnit.DAYS) > 0);
+ }
}
diff --git
a/server/manager/src/main/java/org/apache/accumulo/manager/EventQueue.java
b/server/manager/src/main/java/org/apache/accumulo/manager/EventQueue.java
index 53d22235c8..17a52e6863 100644
--- a/server/manager/src/main/java/org/apache/accumulo/manager/EventQueue.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/EventQueue.java
@@ -18,6 +18,7 @@
*/
package org.apache.accumulo.manager;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
@@ -126,9 +127,8 @@ public class EventQueue {
private static final List<Event> ALL_LEVELS = List.of(new Event());
- public synchronized List<Event> poll(long duration, TimeUnit timeUnit)
- throws InterruptedException {
- CountDownTimer timer = CountDownTimer.startNew(duration, timeUnit);
+ public synchronized List<Event> poll(Duration duration) throws
InterruptedException {
+ CountDownTimer timer = CountDownTimer.startNew(duration);
while (!allLevels && levels.isEmpty() && !timer.isExpired()) {
wait(Math.max(1, timer.timeLeft(TimeUnit.MILLISECONDS)));
}
@@ -149,6 +149,6 @@ public class EventQueue {
}
public List<Event> take() throws InterruptedException {
- return poll(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
+ return poll(Duration.ofNanos(Long.MAX_VALUE));
}
}
diff --git
a/server/manager/src/main/java/org/apache/accumulo/manager/TabletGroupWatcher.java
b/server/manager/src/main/java/org/apache/accumulo/manager/TabletGroupWatcher.java
index fb2eaa62a9..78ae87d42f 100644
---
a/server/manager/src/main/java/org/apache/accumulo/manager/TabletGroupWatcher.java
+++
b/server/manager/src/main/java/org/apache/accumulo/manager/TabletGroupWatcher.java
@@ -128,6 +128,7 @@ import io.micrometer.core.instrument.MeterRegistry;
abstract class TabletGroupWatcher extends AccumuloDaemonThread {
private static final Logger LOG =
LoggerFactory.getLogger(TabletGroupWatcher.class);
+ private static final Duration EVENT_POLL_TIMEOUT = Duration.ofMillis(100);
private static final EscalatingLogger TABLET_UNLOAD_LOGGER =
new EscalatingLogger(Manager.log, Duration.ofMinutes(5), 1000,
Logger::info);
@@ -314,7 +315,7 @@ abstract class TabletGroupWatcher extends
AccumuloDaemonThread {
public void run() {
try {
while (manager.stillManager()) {
- var events = eventQueue.poll(100, TimeUnit.MILLISECONDS);
+ var events = eventQueue.poll(EVENT_POLL_TIMEOUT);
if (events.isEmpty()) {
// check to see if still the manager
diff --git
a/server/manager/src/main/java/org/apache/accumulo/manager/fate/FateWorkerEnv.java
b/server/manager/src/main/java/org/apache/accumulo/manager/fate/FateWorkerEnv.java
index f442948054..52369a2725 100644
---
a/server/manager/src/main/java/org/apache/accumulo/manager/fate/FateWorkerEnv.java
+++
b/server/manager/src/main/java/org/apache/accumulo/manager/fate/FateWorkerEnv.java
@@ -20,10 +20,10 @@ package org.apache.accumulo.manager.fate;
import static
org.apache.accumulo.core.util.threads.ThreadPoolNames.FILE_RENAME_POOL;
+import java.time.Duration;
import java.util.Collection;
import java.util.Set;
import java.util.concurrent.ExecutorService;
-import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.accumulo.core.client.AccumuloException;
@@ -56,6 +56,7 @@ import org.slf4j.LoggerFactory;
public class FateWorkerEnv implements FateEnv {
private static final Logger log =
LoggerFactory.getLogger(FateWorkerEnv.class);
+ private static final Duration EVENT_POLL_TIMEOUT = Duration.ofMillis(100);
private final ServerContext ctx;
private final ExecutorService refreshPool;
@@ -84,7 +85,7 @@ public class FateWorkerEnv implements FateEnv {
public void run() {
while (!stopped.get()) {
try {
- var events = queue.poll(100, TimeUnit.MILLISECONDS);
+ var events = queue.poll(EVENT_POLL_TIMEOUT);
if (events.isEmpty()) {
continue;
}
diff --git
a/server/manager/src/test/java/org/apache/accumulo/manager/EventQueueTest.java
b/server/manager/src/test/java/org/apache/accumulo/manager/EventQueueTest.java
index e05fb66889..55115032f6 100644
---
a/server/manager/src/test/java/org/apache/accumulo/manager/EventQueueTest.java
+++
b/server/manager/src/test/java/org/apache/accumulo/manager/EventQueueTest.java
@@ -21,10 +21,10 @@ package org.apache.accumulo.manager;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
-import java.util.concurrent.TimeUnit;
import org.apache.accumulo.core.data.TableId;
import org.apache.accumulo.core.dataImpl.KeyExtent;
@@ -93,7 +93,7 @@ public class EventQueueTest {
assertTrue(events.stream()
.anyMatch(e -> e.getScope() == EventScope.DATA_LEVEL && e.getLevel()
== DataLevel.USER));
- assertEquals(0, eventQueue.poll(1, TimeUnit.MILLISECONDS).size());
+ assertEquals(0, eventQueue.poll(Duration.ofMillis(1)).size());
}
}
diff --git
a/test/src/main/java/org/apache/accumulo/test/fate/FateExecutionOrderITBase.java
b/test/src/main/java/org/apache/accumulo/test/fate/FateExecutionOrderITBase.java
index 60badd3025..61665344db 100644
---
a/test/src/main/java/org/apache/accumulo/test/fate/FateExecutionOrderITBase.java
+++
b/test/src/main/java/org/apache/accumulo/test/fate/FateExecutionOrderITBase.java
@@ -36,7 +36,6 @@ import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.ScheduledThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import org.apache.accumulo.core.client.Accumulo;
@@ -299,7 +298,7 @@ public abstract class FateExecutionOrderITBase extends
SharedMiniClusterBase
} finally {
if (fate != null) {
- fate.shutdown(10, TimeUnit.MINUTES);
+ fate.shutdown(Duration.ofMinutes(10));
}
}
}
@@ -404,7 +403,7 @@ public abstract class FateExecutionOrderITBase extends
SharedMiniClusterBase
} finally {
if (fate != null) {
- fate.shutdown(10, TimeUnit.MINUTES);
+ fate.shutdown(Duration.ofMinutes(10));
}
}
}
diff --git a/test/src/main/java/org/apache/accumulo/test/fate/FateITBase.java
b/test/src/main/java/org/apache/accumulo/test/fate/FateITBase.java
index c044b9dd87..ee69051c67 100644
--- a/test/src/main/java/org/apache/accumulo/test/fate/FateITBase.java
+++ b/test/src/main/java/org/apache/accumulo/test/fate/FateITBase.java
@@ -36,13 +36,13 @@ import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ScheduledThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
@@ -313,7 +313,7 @@ public abstract class FateITBase extends
SharedMiniClusterBase implements FateTe
return false;
}, SECONDS.toMillis(30), 10);
} finally {
- fate.shutdown(10, TimeUnit.MINUTES);
+ fate.shutdown(Duration.ofMinutes(10));
}
}
@@ -347,7 +347,7 @@ public abstract class FateITBase extends
SharedMiniClusterBase implements FateTe
fate.delete(fateId);
assertEquals(UNKNOWN, getTxStatus(sctx, fateId));
} finally {
- fate.shutdown(10, TimeUnit.MINUTES);
+ fate.shutdown(Duration.ofMinutes(10));
}
}
@@ -382,7 +382,7 @@ public abstract class FateITBase extends
SharedMiniClusterBase implements FateTe
fate.delete(fateId);
assertEquals(UNKNOWN, getTxStatus(sctx, fateId));
} finally {
- fate.shutdown(10, TimeUnit.MINUTES);
+ fate.shutdown(Duration.ofMinutes(10));
}
}
@@ -414,7 +414,7 @@ public abstract class FateITBase extends
SharedMiniClusterBase implements FateTe
assertFalse(fate.cancel(fateId));
finishCall.countDown();
} finally {
- fate.shutdown(10, TimeUnit.MINUTES);
+ fate.shutdown(Duration.ofMinutes(10));
}
}
@@ -486,7 +486,7 @@ public abstract class FateITBase extends
SharedMiniClusterBase implements FateTe
});
} finally {
- fate.shutdown(10, TimeUnit.MINUTES);
+ fate.shutdown(Duration.ofMinutes(10));
}
}
@@ -543,7 +543,7 @@ public abstract class FateITBase extends
SharedMiniClusterBase implements FateTe
assertEquals(FAILED, fate.waitForCompletion(fateId));
assertTrue(fate.getException(fateId).getMessage().contains("isReady()
failed"));
} finally {
- fate.shutdown(10, TimeUnit.MINUTES);
+ fate.shutdown(Duration.ofMinutes(10));
}
}
@@ -631,7 +631,7 @@ public abstract class FateITBase extends
SharedMiniClusterBase implements FateTe
assertEquals(IN_PROGRESS, getTxStatus(sctx, txid));
// shutdown fate
- fate.shutdown(0, SECONDS);
+ fate.shutdown(Duration.ZERO);
// tell the op to exit the method
Wait.waitFor(() -> interruptedException.get() != null);
@@ -659,7 +659,7 @@ public abstract class FateITBase extends
SharedMiniClusterBase implements FateTe
}
assertNull(interruptedException.get());
} finally {
- fate.shutdown(10, TimeUnit.MINUTES);
+ fate.shutdown(Duration.ofMinutes(10));
}
}
@@ -783,7 +783,7 @@ public abstract class FateITBase extends
SharedMiniClusterBase implements FateTe
}
}
} finally {
- fate.shutdown(10, TimeUnit.MINUTES);
+ fate.shutdown(Duration.ofMinutes(10));
}
}
diff --git
a/test/src/main/java/org/apache/accumulo/test/fate/FateOpsCommandsITBase.java
b/test/src/main/java/org/apache/accumulo/test/fate/FateOpsCommandsITBase.java
index ba97473735..8236a62c91 100644
---
a/test/src/main/java/org/apache/accumulo/test/fate/FateOpsCommandsITBase.java
+++
b/test/src/main/java/org/apache/accumulo/test/fate/FateOpsCommandsITBase.java
@@ -43,7 +43,6 @@ import java.util.Optional;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ScheduledThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
import java.util.function.Predicate;
import java.util.stream.Collectors;
@@ -322,7 +321,7 @@ public abstract class FateOpsCommandsITBase extends
SharedMiniClusterBase
validateFateDetails(report.getFateDetails(), 2, fateIdsStarted);
}
} finally {
- fate.shutdown(1, TimeUnit.MINUTES);
+ fate.shutdown(Duration.ofMinutes(1));
}
}
@@ -353,7 +352,7 @@ public abstract class FateOpsCommandsITBase extends
SharedMiniClusterBase
"Fate ID Filters: [" + fateId2.canonical() + ", " +
fateId1.canonical() + "]"));
assertTrue(result.contains("Instance Types Filters: [" +
store.type().name() + "]"));
} finally {
- fate.shutdown(1, TimeUnit.MINUTES);
+ fate.shutdown(Duration.ofMinutes(1));
}
}
@@ -468,7 +467,7 @@ public abstract class FateOpsCommandsITBase extends
SharedMiniClusterBase
assertTrue(fateIdsFromResult.isEmpty());
}
} finally {
- fate.shutdown(1, TimeUnit.MINUTES);
+ fate.shutdown(Duration.ofMinutes(1));
}
}
@@ -595,7 +594,7 @@ public abstract class FateOpsCommandsITBase extends
SharedMiniClusterBase
fateIdsFromSummary = getFateIdsFromSummary();
assertEquals(expected, fateIdsFromSummary);
} finally {
- fate.shutdown(1, TimeUnit.MINUTES);
+ fate.shutdown(Duration.ofMinutes(1));
}
}
@@ -645,7 +644,7 @@ public abstract class FateOpsCommandsITBase extends
SharedMiniClusterBase
} finally {
// Finish work and shutdown
env.workersLatch.countDown();
- fate.shutdown(1, TimeUnit.MINUTES);
+ fate.shutdown(Duration.ofMinutes(1));
}
}
@@ -683,7 +682,7 @@ public abstract class FateOpsCommandsITBase extends
SharedMiniClusterBase
|| fateIdsFromSummary
.equals(Map.of(fateId1.canonical(), "FAILED",
fateId2.canonical(), "NEW")));
} finally {
- fate.shutdown(1, TimeUnit.MINUTES);
+ fate.shutdown(Duration.ofMinutes(1));
}
}
@@ -733,7 +732,7 @@ public abstract class FateOpsCommandsITBase extends
SharedMiniClusterBase
} finally {
// Finish work and shutdown
env.workersLatch.countDown();
- fate.shutdown(1, TimeUnit.MINUTES);
+ fate.shutdown(Duration.ofMinutes(1));
}
}
@@ -768,7 +767,7 @@ public abstract class FateOpsCommandsITBase extends
SharedMiniClusterBase
fateIdsFromSummary = getFateIdsFromSummary();
assertEquals(Map.of(fateId2.canonical(), "NEW"), fateIdsFromSummary);
} finally {
- fate.shutdown(1, TimeUnit.MINUTES);
+ fate.shutdown(Duration.ofMinutes(1));
}
}
diff --git
a/test/src/main/java/org/apache/accumulo/test/fate/FatePoolsWatcherITBase.java
b/test/src/main/java/org/apache/accumulo/test/fate/FatePoolsWatcherITBase.java
index cddeb1e1c1..75e4c94c60 100644
---
a/test/src/main/java/org/apache/accumulo/test/fate/FatePoolsWatcherITBase.java
+++
b/test/src/main/java/org/apache/accumulo/test/fate/FatePoolsWatcherITBase.java
@@ -20,6 +20,7 @@ package org.apache.accumulo.test.fate;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import java.time.Duration;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
@@ -204,7 +205,7 @@ public abstract class FatePoolsWatcherITBase extends
SharedMiniClusterBase
log.error("Failure in test", e);
throw e;
} finally {
- fate.shutdown(30, TimeUnit.SECONDS);
+ fate.shutdown(Duration.ofSeconds(30));
assertEquals(0, fate.getTotalTxRunnersActive());
}
}
@@ -282,7 +283,7 @@ public abstract class FatePoolsWatcherITBase extends
SharedMiniClusterBase
log.error("Failure in test", e);
throw e;
} finally {
- fate.shutdown(30, TimeUnit.SECONDS);
+ fate.shutdown(Duration.ofSeconds(30));
assertEquals(0, fate.getTotalTxRunnersActive());
}
}
@@ -421,7 +422,7 @@ public abstract class FatePoolsWatcherITBase extends
SharedMiniClusterBase
log.error("Failure in test", e);
throw e;
} finally {
- fate.shutdown(30, TimeUnit.SECONDS);
+ fate.shutdown(Duration.ofSeconds(30));
assertEquals(0, fate.getTotalTxRunnersActive());
}
}
@@ -457,7 +458,7 @@ public abstract class FatePoolsWatcherITBase extends
SharedMiniClusterBase
log.error("Failure in test", e);
throw e;
} finally {
- fate.shutdown(30, TimeUnit.SECONDS);
+ fate.shutdown(Duration.ofSeconds(30));
assertEquals(0, fate.getTotalTxRunnersActive());
}
}
@@ -635,7 +636,7 @@ public abstract class FatePoolsWatcherITBase extends
SharedMiniClusterBase
log.error("Failure in test", e);
throw e;
} finally {
- fate.shutdown(30, TimeUnit.SECONDS);
+ fate.shutdown(Duration.ofSeconds(30));
assertEquals(0, fate.getTotalTxRunnersActive());
}
}
@@ -689,7 +690,7 @@ public abstract class FatePoolsWatcherITBase extends
SharedMiniClusterBase
log.error("Failure in test", e);
throw e;
} finally {
- fate.shutdown(30, TimeUnit.SECONDS);
+ fate.shutdown(Duration.ofSeconds(30));
assertEquals(0, fate.getTotalTxRunnersActive());
}
}
diff --git
a/test/src/main/java/org/apache/accumulo/test/fate/MultipleStoresITBase.java
b/test/src/main/java/org/apache/accumulo/test/fate/MultipleStoresITBase.java
index 11f97461a4..2fc74f1d08 100644
--- a/test/src/main/java/org/apache/accumulo/test/fate/MultipleStoresITBase.java
+++ b/test/src/main/java/org/apache/accumulo/test/fate/MultipleStoresITBase.java
@@ -34,7 +34,6 @@ import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ScheduledThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Predicate;
@@ -283,8 +282,8 @@ public abstract class MultipleStoresITBase extends
SharedMiniClusterBase {
assertTrue(Collections.disjoint(testEnv1.executedOps,
testEnv2.executedOps));
assertEquals(allIds, Sets.union(testEnv1.executedOps,
testEnv2.executedOps));
} finally {
- fate1.shutdown(1, TimeUnit.MINUTES);
- fate2.shutdown(1, TimeUnit.MINUTES);
+ fate1.shutdown(Duration.ofMinutes(1));
+ fate2.shutdown(Duration.ofMinutes(1));
}
}
}
@@ -387,10 +386,10 @@ public abstract class MultipleStoresITBase extends
SharedMiniClusterBase {
testEnv1.workersLatch.countDown();
testEnv2.workersLatch.countDown();
if (fate1 != null) {
- fate1.shutdown(1, TimeUnit.MINUTES);
+ fate1.shutdown(Duration.ofMinutes(1));
}
if (fate2 != null) {
- fate2.shutdown(1, TimeUnit.MINUTES);
+ fate2.shutdown(Duration.ofMinutes(1));
}
}
}
