Feature: Add WHILE Task

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;
+ }
 }

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.

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) {

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(

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>

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);
+ }
+}