Skip to content

Fix leaked triggerer thread in test_trigger_logger_fd_closed_when_removed - #70941

Merged
amoghrajesh merged 1 commit into
apache:mainfrom
amoghrajesh:fix-flaky-triggerer-test
Aug 3, 2026
Merged

Fix leaked triggerer thread in test_trigger_logger_fd_closed_when_removed#70941
amoghrajesh merged 1 commit into
apache:mainfrom
amoghrajesh:fix-flaky-triggerer-test

Conversation

@amoghrajesh

@amoghrajesh amoghrajesh commented Aug 2, 2026

Copy link
Copy Markdown
Contributor
Was generative AI tooling used to co-author this PR?
  • Yes (please specify the tool below)

Noticed in https://github.com/apache/airflow/actions/runs/30739444883

Error:

____________________________________________________________________________ test_update_triggers_delegates_workload_creation ____________________________________________________________________________
/usr/python/lib/python3.10/unittest/mock.py:940: in assert_called_once_with
    raise AssertionError(msg)
E   AssertionError: Expected 'monotonic' to be called once. Called 9 times.
E   Calls: [call(), call(), call(), call(), call(), call(), call(), call(), call()].

During handling of the above exception, another exception occurred:
airflow-core/tests/unit/jobs/test_triggerer_job.py:2487: in test_update_triggers_delegates_workload_creation
    monotonic.assert_called_once_with()
E   AssertionError: Expected 'monotonic' to be called once. Called 9 times.
E   Calls: [call(), call(), call(), call(), call(), call(), call(), call(), call()].
------------------------------------------------------------------------------------------ Captured stderr call ------------------------------------------------------------------------------------------
--- Supervised process Last chance exception handler ---
Traceback (most recent call last):
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 1266, in arun
    await self.sync_state_to_supervisor(finished_ids)
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 1551, in sync_state_to_supervisor
    resp = await self.asend(msg)
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 1564, in asend
    response = await self.comms_decoder.asend(msg)
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 1152, in asend
    return self._from_frame(await future)
asyncio.exceptions.CancelledError: Reader loop exited

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/supervisor.py", line 462, in _fork_main
    target()
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 1061, in run_in_process
    TriggerRunner().run()
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 1232, in run
    asyncio.run(self.arun())
  File "/usr/python/lib/python3.10/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/usr/python/lib/python3.10/asyncio/base_events.py", line 649, in run_until_complete
    return future.result()
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 1289, in arun
    await reader_task
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 1102, in _reader_loop
    frame = await self._aread_frame()
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 1089, in _aread_frame
    len_bytes = await self._async_reader.readexactly(4)
  File "/usr/python/lib/python3.10/asyncio/streams.py", line 708, in readexactly
    await self._wait_for_data('readexactly')
  File "/usr/python/lib/python3.10/asyncio/streams.py", line 501, in _wait_for_data
    await self._waiter
  File "/usr/python/lib/python3.10/asyncio/selector_events.py", line 862, in _read_ready__data_received
    data = self._sock.recv(self.max_size)
ConnectionResetError: [Errno 104] Connection reset by peer
--- Supervised process Last chance exception handler ---
Traceback (most recent call last):
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 1266, in arun
    await self.sync_state_to_supervisor(finished_ids)
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 1551, in sync_state_to_supervisor
    resp = await self.asend(msg)
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 1564, in asend
    response = await self.comms_decoder.asend(msg)
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 1152, in asend
    return self._from_frame(await future)
asyncio.exceptions.CancelledError: Reader loop exited

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/supervisor.py", line 462, in _fork_main
    target()
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 1061, in run_in_process
    TriggerRunner().run()
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 1232, in run
    asyncio.run(self.arun())
  File "/usr/python/lib/python3.10/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/usr/python/lib/python3.10/asyncio/base_events.py", line 649, in run_until_complete
    return future.result()
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 1289, in arun
    await reader_task
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 1102, in _reader_loop
    frame = await self._aread_frame()
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 1089, in _aread_frame
    len_bytes = await self._async_reader.readexactly(4)
  File "/usr/python/lib/python3.10/asyncio/streams.py", line 708, in readexactly
    await self._wait_for_data('readexactly')
  File "/usr/python/lib/python3.10/asyncio/streams.py", line 501, in _wait_for_data
    await self._waiter
  File "/usr/python/lib/python3.10/asyncio/selector_events.py", line 862, in _read_ready__data_received
    data = self._sock.recv(self.max_size)
ConnectionResetError: [Errno 104] Connection reset by peer
--- Supervised process Last chance exception handler ---
Traceback (most recent call last):
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 1266, in arun
    await self.sync_state_to_supervisor(finished_ids)
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 1551, in sync_state_to_supervisor
    resp = await self.asend(msg)
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 1564, in asend
    response = await self.comms_decoder.asend(msg)
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 1152, in asend
    return self._from_frame(await future)
asyncio.exceptions.CancelledError: Reader loop exited

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/supervisor.py", line 462, in _fork_main
    target()
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 1061, in run_in_process
    TriggerRunner().run()
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 1232, in run
    asyncio.run(self.arun())
  File "/usr/python/lib/python3.10/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/usr/python/lib/python3.10/asyncio/base_events.py", line 649, in run_until_complete
    return future.result()
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 1289, in arun
    await reader_task
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 1102, in _reader_loop
    frame = await self._aread_frame()
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 1089, in _aread_frame
    len_bytes = await self._async_reader.readexactly(4)
  File "/usr/python/lib/python3.10/asyncio/streams.py", line 708, in readexactly
    await self._wait_for_data('readexactly')
  File "/usr/python/lib/python3.10/asyncio/streams.py", line 501, in _wait_for_data
    await self._waiter
  File "/usr/python/lib/python3.10/asyncio/selector_events.py", line 862, in _read_ready__data_received
    data = self._sock.recv(self.max_size)
ConnectionResetError: [Errno 104] Connection reset by peer

The test started a real TriggerRunnerSupervisor subprocess but only killed it after the final assertion, so a failed or slow assertion left the subprocess running. The leaked process kept calling time.monotonic() in the background, polluting mock call counts in unrelated tests that patch the same target later in the same run.

  • Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
  • For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
  • When adding dependency, check compliance with the ASF 3rd Party License Policy.
  • For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

@amoghrajesh amoghrajesh self-assigned this Aug 2, 2026
@amoghrajesh
amoghrajesh merged commit bdc7bde into apache:main Aug 3, 2026
77 checks passed
dabla pushed a commit to dabla/airflow that referenced this pull request Aug 14, 2026
@amoghrajesh @dabla
Fix leaked triggerer thread in test_trigger_logger_fd_closed_when_rem…
95e4515 
…oved (apache#70941)
imrichardwu pushed a commit to imrichardwu/airflow that referenced this pull request Sep 11, 2026
@amoghrajesh @imrichardwu
Fix leaked triggerer thread in test_trigger_logger_fd_closed_when_rem…
7abe719 
…oved (apache#70941)
@FrankYang0529 FrankYang0529 mentioned this pull request Sep 11, 2026
Merged
1 task done
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@eladkal eladkal eladkal approved these changes
@potiuk potiuk Awaiting requested review from potiuk
@jason810496 jason810496 Awaiting requested review from jason810496
@bugraoz93 bugraoz93 Awaiting requested review from bugraoz93
@shahar1 shahar1 Awaiting requested review from shahar1
@henry3260 henry3260 Awaiting requested review from henry3260

Assignees

@amoghrajesh amoghrajesh

Labels

area:Triggerer

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

2 participants

@amoghrajesh @eladkal