Add setWorkflowIdConflictPolicy

temporal-sdk/src/main/java/io/temporal/client/WorkflowOptions.java

@@ -21,6 +21,7 @@
 package io.temporal.client;
 
 import com.google.common.base.Objects;
+import io.temporal.api.enums.v1.WorkflowIdConflictPolicy;
 import io.temporal.api.enums.v1.WorkflowIdReusePolicy;
 import io.temporal.common.CronSchedule;
 import io.temporal.common.MethodRetry;
@@ -77,6 +78,7 @@ public static WorkflowOptions merge(
         .setContextPropagators(o.getContextPropagators())
         .setDisableEagerExecution(o.isDisableEagerExecution())
         .setStartDelay(o.getStartDelay())
+        .setWorkflowIdConflictPolicy(o.getWorkflowIdConflictPolicy())
         .validateBuildWithDefaults();
   }
 
@@ -110,6 +112,8 @@ public static final class Builder {
 
     private Duration startDelay;
 
+    private WorkflowIdConflictPolicy workflowIdConflictpolicy;
+
     private Builder() {}
 
     private Builder(WorkflowOptions options) {
@@ -130,6 +134,7 @@ private Builder(WorkflowOptions options) {
       this.contextPropagators = options.contextPropagators;
       this.disableEagerExecution = options.disableEagerExecution;
       this.startDelay = options.startDelay;
+      this.workflowIdConflictpolicy = options.workflowIdConflictpolicy;
     }
 
     /**
@@ -145,7 +150,8 @@ public Builder setWorkflowId(String workflowId) {
     /**
      * Specifies server behavior if a completed workflow with the same id exists. Note that under no
      * conditions Temporal allows two workflows with the same namespace and workflow id run
-     * simultaneously.
+     * simultaneously. See {@line setWorkflowIdConflictPolicy} for handling a workflow id
+     * duplication with a <b>Running</b> workflow.
      *
      * <p>Default value if not set: <b>AllowDuplicate</b>
      *
@@ -165,6 +171,25 @@ public Builder setWorkflowIdReusePolicy(WorkflowIdReusePolicy workflowIdReusePol
       return this;
     }
 
+    /**
+     * Specifies server behavior if a <b>Running</b> workflow with the same id exists. See {@link
+     * #setWorkflowIdReusePolicy} for handling a workflow id duplication with a <b>Closed</b>
+     * workflow. Cannot be set when {@link #getWorkflowIdReusePolicy()} is {@link
+     * WorkflowIdReusePolicy#WORKFLOW_ID_REUSE_POLICY_TERMINATE_IF_RUNNING}.
+     *
+     * <ul>
+     *   <li><b>Fail</b> Don't start a new workflow; instead return {@link
+     *       WorkflowExecutionAlreadyStarted}
+     *   <li><b>UseExisting</b> Don't start a new workflow; instead return the handle for the
+     *       running workflow.
+     *   <li><b>TerminateExisting</b> Terminate the running workflow before starting a new one.
+     * </ul>
+     */
+    public Builder setWorkflowIdConflictPolicy(WorkflowIdConflictPolicy workflowIdConflictpolicy) {
+      this.workflowIdConflictpolicy = workflowIdConflictpolicy;
+      return this;
+    }
+
     /**
      * The time after which a workflow run is automatically terminated by Temporal service with
      * WORKFLOW_EXECUTION_TIMED_OUT status.
@@ -374,7 +399,8 @@ public WorkflowOptions build() {
           typedSearchAttributes,
           contextPropagators,
           disableEagerExecution,
-          startDelay);
+          startDelay,
+          workflowIdConflictpolicy);
     }
 
     /**
@@ -395,7 +421,8 @@ public WorkflowOptions validateBuildWithDefaults() {
           typedSearchAttributes,
           contextPropagators,
           disableEagerExecution,
-          startDelay);
+          startDelay,
+          workflowIdConflictpolicy);
     }
   }
 
@@ -427,6 +454,8 @@ public WorkflowOptions validateBuildWithDefaults() {
 
   private final Duration startDelay;
 
+  private final WorkflowIdConflictPolicy workflowIdConflictpolicy;
+
   private WorkflowOptions(
       String workflowId,
       WorkflowIdReusePolicy workflowIdReusePolicy,
@@ -441,7 +470,8 @@ private WorkflowOptions(
       SearchAttributes typedSearchAttributes,
       List<ContextPropagator> contextPropagators,
       boolean disableEagerExecution,
-      Duration startDelay) {
+      Duration startDelay,
+      WorkflowIdConflictPolicy workflowIdConflictpolicy) {
     this.workflowId = workflowId;
     this.workflowIdReusePolicy = workflowIdReusePolicy;
     this.workflowRunTimeout = workflowRunTimeout;
@@ -456,6 +486,7 @@ private WorkflowOptions(
     this.contextPropagators = contextPropagators;
     this.disableEagerExecution = disableEagerExecution;
     this.startDelay = startDelay;
+    this.workflowIdConflictpolicy = workflowIdConflictpolicy;
   }
 
   public String getWorkflowId() {
@@ -523,6 +554,10 @@ public boolean isDisableEagerExecution() {
     return startDelay;
   }
 
+  public WorkflowIdConflictPolicy getWorkflowIdConflictPolicy() {
+    return workflowIdConflictpolicy;
+  }
+
   public Builder toBuilder() {
     return new Builder(this);
   }
@@ -545,7 +580,8 @@ public boolean equals(Object o) {
         && Objects.equal(typedSearchAttributes, that.typedSearchAttributes)
         && Objects.equal(contextPropagators, that.contextPropagators)
         && Objects.equal(disableEagerExecution, that.disableEagerExecution)
-        && Objects.equal(startDelay, that.startDelay);
+        && Objects.equal(startDelay, that.startDelay)
+        && Objects.equal(workflowIdConflictpolicy, that.workflowIdConflictpolicy);
   }
 
   @Override
@@ -564,7 +600,8 @@ public int hashCode() {
         typedSearchAttributes,
         contextPropagators,
         disableEagerExecution,
-        startDelay);
+        startDelay,
+        workflowIdConflictpolicy);
   }
 
   @Override
@@ -601,6 +638,8 @@ public String toString() {
         + disableEagerExecution
         + ", startDelay="
         + startDelay
+        + ", workflowIdConflictpolicy="
+        + workflowIdConflictpolicy
         + '}';
   }
 }

temporal-sdk/src/main/java/io/temporal/internal/client/ScheduleProtoUtil.java

@@ -103,6 +103,10 @@ public ScheduleAction actionToProto(io.temporal.client.schedules.ScheduleAction
       if (wfOptions.getCronSchedule() != null) {
         throw new IllegalArgumentException("Cron schedule cannot be set on scheduled workflow");
       }
+      if (wfOptions.getWorkflowIdConflictPolicy() != null) {
+        throw new IllegalArgumentException(
+            "ID conflict policy cannot change from default for scheduled workflow");
+      }
       // Validate required options
       if (wfOptions.getWorkflowId() == null || wfOptions.getWorkflowId().isEmpty()) {
         throw new IllegalArgumentException("ID required on workflow action");

temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientRequestFactory.java

@@ -82,6 +82,10 @@ StartWorkflowExecutionRequest.Builder newStartWorkflowExecutionRequest(
       request.setWorkflowIdReusePolicy(options.getWorkflowIdReusePolicy());
     }
 
+    if (options.getWorkflowIdConflictPolicy() != null) {
+      request.setWorkflowIdConflictPolicy(options.getWorkflowIdConflictPolicy());
+    }
+
     String taskQueue = options.getTaskQueue();
     if (taskQueue != null && !taskQueue.isEmpty()) {
       request.setTaskQueue(TaskQueue.newBuilder().setName(taskQueue).build());
@@ -138,6 +142,7 @@ SignalWithStartWorkflowExecutionRequest.Builder newSignalWithStartWorkflowExecut
             .setWorkflowTaskTimeout(startParameters.getWorkflowTaskTimeout())
             .setWorkflowType(startParameters.getWorkflowType())
             .setWorkflowIdReusePolicy(startParameters.getWorkflowIdReusePolicy())
+            .setWorkflowIdConflictPolicy(startParameters.getWorkflowIdConflictPolicy())
             .setCronSchedule(startParameters.getCronSchedule());
 
     String workflowId = startParameters.getWorkflowId();

temporal-sdk/src/test/java/io/temporal/client/functional/WorkflowIdConflictPolicyTest.java

@@ -0,0 +1,196 @@
+/*
+ * Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
+ *
+ * Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this material 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.temporal.client.functional;
+
+import static org.junit.Assert.*;
+
+import io.temporal.api.common.v1.WorkflowExecution;
+import io.temporal.api.enums.v1.WorkflowIdConflictPolicy;
+import io.temporal.client.*;
+import io.temporal.testing.internal.SDKTestOptions;
+import io.temporal.testing.internal.SDKTestWorkflowRule;
+import io.temporal.workflow.Workflow;
+import io.temporal.workflow.shared.TestWorkflows;
+import java.util.Optional;
+import java.util.UUID;
+import org.junit.Rule;
+import org.junit.Test;
+
+public class WorkflowIdConflictPolicyTest {
+  @Rule
+  public SDKTestWorkflowRule testWorkflowRule =
+      SDKTestWorkflowRule.newBuilder().setWorkflowTypes(TestWorkflowImpl2.class).build();
+
+  @Test
+  public void policyTerminateExisting() {
+    String workflowId = UUID.randomUUID().toString();
+
+    WorkflowOptions.Builder workflowOptionsBuilder =
+        SDKTestOptions.newWorkflowOptionsWithTimeouts(testWorkflowRule.getTaskQueue()).toBuilder()
+            .setWorkflowIdConflictPolicy(
+                WorkflowIdConflictPolicy.WORKFLOW_ID_CONFLICT_POLICY_TERMINATE_EXISTING)
+            .setWorkflowId(workflowId);
+
+    TestWorkflows.TestSignaledWorkflow workflow1 =
+        testWorkflowRule
+            .getWorkflowClient()
+            .newWorkflowStub(
+                TestWorkflows.TestSignaledWorkflow.class, workflowOptionsBuilder.build());
+    WorkflowStub workflowStub1 = WorkflowStub.fromTyped(workflow1);
+    WorkflowExecution workflowExecution1 = workflowStub1.start();
+    assertNotNull(workflowStub1.getExecution());
+
+    TestWorkflows.TestSignaledWorkflow workflow2 =
+        testWorkflowRule
+            .getWorkflowClient()
+            .newWorkflowStub(
+                TestWorkflows.TestSignaledWorkflow.class, workflowOptionsBuilder.build());
+    WorkflowStub workflowStub2 = WorkflowStub.fromTyped(workflow2);
+    workflowStub2.start();
+    assertNotNull(workflowStub2.getExecution());
+
+    // WORKFLOW_ID_CONFLICT_POLICY_TERMINATE_EXISTING means that calling start with a workflow ID
+    // that already has a running workflow will terminate the existing execution.
+    assertNotEquals(workflowStub1.getExecution(), workflowStub2.getExecution());
+    workflow2.signal("test");
+    assertEquals("done", workflowStub2.getResult(String.class));
+    assertThrows(
+        WorkflowFailedException.class,
+        () ->
+            testWorkflowRule
+                .getWorkflowClient()
+                .newUntypedWorkflowStub(
+                    workflowExecution1,
+                    Optional.of(TestWorkflows.TestSignaledWorkflow.class.toString()))
+                .getResult(String.class));
+  }
+
+  @Test
+  public void policyUseExisting() {
+    String workflowId = UUID.randomUUID().toString();
+
+    WorkflowOptions.Builder workflowOptionsBuilder =
+        SDKTestOptions.newWorkflowOptionsWithTimeouts(testWorkflowRule.getTaskQueue()).toBuilder()
+            .setWorkflowIdConflictPolicy(
+                WorkflowIdConflictPolicy.WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING)
+            .setWorkflowId(workflowId);
+
+    TestWorkflows.TestSignaledWorkflow workflow1 =
+        testWorkflowRule
+            .getWorkflowClient()
+            .newWorkflowStub(
+                TestWorkflows.TestSignaledWorkflow.class, workflowOptionsBuilder.build());
+    WorkflowStub workflowStub1 = WorkflowStub.fromTyped(workflow1);
+    workflowStub1.start();
+    assertNotNull(workflowStub1.getExecution());
+
+    TestWorkflows.TestSignaledWorkflow workflow2 =
+        testWorkflowRule
+            .getWorkflowClient()
+            .newWorkflowStub(
+                TestWorkflows.TestSignaledWorkflow.class, workflowOptionsBuilder.build());
+    WorkflowStub workflowStub2 = WorkflowStub.fromTyped(workflow2);
+    workflowStub2.start();
+    assertNotNull(workflowStub1.getExecution());
+
+    // WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING means that calling start with a workflow ID
+    // that already has a running workflow will return the already running execution.
+    assertEquals(workflowStub1.getExecution(), workflowStub2.getExecution());
+    workflow2.signal("test");
+    assertEquals("done", workflowStub1.getResult(String.class));
+    assertEquals("done", workflowStub2.getResult(String.class));
+  }
+
+  @Test
+  public void policyFail() {
+    String workflowId = UUID.randomUUID().toString();
+
+    WorkflowOptions.Builder workflowOptionsBuilder =
+        SDKTestOptions.newWorkflowOptionsWithTimeouts(testWorkflowRule.getTaskQueue()).toBuilder()
+            .setWorkflowIdConflictPolicy(WorkflowIdConflictPolicy.WORKFLOW_ID_CONFLICT_POLICY_FAIL)
+            .setWorkflowId(workflowId);
+
+    TestWorkflows.TestSignaledWorkflow workflow1 =
+        testWorkflowRule
+            .getWorkflowClient()
+            .newWorkflowStub(
+                TestWorkflows.TestSignaledWorkflow.class, workflowOptionsBuilder.build());
+    WorkflowStub workflowStub1 = WorkflowStub.fromTyped(workflow1);
+    workflowStub1.start();
+    assertNotNull(workflowStub1.getExecution());
+
+    TestWorkflows.TestSignaledWorkflow workflow2 =
+        testWorkflowRule
+            .getWorkflowClient()
+            .newWorkflowStub(
+                TestWorkflows.TestSignaledWorkflow.class, workflowOptionsBuilder.build());
+    WorkflowStub workflowStub2 = WorkflowStub.fromTyped(workflow2);
+    // WORKFLOW_ID_CONFLICT_POLICY_FAIL means that calling start with a workflow ID
+    // that already has a running workflow will fail.
+    assertThrows(WorkflowExecutionAlreadyStarted.class, () -> workflowStub2.start());
+    workflow1.signal("test");
+    assertEquals("done", workflowStub1.getResult(String.class));
+  }
+
+  @Test
+  public void policyDefault() {
+    String workflowId = UUID.randomUUID().toString();
+
+    WorkflowOptions.Builder workflowOptionsBuilder =
+        SDKTestOptions.newWorkflowOptionsWithTimeouts(testWorkflowRule.getTaskQueue()).toBuilder()
+            .setWorkflowId(workflowId);
+
+    TestWorkflows.TestSignaledWorkflow workflow1 =
+        testWorkflowRule
+            .getWorkflowClient()
+            .newWorkflowStub(
+                TestWorkflows.TestSignaledWorkflow.class, workflowOptionsBuilder.build());
+    WorkflowStub workflowStub1 = WorkflowStub.fromTyped(workflow1);
+    workflowStub1.start();
+    assertNotNull(workflowStub1.getExecution());
+
+    TestWorkflows.TestSignaledWorkflow workflow2 =
+        testWorkflowRule
+            .getWorkflowClient()
+            .newWorkflowStub(
+                TestWorkflows.TestSignaledWorkflow.class, workflowOptionsBuilder.build());
+    WorkflowStub workflowStub2 = WorkflowStub.fromTyped(workflow2);
+    // Default policy is WORKFLOW_ID_CONFLICT_POLICY_FAIL
+    assertThrows(WorkflowExecutionAlreadyStarted.class, () -> workflowStub2.start());
+    workflow1.signal("test");
+    assertEquals("done", workflowStub1.getResult(String.class));
+  }
+
+  public static class TestWorkflowImpl2 implements TestWorkflows.TestSignaledWorkflow {
+    boolean done = false;
+
+    @Override
+    public String execute() {
+      Workflow.await(() -> done);
+      return "done";
+    }
+
+    @Override
+    public void signal(String arg) {
+      done = true;
+    }
+  }
+}