imbajin commented on code in PR #357:
URL:
https://github.com/apache/hugegraph-computer/pull/357#discussion_r3668267459
##########
computer/computer-test/src/main/java/org/apache/hugegraph/computer/suite/integrate/SenderIntegrateTest.java:
##########
@@ -68,10 +78,247 @@ public static void clear() {
// pass
}
+ @Test
+ public void testWaitForServicesFailsFast() {
+ CompletableFuture<Void> failedWorker = new CompletableFuture<>();
+ CompletableFuture<Void> waitingMaster = new CompletableFuture<>();
+ IllegalStateException cause = new IllegalStateException("worker
failed");
+ failedWorker.completeExceptionally(cause);
+
+ try {
+ waitForServices(Arrays.asList(failedWorker, waitingMaster));
+ Assert.fail("Expected worker failure to stop service wait");
+ } catch (ComputerException e) {
+ Assert.assertSame(cause, e.getCause());
+ }
+ }
+
+ @Test
+ public void testCleanupFailureDoesNotHideWaitFailure() {
+ ServiceLifecycle lifecycle = new ServiceLifecycle();
+ CompletableFuture<Void> failedWorker = new CompletableFuture<>();
+ IllegalStateException waitFailure =
+ new IllegalStateException("worker failed");
+ IllegalStateException cleanupFailure =
+ new IllegalStateException("worker close failed");
+ failedWorker.completeExceptionally(waitFailure);
+ lifecycle.registerWorker(() -> {
+ throw cleanupFailure;
+ });
+
+ try {
+ waitForServicesAndClose(lifecycle, Arrays.asList(failedWorker),
+ new ArrayList<>(), null);
+ Assert.fail("Expected worker failure to be preserved");
+ } catch (ComputerException e) {
+ Assert.assertSame(waitFailure, e.getCause());
+ Assert.assertEquals(1, e.getSuppressed().length);
+ Assert.assertSame(cleanupFailure, e.getSuppressed()[0].getCause());
+ }
+ }
+
+ @Test
+ public void testMasterErrorCompletesServiceFuture() throws Exception {
+ CompletableFuture<Void> masterFuture = new CompletableFuture<>();
+ Error cause = new AssertionError("master failed");
+ Thread masterThread = new Thread(() -> this.executeMasterTask(
+ masterFuture, () -> {
+ throw cause;
+ }));
+ masterThread.start();
+
+ try {
+ masterFuture.get(1, TimeUnit.SECONDS);
+ Assert.fail("Expected master error to fail the service future");
+ } catch (ExecutionException e) {
+ Assert.assertSame(cause, e.getCause());
+ } catch (TimeoutException e) {
+ Assert.fail("Timed out to wait for master error");
+ } finally {
+ interruptAndJoinThreads(Arrays.asList(masterThread),
+ TEST_THREAD_JOIN_TIMEOUT);
+ }
+ }
+
+ @Test
+ public void testCiTimeoutsAllowHeavyInputStep() {
+ Assert.assertEquals(TimeUnit.MINUTES.toMillis(5L), BSP_WAIT_TIMEOUT);
+ Assert.assertEquals(BSP_WAIT_TIMEOUT + TimeUnit.SECONDS.toMillis(10L),
+ SERVICE_WAIT_TIMEOUT);
+ }
+
+ @Test
+ public void testServiceLifecycleClosesLateRegisteredService() {
+ ServiceLifecycle lifecycle = new ServiceLifecycle();
+ AtomicBoolean closed = new AtomicBoolean();
+
+ lifecycle.closeAll();
+
+ Assert.assertFalse(lifecycle.registerWorker(() -> closed.set(true)));
+ Assert.assertTrue(closed.get());
+ }
+
+ @Test
+ public void testServiceLifecycleClosesWorkersBeforeMasterAfterFailure() {
+ ServiceLifecycle lifecycle = new ServiceLifecycle();
+ AtomicBoolean activeWorkerClosed = new AtomicBoolean();
+ AtomicBoolean masterClosed = new AtomicBoolean();
+ RuntimeException workerFailure =
+ new IllegalStateException("worker close failed");
+
+ lifecycle.registerMaster(() -> {
+ Assert.assertTrue(activeWorkerClosed.get());
+ masterClosed.set(true);
+ });
+ lifecycle.registerWorker(() -> {
+ throw workerFailure;
+ });
+ lifecycle.registerWorker(() -> activeWorkerClosed.set(true));
+
+ Throwable failure = lifecycle.closeAll();
+
+ Assert.assertSame(workerFailure, failure);
+ Assert.assertTrue(activeWorkerClosed.get());
+ Assert.assertTrue(masterClosed.get());
+ }
+
+ @Test
+ public void testInitializeServiceClosesPartiallyInitializedService() {
+ AtomicBoolean closed = new AtomicBoolean();
+ RuntimeException cause = new IllegalStateException("init failed");
+
+ try {
+ initializeService(new Object(), service -> {
+ throw cause;
+ }, service -> closed.set(true));
+ Assert.fail("Expected initialization to fail");
+ } catch (RuntimeException e) {
+ Assert.assertSame(cause, e);
+ }
+
+ Assert.assertTrue(closed.get());
+ }
+
+ @Test
+ public void testCloseServicesAndJoinStopsSpawnedThreads() throws Exception
{
+ ServiceLifecycle lifecycle = new ServiceLifecycle();
+ AtomicBoolean closed = new AtomicBoolean();
+ CountDownLatch started = new CountDownLatch(1);
+ Thread thread = new Thread(() -> {
+ started.countDown();
+ try {
+ Thread.sleep(Long.MAX_VALUE);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ lifecycle.registerWorker(() -> closed.set(true));
+ try {
+ thread.start();
+ Assert.assertTrue(started.await(1, TimeUnit.SECONDS));
+
+ closeServicesAndJoin(lifecycle, Arrays.asList(thread), null);
+
+ Assert.assertTrue(closed.get());
+ Assert.assertFalse(thread.isAlive());
+ } finally {
+ lifecycle.closeAll();
+ interruptAndJoinThreads(Arrays.asList(thread),
+ TEST_THREAD_JOIN_TIMEOUT);
+ }
+ }
+
+ @Test
+ public void testCloseServicesAndJoinUsesOneTimeoutBudget()
+ throws Exception {
+ long timeout = 200L;
+ AtomicBoolean stopping = new AtomicBoolean();
+ CountDownLatch started = new CountDownLatch(3);
+ Thread firstWorker = newInterruptIgnoringThread(stopping, started);
+ Thread secondWorker = newInterruptIgnoringThread(stopping, started);
+ Thread master = newInterruptIgnoringThread(stopping, started);
+ firstWorker.start();
+ secondWorker.start();
+ master.start();
+ Assert.assertTrue(started.await(1, TimeUnit.SECONDS));
+
+ long start = System.nanoTime();
+ try {
+ closeServicesAndJoin(new ServiceLifecycle(),
+ Arrays.asList(firstWorker, secondWorker),
+ master, timeout);
+ Assert.fail("Expected service threads to time out");
+ } catch (ComputerException ignored) {
+ // The timeout is expected; the assertion below verifies its budget
+ } finally {
+ stopping.set(true);
+ firstWorker.interrupt();
+ secondWorker.interrupt();
+ master.interrupt();
+ firstWorker.join(TimeUnit.SECONDS.toMillis(1L));
+ secondWorker.join(TimeUnit.SECONDS.toMillis(1L));
+ master.join(TimeUnit.SECONDS.toMillis(1L));
+ }
+ long elapsed = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
start);
+ Assert.assertTrue("Cleanup exceeded its shared timeout budget: " +
elapsed,
Review Comment:
🧹 `elapsed` includes the `finally` cleanup and three bounded joins, not just
`closeServicesAndJoin()`. A scheduler or GC pause during that cleanup can push
an otherwise correct implementation past the strict 400 ms threshold and make
CI flaky. Please capture the production-call duration before entering cleanup
and use a scheduling-tolerant bound (or a controllable clock) for this
assertion.
##########
computer/computer-core/src/main/java/org/apache/hugegraph/computer/core/sender/QueuedMessageSender.java:
##########
@@ -137,11 +142,19 @@ public void run() {
++emptyQueueCount;
continue;
}
- if (channel.doSend(message)) {
- // Only consume the message after it is sent
+ try {
+ if (channel.doSend(message)) {
+ // Only consume the message after it is sent
+ channel.queue.take();
+ } else {
+ ++busyClientCount;
+ }
+ } catch (TransportException | RuntimeException e) {
+ channel.failControlFuture(e);
Review Comment:
‼️ This only fails a control future that is already registered. During
normal data streaming `controlFutureRef` is null: `finishSend()` waits for
buffers to be enqueued and registers FINISH afterward, so a synchronous
data-send failure can reach this catch first, make `failControlFuture()` a
no-op, and discard the message; the later FINISH can then succeed with missing
graph data. The new regression avoids this window by queuing FINISH before
unblocking the failed send. Please retain a per-channel send failure (or
propagate it to `MessageSendManager`) so the subsequent FINISH/job fails, and
add a regression where data sending fails before FINISH is submitted.
--
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]
