kaxil commented on code in PR #68492:
URL: https://github.com/apache/airflow/pull/68492#discussion_r3408795191
##########
task-sdk/src/airflow/sdk/definitions/dag.py:
##########
@@ -1400,8 +1400,23 @@ def test(
# triggerer may mark tasks scheduled so we read from DB
all_tis = set(dr.get_task_instances(session=session))
scheduled_tis = {x for x in all_tis if x.state ==
TaskInstanceState.SCHEDULED}
- ids_unrunnable = {x for x in all_tis if x.state not in
FINISHED_STATES} - scheduled_tis
- if not scheduled_tis and ids_unrunnable:
+ awaiting_input_tis = {x for x in all_tis if x.state ==
TaskInstanceState.AWAITING_INPUT}
+ ids_unrunnable = (
+ {x for x in all_tis if x.state not in FINISHED_STATES}
+ - scheduled_tis
+ - awaiting_input_tis
+ )
+ if not scheduled_tis and awaiting_input_tis:
+ # Human-in-the-loop tasks stay parked in AWAITING_INPUT:
dag.test() never
+ # resolves them itself. Keep the run alive until a
response recorded from
+ # outside -- the `airflow dags test` console prompt, or
the HITL REST API of
+ # a deployment sharing this metadata DB -- flips them back
to SCHEDULED.
+ log.info(
+ "Waiting for Human-in-the-loop input for tasks: %s",
+ sorted(x.task_id for x in awaiting_input_tis),
+ )
Review Comment:
Reducing anything more than that will just cause cpu churn needlessly, when
by design Approval step can last long - by design.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
