smengcl commented on code in PR #10733:
URL: https://github.com/apache/ozone/pull/10733#discussion_r3568171920
##########
hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestDeleteBlocksCommandHandler.java:
##########
@@ -315,6 +321,109 @@ public void
testDeleteBlocksCommandHandlerExceptionShouldNotInterrupt() throws E
assertEquals(1, deleteBlockTransactionResults.size());
}
+ @Test
+ public void testDeleteBlocksCommandHandlerErrorShouldInterrupt() throws
Exception {
+ setup();
+ Error error = new AssertionError("Simulated Error");
+ CompletableFuture<DeleteBlockTransactionExecutionResult> failed =
+ new CompletableFuture<>();
+ failed.completeExceptionally(error);
+ CompletableFuture<DeleteBlockTransactionExecutionResult> unprocessed =
+ CompletableFuture.completedFuture(
+ new DeleteBlockTransactionExecutionResult(null, false));
+ AtomicInteger processed = new AtomicInteger();
+
+ Error thrown = assertThrows(Error.class, () -> handler.handleTasksResults(
+ Arrays.asList(failed, unprocessed), result ->
processed.incrementAndGet()));
+
+ assertSame(error, thrown);
+ assertEquals(0, processed.get());
+ }
+
+ @Test
+ public void testDeleteBlocksCommandHandlerErrorOnRetryShouldInterrupt()
+ throws Exception {
+ setup();
+ DeletedBlocksTransaction transaction =
+ createDeletedBlocksTransaction(1, 1);
+ DeleteBlockTransactionResult retryResult = DeleteBlockTransactionResult
+ .newBuilder()
+ .setTxID(transaction.getTxID())
+ .setContainerID(transaction.getContainerID())
+ .setSuccess(false)
+ .build();
+ CompletableFuture<DeleteBlockTransactionExecutionResult> retry =
+ CompletableFuture.completedFuture(
+ new DeleteBlockTransactionExecutionResult(retryResult, true));
+ Error error = new AssertionError("Simulated retry Error");
+ CompletableFuture<DeleteBlockTransactionExecutionResult> failed =
+ new CompletableFuture<>();
+ failed.completeExceptionally(error);
+ AtomicInteger invocation = new AtomicInteger();
+ doAnswer(ignored -> invocation.getAndIncrement() == 0
+ ? Collections.singletonList(retry)
+ : Collections.singletonList(failed))
+ .when(handler).submitTasks(any());
+
+ Error thrown = assertThrows(Error.class,
+ () -> handler.executeCmdWithRetry(
+ Collections.singletonList(transaction)));
+
+ assertSame(error, thrown);
+ assertEquals(2, invocation.get());
+ }
+
+ @Test
+ public void testDeleteCmdWorkerTerminatesOnError() throws Exception {
+ setup();
+ ExitUtils.disableSystemExit();
+ Container<?> container = containerSet.getContainer(1);
+ String schemaVersionOrDefault = ((KeyValueContainerData)
+ container.getContainerData()).getSupportedSchemaVersionOrDefault();
+ Error error = new AssertionError("Simulated worker Error");
+ SchemaHandler schemaHandler =
+ handler.getSchemaHandlers().get(schemaVersionOrDefault);
+ CountDownLatch processingStarted = new CountDownLatch(1);
+ CountDownLatch failProcessing = new CountDownLatch(1);
+ doAnswer(ignored -> {
+ processingStarted.countDown();
+ assertTrue(failProcessing.await(5, TimeUnit.SECONDS));
+ throw error;
+ }).when(schemaHandler).handle(any(), any());
+
+ OzoneConfiguration conf = new OzoneConfiguration();
+ DatanodeStateMachine stateMachine = mock(DatanodeStateMachine.class);
+ DatanodeDetails datanodeDetails =
+ MockDatanodeDetails.randomDatanodeDetails();
+ when(stateMachine.getDatanodeDetails()).thenReturn(datanodeDetails);
+ StateContext context = new StateContext(conf,
+ DatanodeStateMachine.DatanodeStates.RUNNING, stateMachine, "");
+ DeleteBlocksCommand fatalCommand = new DeleteBlocksCommand(
+ Collections.singletonList(createDeletedBlocksTransaction(1, 1)));
+ DeleteBlocksCommand queuedCommand = new DeleteBlocksCommand(emptyList());
+ context.addCommand(fatalCommand);
+ context.addCommand(queuedCommand);
+
+ handler.handle(fatalCommand, mock(OzoneContainer.class), context,
+ mock(SCMConnectionManager.class));
+ assertTrue(processingStarted.await(5, TimeUnit.SECONDS));
+ try {
+ handler.handle(queuedCommand, mock(OzoneContainer.class), context,
+ mock(SCMConnectionManager.class));
+ } finally {
+ failProcessing.countDown();
+ }
+
+ GenericTestUtils.waitFor(ExitUtils::isTerminated, 10, 5000);
+
+ assertSame(error, ExitUtils.getFirstExitException().getCause());
+ CommandStatus fatalStatus = context.getCmdStatus(fatalCommand.getId());
+ assertEquals(Status.FAILED, fatalStatus.getStatus());
+ assertFalse(fatalStatus.getProtoBufMessage().hasBlockDeletionAck());
+ assertEquals(Status.PENDING,
+ context.getCmdStatus(queuedCommand.getId()).getStatus());
Review Comment:
nit: this assert can fit on one line, with 120 char limit
```suggestion
assertEquals(Status.PENDING,
context.getCmdStatus(queuedCommand.getId()).getStatus());
```
--
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]
