Merge branch 'main' into sdk_consolidation

Author: v1r3n

conductor-client/src/main/java/com/netflix/conductor/common/metadata/tasks/Task.java

@@ -20,6 +20,7 @@
 import org.apache.commons.lang3.StringUtils;
 
 import com.netflix.conductor.common.metadata.workflow.WorkflowTask;
+import com.netflix.conductor.common.run.tasks.TypedTask;
 
 import lombok.*;
 
@@ -334,4 +335,35 @@ public boolean equals(Object o) {
     public int hashCode() {
         return Objects.hash(getTaskType(), getStatus(), getInputData(), getReferenceTaskName(), getWorkflowPriority(), getRetryCount(), getSeq(), getCorrelationId(), getPollCount(), getTaskDefName(), getScheduledTime(), getStartTime(), getEndTime(), getUpdateTime(), getStartDelayInSeconds(), getRetriedTaskId(), isRetried(), isExecuted(), isCallbackFromWorker(), getResponseTimeoutSeconds(), getWorkflowInstanceId(), getWorkflowType(), getTaskId(), getReasonForIncompletion(), getCallbackAfterSeconds(), getWorkerId(), getOutputData(), getWorkflowTask(), getDomain(), getRateLimitPerFrequency(), getRateLimitFrequencyInSeconds(), getExternalInputPayloadStoragePath(), getExternalOutputPayloadStoragePath(), getIsolationGroupId(), getExecutionNameSpace(), getParentTaskId(), getFirstStartTime());
     }
-}
+
+    /**
+     * Converts this task to a typed task wrapper.
+     *
+     * <p>Example usage:
+     * <pre>{@code
+     * WaitTask wait = task.as(WaitTask.class);
+     * if (wait.isDurationBased()) {
+     *     Duration d = wait.getDuration().orElseThrow();
+     * }
+     * }</pre>
+     *
+     * @param type the typed task class to convert to
+     * @param <T> the typed task type
+     * @return a new instance of the typed task wrapping this task
+     * @throws IllegalArgumentException if this task's type doesn't match the expected type
+     * @throws RuntimeException if the typed task class cannot be instantiated
+     */
+    public <T extends TypedTask> T as(Class<T> type) {
+        try {
+            return type.getConstructor(Task.class).newInstance(this);
+        } catch (java.lang.reflect.InvocationTargetException e) {
+            Throwable cause = e.getCause();
+            if (cause instanceof IllegalArgumentException) {
+                throw (IllegalArgumentException) cause;
+            }
+            throw new RuntimeException("Failed to create typed task: " + type.getName(), cause);
+        } catch (Exception e) {
+            throw new RuntimeException("Failed to create typed task: " + type.getName(), e);
+        }
+    }
+}

conductor-client/src/main/java/com/netflix/conductor/common/run/Workflow.java

@@ -20,6 +20,7 @@
 import com.netflix.conductor.common.metadata.Auditable;
 import com.netflix.conductor.common.metadata.tasks.Task;
 import com.netflix.conductor.common.metadata.workflow.WorkflowDef;
+import com.netflix.conductor.common.run.tasks.TypedTask;
 
 import lombok.*;
 
@@ -217,6 +218,31 @@ public Task getTaskByRefName(String refName) {
         return found.getLast();
     }
 
+    /**
+     * Returns a typed task by its reference name.
+     *
+     * <p>Example usage:
+     * <pre>{@code
+     * WaitTask wait = workflow.getTaskByRefName("myWait", WaitTask.class);
+     * if (wait.isDurationBased()) {
+     *     Duration d = wait.getDuration().orElseThrow();
+     * }
+     * }</pre>
+     *
+     * @param refName the reference task name
+     * @param type the typed task class to convert to
+     * @param <T> the typed task type
+     * @return the typed task, or null if no task with the given reference name exists
+     * @throws IllegalArgumentException if the task's type doesn't match the expected type
+     */
+    public <T extends TypedTask> T getTaskByRefName(String refName, Class<T> type) {
+        Task task = getTaskByRefName(refName);
+        if (task == null) {
+            return null;
+        }
+        return task.as(type);
+    }
+
     /**
      * @return a deep copy of the workflow instance
      */

