uranusjr commented on code in PR #70805:
URL: https://github.com/apache/airflow/pull/70805#discussion_r3732665102
##########
task-sdk/src/airflow/sdk/coordinators/_subprocess.py:
##########
@@ -414,18 +554,45 @@ def execute_task(
subprocess_logs_to_stdout: bool,
**kwargs,
) -> BaseCoordinator.ExecutionResult:
- command, subprocess_schema_version =
self._build_execute_task_command(what=what)
- process = _PopenActivitySubprocess.start(
- what=what,
- dag_rel_path=dag_rel_path,
- bundle_info=bundle_info,
- client=client,
- logger=logger,
- subprocess_logs_to_stdout=subprocess_logs_to_stdout,
- sentry_integration=sentry_integration,
- command=command,
- subprocess_schema_version=subprocess_schema_version,
- startup_timeout=self.task_startup_timeout,
- )
- exit_code = process.wait()
- return self.ExecutionResult(exit_code, process.final_state)
+ task_logger = logger or log
+ with contextlib.ExitStack() as stack:
+ stack.enter_context(self._set_current_bundle(bundle_info))
+ roots, resolved_bundle = self._init_root_source(task_logger)
+ if resolved_bundle is not None:
+ # Hold the version lock across start()/wait() so bundle cleanup
+ # cannot rmtree a version this task is still reading from,
+ # mirroring task_runner.main() for the Python task path.
+ from airflow.dag_processing.bundles.base import ( # noqa:
SDK002
+ BundleVersionLock,
+ unpack_bundle_version,
+ )
+
+ # NAMED_BUNDLE resolves "latest" with no pinned version, so
+ # resolved_bundle.version is None and the lock would be a
silent
+ # no-op. Pin the concrete current version so the lock protects
the read.
+ lock_version = resolved_bundle.version
+ if lock_version is None:
+ lock_version, _ = unpack_bundle_version(
+ resolved_bundle.get_current_version(), resolved_bundle
+ )
+ stack.enter_context(
+ BundleVersionLock(
+ bundle_name=resolved_bundle.name,
+ bundle_version=lock_version,
+ )
+ )
Review Comment:
`resolved_bundle` is built with `version=None` (line 493), so `bundle.path`
is the version-less checkout. That's what roots at line 560 points at.
Resolving the version string doesn't move `bundle.path`, so the lock guards
`versions/<sha>` while the JVM reads `tracking_repo`, so the lock is protecting
a wrong directory. Same when `TASK_BUNDLE` mode gets a run with
`bundle_version=None`.
It also leaves a tracking file for a version directory that was never
created, which sends `_remove_stale_bundle` into rmtree on a missing path.
Would re-resolving the bundle at the concrete version work, so `bundle.path`
is `versions/<sha>`, and the lock covers the tree actually passed to the
subprocess?
--
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]
