add asyncComplete support to sdk Task base class

conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/def/tasks/Task.java

@@ -39,6 +39,8 @@ public abstract class Task<T> {
 
     private int startDelay;
 
+    private boolean asyncComplete;
+
     private final TaskType type;
 
     private Map<String, Object> input = new HashMap<>();
@@ -69,6 +71,7 @@ protected Task(WorkflowTask workflowTask) {
         this.input = workflowTask.getInputParameters();
         this.description = workflowTask.getDescription();
         this.name = workflowTask.getName();
+        this.asyncComplete = Boolean.TRUE.equals(workflowTask.isAsyncComplete());
     }
 
     public T name(String name) {
@@ -81,6 +84,15 @@ public T description(String description) {
         return (T) this;
     }
 
+    public T asyncComplete(boolean asyncComplete) {
+        this.asyncComplete = asyncComplete;
+        return (T) this;
+    }
+
+    public boolean isAsyncComplete() {
+        return asyncComplete;
+    }
+
     public T input(String key, boolean value) {
         input.put(key, value);
         return (T) this;
@@ -210,6 +222,7 @@ protected final WorkflowTask toWorkflowTask() {
         workflowTask.setInputParameters(input);
         workflowTask.setStartDelay(startDelay);
         workflowTask.setOptional(optional);
+        workflowTask.setAsyncComplete(asyncComplete);
 
         // Let the sub-classes enrich the workflow task before returning back
         updateWorkflowTask(workflowTask);

conductor-client/src/test/java/com/netflix/conductor/sdk/workflow/def/TaskConversionsTests.java

@@ -460,6 +460,19 @@ public void testHttpConverter() {
         System.out.println(taskFromWorkflowTask.getInput());
     }
 
+    @Test
+    public void testHttpAsyncComplete() {
+        Http httpTask = new Http("http_ref");
+        httpTask.asyncComplete(true);
+
+        WorkflowTask workflowTask = httpTask.getWorkflowDefTasks().get(0);
+        assertTrue(workflowTask.isAsyncComplete());
+
+        Task fromWorkflowTask = TaskRegistry.getTask(workflowTask);
+        assertTrue(fromWorkflowTask instanceof Http);
+        assertTrue(((Http) fromWorkflowTask).isAsyncComplete());
+    }
+
     @Test
     public void testJQTaskConversion() {
         JQ jqTask = new JQ("task_name", "{ key3: (.key1.value1 + .key2.value2) }");