Feature: Use TaskStatusListener.onTaskInProgress() (#58)

* apply task-task-status-listener * fix test error * prevent NPE
conductor-oss · Mar 5, 2024 · 700619a · 700619a
1 parent 26af4cb
commit 700619a
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 2 deletions.
diff --git a/core/src/main/java/com/netflix/conductor/service/ExecutionService.java b/core/src/main/java/com/netflix/conductor/service/ExecutionService.java
@@ -35,6 +35,7 @@
 import com.netflix.conductor.core.exception.NotFoundException;
 import com.netflix.conductor.core.execution.WorkflowExecutor;
 import com.netflix.conductor.core.execution.tasks.SystemTaskRegistry;
+import com.netflix.conductor.core.listener.TaskStatusListener;
 import com.netflix.conductor.core.utils.QueueUtils;
 import com.netflix.conductor.core.utils.Utils;
 import com.netflix.conductor.dao.QueueDAO;
@@ -52,6 +53,7 @@ public class ExecutionService {
     private final QueueDAO queueDAO;
     private final ExternalPayloadStorage externalPayloadStorage;
     private final SystemTaskRegistry systemTaskRegistry;
+    private final TaskStatusListener taskStatusListener;
 
     private final long queueTaskMessagePostponeSecs;
 
@@ -65,7 +67,8 @@ public ExecutionService(
             QueueDAO queueDAO,
             ConductorProperties properties,
             ExternalPayloadStorage externalPayloadStorage,
-            SystemTaskRegistry systemTaskRegistry) {
+            SystemTaskRegistry systemTaskRegistry,
+            TaskStatusListener taskStatusListener) {
         this.workflowExecutor = workflowExecutor;
         this.executionDAOFacade = executionDAOFacade;
         this.queueDAO = queueDAO;
@@ -74,6 +77,7 @@ public ExecutionService(
         this.queueTaskMessagePostponeSecs =
                 properties.getTaskExecutionPostponeDuration().getSeconds();
         this.systemTaskRegistry = systemTaskRegistry;
+        this.taskStatusListener = taskStatusListener;
     }
 
     public Task poll(String taskType, String workerId) {
@@ -181,6 +185,11 @@ public List<Task> poll(
                 queueDAO.postpone(queueName, taskId, 0, queueTaskMessagePostponeSecs);
             }
         }
+        taskIds.stream()
+                .map(executionDAOFacade::getTaskModel)
+                .filter(Objects::nonNull)
+                .filter(task -> TaskModel.Status.IN_PROGRESS.equals(task.getStatus()))
+                .forEach(taskStatusListener::onTaskInProgress);
         executionDAOFacade.updateTaskLastPoll(taskType, domain, workerId);
         Monitors.recordTaskPoll(queueName);
         tasks.forEach(this::ackTaskReceived);

diff --git a/core/src/test/java/com/netflix/conductor/service/ExecutionServiceTest.java b/core/src/test/java/com/netflix/conductor/service/ExecutionServiceTest.java
@@ -34,6 +34,7 @@
 import com.netflix.conductor.core.dal.ExecutionDAOFacade;
 import com.netflix.conductor.core.execution.WorkflowExecutor;
 import com.netflix.conductor.core.execution.tasks.SystemTaskRegistry;
+import com.netflix.conductor.core.listener.TaskStatusListener;
 import com.netflix.conductor.dao.QueueDAO;
 
 import static junit.framework.TestCase.assertEquals;
@@ -48,6 +49,7 @@ public class ExecutionServiceTest {
     @Mock private ConductorProperties conductorProperties;
     @Mock private ExternalPayloadStorage externalPayloadStorage;
     @Mock private SystemTaskRegistry systemTaskRegistry;
+    @Mock private TaskStatusListener taskStatusListener;
 
     private ExecutionService executionService;
 
@@ -68,7 +70,8 @@ public void setup() {
                         queueDAO,
                         conductorProperties,
                         externalPayloadStorage,
-                        systemTaskRegistry);
+                        systemTaskRegistry,
+                        taskStatusListener);
         WorkflowDef workflowDef = new WorkflowDef();
         workflow1 = new Workflow();
         workflow1.setWorkflowId("wf1");