conductor-oss / conductor-oss/conductor
[BUG] TaskStatusListener.onTaskInProgress not fired for sync system tasks (HUMAN) — engine-driven transitions to IN_PROGRESS bypass the listener
- Dominant language
- Java
- Stars
- 32.2k
- Forks
- 1k
- Avg merge
- 2d 5h
- Merged PRs (30d)
- 41
Description
### Describe the bug
When a workflow contains a sync system task (`WorkflowSystemTask.isAsync() == false`) that transitions itself to `IN_PROGRESS` from `WorkflowExecutor.scheduleTask(...)` — `HUMAN` is the canonical example — the registered `TaskStatusListener` never receives an `onTaskInProgress` event. The terminal `COMPLETED` event still fires because human completion arrives via `POST /api/tasks` → `WorkflowExecutor.updateTask(TaskResult)`, which is the only place that calls `notifyTaskStatusListener(task)` today.
Where the gap is, in `core/src/main/java/com/netflix/conductor/core/execution/WorkflowExecutorOps.java` (sync branch of `scheduleTask`):
```java
for (TaskModel task : systemTasks) {
WorkflowSystemTask workflowSystemTask = systemTaskRegistry.get(task.getTaskType());
...
if (!workflowSystemTask.isAsync()) {
try {
workflowSystemTask.start(workflow, task, this); // Human.start() -> setStatus(IN_PROGRESS)
} catch (Exception e) { ... }
startedSystemTasks = true;
executionDAOFacade.updateTask(task); // persisted, but listener never notified
} else {
tasksToBeQueued.add(task); // async path - SCHEDULED fires via addTaskToQueue
}
}
```
The same shape exists for async system tasks in `AsyncSystemTaskExecutor.executeSystemTask(...)` -`systemTask.start(...)` flips the task to `IN_PROGRESS`, the `finally` block calls `executionDAOFacade.updateTask(task)`, and the listener is again never invoked. We have not independently reproduced this on `WAIT` yet, but the code path is structurally identical so the same fix applies.
**Reproduction (verified on v3.30.2)**
Wire up `task-status-listener` with the HTTP publisher pointing at any receiver:
```properties
conductor.task-status-listener.type=task_publisher
conductor.status-notifier.notification.url=http://my-listener:8080
conductor.status-notifier.notification.endpointTask=task_status
conductor.status-notifier.notification.subscribedTaskStatuses=SCHEDULED,IN_PROGRESS,COMPLETED,FAILED,CANCELED,TIMED_OUT
```
Workflow definition with a single HUMAN task:
```json
{
"name": "human_demo",
"version": 1,
"tasks": [
{ "name": "review", "taskReferenceName": "review", "type": "HUMAN" }
]
}
```
Start the workflow, wait for the human task, then complete it via:
```bash
curl -X POST http://localhost:8080/api/tasks \
-H 'Content-Type: application/json' \
-d '{"taskId":"","status":"COMPLETED","workflowInstanceId":""}'
```
Notifications observed at the receiver:
```
POST /task_status -> { "taskId": "...", "status": "COMPLETED" }
# no IN_PROGRESS event
```
Expected: `IN_PROGRESS` is also emitted at the moment the engine flips the human task from `SCHEDULED` to `IN_PROGRESS` (i.e. when the workflow scheduler runs `Human.start(...)`).
**Suggested fix**
Notify the listener after the engine persists a sync system task transition:
```diff
--- a/core/src/main/java/com/netflix/conductor/core/execution/WorkflowExecutorOps.java
+++ b/core/src/main/java/com/netflix/conductor/core/execution/WorkflowExecutorOps.java
@@ scheduleTask(...)
if (!workflowSystemTask.isAsync()) {
try {
workflowSystemTask.start(workflow, task, this);
} catch (Exception e) { ... }
startedSystemTasks = true;
executionDAOFacade.updateTask(task);
+ try {
+ notifyTaskStatusListener(task);
+ } catch (Exception e) {
+ LOGGER.error("Error notifying TaskStatusListener for sync system task {}",
+ task.getTaskId(), e);
+ }
} else {
tasksToBeQueued.add(task);
}
```
And the symmetric call in the async path:
```diff
--- a/core/src/main/java/com/netflix/conductor/core/execution/AsyncSystemTaskExecutor.java
+++ b/core/src/main/java/com/netflix/conductor/core/execution/AsyncSystemTaskExecutor.java
@@ executeSystemTask(...) finally
} finally {
executionDAOFacade.updateTask(task);
+ try {
+ // notify listener for engine-driven system-task transitions
+ // (mirrors notifyTaskStatusListener in WorkflowExecutor.updateTask)
+ } catch (Exception e) { ... }
if (shouldRemoveTaskFromQueue) { ... }
```
(Or, cleaner: fold listener notification into a single helper that wraps every `executionDAOFacade.updateTask(task)` site so engine-driven and worker-driven transitions emit consistently.)
Happy to PR — would also like guidance on whether the preferred shape is per-call-site notification or a centralized notify-on-persist helper.
Contributor guide
Research direction
Start in core/src/main/java/com/netflix/conductor/core/execution/WorkflowExecutorOps.java at the sync branch of scheduleTask, then compare it with AsyncSystemTaskExecutor.executeSystemTask(...) and WorkflowExecutor.updateTask(TaskResult). Reproduce with the HUMAN workflow and task-status-listener configuration described in the issue. Done means engine-driven IN_PROGRESS transitions notify TaskStatusListener consistently for sync and async system tasks without breaking terminal notifications.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- backend, distributed-systems
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Quiet
- Clarity
- Clearly specified
- Newbie friendliness
- 74/100