feat: throw WorkflowInstanceAlreadyExistsException when scheduling a workflow with an active instance ID

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 <T extends Workflow> String scheduleNewWorkflow(Class<T> clazz) {
    * @return the randomly-generated instance ID for new Workflow instance.
    */
   public <T extends Workflow> String scheduleNewWorkflow(String name) {
-    return this.innerClient.scheduleNewOrchestrationInstance(name);
+    return mapAlreadyExists(null, () -> this.innerClient.scheduleNewOrchestrationInstance(name));
   }
 
   /**
@@ -133,7 +136,7 @@ public <T extends Workflow> String scheduleNewWorkflow(Class<T> clazz, Object in
    * @return the randomly-generated instance ID for new Workflow instance.
    */
   public <T extends Workflow> 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 <T extends Workflow> 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 <code>instanceId</code> parameter value.
+   * @throws WorkflowInstanceAlreadyExistsException if an active workflow with <code>instanceId</code> already exists.
    */
   public <T extends Workflow> String scheduleNewWorkflow(Class<T> clazz, Object input, String instanceId) {
     return this.scheduleNewWorkflow(clazz.getCanonicalName(), input, instanceId);
@@ -157,9 +161,11 @@ public <T extends Workflow> String scheduleNewWorkflow(Class<T> 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 <code>instanceId</code> parameter value.
+   * @throws WorkflowInstanceAlreadyExistsException if an active workflow with <code>instanceId</code> already exists.
    */
   public <T extends Workflow> 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 <T extends Workflow> 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 <code>instanceId</code> parameter value.
+   * @throws WorkflowInstanceAlreadyExistsException if an active workflow with the requested instance ID
+   *                                                already exists.
    */
   public <T extends Workflow> String scheduleNewWorkflow(Class<T> clazz, NewWorkflowOptions options) {
     return this.scheduleNewWorkflow(clazz.getCanonicalName(), options);
@@ -181,11 +189,13 @@ public <T extends Workflow> String scheduleNewWorkflow(Class<T> 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 <code>instanceId</code> parameter value.
+   * @throws WorkflowInstanceAlreadyExistsException if an active workflow with the requested instance ID
+   *                                                already exists.
    */
   public <T extends Workflow> 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,34 @@ 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<String> 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 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("an active workflow with ID");
+  }
+
   private static NewOrchestrationInstanceOptions fromNewWorkflowOptions(NewWorkflowOptions options) {
     NewOrchestrationInstanceOptions instanceOptions = new NewOrchestrationInstanceOptions();
 

sdk-workflows/src/main/java/io/dapr/workflows/client/WorkflowInstanceAlreadyExistsException.java

@@ -0,0 +1,53 @@
+/*
+ * 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.
+ *
+ * <p>The Dapr runtime only rejects duplicate instance IDs of <em>active</em> 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 {
+
+  @Nullable
+  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;
+  }
+}

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,94 @@ 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());
+
+    ArgumentCaptor<NewOrchestrationInstanceOptions> 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
+  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();