himanshuchawla0807-gif commented on issue #75497:
URL: https://github.com/apache/airflow/issues/75497#issuecomment-6370015240
I reproduced the `start_from_trigger=True` path on current main
(`d81ba9c7f4e4521f24e5dc7b4e9bc285f0104d63`) using a real Breeze metadata
database and a serialized Dag. With `slots=1` and `include_deferred=True`,
`DagRun.update_state()` followed by `schedule_tis()` deferred all 20 tasks and
`pool.open_slots()` returned `-19`. The `start_from_trigger=False` control left
all 20 scheduled, with zero deferred and one open slot at this stage.
This is a scheduler-method reproduction, not an end-to-end
triggerer/executor run or a test of the reported 3.3.2 installation. It shows
that enabling deferred-slot accounting alone does not prevent the immediate
deferral path on this main revision. `schedule_tis()` calls `ti.defer_task()`
before the ordinary queued-task pool selection.
The issue's pasted example uses `start_from_trigger=False`, unlike the
title. Please confirm the value actually used and the pool's `slots` /
`include_deferred` settings so the release-specific case is unambiguous.
<details>
<summary>Two-case reproduction appended to
airflow-core/tests/unit/models/test_dagrun.py</summary>
```python
@pytest.mark.parametrize("start_from_trigger", [False, True])
def test_codex_start_from_trigger_pool_probe(dag_maker, session,
start_from_trigger):
from airflow.models.pool import Pool
pool = Pool(pool=f"codex_single_slot_{start_from_trigger}", slots=1,
include_deferred=True)
session.add(pool)
session.flush()
with dag_maker(session=session, serialized=True):
for index in range(20):
task = MockOperator(task_id=f"task_{index}", pool=pool.pool)
task.start_from_trigger = start_from_trigger
task.start_trigger_args = StartTriggerArgs(
trigger_cls="airflow.triggers.testing.SuccessTrigger",
next_method="execute_complete",
)
dr = dag_maker.create_dagrun(session=session)
ready, _ = dr.update_state(session=session)
assert len(ready) == 20
dr.schedule_tis(ready, session=session)
session.flush()
states = [ti.state for ti in dr.get_task_instances(session=session)]
deferred = states.count(TaskInstanceState.DEFERRED)
print(f"POOL_PROBE start_from_trigger={start_from_trigger}
deferred={deferred} open_slots={pool.open_slots(session=session)}")
assert deferred == (20 if start_from_trigger else 0)
assert pool.open_slots(session=session) == (-19 if start_from_trigger
else 1)
```
Run with `breeze run --skip-image-upgrade-check pytest
airflow-core/tests/unit/models/test_dagrun.py::test_codex_start_from_trigger_pool_probe
-q -s`. Both diagnostic cases pass, asserting the observed behavior, not the
desired fix.
</details>
---
Drafted-by: OpenAI Codex (no human review before posting)
--
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]
