imbajin commented on code in PR #357:
URL:
https://github.com/apache/hugegraph-computer/pull/357#discussion_r3654915138
##########
computer/computer-test/src/main/java/org/apache/hugegraph/computer/suite/integrate/SenderIntegrateTest.java:
##########
@@ -129,19 +297,18 @@ public void testOneWorker() {
workerThread.start();
try {
- CompletableFuture.allOf(workerFuture, masterFuture).join();
+ waitForServices(Arrays.asList(workerFuture, masterFuture));
} finally {
- workerServiceRef.get().close();
- masterServiceRef.get().close();
+ closeServicesAndJoin(lifecycle, Arrays.asList(workerThread),
Review Comment:
⚠️ If `waitForServices()` throws and `closeServicesAndJoin()` also throws
from this `finally`, Java replaces the original service failure with the
cleanup exception; the same pattern appears in the other integration cases.
That hides the root cause precisely when teardown also fails. Please keep the
change narrow: preserve the primary exception at these existing call sites and
attach any cleanup failure with `addSuppressed()`, plus one focused regression
where both operations fail. No new exception hierarchy or lifecycle state is
needed.
##########
computer/computer-test/src/main/java/org/apache/hugegraph/computer/suite/integrate/SenderIntegrateTest.java:
##########
@@ -319,16 +518,194 @@ private MasterService initMaster(String[] args) {
Config config = ComputerContextUtil.initContext(
ComputerContextUtil.convertToMap(args));
MasterService service = new MasterService();
- service.init(config);
- return service;
+ return initializeService(service, s -> s.init(config),
+ SenderIntegrateTest::closeMaster);
}
private WorkerService initWorker(String[] args) {
Config config = ComputerContextUtil.initContext(
ComputerContextUtil.convertToMap(args));
WorkerService service = new WorkerService();
- service.init(config);
- return service;
+ return initializeService(service, s -> s.init(config),
+ SenderIntegrateTest::closeWorker);
+ }
+
+ private static <T> T initializeService(T service, Consumer<T> initializer,
+ Consumer<T> closer) {
+ try {
+ initializer.accept(service);
+ return service;
+ } catch (RuntimeException | Error e) {
+ try {
+ closer.accept(service);
+ } catch (RuntimeException | Error closeFailure) {
+ e.addSuppressed(closeFailure);
+ }
+ throw e;
+ }
+ }
+
+ private static void waitForServices(List<CompletableFuture<Void>> futures)
{
+ CompletableFuture<Void> result = new CompletableFuture<>();
+ for (CompletableFuture<Void> future : futures) {
+ future.whenComplete((r, e) -> {
+ if (e != null) {
+ result.completeExceptionally(e);
+ }
+ });
+ }
+ CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
+ .whenComplete((r, e) -> {
+ if (e == null) {
+ result.complete(null);
+ } else {
+ result.completeExceptionally(e);
+ }
+ });
+ try {
+ result.get(SERVICE_WAIT_TIMEOUT, TimeUnit.MILLISECONDS);
+ } catch (TimeoutException e) {
+ throw new ComputerException("Timed out to wait for master and " +
+ "worker services", e);
+ } catch (ExecutionException e) {
+ throw new ComputerException("Failed to wait for master and " +
+ "worker services", e.getCause());
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new ComputerException("Interrupted when waiting for master "
+
+ "and worker services", e);
+ }
+ }
+
+ private static void closeServicesAndJoin(ServiceLifecycle lifecycle,
+ List<Thread> workerThreads,
+ Thread masterThread) {
+ Throwable closeFailure = lifecycle.closeAll();
+ Throwable workerFailure = interruptAndJoinThreads(workerThreads);
+ Throwable masterFailure = null;
+ if (masterThread != null) {
+ masterFailure = interruptAndJoinThreads(
+ Arrays.asList(masterThread));
+ }
+ if (closeFailure != null) {
+ addFailure(closeFailure, workerFailure);
+ addFailure(closeFailure, masterFailure);
+ throw new ComputerException("Failed to close service",
closeFailure);
+ }
+ if (workerFailure != null) {
+ addFailure(workerFailure, masterFailure);
+ throw new ComputerException("Failed to close worker service
thread",
+ workerFailure);
+ }
+ if (masterFailure != null) {
+ throw new ComputerException("Failed to close master service
thread",
+ masterFailure);
+ }
+ }
+
+ private static Throwable interruptAndJoinThreads(List<Thread> threads) {
+ for (Thread thread : threads) {
+ thread.interrupt();
+ }
+ Throwable failure = null;
+ for (Thread thread : threads) {
+ try {
+ thread.join(SERVICE_WAIT_TIMEOUT);
Review Comment:
⚠️ `join(SERVICE_WAIT_TIMEOUT)` applies the full 310-second budget
separately to every thread. `testMultiWorkers()` can therefore spend 3 × 310
seconds on workers and another 310 seconds on the master—over 20 minutes—before
cleanup finally fails, which defeats the new fail-fast behavior. Please keep
the fix local: compute one absolute deadline in `closeServicesAndJoin()`, pass
only the remaining milliseconds to the existing join helper, and add one
short-timeout multi-thread regression. This does not need a new lifecycle
abstraction or concurrent shutdown model.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
