From ef89a24432ccd259ff55d0d3abfe14d75b69096b Mon Sep 17 00:00:00 2001 From: Javier Aliaga Date: Mon, 6 Jul 2026 17:06:35 +0200 Subject: [PATCH 1/2] feat: throw WorkflowInstanceAlreadyExistsException when scheduling a workflow with an active instance ID Signed-off-by: Javier Aliaga --- .../workflows/client/DaprWorkflowClient.java | 45 ++++++++++-- ...orkflowInstanceAlreadyExistsException.java | 52 +++++++++++++ .../client/DaprWorkflowClientTest.java | 73 +++++++++++++++++++ 3 files changed, 165 insertions(+), 5 deletions(-) create mode 100644 sdk-workflows/src/main/java/io/dapr/workflows/client/WorkflowInstanceAlreadyExistsException.java diff --git a/sdk-workflows/src/main/java/io/dapr/workflows/client/DaprWorkflowClient.java b/sdk-workflows/src/main/java/io/dapr/workflows/client/DaprWorkflowClient.java index bb4e63d76d..87cdbad5fc 100644 --- a/sdk-workflows/src/main/java/io/dapr/workflows/client/DaprWorkflowClient.java +++ b/sdk-workflows/src/main/java/io/dapr/workflows/client/DaprWorkflowClient.java @@ -26,12 +26,15 @@ import io.dapr.workflows.runtime.DefaultWorkflowState; import io.grpc.ClientInterceptor; import io.grpc.ManagedChannel; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; import javax.annotation.Nullable; import java.time.Duration; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.function.Supplier; /** * Defines client operations for managing Dapr Workflow instances. @@ -109,7 +112,7 @@ public String scheduleNewWorkflow(Class clazz) { * @return the randomly-generated instance ID for new Workflow instance. */ public String scheduleNewWorkflow(String name) { - return this.innerClient.scheduleNewOrchestrationInstance(name); + return mapAlreadyExists(null, () -> this.innerClient.scheduleNewOrchestrationInstance(name)); } /** @@ -133,7 +136,7 @@ public String scheduleNewWorkflow(Class clazz, Object in * @return the randomly-generated instance ID for new Workflow instance. */ public String scheduleNewWorkflow(String name, Object input) { - return this.innerClient.scheduleNewOrchestrationInstance(name, input); + return mapAlreadyExists(null, () -> this.innerClient.scheduleNewOrchestrationInstance(name, input)); } /** @@ -144,6 +147,7 @@ public String scheduleNewWorkflow(String name, Object input * @param input the input to pass to the scheduled orchestration instance. Must be serializable. * @param instanceId the unique ID of the orchestration instance to schedule * @return the instanceId parameter value. + * @throws WorkflowInstanceAlreadyExistsException if an active workflow with instanceId already exists. */ public String scheduleNewWorkflow(Class clazz, Object input, String instanceId) { return this.scheduleNewWorkflow(clazz.getCanonicalName(), input, instanceId); @@ -157,9 +161,11 @@ public String scheduleNewWorkflow(Class clazz, Object in * @param input the input to pass to the scheduled orchestration instance. Must be serializable. * @param instanceId the unique ID of the orchestration instance to schedule * @return the instanceId parameter value. + * @throws WorkflowInstanceAlreadyExistsException if an active workflow with instanceId already exists. */ public String scheduleNewWorkflow(String name, Object input, String instanceId) { - return this.innerClient.scheduleNewOrchestrationInstance(name, input, instanceId); + return mapAlreadyExists(instanceId, () -> this.innerClient.scheduleNewOrchestrationInstance(name, input, + instanceId)); } /** @@ -169,6 +175,8 @@ public String scheduleNewWorkflow(String name, Object input * @param clazz Class extending Workflow to start an instance of. * @param options the options for the new workflow, including input, instance ID, etc. * @return the instanceId parameter value. + * @throws WorkflowInstanceAlreadyExistsException if an active workflow with the requested instance ID + * already exists. */ public String scheduleNewWorkflow(Class clazz, NewWorkflowOptions options) { return this.scheduleNewWorkflow(clazz.getCanonicalName(), options); @@ -181,11 +189,13 @@ public String scheduleNewWorkflow(Class clazz, NewWorkfl * @param name name of the workflow to schedule * @param options the options for the new workflow, including input, instance ID, etc. * @return the instanceId parameter value. + * @throws WorkflowInstanceAlreadyExistsException if an active workflow with the requested instance ID + * already exists. */ public String scheduleNewWorkflow(String name, NewWorkflowOptions options) { NewOrchestrationInstanceOptions orchestrationInstanceOptions = fromNewWorkflowOptions(options); - return this.innerClient.scheduleNewOrchestrationInstance(name, - orchestrationInstanceOptions); + return mapAlreadyExists(options.getInstanceId(), () -> this.innerClient.scheduleNewOrchestrationInstance(name, + orchestrationInstanceOptions)); } /** @@ -452,6 +462,31 @@ private static DurableTaskClient createDurableTaskClient(ManagedChannel grpcChan .build(); } + /** + * Runs the given scheduling call, translating the sidecar's rejection of a duplicate active + * instance ID into a {@link WorkflowInstanceAlreadyExistsException}. + */ + private static String mapAlreadyExists(@Nullable String instanceId, Supplier schedule) { + try { + return schedule.get(); + } catch (StatusRuntimeException e) { + if (isAlreadyExists(e)) { + throw new WorkflowInstanceAlreadyExistsException(instanceId, e); + } + throw e; + } + } + + private static boolean isAlreadyExists(StatusRuntimeException e) { + if (e.getStatus().getCode() == Status.Code.ALREADY_EXISTS) { + return true; + } + // Runtimes that predate the AlreadyExists status code only identify the collision + // through the error message. + String description = e.getStatus().getDescription(); + return description != null && description.contains("already exists"); + } + private static NewOrchestrationInstanceOptions fromNewWorkflowOptions(NewWorkflowOptions options) { NewOrchestrationInstanceOptions instanceOptions = new NewOrchestrationInstanceOptions(); diff --git a/sdk-workflows/src/main/java/io/dapr/workflows/client/WorkflowInstanceAlreadyExistsException.java b/sdk-workflows/src/main/java/io/dapr/workflows/client/WorkflowInstanceAlreadyExistsException.java new file mode 100644 index 0000000000..13b5244f9e --- /dev/null +++ b/sdk-workflows/src/main/java/io/dapr/workflows/client/WorkflowInstanceAlreadyExistsException.java @@ -0,0 +1,52 @@ +/* + * Copyright 2026 The Dapr Authors + * Licensed 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 io.dapr.workflows.client; + +import javax.annotation.Nullable; + +/** + * Exception thrown when scheduling a new workflow with an instance ID that is already in use + * by an active workflow instance. + * + *

The Dapr runtime only rejects duplicate instance IDs of active instances: scheduling + * with the instance ID of a workflow that already reached a terminal state (completed, failed or + * terminated) succeeds and re-runs the workflow with fresh state. + */ +public class WorkflowInstanceAlreadyExistsException extends RuntimeException { + + private final String instanceId; + + /** + * Constructor for WorkflowInstanceAlreadyExistsException. + * + * @param instanceId the instance ID that is already in use, or null when not known. + * @param cause the underlying gRPC exception returned by the sidecar. + */ + public WorkflowInstanceAlreadyExistsException(@Nullable String instanceId, Throwable cause) { + super(instanceId == null + ? "an active workflow with the requested instance ID already exists" + : String.format("an active workflow with ID '%s' already exists", instanceId), cause); + this.instanceId = instanceId; + } + + /** + * Returns the workflow instance ID that is already in use. + * + * @return the instance ID, or null when the collision was reported for a generated ID. + */ + @Nullable + public String getInstanceId() { + return instanceId; + } +} diff --git a/sdk-workflows/src/test/java/io/dapr/workflows/client/DaprWorkflowClientTest.java b/sdk-workflows/src/test/java/io/dapr/workflows/client/DaprWorkflowClientTest.java index 2faf1e1080..2b1bc3bb6f 100644 --- a/sdk-workflows/src/test/java/io/dapr/workflows/client/DaprWorkflowClientTest.java +++ b/sdk-workflows/src/test/java/io/dapr/workflows/client/DaprWorkflowClientTest.java @@ -21,6 +21,8 @@ import io.dapr.workflows.WorkflowContext; import io.dapr.workflows.WorkflowStub; import io.grpc.ManagedChannel; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -35,6 +37,8 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; @@ -115,6 +119,75 @@ public void scheduleNewWorkflowWithArgsNameInputInstance() { .scheduleNewOrchestrationInstance(expectedName, expectedInput, expectedInstanceId); } + @Test + public void scheduleNewWorkflowThrowsWhenInstanceAlreadyExists() { + String expectedName = TestWorkflow.class.getCanonicalName(); + Object expectedInput = new Object(); + String expectedInstanceId = "duplicateInstance"; + StatusRuntimeException grpcException = new StatusRuntimeException( + Status.ALREADY_EXISTS.withDescription( + "an active workflow with ID 'duplicateInstance' already exists")); + when(mockInnerClient.scheduleNewOrchestrationInstance(expectedName, expectedInput, expectedInstanceId)) + .thenThrow(grpcException); + + WorkflowInstanceAlreadyExistsException exception = assertThrows(WorkflowInstanceAlreadyExistsException.class, + () -> client.scheduleNewWorkflow(TestWorkflow.class, expectedInput, expectedInstanceId)); + + assertEquals(expectedInstanceId, exception.getInstanceId()); + assertSame(grpcException, exception.getCause()); + } + + @Test + public void scheduleNewWorkflowThrowsWhenInstanceAlreadyExistsOnLegacyRuntime() { + // Runtimes that predate the AlreadyExists status code report the collision + // as UNKNOWN with only the message identifying the failure. + String expectedName = TestWorkflow.class.getCanonicalName(); + String expectedInstanceId = "duplicateInstance"; + StatusRuntimeException grpcException = new StatusRuntimeException( + Status.UNKNOWN.withDescription( + "failed to create workflow instance: an active workflow with ID 'duplicateInstance' already exists")); + when(mockInnerClient.scheduleNewOrchestrationInstance(expectedName, null, expectedInstanceId)) + .thenThrow(grpcException); + + WorkflowInstanceAlreadyExistsException exception = assertThrows(WorkflowInstanceAlreadyExistsException.class, + () -> client.scheduleNewWorkflow(TestWorkflow.class, null, expectedInstanceId)); + + assertEquals(expectedInstanceId, exception.getInstanceId()); + assertSame(grpcException, exception.getCause()); + } + + @Test + public void scheduleNewWorkflowWithOptionsThrowsWhenInstanceAlreadyExists() { + String expectedName = TestWorkflow.class.getCanonicalName(); + String expectedInstanceId = "duplicateInstance"; + NewWorkflowOptions options = new NewWorkflowOptions().setInstanceId(expectedInstanceId); + StatusRuntimeException grpcException = new StatusRuntimeException( + Status.ALREADY_EXISTS.withDescription( + "an active workflow with ID 'duplicateInstance' already exists")); + when(mockInnerClient.scheduleNewOrchestrationInstance(eq(expectedName), any(NewOrchestrationInstanceOptions.class))) + .thenThrow(grpcException); + + WorkflowInstanceAlreadyExistsException exception = assertThrows(WorkflowInstanceAlreadyExistsException.class, + () -> client.scheduleNewWorkflow(TestWorkflow.class, options)); + + assertEquals(expectedInstanceId, exception.getInstanceId()); + assertSame(grpcException, exception.getCause()); + } + + @Test + public void scheduleNewWorkflowRethrowsUnrelatedGrpcErrors() { + String expectedName = TestWorkflow.class.getCanonicalName(); + StatusRuntimeException grpcException = new StatusRuntimeException( + Status.UNAVAILABLE.withDescription("sidecar unavailable")); + when(mockInnerClient.scheduleNewOrchestrationInstance(expectedName, null, "myInstance")) + .thenThrow(grpcException); + + StatusRuntimeException exception = assertThrows(StatusRuntimeException.class, + () -> client.scheduleNewWorkflow(TestWorkflow.class, null, "myInstance")); + + assertSame(grpcException, exception); + } + @Test public void scheduleNewWorkflowWithNewWorkflowOption() { String expectedName = TestWorkflow.class.getCanonicalName(); From 610c55ebd3ad7711227b77a139752d558907f2a5 Mon Sep 17 00:00:00 2001 From: Javier Aliaga Date: Mon, 6 Jul 2026 17:21:26 +0200 Subject: [PATCH 2/2] fix: address Copilot review feedback on already-exists mapping Annotate nullable instanceId field, restrict the legacy fallback to UNKNOWN status with the runtime's specific collision message, and assert instance ID propagation into NewOrchestrationInstanceOptions. Signed-off-by: Javier Aliaga --- .../workflows/client/DaprWorkflowClient.java | 9 ++++++--- ...orkflowInstanceAlreadyExistsException.java | 1 + .../client/DaprWorkflowClientTest.java | 19 +++++++++++++++++++ 3 files changed, 26 insertions(+), 3 deletions(-) diff --git a/sdk-workflows/src/main/java/io/dapr/workflows/client/DaprWorkflowClient.java b/sdk-workflows/src/main/java/io/dapr/workflows/client/DaprWorkflowClient.java index 87cdbad5fc..de9fe466e1 100644 --- a/sdk-workflows/src/main/java/io/dapr/workflows/client/DaprWorkflowClient.java +++ b/sdk-workflows/src/main/java/io/dapr/workflows/client/DaprWorkflowClient.java @@ -481,10 +481,13 @@ private static boolean isAlreadyExists(StatusRuntimeException e) { if (e.getStatus().getCode() == Status.Code.ALREADY_EXISTS) { return true; } - // Runtimes that predate the AlreadyExists status code only identify the collision - // through the error message. + // Runtimes that predate the AlreadyExists status code report the collision as UNKNOWN and + // only identify it through the error message. + if (e.getStatus().getCode() != Status.Code.UNKNOWN) { + return false; + } String description = e.getStatus().getDescription(); - return description != null && description.contains("already exists"); + return description != null && description.contains("an active workflow with ID"); } private static NewOrchestrationInstanceOptions fromNewWorkflowOptions(NewWorkflowOptions options) { diff --git a/sdk-workflows/src/main/java/io/dapr/workflows/client/WorkflowInstanceAlreadyExistsException.java b/sdk-workflows/src/main/java/io/dapr/workflows/client/WorkflowInstanceAlreadyExistsException.java index 13b5244f9e..c346beee94 100644 --- a/sdk-workflows/src/main/java/io/dapr/workflows/client/WorkflowInstanceAlreadyExistsException.java +++ b/sdk-workflows/src/main/java/io/dapr/workflows/client/WorkflowInstanceAlreadyExistsException.java @@ -25,6 +25,7 @@ */ public class WorkflowInstanceAlreadyExistsException extends RuntimeException { + @Nullable private final String instanceId; /** diff --git a/sdk-workflows/src/test/java/io/dapr/workflows/client/DaprWorkflowClientTest.java b/sdk-workflows/src/test/java/io/dapr/workflows/client/DaprWorkflowClientTest.java index 2b1bc3bb6f..208cd66dc3 100644 --- a/sdk-workflows/src/test/java/io/dapr/workflows/client/DaprWorkflowClientTest.java +++ b/sdk-workflows/src/test/java/io/dapr/workflows/client/DaprWorkflowClientTest.java @@ -172,6 +172,25 @@ public void scheduleNewWorkflowWithOptionsThrowsWhenInstanceAlreadyExists() { assertEquals(expectedInstanceId, exception.getInstanceId()); assertSame(grpcException, exception.getCause()); + + ArgumentCaptor captor = + ArgumentCaptor.forClass(NewOrchestrationInstanceOptions.class); + verify(mockInnerClient, times(1)).scheduleNewOrchestrationInstance(eq(expectedName), captor.capture()); + assertEquals(expectedInstanceId, captor.getValue().getInstanceId()); + } + + @Test + public void scheduleNewWorkflowRethrowsUnknownErrorsWithoutCollisionMessage() { + String expectedName = TestWorkflow.class.getCanonicalName(); + StatusRuntimeException grpcException = new StatusRuntimeException( + Status.UNKNOWN.withDescription("failed to create workflow instance: state store unreachable")); + when(mockInnerClient.scheduleNewOrchestrationInstance(expectedName, null, "myInstance")) + .thenThrow(grpcException); + + StatusRuntimeException exception = assertThrows(StatusRuntimeException.class, + () -> client.scheduleNewWorkflow(TestWorkflow.class, null, "myInstance")); + + assertSame(grpcException, exception); } @Test