conductor-client/src/main/java/com/netflix/conductor/common/run/tasks/DoWhileTask.java

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2024 Conductor Authors.
+ * <p>
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 com.netflix.conductor.common.run.tasks;
+
+import com.netflix.conductor.common.metadata.tasks.Task;
+import com.netflix.conductor.common.metadata.tasks.TaskType;
+
+/**
+ * Typed wrapper for DO_WHILE tasks providing convenient access to loop properties.
+ *
+ * <p>DO_WHILE tasks execute a set of tasks repeatedly until a condition is met.
+ *
+ * <p>Example usage:
+ * <pre>{@code
+ * Task task = workflow.getTaskByRefName("myLoop");
+ * DoWhileTask doWhile = new DoWhileTask(task);
+ *
+ * int iteration = doWhile.getCurrentIteration();
+ * String condition = doWhile.getLoopCondition();
+ * }</pre>
+ */
+public class DoWhileTask extends TypedTask {
+
+    public static final String LOOP_CONDITION_INPUT = "loopCondition";
+    public static final String ITERATION_OUTPUT = "iteration";
+
+    /**
+     * Creates a DoWhileTask wrapper.
+     *
+     * @param task the underlying task to wrap
+     * @throws IllegalArgumentException if task is null or not a DO_WHILE task
+     */
+    public DoWhileTask(Task task) {
+        super(task, TaskType.TASK_TYPE_DO_WHILE);
+    }
+
+    /**
+     * Checks if the given task is a DO_WHILE task.
+     */
+    public static boolean isDoWhileTask(Task task) {
+        return task != null && TaskType.TASK_TYPE_DO_WHILE.equals(task.getTaskType());
+    }
+
+    /**
+     * Returns the loop condition expression, or null if not set.
+     */
+    public String getLoopCondition() {
+        return getInputString(LOOP_CONDITION_INPUT);
+    }
+
+    /**
+     * Returns the current iteration number (1-based from the task's iteration field).
+     */
+    public int getCurrentIteration() {
+        return task.getIteration();
+    }
+
+    /**
+     * Returns the iteration count from output (may differ from current iteration), or null if not available.
+     */
+    public Integer getIterationCount() {
+        return getOutputInteger(ITERATION_OUTPUT);
+    }
+
+    /**
+     * Returns true if this is the first iteration.
+     */
+    public boolean isFirstIteration() {
+        return getCurrentIteration() <= 1;
+    }
+
+}

conductor-client/src/main/java/com/netflix/conductor/common/run/tasks/EventTask.java

@@ -0,0 +1,121 @@
+/*
+ * Copyright 2024 Conductor Authors.
+ * <p>
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 com.netflix.conductor.common.run.tasks;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import com.netflix.conductor.common.metadata.tasks.Task;
+import com.netflix.conductor.common.metadata.tasks.TaskType;
+
+/**
+ * Typed wrapper for EVENT tasks providing convenient access to event properties.
+ *
+ * <p>EVENT tasks publish messages to external event systems like SQS, Kafka, NATS, etc.
+ *
+ * <p>Example usage:
+ * <pre>{@code
+ * Task task = workflow.getTaskByRefName("myEvent");
+ * EventTask event = new EventTask(task);
+ *
+ * String sink = event.getSink();
+ * Map<String, Object> payload = event.getEventPayload();
+ * }</pre>
+ */
+public class EventTask extends TypedTask {
+
+    public static final String SINK_INPUT = "sink";
+    public static final String EVENT_PRODUCED_OUTPUT = "event_produced";
+
+    /**
+     * Creates an EventTask wrapper.
+     *
+     * @param task the underlying task to wrap
+     * @throws IllegalArgumentException if task is null or not an EVENT task
+     */
+    public EventTask(Task task) {
+        super(task, TaskType.TASK_TYPE_EVENT);
+    }
+
+    /**
+     * Checks if the given task is an EVENT task.
+     */
+    public static boolean isEventTask(Task task) {
+        return task != null && TaskType.TASK_TYPE_EVENT.equals(task.getTaskType());
+    }
+
+    /**
+     * Returns the event sink (e.g., "sqs:queue-name", "kafka:topic-name"), or null if not set.
+     */
+    public String getSink() {
+        return getInputString(SINK_INPUT);
+    }
+
+    /**
+     * Returns the sink type (e.g., "sqs", "kafka", "nats"), or null if sink is not set.
+     */
+    public String getSinkType() {
+        String sink = getSink();
+        if (sink == null) {
+            return null;
+        }
+        int colonIndex = sink.indexOf(':');
+        return colonIndex > 0 ? sink.substring(0, colonIndex) : sink;
+    }
+
+    /**
+     * Returns the sink target (e.g., queue name, topic name), or null/empty if sink is not set.
+     */
+    public String getSinkTarget() {
+        String sink = getSink();
+        if (sink == null) {
+            return null;
+        }
+        int colonIndex = sink.indexOf(':');
+        return colonIndex > 0 ? sink.substring(colonIndex + 1) : "";
+    }
+
+    /**
+     * Returns the event payload (all input data except sink).
+     */
+    public Map<String, Object> getEventPayload() {
+        Map<String, Object> inputData = task.getInputData();
+        if (inputData == null || inputData.isEmpty()) {
+            return Collections.emptyMap();
+        }
+        // The payload is typically the entire input except for the sink
+        HashMap<String, Object> payload = new HashMap<>(inputData);
+        payload.remove(SINK_INPUT);
+        return payload;
+    }
+
+    /**
+     * Returns the produced event details (available after the event is sent).
+     */
+    @SuppressWarnings("unchecked")
+    public Map<String, Object> getEventProduced() {
+        Object produced = task.getOutputData().get(EVENT_PRODUCED_OUTPUT);
+        if (produced instanceof Map) {
+            return (Map<String, Object>) produced;
+        }
+        return Collections.emptyMap();
+    }
+
+    /**
+     * Returns true if the event has been published.
+     */
+    public boolean isEventPublished() {
+        return !getEventProduced().isEmpty();
+    }
+}