Sync from main

core/src/main/java/com/netflix/conductor/core/config/ConductorCoreConfiguration.java

@@ -105,15 +105,23 @@ public ExecutorService executorService(ConductorProperties conductorProperties)
     @Bean
     @Qualifier("taskMappersByTaskType")
     public Map<String, TaskMapper> getTaskMappers(List<TaskMapper> taskMappers) {
-        return taskMappers.stream().collect(Collectors.toMap(TaskMapper::getTaskType, identity()));
+        // Return mutable map so annotated task mappers can be added
+        return taskMappers.stream()
+                .collect(
+                        Collectors.toMap(
+                                TaskMapper::getTaskType,
+                                identity(),
+                                (a, b) -> a,
+                                java.util.HashMap::new));
     }
 
     @Bean
     @Qualifier(ASYNC_SYSTEM_TASKS_QUALIFIER)
     public Set<WorkflowSystemTask> asyncSystemTasks(Set<WorkflowSystemTask> allSystemTasks) {
+        // Return mutable set so annotated tasks can be added
         return allSystemTasks.stream()
                 .filter(WorkflowSystemTask::isAsync)
-                .collect(Collectors.toUnmodifiableSet());
+                .collect(Collectors.toCollection(java.util.HashSet::new));
     }
 
     @Bean

core/src/main/java/com/netflix/conductor/core/execution/DeciderService.java

