diff --git a/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcClient.java b/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcClient.java index 4f27bdef94..f258525915 100644 --- a/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcClient.java +++ b/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcClient.java @@ -204,6 +204,10 @@ public String scheduleNewOrchestrationInstance( builder.setScheduledStartTimestamp(ts); } + if (options.isEnforceUniqueInstanceId()) { + builder.setEnforceUniqueInstanceId(true); + } + Span span = null; if (this.tracer != null) { span = this.tracer.spanBuilder("create_orchestration:" + orchestratorName) diff --git a/durabletask-client/src/main/java/io/dapr/durabletask/NewOrchestrationInstanceOptions.java b/durabletask-client/src/main/java/io/dapr/durabletask/NewOrchestrationInstanceOptions.java index 32639e41d1..cac3c421ca 100644 --- a/durabletask-client/src/main/java/io/dapr/durabletask/NewOrchestrationInstanceOptions.java +++ b/durabletask-client/src/main/java/io/dapr/durabletask/NewOrchestrationInstanceOptions.java @@ -24,6 +24,7 @@ public final class NewOrchestrationInstanceOptions { private Object input; private Instant startTime; private String appID; // Target app ID for cross-app workflow routing + private boolean enforceUniqueInstanceId; /** * Default constructor for the {@link NewOrchestrationInstanceOptions} class. @@ -91,6 +92,20 @@ public NewOrchestrationInstanceOptions setAppID(String appID) { return this; } + /** + * Sets whether the instance ID of the new orchestration must be unique. + * When enabled, scheduling fails with an {@code ALREADY_EXISTS} gRPC status if an orchestration + * instance with the same ID already exists, regardless of whether that instance is still running + * or has already completed. By default, scheduling over a completed instance restarts it. + * + * @param enforceUniqueInstanceId whether to reject instance IDs that already exist + * @return this {@link NewOrchestrationInstanceOptions} object + */ + public NewOrchestrationInstanceOptions setEnforceUniqueInstanceId(boolean enforceUniqueInstanceId) { + this.enforceUniqueInstanceId = enforceUniqueInstanceId; + return this; + } + /** * Gets the user-specified version of the new orchestration. * @@ -136,6 +151,15 @@ public String getAppID() { return this.appID; } + /** + * Gets whether the instance ID of the new orchestration must be unique. + * + * @return true if instance IDs that already exist are rejected, false otherwise + */ + public boolean isEnforceUniqueInstanceId() { + return this.enforceUniqueInstanceId; + } + /** * Checks if an app ID is configured for cross-app routing. * diff --git a/durabletask-client/src/test/java/io/dapr/durabletask/DurableTaskGrpcClientScheduleTest.java b/durabletask-client/src/test/java/io/dapr/durabletask/DurableTaskGrpcClientScheduleTest.java new file mode 100644 index 0000000000..4c9cc74775 --- /dev/null +++ b/durabletask-client/src/test/java/io/dapr/durabletask/DurableTaskGrpcClientScheduleTest.java @@ -0,0 +1,135 @@ +/* + * 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.durabletask; + +import io.dapr.durabletask.implementation.protobuf.OrchestratorService; +import io.dapr.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc; +import io.grpc.ManagedChannel; +import io.grpc.Server; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; +import io.grpc.inprocess.InProcessChannelBuilder; +import io.grpc.inprocess.InProcessServerBuilder; +import io.grpc.stub.StreamObserver; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests that {@link NewOrchestrationInstanceOptions} are mapped onto the + * {@code CreateInstanceRequest} sent to the sidecar. + */ +class DurableTaskGrpcClientScheduleTest { + + private static final String ORCHESTRATION_NAME = "TestOrchestration"; + + private Server server; + private ManagedChannel channel; + private DurableTaskClient client; + private final AtomicReference capturedRequest = new AtomicReference<>(); + private final AtomicReference responseStatus = new AtomicReference<>(Status.OK); + + @BeforeEach + void setUp() throws Exception { + String serverName = InProcessServerBuilder.generateName(); + server = InProcessServerBuilder.forName(serverName) + .directExecutor() + .addService(new TaskHubSidecarServiceGrpc.TaskHubSidecarServiceImplBase() { + @Override + public void startInstance( + OrchestratorService.CreateInstanceRequest request, + StreamObserver responseObserver) { + capturedRequest.set(request); + Status status = responseStatus.get(); + if (!status.isOk()) { + responseObserver.onError(status.asRuntimeException()); + return; + } + responseObserver.onNext(OrchestratorService.CreateInstanceResponse.newBuilder() + .setInstanceId(request.getInstanceId()) + .build()); + responseObserver.onCompleted(); + } + }) + .build() + .start(); + channel = InProcessChannelBuilder.forName(serverName).directExecutor().build(); + client = new DurableTaskGrpcClientBuilder() + .grpcChannel(channel) + .build(); + } + + @AfterEach + void tearDown() throws Exception { + if (client != null) { + client.close(); + } + if (channel != null) { + channel.shutdownNow().awaitTermination(5, TimeUnit.SECONDS); + } + if (server != null) { + server.shutdownNow().awaitTermination(5, TimeUnit.SECONDS); + } + } + + @Test + void scheduleWithoutOptionsDoesNotEnforceUniqueInstanceId() { + client.scheduleNewOrchestrationInstance(ORCHESTRATION_NAME); + + assertFalse(capturedRequest.get().getEnforceUniqueInstanceId()); + } + + @Test + void scheduleWithDefaultOptionsDoesNotEnforceUniqueInstanceId() { + client.scheduleNewOrchestrationInstance(ORCHESTRATION_NAME, new NewOrchestrationInstanceOptions()); + + assertFalse(capturedRequest.get().getEnforceUniqueInstanceId()); + } + + @Test + void scheduleWithEnforceUniqueInstanceIdSetsRequestField() { + NewOrchestrationInstanceOptions options = new NewOrchestrationInstanceOptions() + .setInstanceId("myInstance") + .setEnforceUniqueInstanceId(true); + + String instanceId = client.scheduleNewOrchestrationInstance(ORCHESTRATION_NAME, options); + + OrchestratorService.CreateInstanceRequest request = capturedRequest.get(); + assertEquals("myInstance", instanceId); + assertEquals("myInstance", request.getInstanceId()); + assertTrue(request.getEnforceUniqueInstanceId()); + } + + @Test + void scheduleWithEnforceUniqueInstanceIdSurfacesAlreadyExists() { + responseStatus.set(Status.ALREADY_EXISTS.withDescription("a workflow with ID 'myInstance' already exists")); + NewOrchestrationInstanceOptions options = new NewOrchestrationInstanceOptions() + .setInstanceId("myInstance") + .setEnforceUniqueInstanceId(true); + + StatusRuntimeException exception = assertThrows(StatusRuntimeException.class, + () -> client.scheduleNewOrchestrationInstance(ORCHESTRATION_NAME, options)); + + assertEquals(Status.Code.ALREADY_EXISTS, exception.getStatus().getCode()); + assertTrue(capturedRequest.get().getEnforceUniqueInstanceId()); + } +} 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 d97b1e288b..3880c770e0 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 @@ -533,6 +533,8 @@ private static NewOrchestrationInstanceOptions fromNewWorkflowOptions(NewWorkflo instanceOptions.setStartTime(options.getStartTime()); } + instanceOptions.setEnforceUniqueInstanceId(options.isEnforceUniqueInstanceId()); + return instanceOptions; } diff --git a/sdk-workflows/src/main/java/io/dapr/workflows/client/NewWorkflowOptions.java b/sdk-workflows/src/main/java/io/dapr/workflows/client/NewWorkflowOptions.java index f808c19024..088f731127 100644 --- a/sdk-workflows/src/main/java/io/dapr/workflows/client/NewWorkflowOptions.java +++ b/sdk-workflows/src/main/java/io/dapr/workflows/client/NewWorkflowOptions.java @@ -24,6 +24,7 @@ public class NewWorkflowOptions { private String instanceId; private Object input; private Instant startTime; + private boolean enforceUniqueInstanceId; /** * Sets the version of the workflow to start. @@ -76,6 +77,24 @@ public NewWorkflowOptions setStartTime(Instant startTime) { return this; } + /** + * Sets whether the instance ID of the new workflow must be unique. + * + *

When enabled, scheduling fails with a {@link WorkflowInstanceAlreadyExistsException} + * if a workflow instance with the same ID already exists, regardless of whether that + * instance is still running or has already completed. By default, scheduling a workflow + * with the instance ID of a completed instance restarts that instance. + * + *

Requires a Dapr runtime that supports this option. + * + * @param enforceUniqueInstanceId whether to reject instance IDs that already exist + * @return this {@link NewWorkflowOptions} object + */ + public NewWorkflowOptions setEnforceUniqueInstanceId(boolean enforceUniqueInstanceId) { + this.enforceUniqueInstanceId = enforceUniqueInstanceId; + return this; + } + /** * Gets the user-specified version of the new workflow. * @@ -112,4 +131,13 @@ public Instant getStartTime() { return this.startTime; } + /** + * Gets whether the instance ID of the new workflow must be unique. + * + * @return true if instance IDs that already exist are rejected, false otherwise + */ + public boolean isEnforceUniqueInstanceId() { + return this.enforceUniqueInstanceId; + } + } 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 c346beee94..a24827abd3 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 @@ -16,12 +16,21 @@ 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. + * Exception thrown when scheduling a new workflow with an instance ID that the Dapr runtime + * rejects because a workflow instance with that ID already exists. * - *

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. + *

Which existing instances cause the rejection depends on + * {@link NewWorkflowOptions#setEnforceUniqueInstanceId(boolean)}: + * + *

    + *
  • By default (option disabled), the runtime only rejects instance IDs that belong to an + * active instance. 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.
    • + *
  • When the option is enabled, the runtime rejects the instance ID if an instance with that ID + * exists in any status, including terminal ones. The existing instance is left + * untouched.
    • + *
*/ public class WorkflowInstanceAlreadyExistsException extends RuntimeException { @@ -32,12 +41,13 @@ public class WorkflowInstanceAlreadyExistsException extends RuntimeException { * 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. + * @param cause the underlying gRPC exception returned by the sidecar. Its status description + * carries the runtime's own explanation of the collision. */ 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); + ? "a workflow with the requested instance ID already exists" + : String.format("a workflow with ID '%s' already exists", instanceId), cause); this.instanceId = 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 f88a7dbcc0..9c76c4ca43 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 @@ -39,10 +39,12 @@ import java.util.concurrent.TimeoutException; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; 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.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; @@ -264,6 +266,28 @@ public void scheduleNewWorkflowWithNewWorkflowOption() { assertEquals(expectedStartTime, captor.getValue().getStartTime()); assertEquals(expectedInput, captor.getValue().getInput()); + assertFalse(captor.getValue().isEnforceUniqueInstanceId()); + } + + @Test + public void scheduleNewWorkflowWithEnforceUniqueInstanceId() { + String expectedName = TestWorkflow.class.getCanonicalName(); + String expectedInstanceId = "uniqueInstance"; + NewWorkflowOptions newWorkflowOptions = new NewWorkflowOptions() + .setInstanceId(expectedInstanceId) + .setEnforceUniqueInstanceId(true); + + client.scheduleNewWorkflow(TestWorkflow.class, newWorkflowOptions); + + ArgumentCaptor captor = ArgumentCaptor.forClass( + NewOrchestrationInstanceOptions.class + ); + + verify(mockInnerClient, times(1)) + .scheduleNewOrchestrationInstance(eq(expectedName), captor.capture()); + + assertEquals(expectedInstanceId, captor.getValue().getInstanceId()); + assertTrue(captor.getValue().isEnforceUniqueInstanceId()); } @Test diff --git a/sdk-workflows/src/test/java/io/dapr/workflows/client/NewWorkflowOptionsTest.java b/sdk-workflows/src/test/java/io/dapr/workflows/client/NewWorkflowOptionsTest.java index e1d10c68ce..4c02874e6a 100644 --- a/sdk-workflows/src/test/java/io/dapr/workflows/client/NewWorkflowOptionsTest.java +++ b/sdk-workflows/src/test/java/io/dapr/workflows/client/NewWorkflowOptionsTest.java @@ -18,11 +18,18 @@ void testNewWorkflowOption() { workflowOption.setVersion(version) .setInstanceId(instanceId) .setInput(input) - .setStartTime(startTime); + .setStartTime(startTime) + .setEnforceUniqueInstanceId(true); Assertions.assertEquals(version, workflowOption.getVersion()); Assertions.assertEquals(instanceId, workflowOption.getInstanceId()); Assertions.assertEquals(input, workflowOption.getInput()); Assertions.assertEquals(startTime, workflowOption.getStartTime()); + Assertions.assertTrue(workflowOption.isEnforceUniqueInstanceId()); + } + + @Test + void testEnforceUniqueInstanceIdDefaultsToFalse() { + Assertions.assertFalse(new NewWorkflowOptions().isEnforceUniqueInstanceId()); } } diff --git a/sdk-workflows/src/test/java/io/dapr/workflows/client/WorkflowInstanceAlreadyExistsExceptionTest.java b/sdk-workflows/src/test/java/io/dapr/workflows/client/WorkflowInstanceAlreadyExistsExceptionTest.java new file mode 100644 index 0000000000..a8cc28fd31 --- /dev/null +++ b/sdk-workflows/src/test/java/io/dapr/workflows/client/WorkflowInstanceAlreadyExistsExceptionTest.java @@ -0,0 +1,49 @@ +/* + * 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 io.grpc.Status; +import io.grpc.StatusRuntimeException; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; + +public class WorkflowInstanceAlreadyExistsExceptionTest { + + @Test + public void messageIncludesInstanceIdWhenKnown() { + StatusRuntimeException cause = new StatusRuntimeException(Status.ALREADY_EXISTS); + + WorkflowInstanceAlreadyExistsException exception = + new WorkflowInstanceAlreadyExistsException("myInstance", cause); + + assertEquals("a workflow with ID 'myInstance' already exists", exception.getMessage()); + assertEquals("myInstance", exception.getInstanceId()); + assertSame(cause, exception.getCause()); + } + + @Test + public void messageIsGenericWhenInstanceIdIsUnknown() { + StatusRuntimeException cause = new StatusRuntimeException(Status.ALREADY_EXISTS); + + WorkflowInstanceAlreadyExistsException exception = + new WorkflowInstanceAlreadyExistsException(null, cause); + + assertEquals("a workflow with the requested instance ID already exists", exception.getMessage()); + assertNull(exception.getInstanceId()); + assertSame(cause, exception.getCause()); + } +}