Fix remote log handlers crashing triggerer logging thread on concurrent close

[ZhaoMJ]

Remote task log handlers (S3, GCS, WASB, OSS, Elasticsearch, HDFS) reclaim disk on upload by calling shutil.rmtree(os.path.dirname(local_loc)) when delete_local_copy is enabled — removing the log file's parent directory rather than the file itself.

The triggerer runs many trigger log handlers concurrently in a single process, and handler close is not atomic: the main thread's logging.shutdown() and the QueueListener monitor thread draining a trigger_end record can both call close() → upload() on the same handler. Two rmtree calls then run against the same directory — the first removes the tree, the second walks a path that just vanished and raises FileNotFoundError. Because it is raised on the logging monitor thread, it is unhandled and kills the listener thread, silently stopping all trigger log delivery for the remainder of the process's life.

Observed traceback (provider amazon 9.2.0, Airflow 2.11.0):

Exception in thread Thread-1 (_monitor):
  File ".../logging/handlers.py", line 1598, in _monitor
    self.handle(record)
  File ".../airflow/utils/log/trigger_handler.py", line 113, in close_one
    h.close()
  File ".../airflow/providers/amazon/aws/log/s3_task_handler.py", line 102, in close
    shutil.rmtree(os.path.dirname(local_loc))
  ...
FileNotFoundError: [Errno 2] No such file or directory: '.../task_id=.../map_index=0'

Fix

Delete only the uploaded file (unlink(missing_ok=True)) and then prune now-empty parent directories, stopping at the first non-empty parent and at base_log_folder. This is the same pattern the OpenSearch handler already uses (added in #64364). It:

• is idempotent under concurrent/double close (missing_ok=True + contextlib.suppress(OSError)),
• never removes a directory a concurrent sibling handler is still using (if any(parent.iterdir()): break),
• still reclaims disk (the log bytes are removed; only empty dirs are pruned).

Applied to all six affected handlers (opensearch already had it).

Tests

Added a regression test for the S3 handler asserting the delete is idempotent and that a shared parent directory with a sibling log is preserved.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Code

---

Important

🛠️ Maintainer triage note for @ZhaoMJ · by @potiuk · 2026-07-28 16:11 UTC

Helpful heads-up from the maintainers — please address before this PR can be reviewed:

• ❌ Provider tests. See docs.

The ball is in your court — you've been assigned to this PR. Fix the above, then mark it Ready for review.

See the Pull Request quality criteria for how to fix each item. There is no rush.

_{Automated triage — may be imperfect; a maintainer takes the next look. We use this two-stage triage process so maintainers' limited time goes to the conversation with you.}