conductor-oss · Young-Zen · Dec 19, 2023
diff --git a/common/src/main/java/com/netflix/conductor/common/metadata/tasks/TaskType.java b/common/src/main/java/com/netflix/conductor/common/metadata/tasks/TaskType.java
@@ -13,6 +13,7 @@
 package com.netflix.conductor.common.metadata.tasks;
 
 import java.util.HashSet;
+import java.util.Objects;
 import java.util.Set;
 
 import com.netflix.conductor.annotations.protogen.ProtoEnum;
@@ -27,6 +28,7 @@ public enum TaskType {
     SWITCH,
     JOIN,
     DO_WHILE,
+    WHILE,
     SUB_WORKFLOW,
     START_WORKFLOW,
     EVENT,
@@ -53,6 +55,7 @@ public enum TaskType {
     public static final String TASK_TYPE_DYNAMIC = "DYNAMIC";
     public static final String TASK_TYPE_JOIN = "JOIN";
     public static final String TASK_TYPE_DO_WHILE = "DO_WHILE";
+    public static final String TASK_TYPE_WHILE = "WHILE";
     public static final String TASK_TYPE_FORK_JOIN_DYNAMIC = "FORK_JOIN_DYNAMIC";
     public static final String TASK_TYPE_EVENT = "EVENT";
     public static final String TASK_TYPE_WAIT = "WAIT";
@@ -81,6 +84,7 @@ public enum TaskType {
         BUILT_IN_TASKS.add(TASK_TYPE_JOIN);
         BUILT_IN_TASKS.add(TASK_TYPE_EXCLUSIVE_JOIN);
         BUILT_IN_TASKS.add(TASK_TYPE_DO_WHILE);
+        BUILT_IN_TASKS.add(TASK_TYPE_WHILE);
     }
 
     /**
@@ -104,4 +108,13 @@ public static TaskType of(String taskType) {
     public static boolean isBuiltIn(String taskType) {
         return BUILT_IN_TASKS.contains(taskType);
     }
+
+    public static boolean isLoopTask(String taskType) {
+        return Objects.equals(TASK_TYPE_DO_WHILE, taskType)
+                || Objects.equals(TASK_TYPE_WHILE, taskType);
+    }
+
+    public static boolean isLoopTask(TaskType taskType) {
+        return taskType == DO_WHILE || taskType == WHILE;
+    }
 }
diff --git a/common/src/main/java/com/netflix/conductor/common/metadata/workflow/WorkflowDef.java b/common/src/main/java/com/netflix/conductor/common/metadata/workflow/WorkflowDef.java
@@ -341,7 +341,7 @@ public WorkflowTask getNextTask(String taskReferenceName) {
             WorkflowTask nextTask = task.next(taskReferenceName, null);
             if (nextTask != null) {
                 return nextTask;
-            } else if (TaskType.DO_WHILE.name().equals(task.getType())
+            } else if (TaskType.isLoopTask(task.getType())
                     && !task.getTaskReferenceName().equals(taskReferenceName)
                     && task.has(taskReferenceName)) {
                 // If the task is child of Loop Task and at last position, return null.

diff --git a/common/src/main/java/com/netflix/conductor/common/metadata/workflow/WorkflowTask.java b/common/src/main/java/com/netflix/conductor/common/metadata/workflow/WorkflowTask.java
@@ -560,6 +560,7 @@ private Collection<List<WorkflowTask>> children() {
                 workflowTaskLists.addAll(forkTasks);
                 break;
             case DO_WHILE:
+            case WHILE:
                 workflowTaskLists.add(loopOver);
                 break;
             default:
@@ -584,6 +585,7 @@ public WorkflowTask next(String taskReferenceName, WorkflowTask parent) {
 
         switch (taskType) {
             case DO_WHILE:
+            case WHILE:
             case DECISION:
             case SWITCH:
                 for (List<WorkflowTask> workflowTasks : children()) {
@@ -605,13 +607,12 @@ public WorkflowTask next(String taskReferenceName, WorkflowTask parent) {
                         return iterator.next();
                     }
                 }
-                if (taskType == TaskType.DO_WHILE && this.has(taskReferenceName)) {
-                    // come here means this is DO_WHILE task and `taskReferenceName` is the last
-                    // task in
-                    // this DO_WHILE task, because DO_WHILE task need to be executed to decide
-                    // whether to
-                    // schedule next iteration, so we just return the DO_WHILE task, and then ignore
-                    // generating this task again in deciderService.getNextTask()
+                if (TaskType.isLoopTask(taskType) && this.has(taskReferenceName)) {
+                    // come here means this is DO_WHILE/WHILE task and `taskReferenceName` is the
+                    // last task in this DO_WHILE/WHILE task, because DO_WHILE/WHILE task need to be
+                    // executed to decide whether to schedule next iteration, so we just return the
+                    // DO_WHILE/WHILE task, and then ignore generating this task again in
+                    // deciderService.getNextTask()
                     return this;
                 }
                 break;
@@ -663,6 +664,7 @@ public boolean has(String taskReferenceName) {
             case DECISION:
             case SWITCH:
             case DO_WHILE:
+            case WHILE:
             case FORK_JOIN:
                 for (List<WorkflowTask> childx : children()) {
                     for (WorkflowTask child : childx) {

diff --git a/core/src/main/java/com/netflix/conductor/core/execution/DeciderService.java b/core/src/main/java/com/netflix/conductor/core/execution/DeciderService.java
@@ -218,7 +218,7 @@ private DeciderOutcome decide(final WorkflowModel workflow, List<TaskModel> preS
                 pendingTask.setExecuted(true);
                 List<TaskModel> nextTasks = getNextTask(workflow, pendingTask);
                 if (pendingTask.isLoopOverTask()
-                        && !TaskType.DO_WHILE.name().equals(pendingTask.getTaskType())
+                        && !TaskType.isLoopTask(pendingTask.getTaskType())
                         && !nextTasks.isEmpty()) {
                     nextTasks = filterNextLoopOverTasks(nextTasks, pendingTask, workflow);
                 }
@@ -476,8 +476,8 @@ List<TaskModel> getNextTask(WorkflowModel workflow, TaskModel task) {
         while (isTaskSkipped(taskToSchedule, workflow)) {
             taskToSchedule = workflowDef.getNextTask(taskToSchedule.getTaskReferenceName());
         }
-        if (taskToSchedule != null && TaskType.DO_WHILE.name().equals(taskToSchedule.getType())) {
-            // check if already has this DO_WHILE task, ignore it if it already exists
+        if (taskToSchedule != null && TaskType.isLoopTask(taskToSchedule.getType())) {
+            // check if already has this DO_WHILE/WHILE task, ignore it if it already exists
             String nextTaskReferenceName = taskToSchedule.getTaskReferenceName();
             if (workflow.getTasks().stream()
                     .anyMatch(

diff --git a/core/src/main/java/com/netflix/conductor/core/execution/WorkflowExecutor.java b/core/src/main/java/com/netflix/conductor/core/execution/WorkflowExecutor.java
@@ -342,7 +342,7 @@ private void retry(WorkflowModel workflow) {
                     break;
                 case CANCELED:
                     if (task.getTaskType().equalsIgnoreCase(TaskType.JOIN.toString())
-                            || task.getTaskType().equalsIgnoreCase(TaskType.DO_WHILE.toString())) {
+                            || TaskType.isLoopTask(task.getTaskType())) {
                         task.setStatus(IN_PROGRESS);
                         addTaskToQueue(task);
                         // Task doesn't have to be updated yet. Will be updated along with other
@@ -934,7 +934,7 @@ private void extendLease(TaskResult taskResult) {
      * Determines if a workflow can be lazily evaluated, if it meets any of these criteria
      *
      * <ul>
-     *   <li>The task is NOT a loop task within DO_WHILE
+     *   <li>The task is NOT a loop task within DO_WHILE or WHILE
      *   <li>The task is one of the intermediate tasks in a branch within a FORK_JOIN
      *   <li>The task is forked from a FORK_JOIN_DYNAMIC
      * </ul>

diff --git a/core/src/main/java/com/netflix/conductor/core/execution/mapper/WhileTaskMapper.java b/core/src/main/java/com/netflix/conductor/core/execution/mapper/WhileTaskMapper.java
@@ -0,0 +1,104 @@
+/*
+ * Copyright 2023 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.core.execution.mapper;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import com.netflix.conductor.common.metadata.tasks.TaskDef;
+import com.netflix.conductor.common.metadata.tasks.TaskType;
+import com.netflix.conductor.common.metadata.workflow.WorkflowDef;
+import com.netflix.conductor.common.metadata.workflow.WorkflowTask;
+import com.netflix.conductor.core.utils.ParametersUtils;
+import com.netflix.conductor.dao.MetadataDAO;
+import com.netflix.conductor.model.TaskModel;
+import com.netflix.conductor.model.WorkflowModel;
+
+/**
+ * An implementation of {@link TaskMapper} to map a {@link WorkflowTask} of type {@link
+ * TaskType#WHILE} to a {@link TaskModel} of type {@link TaskType#WHILE}
+ */
+@Component
+public class WhileTaskMapper implements TaskMapper {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(WhileTaskMapper.class);
+
+    private final MetadataDAO metadataDAO;
+    private final ParametersUtils parametersUtils;
+
+    @Autowired
+    public WhileTaskMapper(MetadataDAO metadataDAO, ParametersUtils parametersUtils) {
+        this.metadataDAO = metadataDAO;
+        this.parametersUtils = parametersUtils;
+    }
+
+    @Override
+    public String getTaskType() {
+        return TaskType.WHILE.name();
+    }
+
+    /**
+     * This method maps {@link TaskMapper} to map a {@link WorkflowTask} of type {@link
+     * TaskType#WHILE} to a {@link TaskModel} of type {@link TaskType#WHILE} with a status of {@link
+     * TaskModel.Status#IN_PROGRESS}
+     *
+     * @param taskMapperContext: A wrapper class containing the {@link WorkflowTask}, {@link
+     *     WorkflowDef}, {@link WorkflowModel} and a string representation of the TaskId
+     * @return: A {@link TaskModel} of type {@link TaskType#WHILE} in a List
+     */
+    @Override
+    public List<TaskModel> getMappedTasks(TaskMapperContext taskMapperContext) {
+        LOGGER.debug("TaskMapperContext {} in WhileTaskMapper", taskMapperContext);
+
+        WorkflowTask workflowTask = taskMapperContext.getWorkflowTask();
+        WorkflowModel workflowModel = taskMapperContext.getWorkflowModel();
+
+        TaskModel task = workflowModel.getTaskByRefName(workflowTask.getTaskReferenceName());
+        if (task != null && task.getStatus().isTerminal()) {
+            // Since loopTask is already completed no need to schedule task again.
+            return List.of();
+        }
+
+        TaskDef taskDefinition =
+                Optional.ofNullable(taskMapperContext.getTaskDefinition())
+                        .orElseGet(
+                                () ->
+                                        Optional.ofNullable(
+                                                        metadataDAO.getTaskDef(
+                                                                workflowTask.getName()))
+                                                .orElseGet(TaskDef::new));
+
+        TaskModel whileTask = taskMapperContext.createTaskModel();
+        whileTask.setTaskType(TaskType.TASK_TYPE_WHILE);
+        whileTask.setStatus(TaskModel.Status.IN_PROGRESS);
+        whileTask.setStartTime(System.currentTimeMillis());
+        whileTask.setRateLimitPerFrequency(taskDefinition.getRateLimitPerFrequency());
+        whileTask.setRateLimitFrequencyInSeconds(taskDefinition.getRateLimitFrequencyInSeconds());
+        whileTask.setRetryCount(taskMapperContext.getRetryCount());
+
+        Map<String, Object> taskInput =
+                parametersUtils.getTaskInputV2(
+                        workflowTask.getInputParameters(),
+                        workflowModel,
+                        whileTask.getTaskId(),
+                        taskDefinition);
+        whileTask.setInputData(taskInput);
+        return List.of(whileTask);
+    }
+}