@@ -75,6 +75,7 @@ public DeciderService(
             ExternalPayloadStorageUtils externalPayloadStorageUtils,
             SystemTaskRegistry systemTaskRegistry,
             @Qualifier("taskMappersByTaskType") Map<String, TaskMapper> taskMappers,
+            @Qualifier("annotatedTaskSystems") Map<String, TaskMapper> annotatedTaskSystems,
             @Value("${conductor.app.taskPendingTimeThreshold:60m}")
                     Duration taskPendingTimeThreshold) {
         this.idGenerator = idGenerator;
@@ -84,6 +85,19 @@ public DeciderService(
         this.externalPayloadStorageUtils = externalPayloadStorageUtils;
         this.taskPendingTimeThresholdMins = taskPendingTimeThreshold.toMinutes();
         this.systemTaskRegistry = systemTaskRegistry;
+        LOGGER.info("taskMappers: {}", taskMappers.keySet());
+        LOGGER.info("annotatedTaskMappers: {}", annotatedTaskSystems.keySet());
+        // Add annotated mappers only if no existing mapper is registered for that task
+        // type
+        // Existing mappers take precedence over annotated ones
+        annotatedTaskSystems.forEach(
+                (taskType, mapper) -> {
+                    if (this.taskMappers.putIfAbsent(taskType, mapper) != null) {
+                        LOGGER.info(
+                                "Skipping annotated mapper for '{}' - existing mapper already registered",
+                                taskType);
+                    }
+                });
     }
 
     public DeciderOutcome decide(WorkflowModel workflow) throws TerminateWorkflowException {
@@ -135,8 +149,10 @@ private DeciderOutcome decide(final WorkflowModel workflow, List<TaskModel> preS
         boolean hasSuccessfulTerminateTask = false;
         for (TaskModel task : workflow.getTasks()) {
 
-            // Filter the list of tasks and include only tasks that are not retried, not executed
-            // marked to be skipped and not part of System tasks that is DECISION, FORK, JOIN
+            // Filter the list of tasks and include only tasks that are not retried, not
+            // executed
+            // marked to be skipped and not part of System tasks that is DECISION, FORK,
+            // JOIN
             // This list will be empty for a new workflow being started
             if (!task.isRetried() && !task.getStatus().equals(SKIPPED) && !task.isExecuted()) {
                 pendingTasks.add(task);
@@ -186,7 +202,8 @@ private DeciderOutcome decide(final WorkflowModel workflow, List<TaskModel> preS
             if (taskDefinition.isPresent()) {
                 checkTaskTimeout(taskDefinition.get(), pendingTask);
                 checkTaskPollTimeout(taskDefinition.get(), pendingTask);
-                // If the task has not been updated for "responseTimeoutSeconds" then mark task as
+                // If the task has not been updated for "responseTimeoutSeconds" then mark task
+                // as
                 // TIMED_OUT
                 if (isResponseTimedOut(taskDefinition.get(), pendingTask)) {
                     timeoutTask(taskDefinition.get(), pendingTask);
@@ -448,7 +465,8 @@ public boolean checkForWorkflowCompletion(final WorkflowModel workflow)
                 return false;
             }
 
-            // If there is a TERMINATE task that has been executed successfuly then the workflow
+            // If there is a TERMINATE task that has been executed successfuly then the
+            // workflow
             // should be marked as completed.
             if (TERMINATE.name().equals(task.getTaskType())
                     && task.getStatus().isTerminal()
@@ -869,7 +887,7 @@ public List<TaskModel> getTasksToBeScheduled(
 
         String type = taskToSchedule.getType();
 
-        // get tasks already scheduled (in progress/terminal) for  this workflow instance
+        // get tasks already scheduled (in progress/terminal) for this workflow instance
         List<String> tasksInWorkflow =
                 workflow.getTasks().stream()
                         .filter(
@@ -892,10 +910,13 @@ public List<TaskModel> getTasksToBeScheduled(
                         .withDeciderService(this)
                         .build();
 
-        // For static forks, each branch of the fork creates a join task upon completion for
-        // dynamic forks, a join task is created with the fork and also with each branch of the
+        // For static forks, each branch of the fork creates a join task upon completion
+        // for
+        // dynamic forks, a join task is created with the fork and also with each branch
+        // of the
         // fork.
-        // A new task must only be scheduled if a task, with the same reference name is not already
+        // A new task must only be scheduled if a task, with the same reference name is
+        // not already
         // in this workflow instance
         return taskMappers
                 .getOrDefault(type, taskMappers.get(USER_DEFINED.name()))

core/src/main/java/com/netflix/conductor/core/execution/tasks/SystemTaskRegistry.java

@@ -18,13 +18,15 @@
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
+import org.springframework.context.annotation.DependsOn;
 import org.springframework.stereotype.Component;
 
 /**
  * A container class that holds a mapping of system task types {@link
  * com.netflix.conductor.common.metadata.tasks.TaskType} to {@link WorkflowSystemTask} instances.
  */
 @Component
+@DependsOn("workerTaskAnnotationScanner")
 public class SystemTaskRegistry {
 
     public static final String ASYNC_SYSTEM_TASKS_QUALIFIER = "asyncSystemTasks";

core/src/main/java/com/netflix/conductor/core/reconciliation/WorkflowSweeper.java

@@ -19,7 +19,9 @@
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.scheduling.annotation.Async;
+import org.springframework.stereotype.Component;
 
 import com.netflix.conductor.annotations.VisibleForTesting;
 import com.netflix.conductor.common.metadata.tasks.TaskDef;
@@ -39,7 +41,13 @@
 import static com.netflix.conductor.core.config.SchedulerConfiguration.SWEEPER_EXECUTOR_NAME;
 import static com.netflix.conductor.core.utils.Utils.DECIDER_QUEUE;
 
-// @Component
+// Deprecated in favor of org.conductoross.conductor.core.execution.WorkflowSweeper
+@Deprecated(forRemoval = true)
+@Component
+@ConditionalOnProperty(
+        name = "conductor.app.legacy.sweeper.enabled",
+        havingValue = "true",
+        matchIfMissing = false)
 public class WorkflowSweeper {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(WorkflowSweeper.class);

core/src/main/java/com/netflix/conductor/sdk/workflow/task/InputParam.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2021 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.sdk.workflow.task;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.PARAMETER)
+public @interface InputParam {
+    String value();
+
+    boolean required() default false;
+}

core/src/main/java/com/netflix/conductor/sdk/workflow/task/OutputParam.java

@@ -0,0 +1,24 @@
+/*
+ * Copyright 2021 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.sdk.workflow.task;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.TYPE_USE)
+public @interface OutputParam {
+    String value();
+}

core/src/main/java/com/netflix/conductor/sdk/workflow/task/WorkerTask.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2021 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.sdk.workflow.task;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/** Identifies a simple worker task. */
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.METHOD})
+public @interface WorkerTask {
+    String value();
+
+    // No. of threads to use for executing the task
+    int threadCount() default 1;
+
+    int pollingInterval() default 100;
+
+    String domain() default "";
+
+    // In millis
+    int pollTimeout() default 100;
+
+    // number of task pollers
+    // default is 1 which is good enough for most use cases
+    // a number higher than 1 will have concurrent pollers doing poll and execute
+    int pollerCount() default 1;
+}