Add NewWorkflowOption to config version, time, input, etc when start a new workflow

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

@@ -119,6 +119,19 @@ public <T extends Workflow> String scheduleNewWorkflow(Class<T> clazz, Object in
     return this.innerClient.scheduleNewOrchestrationInstance(clazz.getCanonicalName(), input, instanceId);
   }
 
+  /**
+   * Schedules a new workflow with a specified set of options for execution.
+   *
+   * @param <T>        any Workflow type
+   * @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.
+   */
+  public <T extends Workflow> String scheduleNewWorkflow(Class<T> clazz, NewWorkflowOption options) {
+    return this.innerClient.scheduleNewOrchestrationInstance(clazz.getCanonicalName(),
+        options.getNewOrchestrationInstanceOptions());
+  }
+
   /**
    * Terminates the workflow associated with the provided instance id.
    *

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

@@ -0,0 +1,114 @@
+/*
+ * Copyright 2023 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 com.microsoft.durabletask.NewOrchestrationInstanceOptions;
+
+import java.time.Instant;
+
+/**
+ * Options for starting a new instance of a workflow.
+ */
+public class NewWorkflowOption {
+  private final NewOrchestrationInstanceOptions newOrchestrationInstanceOptions = new NewOrchestrationInstanceOptions();
+
+  /**
+   * Sets the version of the workflow to start.
+   *
+   * @param version the user-defined version of workflow
+   * @return this {@link NewWorkflowOption} object
+   */
+  public NewWorkflowOption setVersion(String version) {
+    this.newOrchestrationInstanceOptions.setVersion(version);
+    return this;
+  }
+
+  /**
+   * Sets the instance ID of the workflow to start.
+   *
+   * <p>If no instance ID is configured, the workflow will be created with a randomly generated instance ID.
+   *
+   * @param instanceId the ID of the new workflow
+   * @return this {@link NewWorkflowOption} object
+   */
+  public NewWorkflowOption setInstanceId(String instanceId) {
+    this.newOrchestrationInstanceOptions.setInstanceId(instanceId);
+    return this;
+  }
+
+  /**
+   * Sets the input of the workflow to start.
+   *
+   * @param input the input of the new workflow
+   * @return this {@link NewWorkflowOption} object
+   */
+  public NewWorkflowOption setInput(Object input) {
+    this.newOrchestrationInstanceOptions.setInput(input);
+    return this;
+  }
+
+  /**
+   * Sets the start time of the new workflow.
+   *
+   * <p>By default, new workflow instances start executing immediately. This method can be used
+   * to start them at a specific time in the future.
+   *
+   * @param startTime the start time of the new workflow
+   * @return this {@link NewWorkflowOption} object
+   */
+  public NewWorkflowOption setStartTime(Instant startTime) {
+    this.newOrchestrationInstanceOptions.setStartTime(startTime);
+    return this;
+  }
+
+  /**
+   * Gets the user-specified version of the new workflow.
+   *
+   * @return the user-specified version of the new workflow.
+   */
+  public String getVersion() {
+    return this.newOrchestrationInstanceOptions.getVersion();
+  }
+
+  /**
+   * Gets the instance ID of the new workflow.
+   *
+   * @return the instance ID of the new workflow.
+   */
+  public String getInstanceId() {
+    return this.newOrchestrationInstanceOptions.getInstanceId();
+  }
+
+  /**
+   * Gets the input of the new workflow.
+   *
+   * @return the input of the new workflow.
+   */
+  public Object getInput() {
+    return this.newOrchestrationInstanceOptions.getInput();
+  }
+
+  /**
+   * Gets the configured start time of the new workflow instance.
+   *
+   * @return the configured start time of the new workflow instance.
+   */
+  public Instant getStartTime() {
+    return this.newOrchestrationInstanceOptions.getStartTime();
+  }
+
+  public NewOrchestrationInstanceOptions getNewOrchestrationInstanceOptions() {
+    return newOrchestrationInstanceOptions;
+  }
+}

sdk-workflows/src/test/java/io/dapr/workflows/client/DaprWorkflowClientTest.java

@@ -26,6 +26,7 @@
 
 import java.lang.reflect.Constructor;
 import java.time.Duration;
+import java.time.Instant;
 import java.util.Arrays;
 import java.util.concurrent.TimeoutException;
 
@@ -106,6 +107,19 @@ public void scheduleNewWorkflowWithArgsNameInputInstance() {
         .scheduleNewOrchestrationInstance(expectedName, expectedInput, expectedInstanceId);
   }
 
+  @Test
+  public void scheduleNewWorkflowWithNewWorkflowOption() {
+    String expectedName = TestWorkflow.class.getCanonicalName();
+    Object expectedInput = new Object();
+    NewWorkflowOption newWorkflowOption = new NewWorkflowOption();
+    newWorkflowOption.setInput(expectedInput).setStartTime(Instant.now());
+
+    client.scheduleNewWorkflow(TestWorkflow.class, newWorkflowOption);
+
+    verify(mockInnerClient, times(1))
+        .scheduleNewOrchestrationInstance(expectedName, newWorkflowOption.getNewOrchestrationInstanceOptions());
+  }
+
   @Test
   public void terminateWorkflow() {
     String expectedArgument = "TestWorkflowInstanceId";