This is an automated email from the ASF dual-hosted git repository.
DomGarguilo 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 e63a13ac75 Stop MAC servers in parallel (#6575)
e63a13ac75 is described below
commit e63a13ac755b757532ab6f04b933f77f79739e84
Author: Dom G. <[email protected]>
AuthorDate: Thu Oct 8 10:29:59 2026 -0400
Stop MAC servers in parallel (#6575)
* Stop MAC servers in parallel
* New method to wait for all procs under single timeout
---
.../MiniAccumuloClusterControl.java | 33 ++++------------------
.../miniclusterImpl/MiniAccumuloClusterImpl.java | 27 ++++++++++++++++++
2 files changed, 33 insertions(+), 27 deletions(-)
diff --git
a/minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloClusterControl.java
b/minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloClusterControl.java
index 5b83301611..492c79d417 100644
---
a/minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloClusterControl.java
+++
b/minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloClusterControl.java
@@ -349,15 +349,8 @@ public class MiniAccumuloClusterControl implements
ClusterControl {
case TABLET_SERVER:
synchronized (tabletServerProcesses) {
try {
- for (Process tserver : tabletServerProcesses) {
- try {
- cluster.stopProcessWithTimeout(tserver, 30, TimeUnit.SECONDS);
- } catch (ExecutionException | TimeoutException e) {
- log.warn("TabletServer did not fully stop after 30 seconds",
e);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- }
- }
+ cluster.stopProcessesWithTimeout(ServerType.TABLET_SERVER,
tabletServerProcesses, 30,
+ TimeUnit.SECONDS);
} finally {
tabletServerProcesses.clear();
}
@@ -379,15 +372,8 @@ public class MiniAccumuloClusterControl implements
ClusterControl {
case SCAN_SERVER:
synchronized (scanServerProcesses) {
try {
- for (Process sserver : scanServerProcesses) {
- try {
- cluster.stopProcessWithTimeout(sserver, 30, TimeUnit.SECONDS);
- } catch (ExecutionException | TimeoutException e) {
- log.warn("ScanServer did not fully stop after 30 seconds", e);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- }
- }
+ cluster.stopProcessesWithTimeout(ServerType.SCAN_SERVER,
scanServerProcesses, 30,
+ TimeUnit.SECONDS);
} finally {
scanServerProcesses.clear();
}
@@ -409,15 +395,8 @@ public class MiniAccumuloClusterControl implements
ClusterControl {
case COMPACTOR:
synchronized (compactorProcesses) {
try {
- for (Process compactor : compactorProcesses) {
- try {
- cluster.stopProcessWithTimeout(compactor, 30,
TimeUnit.SECONDS);
- } catch (ExecutionException | TimeoutException e) {
- log.warn("Compactor did not fully stop after 30 seconds", e);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- }
- }
+ cluster.stopProcessesWithTimeout(ServerType.COMPACTOR,
compactorProcesses, 30,
+ TimeUnit.SECONDS);
} finally {
compactorProcesses.clear();
}
diff --git
a/minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloClusterImpl.java
b/minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloClusterImpl.java
index cf550433f2..4bce23d5e9 100644
---
a/minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloClusterImpl.java
+++
b/minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloClusterImpl.java
@@ -82,6 +82,7 @@ import
org.apache.accumulo.core.manager.thrift.ManagerMonitorInfo;
import org.apache.accumulo.core.rpc.clients.ThriftClientTypes;
import org.apache.accumulo.core.trace.TraceUtil;
import org.apache.accumulo.core.util.Pair;
+import org.apache.accumulo.core.util.Timer;
import org.apache.accumulo.manager.state.SetGoalState;
import org.apache.accumulo.minicluster.MiniAccumuloCluster;
import org.apache.accumulo.minicluster.ServerType;
@@ -989,6 +990,32 @@ public class MiniAccumuloClusterImpl implements
AccumuloCluster {
return executor;
}
+ /**
+ * Stops the processes in parallel. The given timeout is the total wait for
all given processes.
+ */
+ public void stopProcessesWithTimeout(final ServerType type, final
List<Process> procs,
+ final long timeout, final TimeUnit unit) {
+
+ // signal all first so they stop in parallel
+ procs.forEach(Process::destroy);
+
+ final Timer timer = Timer.startNew();
+ final long timeoutNanos = unit.toNanos(timeout);
+ try {
+ for (Process proc : procs) {
+ long remainingNanos = timeoutNanos -
timer.elapsed(TimeUnit.NANOSECONDS);
+ boolean stopped = proc.waitFor(remainingNanos, TimeUnit.NANOSECONDS);
+ if (!stopped) {
+ log.warn("{} (pid {}) did not fully stop after {} seconds", type,
proc.pid(),
+ unit.toSeconds(timeout));
+ }
+ }
+ } catch (InterruptedException e) {
+ log.warn("Interrupted while trying to stop {} processes.", type);
+ Thread.currentThread().interrupt();
+ }
+ }
+
public int stopProcessWithTimeout(final Process proc, long timeout, TimeUnit
unit)
throws InterruptedException, ExecutionException, TimeoutException {
FutureTask<Integer> future = new FutureTask<>(() -> {
