imbajin commented on code in PR #357:
URL:
https://github.com/apache/hugegraph-computer/pull/357#discussion_r3684012799
##########
computer/computer-test/src/main/java/org/apache/hugegraph/computer/core/sender/QueuedMessageSenderTest.java:
##########
@@ -64,4 +82,449 @@ public void testInitAndClose() {
Assert.assertTrue(ImmutableSet.of(Thread.State.TERMINATED)
.contains(sendExecutor.getState()));
}
+
+ @Test
+ public void testControlBeforeCompletionFinishes() throws Exception {
+ ControlFutureClient client = new ControlFutureClient();
+ QueuedMessageSender sender = this.newSender(client, new
MockTransportClient());
+
+ CountDownLatch completionStarted = new CountDownLatch(1);
+ CountDownLatch allowCompletion = new CountDownLatch(1);
+ Thread completionThread = null;
+ try {
+ CompletableFuture<Void> startFuture = sender.send(1,
MessageType.START);
+ Assert.assertTrue(await(client.startCalled));
+ startFuture.whenComplete((r, e) -> {
+ completionStarted.countDown();
+ try {
+ allowCompletion.await();
+ } catch (InterruptedException exception) {
+ Thread.currentThread().interrupt();
+ throw new AssertionError(exception);
+ }
+ });
+
+ completionThread = new Thread(
+ () -> client.startFuture.complete(null));
+ completionThread.start();
+ Assert.assertTrue(completionStarted.await(1, TimeUnit.SECONDS));
+
+ CompletableFuture<Void> finishFuture = sender.send(1,
MessageType.FINISH);
+ Assert.assertTrue(await(client.finishCalled));
+ allowCompletion.countDown();
+ completionThread.join(TimeUnit.SECONDS.toMillis(1));
+ Assert.assertFalse(completionThread.isAlive());
+ client.finishFuture.complete(null);
+ finishFuture.get(1, TimeUnit.SECONDS);
+ } finally {
+ allowCompletion.countDown();
+ if (completionThread != null) {
+ completionThread.join(TimeUnit.SECONDS.toMillis(1));
+ }
+ sender.close();
+ }
+ }
+
+ @Test
+ public void testTransportExceptionControlFuture() throws Exception {
+ ControlFutureClient client = new ControlFutureClient();
+ QueuedMessageSender sender = this.newSender(client, new
MockTransportClient());
+
+ try {
+ CompletableFuture<Void> startFuture = sender.send(1,
MessageType.START);
+ Assert.assertTrue(await(client.startCalled));
+
+ TransportException cause = new TransportException("connection
failed");
+ sender.transportExceptionCaught(cause, client.connectionId());
+ assertFutureFailedWith(startFuture, cause);
+
+ CompletableFuture<Void> finishFuture = sender.send(1,
MessageType.FINISH);
+ Assert.assertTrue(await(client.finishCalled));
+ client.startFuture.complete(null);
+ assertFutureFailedWith(startFuture, cause);
+
+ client.finishFuture.complete(null);
+ finishFuture.get(1, TimeUnit.SECONDS);
+ } finally {
+ sender.close();
+ }
+ }
+
+ @Test
+ public void testExceptionalCompletionCasLossFailsNextControl()
+ throws Exception {
+ ControlFutureClient client = new ControlFutureClient();
+ QueuedMessageSender sender = this.newSender(
+ client, new MockTransportClient());
+ CountDownLatch failureObserved = new CountDownLatch(1);
+ CountDownLatch resumeFailure = new CountDownLatch(1);
+ Thread failureThread = null;
+
+ try {
+ CompletableFuture<Void> startFuture = sender.send(
+ 1, MessageType.START);
+ Assert.assertTrue(await(client.startCalled));
+
+ Object[] channels = Whitebox.getInternalState(sender, "channels");
+ Object channel = channels[0];
+ AtomicReference<CompletableFuture<Void>> controlFutureRef =
+ Whitebox.getInternalState(channel, "controlFutureRef");
+ AtomicReference<CompletableFuture<Void>> observedFuture =
+ new AtomicReference<>();
+ TransportException cause =
+ new TransportException("connection failed");
+ failureThread = new Thread(() -> {
+ observedFuture.set(controlFutureRef.get());
+ failureObserved.countDown();
+ try {
+ resumeFailure.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new AssertionError(e);
+ }
+ Whitebox.invoke(channel.getClass(), new Class<?>[] {
+ CompletableFuture.class,
+ Throwable.class},
+ "completeControlFuture", channel,
+ observedFuture.get(), cause);
+ });
+ failureThread.start();
+ Assert.assertTrue(await(failureObserved));
+
+ client.startFuture.complete(null);
+ startFuture.get(1, TimeUnit.SECONDS);
+ CompletableFuture<Void> finishFuture = sender.send(
+ 1, MessageType.FINISH);
+ Assert.assertTrue(await(client.finishCalled));
+
+ resumeFailure.countDown();
+ assertFutureFailedWith(finishFuture, cause);
+ client.finishFuture.complete(null);
+ } finally {
+ resumeFailure.countDown();
+ if (failureThread != null) {
+ failureThread.join(TimeUnit.SECONDS.toMillis(1L));
+ }
+ sender.close();
+ }
+ }
+
+ @Test
+ public void testTransportExceptionDispatch() throws Exception {
+ ControlFutureClient client = new ControlFutureClient();
+ QueuedMessageSender sender = this.newSender(client, new
MockTransportClient());
+
+ client.blockDataSend = true;
+ try {
+ sender.send(1, new QueuedMessage(0, MessageType.MSG,
ByteBuffer.allocate(1)));
+ Assert.assertTrue(await(client.dataSendCalled));
+
+ CompletableFuture<Void> startFuture = sender.send(1,
MessageType.START);
+ TransportException cause = new TransportException("connection
failed before start");
+ sender.transportExceptionCaught(cause, client.connectionId());
+ assertFutureFailedWith(startFuture, cause);
+
+ client.allowDataSend.countDown();
+ Assert.assertFalse(await(client.startCalled));
+ } finally {
+ client.allowDataSend.countDown();
+ sender.close();
+ }
+ }
+
+ @Test
+ public void testExecutorAlive() throws Exception {
+ ControlFutureClient client = new ControlFutureClient();
+ QueuedMessageSender sender = this.newSender(client, new
MockTransportClient());
+
+ try {
+ RuntimeException startCause = new IllegalArgumentException("start
session failed");
+ client.startFailure = startCause;
+ CompletableFuture<Void> startFuture = sender.send(1,
MessageType.START);
+ assertFutureFailedWith(startFuture, startCause);
+
+ RuntimeException finishCause = new
IllegalArgumentException("finish session failed");
+ client.finishFailure = finishCause;
+ CompletableFuture<Void> finishFuture = sender.send(1,
MessageType.FINISH);
+ assertFutureFailedWith(finishFuture, finishCause);
+
+ Thread sendExecutor = Whitebox.getInternalState(sender,
"sendExecutor");
+ sendExecutor.join(TimeUnit.SECONDS.toMillis(1));
+ Assert.assertTrue(sendExecutor.isAlive());
+
+ client.startFailure = null;
+ CompletableFuture<Void> nextStartFuture = sender.send(1,
MessageType.START);
+ Assert.assertTrue(await(client.startCalled));
+ client.startFuture.complete(null);
+ nextStartFuture.get(1, TimeUnit.SECONDS);
+ } finally {
+ sender.close();
+ }
+ }
+
+ @Test
+ public void testOtherClients() throws Exception {
+ ControlFutureClient failedClient = new ControlFutureClient();
+ ControlFutureClient activeClient = new ControlFutureClient();
+ QueuedMessageSender sender = this.newSender(failedClient,
activeClient);
+
+ try {
+ TransportException startCause =
+ new TransportException("start session failed");
+ failedClient.startFailure = startCause;
+ CompletableFuture<Void> failedStart = sender.send(1,
MessageType.START);
+ assertFutureFailedWith(failedStart, startCause);
+
+ CompletableFuture<Void> activeStart = sender.send(2,
MessageType.START);
+ Assert.assertTrue(await(activeClient.startCalled));
+ activeClient.startFuture.complete(null);
+ activeStart.get(1, TimeUnit.SECONDS);
+
+ TransportException finishCause = new TransportException("finish
session failed");
+ failedClient.finishFailure = finishCause;
+ CompletableFuture<Void> failedFinish = sender.send(1,
MessageType.FINISH);
+ assertFutureFailedWith(failedFinish, finishCause);
+
+ CompletableFuture<Void> activeFinish = sender.send(2,
MessageType.FINISH);
+ Assert.assertTrue(await(activeClient.finishCalled));
+ activeClient.finishFuture.complete(null);
+ activeFinish.get(1, TimeUnit.SECONDS);
+
+ Thread sendExecutor = Whitebox.getInternalState(sender,
"sendExecutor");
+ Assert.assertTrue(sendExecutor.isAlive());
+ } finally {
+ sender.close();
+ }
+ }
+
+ @Test
+ public void testQueuedFinish() throws Exception {
+ this.assertSynchronousDataFailureCompletesQueuedFinish(
+ new TransportException("data send failed"));
+ }
+
+ @Test
+ public void testCompletesQueuedFinish() throws Exception {
+ this.assertSynchronousDataFailureCompletesQueuedFinish(
+ new IllegalStateException("data send failed"));
+ }
+
+ @Test
+ public void testFinishFailsFinish() throws Exception {
+ this.assertSynchronousDataFailureBeforeFinishFailsFinish(
+ new TransportException("data send failed before finish"));
+ }
+
+ @Test
+ public void testDataRuntimeFinish() throws Exception {
+ this.assertSynchronousDataFailureBeforeFinishFailsFinish(
+ new IllegalStateException("data send failed before finish"));
+ }
+
+ @Test
+ public void testConflictKeepsSendExecutorAlive() throws Exception {
+ ControlFutureClient client = new ControlFutureClient();
+ QueuedMessageSender sender = this.newSender(client, new
MockTransportClient());
+
+ try {
+ CompletableFuture<Void> startFuture = sender.send(1,
MessageType.START);
+ Assert.assertTrue(await(client.startCalled));
+
+ CompletableFuture<Void> conflictingFinishFuture = sender.send(1,
MessageType.FINISH);
+ assertFutureFailedWithMessage(conflictingFinishFuture, "The origin
future must be null");
+
+ Thread sendExecutor = Whitebox.getInternalState(sender,
"sendExecutor");
+ sendExecutor.join(TimeUnit.SECONDS.toMillis(1));
+ Assert.assertTrue(sendExecutor.isAlive());
+
+ client.startFuture.complete(null);
+ startFuture.get(1, TimeUnit.SECONDS);
+
+ CompletableFuture<Void> finishFuture = sender.send(1,
MessageType.FINISH);
+ Assert.assertTrue(await(client.finishCalled));
+ client.finishFuture.complete(null);
+ finishFuture.get(1, TimeUnit.SECONDS);
+ } finally {
+ sender.close();
+ }
+ }
+
+ private static void assertFutureFailedWith(CompletableFuture<Void> future,
Throwable cause)
+ throws InterruptedException, TimeoutException {
+ try {
+ future.get(1, TimeUnit.SECONDS);
+ Assert.fail("Expected control future to fail");
+ } catch (ExecutionException exception) {
+ Assert.assertSame(cause, exception.getCause());
+ }
+ }
+
+ private static boolean await(CountDownLatch latch) throws
InterruptedException {
+ return latch.await(1, TimeUnit.SECONDS);
+ }
+
+ private void assertSynchronousDataFailureCompletesQueuedFinish(
+ Throwable cause) throws Exception {
+ ControlFutureClient failedClient = new ControlFutureClient();
+ ControlFutureClient activeClient = new ControlFutureClient();
+ QueuedMessageSender sender = this.newSender(failedClient,
activeClient);
+
+ failedClient.blockDataSend = true;
+ try {
+ sender.send(1, new QueuedMessage(0, MessageType.MSG,
+ ByteBuffer.allocate(1)));
+ Assert.assertTrue(await(failedClient.dataSendCalled));
+
+ CompletableFuture<Void> finishFuture = sender.send(1,
MessageType.FINISH);
+ failedClient.dataFailure = cause;
+ failedClient.allowDataSend.countDown();
+ assertFutureFailedWith(finishFuture, cause);
+
+ CompletableFuture<Void> activeStart = sender.send(2,
MessageType.START);
+ Assert.assertTrue(await(activeClient.startCalled));
+ activeClient.startFuture.complete(null);
+ activeStart.get(1, TimeUnit.SECONDS);
+ } finally {
+ failedClient.allowDataSend.countDown();
+ sender.close();
+ }
+ }
+
+ @Test
+ public void testTransportExceptionDuringDataSendFailsLaterFinish()
+ throws Exception {
+ ControlFutureClient client = new ControlFutureClient();
+ QueuedMessageSender sender = this.newSender(
+ client, new MockTransportClient());
+
+ client.blockDataSend = true;
+ try {
+ sender.send(1, new QueuedMessage(0, MessageType.MSG,
+ ByteBuffer.allocate(1)));
+ Assert.assertTrue(await(client.dataSendCalled));
+
+ TransportException cause =
+ new TransportException("connection failed during data");
+ sender.transportExceptionCaught(cause, client.connectionId());
+ client.allowDataSend.countDown();
+ waitForQueueEmpty(sender, 1);
+
+ CompletableFuture<Void> finishFuture = sender.send(
+ 1, MessageType.FINISH);
+ assertFutureFailedWith(finishFuture, cause);
+ Assert.assertFalse(await(client.finishCalled));
+ } finally {
+ client.allowDataSend.countDown();
+ sender.close();
+ }
+ }
+
+ private void assertSynchronousDataFailureBeforeFinishFailsFinish(
+ Throwable cause) throws Exception {
+ ControlFutureClient failedClient = new ControlFutureClient();
+ QueuedMessageSender sender = this.newSender(
+ failedClient, new MockTransportClient());
+
+ failedClient.dataFailure = cause;
+ try {
+ sender.send(1, new QueuedMessage(0, MessageType.MSG,
+ ByteBuffer.allocate(1)));
+ waitForQueueEmpty(sender, 1);
+
+ CompletableFuture<Void> finishFuture = sender.send(1,
MessageType.FINISH);
+ assertFutureFailedWith(finishFuture, cause);
+ Assert.assertFalse(await(failedClient.finishCalled));
+ } finally {
+ sender.close();
+ }
+ }
+
+ private static void waitForQueueEmpty(QueuedMessageSender sender,
+ int workerId)
+ throws InterruptedException {
+ Object[] channels = Whitebox.getInternalState(sender, "channels");
+ MessageQueue queue = Whitebox.getInternalState(channels[workerId - 1],
+ "queue");
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(1L);
+ while (queue.peek() != null && System.nanoTime() < deadline) {
+ Thread.sleep(10L);
+ }
+ Assert.assertTrue("Timed out to wait for sender queue to be empty",
+ queue.peek() == null);
+ }
+
+ private static void assertFutureFailedWithMessage(CompletableFuture<Void>
future,
+ String message)
+ throws InterruptedException, TimeoutException {
+ try {
+ future.get(1, TimeUnit.SECONDS);
+ Assert.fail("Expected control future to fail");
+ } catch (ExecutionException exception) {
+ Assert.assertContains(message, exception.getCause().getMessage());
+ }
+ }
+
+ private static class ControlFutureClient extends MockTransportClient {
Review Comment:
🧹 Both `ControlFutureClient` instances inherit the same `localhost:8080`
`connectionId` from `MockTransportClient`. Because `transportExceptionCaught()`
notifies every channel whose ID matches, injecting a worker-1 failure also
marks worker 2 failed, but these tests never exercise worker 2 afterward; an
implementation that broadcasts failures across workers would therefore still
pass. Please give each test client a distinct `ConnectionId` and assert that
START/FINISH on the unaffected worker still succeeds after the injected
exception.
--
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]
