This is an automated email from the ASF dual-hosted git repository.
smengcl pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 958465f6e28 HDDS-13472. Shut down datanode if Error occurs while
processing delete blocks command (#10733)
958465f6e28 is described below
commit 958465f6e280d980ebc3a193c35b0762e3080b0f
Author: Chi-Hsuan Huang <[email protected]>
AuthorDate: Tue Jul 14 04:04:39 2026 +0800
HDDS-13472. Shut down datanode if Error occurs while processing delete
blocks command (#10733)
Co-authored-by: Siyao Meng <[email protected]>
---
.../commandhandler/DeleteBlocksCommandHandler.java | 7 ++
.../TestDeleteBlocksCommandHandler.java | 108 +++++++++++++++++++++
2 files changed, 115 insertions(+)
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/DeleteBlocksCommandHandler.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/DeleteBlocksCommandHandler.java
index 45ea5bffdec..c947e3a3082 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/DeleteBlocksCommandHandler.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/DeleteBlocksCommandHandler.java
@@ -73,6 +73,7 @@
import org.apache.hadoop.ozone.protocol.commands.SCMCommand;
import org.apache.hadoop.util.Daemon;
import org.apache.hadoop.util.Time;
+import org.apache.ratis.util.ExitUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -254,6 +255,9 @@ public void run() {
DeleteCmdInfo cmd = deleteCommandQueues.poll();
try {
processCmd(cmd);
+ } catch (Error e) {
+ ExitUtils.terminate(1,
+ "Fatal error while processing delete blocks command", e, LOG);
} catch (Throwable e) {
LOG.error("taskProcess failed.", e);
}
@@ -500,6 +504,9 @@ public void handleTasksResults(
DeleteBlockTransactionExecutionResult result = f.get();
handler.accept(result);
} catch (ExecutionException e) {
+ if (e.getCause() instanceof Error) {
+ throw (Error) e.getCause();
+ }
LOG.error("task failed.", e);
} catch (InterruptedException e) {
LOG.error("task interrupted.", e);
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestDeleteBlocksCommandHandler.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestDeleteBlocksCommandHandler.java
index 6ac39d9ba4f..a85f80ca373 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestDeleteBlocksCommandHandler.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestDeleteBlocksCommandHandler.java
@@ -28,6 +28,8 @@
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.eq;
@@ -52,6 +54,7 @@
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
@@ -82,6 +85,8 @@
import org.apache.hadoop.ozone.container.ozoneimpl.OzoneContainer;
import org.apache.hadoop.ozone.protocol.commands.CommandStatus;
import org.apache.hadoop.ozone.protocol.commands.DeleteBlocksCommand;
+import org.apache.ozone.test.GenericTestUtils;
+import org.apache.ratis.util.ExitUtils;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -167,6 +172,7 @@ private void setup() throws Exception {
public void tearDown() {
handler.stop();
BlockDeletingServiceMetrics.unRegister();
+ ExitUtils.clear();
}
@ContainerTestVersionInfo.ContainerTest
@@ -315,6 +321,108 @@ 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());
+ }
+
@ContainerTestVersionInfo.ContainerTest
public void testDeleteCmdWorkerInterval(
ContainerTestVersionInfo versionInfo) throws Exception {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
