This is an automated email from the ASF dual-hosted git repository.
SbloodyS pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
The following commit(s) were added to refs/heads/dev by this push:
new 29a04ecef2 [Fix-18570][Master] Detect wrapped
CommandDuplicateHandleException in bootstrapError (#18570) (#18573)
29a04ecef2 is described below
commit 29a04ecef280b0177b3fabeabb2f0e16679d5422
Author: 丁明亮 <[email protected]>
AuthorDate: Fri Aug 21 10:00:28 2026 +0800
[Fix-18570][Master] Detect wrapped CommandDuplicateHandleException in
bootstrapError (#18570) (#18573)
---
.../master/engine/command/CommandEngine.java | 4 +-
.../master/engine/command/CommandEngineTest.java | 133 +++++++++++++++++++++
2 files changed, 136 insertions(+), 1 deletion(-)
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/CommandEngine.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/CommandEngine.java
index 693139d954..63d41bad74 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/CommandEngine.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/CommandEngine.java
@@ -193,7 +193,9 @@ public class CommandEngine extends BaseDaemonThread
implements AutoCloseable {
}
private Void bootstrapError(Command command, Throwable throwable) {
- if (throwable instanceof CommandDuplicateHandleException) {
+ // The exception is raised inside a CompletableFuture chain, so it
arrives here wrapped in
+ // CompletionException, which a direct instanceof check cannot see
through. See #18570.
+ if (ExceptionUtils.throwableOfType(throwable,
CommandDuplicateHandleException.class) != null) {
log.warn("Handle command failed, the command: {} has been handled
by other master",
command,
throwable);
diff --git
a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/command/CommandEngineTest.java
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/command/CommandEngineTest.java
new file mode 100644
index 0000000000..0a051dbc90
--- /dev/null
+++
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/command/CommandEngineTest.java
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dolphinscheduler.server.master.engine.command;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
+import org.apache.dolphinscheduler.dao.entity.Command;
+import org.apache.dolphinscheduler.dao.repository.WorkflowInstanceDao;
+import
org.apache.dolphinscheduler.server.master.engine.exceptions.CommandDuplicateHandleException;
+import org.apache.dolphinscheduler.service.command.CommandService;
+
+import java.util.concurrent.CompletionException;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.transaction.support.TransactionCallback;
+import org.springframework.transaction.support.TransactionTemplate;
+
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.LENIENT)
+class CommandEngineTest {
+
+ @Mock
+ private WorkflowInstanceDao workflowInstanceDao;
+
+ @Mock
+ private CommandService commandService;
+
+ @Mock
+ private TransactionTemplate transactionTemplate;
+
+ @InjectMocks
+ private CommandEngine commandEngine;
+
+ /**
+ * Execute the transaction callback inline, so that reaching the
non-duplicate branch produces a
+ * clear verification failure instead of a NullPointerException.
+ */
+ @BeforeEach
+ void setUp() {
+ when(transactionTemplate.execute(any()))
+ .thenAnswer(invocation -> {
+ final TransactionCallback<?> callback =
invocation.getArgument(0);
+ return callback.doInTransaction(null);
+ });
+ }
+
+ private static Command duplicatedCommand() {
+ final Command command = new Command();
+ command.setId(520266);
+ command.setWorkflowInstanceId(516982);
+ return command;
+ }
+
+ private void invokeBootstrapError(Command command, Throwable throwable) {
+ ReflectionTestUtils.invokeMethod(commandEngine, "bootstrapError",
command, throwable);
+ }
+
+ private void assertWorkflowWasNotForceFailed() {
+ verify(workflowInstanceDao, never())
+ .forceUpdateWorkflowInstanceState(anyInt(),
any(WorkflowExecutionStatus.class));
+ verify(commandService, never()).moveToErrorCommand(any(Command.class),
anyString());
+ }
+
+ /**
+ * The duplicate-handle exception is raised inside a CompletableFuture
chain and therefore reaches
+ * bootstrapError wrapped in CompletionException. It must still be
recognised, otherwise the healthy
+ * first execution has its workflow instance force-failed underneath it,
can never complete, never
+ * leaves the in-memory workflow repository, and permanently pins the
instance count that
+ * MasterServerLoadProtection uses - deadlocking command consumption. See
#18570.
+ */
+ @Test
+ void bootstrapErrorShouldNotFailWorkflowWhenDuplicateExceptionIsWrapped() {
+ final Command command = duplicatedCommand();
+
+ invokeBootstrapError(command, new CompletionException(new
CommandDuplicateHandleException(command)));
+
+ assertWorkflowWasNotForceFailed();
+ }
+
+ /**
+ * Deeply nested wrapping must be handled as well.
+ */
+ @Test
+ void
bootstrapErrorShouldNotFailWorkflowWhenDuplicateExceptionIsNestedTwice() {
+ final Command command = duplicatedCommand();
+
+ invokeBootstrapError(command,
+ new CompletionException(new RuntimeException(new
CommandDuplicateHandleException(command))));
+
+ assertWorkflowWasNotForceFailed();
+ }
+
+ /**
+ * The unwrapped case must keep behaving exactly as before.
+ */
+ @Test
+ void bootstrapErrorShouldNotFailWorkflowWhenDuplicateExceptionIsDirect() {
+ final Command command = duplicatedCommand();
+
+ invokeBootstrapError(command, new
CommandDuplicateHandleException(command));
+
+ assertWorkflowWasNotForceFailed();
+ }
+}
