Fix issue with timed out tasks not being reported by TaskStatusListener (#6)

Author: boma96

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

@@ -36,6 +36,7 @@
 import com.netflix.conductor.core.execution.mapper.TaskMapper;
 import com.netflix.conductor.core.execution.mapper.TaskMapperContext;
 import com.netflix.conductor.core.execution.tasks.SystemTaskRegistry;
+import com.netflix.conductor.core.listener.TaskStatusListener;
 import com.netflix.conductor.core.utils.ExternalPayloadStorageUtils;
 import com.netflix.conductor.core.utils.IDGenerator;
 import com.netflix.conductor.core.utils.ParametersUtils;
@@ -65,7 +66,7 @@ public class DeciderService {
     private final MetadataDAO metadataDAO;
     private final SystemTaskRegistry systemTaskRegistry;
     private final long taskPendingTimeThresholdMins;
-
+    private final TaskStatusListener taskStatusListener;
     private final Map<String, TaskMapper> taskMappers;
 
     public DeciderService(
@@ -76,14 +77,16 @@ public DeciderService(
             SystemTaskRegistry systemTaskRegistry,
             @Qualifier("taskMappersByTaskType") Map<String, TaskMapper> taskMappers,
             @Value("${conductor.app.taskPendingTimeThreshold:60m}")
-                    Duration taskPendingTimeThreshold) {
+                    Duration taskPendingTimeThreshold,
+            TaskStatusListener taskStatusListener) {
         this.idGenerator = idGenerator;
         this.metadataDAO = metadataDAO;
         this.parametersUtils = parametersUtils;
         this.taskMappers = taskMappers;
         this.externalPayloadStorageUtils = externalPayloadStorageUtils;
         this.taskPendingTimeThresholdMins = taskPendingTimeThreshold.toMinutes();
         this.systemTaskRegistry = systemTaskRegistry;
+        this.taskStatusListener = taskStatusListener;
     }
 
     public DeciderOutcome decide(WorkflowModel workflow) throws TerminateWorkflowException {
@@ -773,10 +776,12 @@ void timeoutTaskWithTimeoutPolicy(String reason, TaskDef taskDef, TaskModel task
             case RETRY:
                 task.setStatus(TIMED_OUT);
                 task.setReasonForIncompletion(reason);
+                taskStatusListener.onTaskTimedOut(task);
                 return;
             case TIME_OUT_WF:
                 task.setStatus(TIMED_OUT);
                 task.setReasonForIncompletion(reason);
+                taskStatusListener.onTaskTimedOut(task);
                 throw new TerminateWorkflowException(reason, WorkflowModel.Status.TIMED_OUT, task);
         }
     }
@@ -851,6 +856,7 @@ private void timeoutTask(TaskDef taskDef, TaskModel task) {
         LOGGER.debug(reason);
         task.setStatus(TIMED_OUT);
         task.setReasonForIncompletion(reason);
+        taskStatusListener.onTaskTimedOut(task);
     }
 
     public List<TaskModel> getTasksToBeScheduled(