Background
In coordinator mode the Java SDK reports every task failure as a terminal TaskState(FAILED) frame and then exits 0, so a TaskInstance with retries configured is recorded FAILED and never retried. The retry decision is made client-side in the SDK; the Java runtime does not implement it yet.
task-sdk/src/airflow/sdk/execution_time/task_runner.py is the source of truth for SDK behavior. _handle_current_task_failed (~L1640):
if ti._ti_context_from_server and ti._ti_context_from_server.should_retry:
return RetryTask(...), UP_FOR_RETRY
return TaskState(state=FAILED, ...), FAILED
The supervisor confirms this is load-bearing — supervisor.py final_state (~L1658): on exit 0 the runtime's frame is taken verbatim, so a TaskState(FAILED) frame becomes terminal FAILED. UP_FOR_RETRY is reachable only via a distinct RetryTask message, or a non-zero exit combined with should_retry (supervisor.py:1359, :1668).
Current Java state:
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Task.kt (~L64-71) maps task-not-found to REMOVED and every caught exception to TaskResult.of(TaskState.State.FAILED); there is no retry branch, no should_retry read, and no RetryTask message.
should_retry already travels on the wire as a field of TIRunContext (StartupDetails.ti_context); see the supervisor schema.json and _generated.py (TIRunContext.should_retry).
What needs to happen
- Read
should_retry (and max_tries) from StartupDetails.ti_context.
- Add a
RetryTask outbound message mirroring the Python wire shape.
- In
Task.kt, on task failure emit RetryTask (→ UP_FOR_RETRY) when should_retry is true, otherwise keep TaskState(FAILED).
- Add tests covering success / fail / retry.
Acceptance criteria
- A failing coordinator-mode task whose TI has retries remaining is recorded
UP_FOR_RETRY, not FAILED.
should_retry from ti_context is parsed and honored.
Context
Drafted-by: Claude Code (Opus 4.7); reviewed by @jason810496 before posting
Background
In coordinator mode the Java SDK reports every task failure as a terminal
TaskState(FAILED)frame and then exits 0, so a TaskInstance with retries configured is recordedFAILEDand never retried. The retry decision is made client-side in the SDK; the Java runtime does not implement it yet.task-sdk/src/airflow/sdk/execution_time/task_runner.pyis the source of truth for SDK behavior._handle_current_task_failed(~L1640):The supervisor confirms this is load-bearing —
supervisor.pyfinal_state(~L1658): on exit 0 the runtime's frame is taken verbatim, so aTaskState(FAILED)frame becomes terminalFAILED.UP_FOR_RETRYis reachable only via a distinctRetryTaskmessage, or a non-zero exit combined withshould_retry(supervisor.py:1359,:1668).Current Java state:
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Task.kt(~L64-71) maps task-not-found toREMOVEDand every caught exception toTaskResult.of(TaskState.State.FAILED); there is no retry branch, noshould_retryread, and noRetryTaskmessage.should_retryalready travels on the wire as a field ofTIRunContext(StartupDetails.ti_context); see the supervisorschema.jsonand_generated.py(TIRunContext.should_retry).What needs to happen
should_retry(andmax_tries) fromStartupDetails.ti_context.RetryTaskoutbound message mirroring the Python wire shape.Task.kt, on task failure emitRetryTask(→UP_FOR_RETRY) whenshould_retryis true, otherwise keepTaskState(FAILED).Acceptance criteria
UP_FOR_RETRY, notFAILED.should_retryfromti_contextis parsed and honored.Context
task-sdk/src/airflow/sdk/execution_time/task_runner.pyandsupervisor.py.RetryPolicy/RetryAction(_apply_retry_policy_or_default) and the other non-success terminals (DeferTask,RescheduleTask,SkipDownstreamTasks).Drafted-by: Claude Code (Opus 4.7); reviewed by @jason810496 before posting