You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "kakatur (via GitHub)" <gi...@apache.org> on 2026/04/05 17:22:31 UTC
[PR] Fix batchoperator deferrable xcom links [airflow]
kakatur opened a new pull request, #64745:
URL: https://github.com/apache/airflow/pull/64745
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull request.
Write a good git commit message following this guide: http://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the documentation.
Feel free to ping (in general) for the review if you do not see reaction for a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- [ ] Yes (please specify the tool below)
<!--
Generated-by: [Tool Name] following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#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](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3041387374
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -222,6 +222,9 @@ def execute(self, context: Context) -> str | None:
if not self.job_id:
raise AirflowException("AWS Batch job - job_id was not found")
+ # Persist operator links before deferring so they're available in the UI
+ self._persist_links(context)
+
job = self.hook.get_job_description(self.job_id)
job_status = job.get("status")
if job_status == self.hook.SUCCESS_STATE:
Review Comment:
In the deferrable path this adds multiple `describe_jobs` calls in quick succession: `_persist_links()` calls `get_job_description()` and then `get_job_all_awslogs_info()` (which calls `get_job_description()` again), and `execute()` then calls `get_job_description()` again for `job_status`. This increases AWS API usage and can hit Batch API throttling. Consider refactoring to reuse a single job description (e.g., have `_persist_links` accept/return `job_desc`) and/or avoid fetching CloudWatch log info before deferring (it is often unavailable immediately).
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -403,6 +395,50 @@ def monitor_job(self, context: Context):
aws_partition=self.hook.conn_partition,
**awslogs[0],
)
+ else:
+ # Persist placeholder to prevent "XCom not found" warnings
+ # CloudWatch logs will be updated when job completes
+ context["task_instance"].xcom_push(
+ key="cloudwatch_events",
+ value=None,
+ )
+
+ def monitor_job(self, context: Context):
+ """
+ Monitor an AWS Batch job.
+
+ This can raise an exception or an AirflowTaskTimeout if the task was
+ created with ``execution_timeout``.
+ """
+ if not self.job_id:
+ raise AirflowException("AWS Batch job - job_id was not found")
+
+ # Persist operator links
+ self._persist_links(context)
+
+ if self.awslogs_enabled:
+ if self.waiters:
+ self.waiters.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
+ else:
+ self.hook.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
+ else:
+ if self.waiters:
+ self.waiters.wait_for_job(self.job_id)
+ else:
+ self.hook.wait_for_job(self.job_id)
+
+ # Log all CloudWatch log stream links for user reference
+ try:
+ awslogs = self.hook.get_job_all_awslogs_info(self.job_id)
+ if awslogs:
+ self.log.info(
+ "AWS Batch job (%s) CloudWatch Events details found. Links to logs:", self.job_id
+ )
+ link_builder = CloudWatchEventsLink()
+ for log in awslogs:
+ self.log.info(link_builder.format_link(**log))
+ except AirflowException as ae:
+ self.log.warning("Cannot determine where to find the AWS logs for this Batch job: %s", ae)
Review Comment:
`monitor_job()` calls `_persist_links()` (which already calls `get_job_all_awslogs_info()`), then after waiting it calls `get_job_all_awslogs_info()` again just to log stream links. This adds extra `describe_jobs` calls and will break existing expectations in unit tests that `get_job_description` is called exactly twice during `execute()` (see `test_execute_without_failures`). Consider fetching/logging/persisting CloudWatch log info only once (e.g., do it after waiting, like the previous implementation, or have `_persist_links` optionally skip CloudWatch work when called pre-wait).
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -257,6 +260,9 @@ def execute_complete(self, context: Context, event: dict[str, Any] | None = None
self.job_id = validated_event["job_id"]
+ # Persist operator links for UI
+ self._persist_links(context)
Review Comment:
`execute_complete()` now unconditionally calls `_persist_links(context)`, which performs AWS API calls and assumes the context contains a task instance. In this repo's unit tests `execute_complete` is invoked with an empty dict (`context={}`), and `_persist_links` can raise (e.g., when it hits the CloudWatch placeholder `context["task_instance"].xcom_push`). Either guard `_persist_links` when required context keys are missing, or update the tests/mocks to pass a real TI context and stub the hook calls.
```suggestion
# Persist operator links for UI when task instance context is available.
if context and context.get("task_instance") is not None:
self._persist_links(context)
```
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -403,6 +395,50 @@ def monitor_job(self, context: Context):
aws_partition=self.hook.conn_partition,
**awslogs[0],
)
+ else:
+ # Persist placeholder to prevent "XCom not found" warnings
+ # CloudWatch logs will be updated when job completes
+ context["task_instance"].xcom_push(
+ key="cloudwatch_events",
+ value=None,
+ )
Review Comment:
This manually pushes an XCom placeholder even when `do_xcom_push=False`, uses a hard-coded key string, and directly indexes `context["task_instance"]` (KeyError if not present). It also isn’t needed for `BaseAwsLink.get_link()` since missing/falsey XCom values already render as no link. Prefer removing the placeholder entirely, or at least respect `operator.do_xcom_push` and use `CloudWatchEventsLink.key` plus the same `context["ti"]` convention used by `BaseAwsLink.persist()`.
```suggestion
```
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -368,33 +375,18 @@ def monitor_job(self, context: Context):
job_queue_arn=job_queue_arn,
)
- if self.awslogs_enabled:
- if self.waiters:
- self.waiters.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
- else:
- self.hook.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
- else:
- if self.waiters:
- self.waiters.wait_for_job(self.job_id)
- else:
- self.hook.wait_for_job(self.job_id)
-
+ # Persist CloudWatch logs link if available
awslogs = []
try:
awslogs = self.hook.get_job_all_awslogs_info(self.job_id)
except AirflowException as ae:
- self.log.warning("Cannot determine where to find the AWS logs for this Batch job: %s", ae)
+ # CloudWatch logs may not be available immediately after job submission
+ self.log.info("CloudWatch logs not yet available for Batch job: %s", ae)
Review Comment:
The `except AirflowException` branch assumes the failure to fetch log info is transient (“not yet available”) and logs at INFO. `get_job_all_awslogs_info()` can raise `AirflowException` for non-transient cases too (e.g. unsupported job type), so this message/level is misleading and can hide real configuration issues. Consider logging a more neutral message (or warning) and differentiating transient “stream not created yet” (which returns `[]`) from actual exceptions.
```suggestion
self.log.warning(
"Unable to retrieve CloudWatch log information for AWS Batch job (%s): %s",
self.job_id,
ae,
)
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3067498056
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -257,6 +260,9 @@ def execute_complete(self, context: Context, event: dict[str, Any] | None = None
self.job_id = validated_event["job_id"]
+ # Persist operator links for UI
+ self._persist_links(context)
+
# Fetch logs if awslogs_enabled
if self.awslogs_enabled:
self.monitor_job(context) # fetch logs, no need to return
Review Comment:
I need to remove the duplicate _persist_links() call.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on PR #64745:
URL: https://github.com/apache/airflow/pull/64745#issuecomment-4230709483
I've fixed all issues raised by Copilot :)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3068815730
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -404,6 +418,33 @@ def monitor_job(self, context: Context):
**awslogs[0],
)
+ def monitor_job(self, context: Context):
+ """
+ Monitor an AWS Batch job.
+
+ This can raise an exception or an AirflowTaskTimeout if the task was
+ created with ``execution_timeout``.
+ """
+ if not self.job_id:
+ raise AirflowException("AWS Batch job - job_id was not found")
+
+ # Persist job definition and queue links
+ self._persist_links(context)
+
+ if self.awslogs_enabled:
+ if self.waiters:
+ self.waiters.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
+ else:
+ self.hook.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
+ else:
+ if self.waiters:
+ self.waiters.wait_for_job(self.job_id)
+ else:
+ self.hook.wait_for_job(self.job_id)
+
+ # After job completes, persist CloudWatch logs
+ self._persist_cloudwatch_link(context)
Review Comment:
Only persists CloudWatch link when awslogs_enabled=True
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3263425331
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -252,14 +258,17 @@ def execute(self, context: Context) -> str | None:
def execute_complete(self, context: Context, event: dict[str, Any] | None = None) -> str:
validated_event = validate_execute_complete_event(event)
+ # Set job_id first so CloudWatch link can be persisted even on failure
+ self.job_id = validated_event["job_id"]
+
+ # Persist CloudWatch logs for both success and failure
+ self._persist_cloudwatch_link(context)
Review Comment:
I’m closing this comment as resolved without making a fix. Please feel free to reopen it or add another comment if you believe this fix is necessary.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3463529628
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -368,33 +398,34 @@ def monitor_job(self, context: Context):
job_queue_arn=job_queue_arn,
)
- if self.awslogs_enabled:
- if self.waiters:
- self.waiters.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
- else:
- self.hook.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
- else:
- if self.waiters:
- self.waiters.wait_for_job(self.job_id)
- else:
- self.hook.wait_for_job(self.job_id)
+ return job_desc
+
+ def _persist_cloudwatch_link(self, context: Context) -> None:
+ """
+ Persist CloudWatch logs link if available.
+
+ :param context: Task context
+ """
+ if not context or "ti" not in context:
+ return
+
+ if not self.job_id:
+ return
Review Comment:
I've added early return when do_xcom_push=False to skip the get_job_all_awslogs_info() API call since BaseAwsLink.persist() would no-op anyway.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3066475109
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -257,6 +260,9 @@ def execute_complete(self, context: Context, event: dict[str, Any] | None = None
self.job_id = validated_event["job_id"]
+ # Persist operator links for UI
+ self._persist_links(context)
+
# Fetch logs if awslogs_enabled
if self.awslogs_enabled:
self.monitor_job(context) # fetch logs, no need to return
Review Comment:
`execute_complete()` persists links and then (when `awslogs_enabled`) calls `monitor_job()`, which immediately calls `_persist_links()` again. This duplicates XCom writes and AWS lookups. Consider removing one of the two calls (e.g., let `monitor_job()` handle persistence for the `awslogs_enabled` branch, or add a parameter/flag to `monitor_job()` to avoid re-persisting when already done).
```suggestion
# Fetch logs if awslogs_enabled. monitor_job() persists operator links,
# so avoid duplicating that work here.
if self.awslogs_enabled:
self.monitor_job(context) # fetch logs, no need to return
else:
# Persist operator links for UI when monitor_job() is not called.
self._persist_links(context)
```
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -403,6 +395,50 @@ def monitor_job(self, context: Context):
aws_partition=self.hook.conn_partition,
**awslogs[0],
)
+ else:
+ # Persist placeholder to prevent "XCom not found" warnings
+ # CloudWatch logs will be updated when job completes
+ context["task_instance"].xcom_push(
+ key="cloudwatch_events",
+ value=None,
+ )
+
+ def monitor_job(self, context: Context):
+ """
+ Monitor an AWS Batch job.
+
+ This can raise an exception or an AirflowTaskTimeout if the task was
+ created with ``execution_timeout``.
+ """
+ if not self.job_id:
+ raise AirflowException("AWS Batch job - job_id was not found")
+
+ # Persist operator links
+ self._persist_links(context)
+
+ if self.awslogs_enabled:
+ if self.waiters:
+ self.waiters.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
+ else:
+ self.hook.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
+ else:
+ if self.waiters:
+ self.waiters.wait_for_job(self.job_id)
+ else:
+ self.hook.wait_for_job(self.job_id)
Review Comment:
In the non-deferrable path, `_persist_links()` now runs *before* waiting for the job. If CloudWatch logs are not available yet, `_persist_links()` pushes a `None` placeholder and the operator link may never be updated to a real CloudWatch link (because persistence is not rerun after `wait_for_job`). Consider persisting job-definition/job-queue links before waiting, then persisting the CloudWatch log link after `wait_for_job` (or call `_persist_links()` again after waiting but avoid redoing the non-log work).
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on PR #64745:
URL: https://github.com/apache/airflow/pull/64745#issuecomment-4228061804
I've implemented the feedback from Guthub copilot.
1. Reduced API calls - 67% reduction (3→1) before deferring
2. Context validation - Handles empty context in unit tests
3. Removed XCom placeholder - Respects do_xcom_push setting
4. Eliminated duplication in monitor_job - CloudWatch fetched once
5. Better error logging - WARNING level for CloudWatch failures
6. No duplication in execute_complete - Conditional logic based on awslogs_enabled
7. Bug fix - Added missing check_job_success() when awslogs_enabled=False
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3068198610
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -330,18 +337,38 @@ def submit_job(self, context: Context):
job_id=self.job_id,
)
- def monitor_job(self, context: Context):
+ def _persist_links(
+ self, context: Context, skip_cloudwatch: bool = False, job_description: dict | None = None
+ ) -> dict:
"""
- Monitor an AWS Batch job.
+ Persist operator links for UI display.
- This can raise an exception or an AirflowTaskTimeout if the task was
- created with ``execution_timeout``.
+ This method retrieves job details and persists the operator links
+ (job definition, job queue, CloudWatch logs) as XCom values so they
+ can be rendered in the Airflow UI.
+
+ :param context: Task context
+ :param skip_cloudwatch: If True, skip fetching CloudWatch logs (useful before deferring)
+ :param job_description: Optional pre-fetched job description to avoid redundant API calls
+ :return: Job description dict
"""
if not self.job_id:
raise AirflowException("AWS Batch job - job_id was not found")
+ # Fetch job description (needed for return value and link persistence)
+ try:
+ job_desc = job_description or self.hook.get_job_description(self.job_id)
+ except KeyError:
+ self.log.warning("AWS Batch job (%s) description not available", self.job_id)
+ return {}
+
+ # Check if we can persist links (requires task_instance in context)
+ task_instance = context.get("task_instance") if context else None
+ if not task_instance:
+ self.log.debug("Skipping link persistence: task_instance not available in context")
+ return job_desc
+
Review Comment:
Removed the context guard entirely (lines 360-364).
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3068197538
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -330,18 +337,38 @@ def submit_job(self, context: Context):
job_id=self.job_id,
)
- def monitor_job(self, context: Context):
+ def _persist_links(
+ self, context: Context, skip_cloudwatch: bool = False, job_description: dict | None = None
+ ) -> dict:
"""
- Monitor an AWS Batch job.
+ Persist operator links for UI display.
- This can raise an exception or an AirflowTaskTimeout if the task was
- created with ``execution_timeout``.
+ This method retrieves job details and persists the operator links
+ (job definition, job queue, CloudWatch logs) as XCom values so they
+ can be rendered in the Airflow UI.
+
+ :param context: Task context
+ :param skip_cloudwatch: If True, skip fetching CloudWatch logs (useful before deferring)
+ :param job_description: Optional pre-fetched job description to avoid redundant API calls
+ :return: Job description dict
"""
if not self.job_id:
raise AirflowException("AWS Batch job - job_id was not found")
+ # Fetch job description (needed for return value and link persistence)
+ try:
+ job_desc = job_description or self.hook.get_job_description(self.job_id)
+ except KeyError:
+ self.log.warning("AWS Batch job (%s) description not available", self.job_id)
+ return {}
Review Comment:
Removed dead except KeyError block (lines 359-362).
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3293393517
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -330,18 +339,25 @@ def submit_job(self, context: Context):
job_id=self.job_id,
)
- def monitor_job(self, context: Context):
+ def _persist_links(
+ self,
+ context: Context,
+ job_description: dict | None = None,
+ ) -> dict:
"""
- Monitor an AWS Batch job.
+ Persist job definition and queue links for UI display.
- This can raise an exception or an AirflowTaskTimeout if the task was
- created with ``execution_timeout``.
+ :param context: Task context
+ :param job_description: Optional pre-fetched job description to avoid redundant API calls
+ :return: Job description dict
"""
if not self.job_id:
- raise AirflowException("AWS Batch job - job_id was not found")
+ raise ValueError("AWS Batch job - job_id was not found")
Review Comment:
Changed _persist_links() to raise AirflowException instead of ValueError for missing job_id
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3289450942
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -330,18 +339,25 @@ def submit_job(self, context: Context):
job_id=self.job_id,
)
- def monitor_job(self, context: Context):
+ def _persist_links(
+ self,
+ context: Context,
+ job_description: dict | None = None,
+ ) -> dict:
"""
- Monitor an AWS Batch job.
+ Persist job definition and queue links for UI display.
- This can raise an exception or an AirflowTaskTimeout if the task was
- created with ``execution_timeout``.
+ :param context: Task context
+ :param job_description: Optional pre-fetched job description to avoid redundant API calls
+ :return: Job description dict
"""
if not self.job_id:
- raise AirflowException("AWS Batch job - job_id was not found")
+ raise ValueError("AWS Batch job - job_id was not found")
Review Comment:
`_persist_links()` now raises `ValueError` for a missing `job_id`, while the rest of the operator (and Airflow operators generally) use `AirflowException` for runtime task failures. This change can alter error classification/handling and is inconsistent with nearby code (e.g., `execute()` already raises `AirflowException` for the same condition). Prefer raising `AirflowException` here for consistency and clearer operator failure semantics.
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -252,14 +258,17 @@ def execute(self, context: Context) -> str | None:
def execute_complete(self, context: Context, event: dict[str, Any] | None = None) -> str:
validated_event = validate_execute_complete_event(event)
+ # Set job_id first so CloudWatch link can be persisted even on failure
+ self.job_id = validated_event["job_id"]
+
+ # Persist CloudWatch logs for both success and failure
+ self._persist_cloudwatch_link(context)
+
if validated_event["status"] != "success":
raise AirflowException(f"Error while running job: {validated_event}")
- self.job_id = validated_event["job_id"]
-
- # Fetch logs if awslogs_enabled
- if self.awslogs_enabled:
- self.monitor_job(context) # fetch logs, no need to return
+ # Check job success (already know status is "success" from above)
+ self.hook.check_job_success(self.job_id)
self.log.info("Job completed successfully for job_id: %s", self.job_id)
return self.job_id
Review Comment:
`awslogs_enabled` is no longer honored in deferrable mode: `execute_complete()` used to call `monitor_job()` to fetch/emit CloudWatch logs when `awslogs_enabled=True`, but now it only persists a UI link and checks success. This is a behavior regression relative to the parameter’s documented purpose (“logs … should be printed or not”). Consider restoring log fetching in `execute_complete()` when `awslogs_enabled` is enabled (or update docs/parameter behavior if intentionally changed).
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -404,6 +424,33 @@ def monitor_job(self, context: Context):
**awslogs[0],
)
+ def monitor_job(self, context: Context):
+ """
+ Monitor an AWS Batch job.
+
+ This can raise an exception or an AirflowTaskTimeout if the task was
+ created with ``execution_timeout``.
+ """
+ if not self.job_id:
+ raise ValueError("AWS Batch job - job_id was not found")
Review Comment:
`monitor_job()` is a public method and previously raised `AirflowException` for missing `job_id`; changing it to `ValueError` is a breaking behavioral change for any callers expecting an `AirflowException` (including consistent task-failure messaging). Recommend keeping `AirflowException` (or another Airflow-specific exception type) for backward compatibility.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #64745:
URL: https://github.com/apache/airflow/pull/64745#issuecomment-4476940112
@kakatur — There are 1 unresolved review thread on this PR from @ferruzzi. Could you either push a fix or reply in each thread explaining why the feedback doesn't apply? Once you believe the feedback is addressed, mark the thread as resolved so the reviewer isn't re-pinged needlessly. Thanks!
---
_Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this [two-stage triage process](https://github.com/apache/airflow/blob/main/contributing-docs/25_maintainer_pr_triage.md#why-the-first-pass-is-automated) so that our maintainers' limited time is spent where it matters most: the conversation with you._
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on PR #64745:
URL: https://github.com/apache/airflow/pull/64745#issuecomment-4203127099
Converting this to draft since there is unaddressed copilot feedback and failing tests. change this to ready for review once all feedback is addressed and the tests are green :slightly_smiling_face:
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3067498571
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -403,6 +395,50 @@ def monitor_job(self, context: Context):
aws_partition=self.hook.conn_partition,
**awslogs[0],
)
+ else:
+ # Persist placeholder to prevent "XCom not found" warnings
+ # CloudWatch logs will be updated when job completes
+ context["task_instance"].xcom_push(
+ key="cloudwatch_events",
+ value=None,
+ )
+
+ def monitor_job(self, context: Context):
+ """
+ Monitor an AWS Batch job.
+
+ This can raise an exception or an AirflowTaskTimeout if the task was
+ created with ``execution_timeout``.
+ """
+ if not self.job_id:
+ raise AirflowException("AWS Batch job - job_id was not found")
+
+ # Persist operator links
+ self._persist_links(context)
+
+ if self.awslogs_enabled:
+ if self.waiters:
+ self.waiters.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
+ else:
+ self.hook.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
+ else:
+ if self.waiters:
+ self.waiters.wait_for_job(self.job_id)
+ else:
+ self.hook.wait_for_job(self.job_id)
Review Comment:
This was aleady fixed in response to the previous feedback.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3067566426
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -330,18 +337,38 @@ def submit_job(self, context: Context):
job_id=self.job_id,
)
- def monitor_job(self, context: Context):
+ def _persist_links(
+ self, context: Context, skip_cloudwatch: bool = False, job_description: dict | None = None
+ ) -> dict:
"""
- Monitor an AWS Batch job.
+ Persist operator links for UI display.
- This can raise an exception or an AirflowTaskTimeout if the task was
- created with ``execution_timeout``.
+ This method retrieves job details and persists the operator links
+ (job definition, job queue, CloudWatch logs) as XCom values so they
+ can be rendered in the Airflow UI.
+
+ :param context: Task context
+ :param skip_cloudwatch: If True, skip fetching CloudWatch logs (useful before deferring)
+ :param job_description: Optional pre-fetched job description to avoid redundant API calls
+ :return: Job description dict
"""
if not self.job_id:
raise AirflowException("AWS Batch job - job_id was not found")
+ # Fetch job description (needed for return value and link persistence)
+ try:
+ job_desc = job_description or self.hook.get_job_description(self.job_id)
+ except KeyError:
+ self.log.warning("AWS Batch job (%s) description not available", self.job_id)
+ return {}
Review Comment:
`BatchClientHook.get_job_description()` raises `AirflowException` on failure (after retries), not `KeyError`, so this `except KeyError` will never run and the warning/empty-dict fallback is effectively dead code. Either catch `AirflowException` here (if you really want to proceed without a description) or remove the try/except and let the exception propagate.
```suggestion
job_desc = job_description or self.hook.get_job_description(self.job_id)
```
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -330,18 +337,38 @@ def submit_job(self, context: Context):
job_id=self.job_id,
)
- def monitor_job(self, context: Context):
+ def _persist_links(
+ self, context: Context, skip_cloudwatch: bool = False, job_description: dict | None = None
+ ) -> dict:
"""
- Monitor an AWS Batch job.
+ Persist operator links for UI display.
- This can raise an exception or an AirflowTaskTimeout if the task was
- created with ``execution_timeout``.
+ This method retrieves job details and persists the operator links
+ (job definition, job queue, CloudWatch logs) as XCom values so they
+ can be rendered in the Airflow UI.
+
+ :param context: Task context
+ :param skip_cloudwatch: If True, skip fetching CloudWatch logs (useful before deferring)
+ :param job_description: Optional pre-fetched job description to avoid redundant API calls
+ :return: Job description dict
"""
if not self.job_id:
raise AirflowException("AWS Batch job - job_id was not found")
+ # Fetch job description (needed for return value and link persistence)
+ try:
+ job_desc = job_description or self.hook.get_job_description(self.job_id)
Review Comment:
`get_job_description()` is called with a positional argument here, but the surrounding codebase/tests treat this as a keyworded call (`job_id=...`). Using the keyword avoids brittle mock assertions and is consistent with the hook signature, especially since unit tests assert `get_job_description(job_id=JOB_ID)`.
```suggestion
job_desc = job_description or self.hook.get_job_description(job_id=self.job_id)
```
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -330,18 +337,38 @@ def submit_job(self, context: Context):
job_id=self.job_id,
)
- def monitor_job(self, context: Context):
+ def _persist_links(
+ self, context: Context, skip_cloudwatch: bool = False, job_description: dict | None = None
+ ) -> dict:
"""
- Monitor an AWS Batch job.
+ Persist operator links for UI display.
- This can raise an exception or an AirflowTaskTimeout if the task was
- created with ``execution_timeout``.
+ This method retrieves job details and persists the operator links
+ (job definition, job queue, CloudWatch logs) as XCom values so they
+ can be rendered in the Airflow UI.
+
+ :param context: Task context
+ :param skip_cloudwatch: If True, skip fetching CloudWatch logs (useful before deferring)
+ :param job_description: Optional pre-fetched job description to avoid redundant API calls
+ :return: Job description dict
"""
if not self.job_id:
raise AirflowException("AWS Batch job - job_id was not found")
+ # Fetch job description (needed for return value and link persistence)
+ try:
+ job_desc = job_description or self.hook.get_job_description(self.job_id)
+ except KeyError:
+ self.log.warning("AWS Batch job (%s) description not available", self.job_id)
+ return {}
+
+ # Check if we can persist links (requires task_instance in context)
+ task_instance = context.get("task_instance") if context else None
+ if not task_instance:
+ self.log.debug("Skipping link persistence: task_instance not available in context")
+ return job_desc
+
Review Comment:
The link-persistence guard is checking `context.get("task_instance")`, but AWS operator links in this codebase persist via `context["ti"].xcom_push` (see `BaseAwsLink.persist`). This means `_persist_links()` can incorrectly skip persistence when the context only contains `ti`, and it also changes behavior for callers that pass `context=None` (existing unit test `test_monitor_job_with_logs` calls `monitor_job(context=None)` and expects the link `persist()` methods to be invoked). Consider removing this guard entirely and letting `BaseAwsLink.persist` no-op when context/ti is missing, or at least key off `context.get("ti")` instead of `task_instance`.
```suggestion
```
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -257,9 +259,14 @@ def execute_complete(self, context: Context, event: dict[str, Any] | None = None
self.job_id = validated_event["job_id"]
- # Fetch logs if awslogs_enabled
+ # Fetch logs and persist links if awslogs_enabled, otherwise just persist links
if self.awslogs_enabled:
- self.monitor_job(context) # fetch logs, no need to return
+ # monitor_job() handles link persistence, log fetching, and success check
+ self.monitor_job(context)
+ else:
+ # Persist operator links and check job success
+ self._persist_links(context)
+ self.hook.check_job_success(self.job_id)
Review Comment:
In `execute_complete()` when `awslogs_enabled` is False, calling `_persist_links(context)` introduces new AWS API calls (e.g. `describe_jobs`, and potentially CloudWatch log parsing) during completion even though the links were already persisted before deferral in `execute()` (and there’s no log link to fetch in this branch). This also breaks existing unit tests that call `execute_complete(context={...})` without stubbing the hook. Consider skipping `_persist_links()` entirely in this branch (just `check_job_success`), or at least guard it behind `context.get("ti")`/`do_xcom_push` and pass `skip_cloudwatch=True` to avoid extra log-related calls/warnings.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on PR #64745:
URL: https://github.com/apache/airflow/pull/64745#issuecomment-4271526633
Could you also please add some context to the description so we know what's being fixed, how we can tell it is fixed, etc
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3300694871
##########
providers/amazon/tests/unit/amazon/aws/operators/test_batch.py:
##########
@@ -149,16 +149,24 @@ def test_init_defaults(self):
def test_template_fields_overrides(self):
validate_template_fields(self.batch)
+ @mock.patch.object(BatchClientHook, "get_job_all_awslogs_info")
@mock.patch.object(BatchClientHook, "get_job_description")
@mock.patch.object(BatchClientHook, "wait_for_job")
@mock.patch.object(BatchClientHook, "check_job_success")
- def test_execute_without_failures(self, check_mock, wait_mock, job_description_mock):
+ def test_execute_without_failures(
+ self, check_mock, wait_mock, job_description_mock, get_job_all_awslogs_info_mock
+ ):
# JOB_ID is in RESPONSE_WITHOUT_FAILURES
self.client_mock.submit_job.return_value = RESPONSE_WITHOUT_FAILURES
self.batch.job_id = None
self.batch.waiters = None # use default wait
+ get_job_all_awslogs_info_mock.return_value = []
+ # Enable xcom push so _persist_cloudwatch_link actually runs
+ self.batch.do_xcom_push = True
+ # Use a real dict so "ti" in context works correctly
Review Comment:
Remove do_xcom_push guard from CloudWatch link persistence to match other links
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3300705921
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -330,18 +339,25 @@ def submit_job(self, context: Context):
job_id=self.job_id,
)
- def monitor_job(self, context: Context):
+ def _persist_links(
+ self,
+ context: Context,
+ job_description: dict | None = None,
+ ) -> dict:
Review Comment:
Changed _persist_links signature from bare dict to dict[str, Any] for better static type checking and consistency with the rest of the codebase.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3068421599
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -404,6 +418,33 @@ def monitor_job(self, context: Context):
**awslogs[0],
)
+ def monitor_job(self, context: Context):
+ """
+ Monitor an AWS Batch job.
+
+ This can raise an exception or an AirflowTaskTimeout if the task was
+ created with ``execution_timeout``.
+ """
+ if not self.job_id:
+ raise AirflowException("AWS Batch job - job_id was not found")
+
+ # Persist job definition and queue links
+ self._persist_links(context)
+
+ if self.awslogs_enabled:
+ if self.waiters:
+ self.waiters.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
+ else:
+ self.hook.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
+ else:
+ if self.waiters:
+ self.waiters.wait_for_job(self.job_id)
+ else:
+ self.hook.wait_for_job(self.job_id)
+
+ # After job completes, persist CloudWatch logs
+ self._persist_cloudwatch_link(context)
Review Comment:
`monitor_job()` now persists the CloudWatch link unconditionally after waiting. This means even when `awslogs_enabled=False` the operator will still call `get_job_all_awslogs_info()` (extra `describe_jobs` call + potential warnings from the hook about missing streams / non-aws log drivers). If `awslogs_enabled` is intended to disable CloudWatch log integration, consider gating `_persist_cloudwatch_link()` behind that flag (or introduce a separate flag for UI link persistence) to avoid unexpected API usage/noisy logs.
```suggestion
# After job completes, persist CloudWatch logs when log integration is enabled.
if self.awslogs_enabled:
self._persist_cloudwatch_link(context)
```
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -368,33 +384,31 @@ def monitor_job(self, context: Context):
job_queue_arn=job_queue_arn,
)
- if self.awslogs_enabled:
- if self.waiters:
- self.waiters.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
- else:
- self.hook.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
- else:
- if self.waiters:
- self.waiters.wait_for_job(self.job_id)
- else:
- self.hook.wait_for_job(self.job_id)
+ return job_desc
+
+ def _persist_cloudwatch_link(self, context: Context) -> None:
+ """
+ Persist CloudWatch logs link if available.
+
+ :param context: Task context
+ """
+ if not self.do_xcom_push:
+ return
Review Comment:
`_persist_cloudwatch_link()` makes AWS API calls (`get_job_all_awslogs_info`) even when the task context can’t persist operator links (e.g. `context` is `{}` or missing `"ti"`). Since `BaseAwsLink.persist()` requires `context["ti"]` (and will no-op via `return_on_error`), consider an early return when `not context` or `"ti" not in context` to avoid unnecessary Batch `describe_jobs` calls and to keep unit tests that call `execute_complete(context={...})` from needing to stub AWS hook methods.
```suggestion
if not context or "ti" not in context:
return
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas merged PR #64745:
URL: https://github.com/apache/airflow/pull/64745
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3463449661
##########
providers/amazon/tests/unit/amazon/aws/operators/test_batch.py:
##########
@@ -545,46 +551,66 @@ def test_monitor_job_with_logs(
@patch.object(BatchOperator, "log", new_callable=MagicMock)
@patch("airflow.providers.amazon.aws.operators.batch.validate_execute_complete_event")
- @patch.object(BatchOperator, "monitor_job")
- def test_execute_complete_success_with_logs(self, mock_monitor_job, mock_validate, mock_log):
+ @patch.object(BatchClientHook, "check_job_success")
+ @patch.object(BatchClientHook, "get_job_all_awslogs_info")
+ @mock.patch("airflow.providers.amazon.aws.hooks.batch_client.AwsBaseHook.get_client_type")
+ def test_execute_complete_success_with_logs(
+ self, mock_client, mock_get_job_all_awslogs_info, mock_check_job_success, mock_validate, mock_log
+ ):
# Setup
mock_validate.return_value = {"status": "success", "job_id": "12345"}
+ mock_get_job_all_awslogs_info.return_value = [
+ {"awslogs_group": "/aws/batch/job", "awslogs_stream_name": "stream1"}
+ ]
batch = BatchOperator(
task_id="test_task",
job_name=JOB_NAME,
job_queue="dummy_queue",
job_definition="dummy_definition",
deferrable=True,
awslogs_enabled=True,
+ do_xcom_push=True, # Enable xcom push so _persist_cloudwatch_link runs
)
+ # Add task instance to context
+ context = {"ti": mock.MagicMock()}
- result = batch.execute_complete(context={}, event={"dummy": "event"})
+ result = batch.execute_complete(context=context, event={"dummy": "event"})
# Assertion
assert result == "12345"
- mock_monitor_job.assert_called_once_with({})
+ mock_get_job_all_awslogs_info.assert_called_once_with("12345")
+ mock_check_job_success.assert_called_once_with("12345")
mock_log.info.assert_called_with("Job completed successfully for job_id: %s", "12345")
@patch.object(BatchOperator, "log", new_callable=MagicMock)
@patch("airflow.providers.amazon.aws.operators.batch.validate_execute_complete_event")
- @patch.object(BatchOperator, "monitor_job")
- def test_execute_complete_success_without_logs(self, mock_monitor_job, mock_validate, mock_log):
+ @patch.object(BatchClientHook, "check_job_success")
+ @patch.object(BatchClientHook, "get_job_all_awslogs_info")
+ @mock.patch("airflow.providers.amazon.aws.hooks.batch_client.AwsBaseHook.get_client_type")
+ def test_execute_complete_success_without_logs(
Review Comment:
@o-nikolas I've added test_execute_complete_failure_without_logs to verify CloudWatch links are still persisted on failure even when awslogs_enabled=False — all 23 tests pass.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3067496690
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -403,6 +395,50 @@ def monitor_job(self, context: Context):
aws_partition=self.hook.conn_partition,
**awslogs[0],
)
+ else:
+ # Persist placeholder to prevent "XCom not found" warnings
+ # CloudWatch logs will be updated when job completes
+ context["task_instance"].xcom_push(
+ key="cloudwatch_events",
+ value=None,
+ )
+
+ def monitor_job(self, context: Context):
+ """
+ Monitor an AWS Batch job.
+
+ This can raise an exception or an AirflowTaskTimeout if the task was
+ created with ``execution_timeout``.
+ """
+ if not self.job_id:
+ raise AirflowException("AWS Batch job - job_id was not found")
+
+ # Persist operator links
+ self._persist_links(context)
+
+ if self.awslogs_enabled:
+ if self.waiters:
+ self.waiters.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
+ else:
+ self.hook.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
+ else:
+ if self.waiters:
+ self.waiters.wait_for_job(self.job_id)
+ else:
+ self.hook.wait_for_job(self.job_id)
+
+ # Log all CloudWatch log stream links for user reference
+ try:
+ awslogs = self.hook.get_job_all_awslogs_info(self.job_id)
+ if awslogs:
+ self.log.info(
+ "AWS Batch job (%s) CloudWatch Events details found. Links to logs:", self.job_id
+ )
+ link_builder = CloudWatchEventsLink()
+ for log in awslogs:
+ self.log.info(link_builder.format_link(**log))
+ except AirflowException as ae:
+ self.log.warning("Cannot determine where to find the AWS logs for this Batch job: %s", ae)
Review Comment:
I will remove the redundant get_job_all_awslogs_info() call
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3067497154
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -368,33 +375,18 @@ def monitor_job(self, context: Context):
job_queue_arn=job_queue_arn,
)
- if self.awslogs_enabled:
- if self.waiters:
- self.waiters.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
- else:
- self.hook.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
- else:
- if self.waiters:
- self.waiters.wait_for_job(self.job_id)
- else:
- self.hook.wait_for_job(self.job_id)
-
+ # Persist CloudWatch logs link if available
awslogs = []
try:
awslogs = self.hook.get_job_all_awslogs_info(self.job_id)
except AirflowException as ae:
- self.log.warning("Cannot determine where to find the AWS logs for this Batch job: %s", ae)
+ # CloudWatch logs may not be available immediately after job submission
+ self.log.info("CloudWatch logs not yet available for Batch job: %s", ae)
Review Comment:
I will change this to WARNING level with neutral message (lines 402-406).
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3068393683
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -404,6 +416,52 @@ def monitor_job(self, context: Context):
**awslogs[0],
)
+ def monitor_job(self, context: Context):
+ """
+ Monitor an AWS Batch job.
+
+ This can raise an exception or an AirflowTaskTimeout if the task was
+ created with ``execution_timeout``.
+ """
+ if not self.job_id:
+ raise AirflowException("AWS Batch job - job_id was not found")
+
+ # Persist job definition and queue links
+ self._persist_links(context)
+
+ if self.awslogs_enabled:
+ if self.waiters:
+ self.waiters.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
+ else:
+ self.hook.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
+ else:
+ if self.waiters:
+ self.waiters.wait_for_job(self.job_id)
+ else:
+ self.hook.wait_for_job(self.job_id)
+
+ # After job completes, fetch and persist CloudWatch logs once
+ try:
+ awslogs = self.hook.get_job_all_awslogs_info(self.job_id)
+ if awslogs:
+ self.log.info(
+ "AWS Batch job (%s) CloudWatch Events details found. Links to logs:", self.job_id
+ )
+ link_builder = CloudWatchEventsLink()
+ for log in awslogs:
+ self.log.info(link_builder.format_link(**log))
+
+ # Persist the first log stream as the UI link
+ CloudWatchEventsLink.persist(
+ context=context,
+ operator=self,
+ region_name=self.hook.conn_region_name,
+ aws_partition=self.hook.conn_partition,
+ **awslogs[0],
+ )
+ except AirflowException as ae:
+ self.log.warning("Cannot determine where to find the AWS logs for this Batch job: %s", ae)
+
Review Comment:
Replaced duplicate CloudWatch logic in monitor_job() with call to _persist_cloudwatch_link(context) (line 444).
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3293394316
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -404,6 +424,33 @@ def monitor_job(self, context: Context):
**awslogs[0],
)
+ def monitor_job(self, context: Context):
+ """
+ Monitor an AWS Batch job.
+
+ This can raise an exception or an AirflowTaskTimeout if the task was
+ created with ``execution_timeout``.
+ """
+ if not self.job_id:
+ raise ValueError("AWS Batch job - job_id was not found")
Review Comment:
Changed monitor_job() to raise AirflowException instead of ValueError for missing job_id
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3455583616
##########
providers/amazon/tests/unit/amazon/aws/operators/test_batch.py:
##########
@@ -545,46 +551,66 @@ def test_monitor_job_with_logs(
@patch.object(BatchOperator, "log", new_callable=MagicMock)
@patch("airflow.providers.amazon.aws.operators.batch.validate_execute_complete_event")
- @patch.object(BatchOperator, "monitor_job")
- def test_execute_complete_success_with_logs(self, mock_monitor_job, mock_validate, mock_log):
+ @patch.object(BatchClientHook, "check_job_success")
+ @patch.object(BatchClientHook, "get_job_all_awslogs_info")
+ @mock.patch("airflow.providers.amazon.aws.hooks.batch_client.AwsBaseHook.get_client_type")
+ def test_execute_complete_success_with_logs(
+ self, mock_client, mock_get_job_all_awslogs_info, mock_check_job_success, mock_validate, mock_log
+ ):
# Setup
mock_validate.return_value = {"status": "success", "job_id": "12345"}
+ mock_get_job_all_awslogs_info.return_value = [
+ {"awslogs_group": "/aws/batch/job", "awslogs_stream_name": "stream1"}
+ ]
batch = BatchOperator(
task_id="test_task",
job_name=JOB_NAME,
job_queue="dummy_queue",
job_definition="dummy_definition",
deferrable=True,
awslogs_enabled=True,
+ do_xcom_push=True, # Enable xcom push so _persist_cloudwatch_link runs
)
+ # Add task instance to context
+ context = {"ti": mock.MagicMock()}
- result = batch.execute_complete(context={}, event={"dummy": "event"})
+ result = batch.execute_complete(context=context, event={"dummy": "event"})
# Assertion
assert result == "12345"
- mock_monitor_job.assert_called_once_with({})
+ mock_get_job_all_awslogs_info.assert_called_once_with("12345")
+ mock_check_job_success.assert_called_once_with("12345")
mock_log.info.assert_called_with("Job completed successfully for job_id: %s", "12345")
@patch.object(BatchOperator, "log", new_callable=MagicMock)
@patch("airflow.providers.amazon.aws.operators.batch.validate_execute_complete_event")
- @patch.object(BatchOperator, "monitor_job")
- def test_execute_complete_success_without_logs(self, mock_monitor_job, mock_validate, mock_log):
+ @patch.object(BatchClientHook, "check_job_success")
+ @patch.object(BatchClientHook, "get_job_all_awslogs_info")
+ @mock.patch("airflow.providers.amazon.aws.hooks.batch_client.AwsBaseHook.get_client_type")
+ def test_execute_complete_success_without_logs(
Review Comment:
We're missing testing of the failure case no? To ensure the log links are still persisted despite failures?
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -252,14 +258,17 @@ def execute(self, context: Context) -> str | None:
def execute_complete(self, context: Context, event: dict[str, Any] | None = None) -> str:
validated_event = validate_execute_complete_event(event)
+ # Set job_id first so CloudWatch link can be persisted even on failure
+ self.job_id = validated_event["job_id"]
+
+ # Persist CloudWatch logs for both success and failure
+ self._persist_cloudwatch_link(context)
+
if validated_event["status"] != "success":
raise AirflowException(f"Error while running job: {validated_event}")
- self.job_id = validated_event["job_id"]
-
- # Fetch logs if awslogs_enabled
- if self.awslogs_enabled:
- self.monitor_job(context) # fetch logs, no need to return
Review Comment:
It looks like this just got dropped in the new changes? How are logs pulled down now?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3300377889
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -330,18 +339,25 @@ def submit_job(self, context: Context):
job_id=self.job_id,
)
- def monitor_job(self, context: Context):
+ def _persist_links(
+ self,
+ context: Context,
+ job_description: dict | None = None,
+ ) -> dict:
Review Comment:
The new helper is typed as returning a bare `dict` and accepts a bare `dict | None`. In this codebase the surrounding methods already use more specific typing (e.g., `dict[str, Any]`). Consider updating this signature to `job_description: Mapping[str, Any] | None` (or `dict[str, Any] | None`) and returning `dict[str, Any]` to improve static checking and make the expected structure clearer.
##########
providers/amazon/tests/unit/amazon/aws/operators/test_batch.py:
##########
@@ -149,16 +149,24 @@ def test_init_defaults(self):
def test_template_fields_overrides(self):
validate_template_fields(self.batch)
+ @mock.patch.object(BatchClientHook, "get_job_all_awslogs_info")
@mock.patch.object(BatchClientHook, "get_job_description")
@mock.patch.object(BatchClientHook, "wait_for_job")
@mock.patch.object(BatchClientHook, "check_job_success")
- def test_execute_without_failures(self, check_mock, wait_mock, job_description_mock):
+ def test_execute_without_failures(
+ self, check_mock, wait_mock, job_description_mock, get_job_all_awslogs_info_mock
+ ):
# JOB_ID is in RESPONSE_WITHOUT_FAILURES
self.client_mock.submit_job.return_value = RESPONSE_WITHOUT_FAILURES
self.batch.job_id = None
self.batch.waiters = None # use default wait
+ get_job_all_awslogs_info_mock.return_value = []
+ # Enable xcom push so _persist_cloudwatch_link actually runs
+ self.batch.do_xcom_push = True
+ # Use a real dict so "ti" in context works correctly
Review Comment:
These tests now depend on setting `do_xcom_push=True` to exercise CloudWatch-link persistence. If link persistence is meant to be independent of return-value XCom behavior (see comment on `_persist_cloudwatch_link`), the tests should avoid coupling to `do_xcom_push` and instead only provide a context with `ti`. Updating the tests accordingly will make them better reflect the intended UI-link behavior.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3068302982
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -404,6 +416,52 @@ def monitor_job(self, context: Context):
**awslogs[0],
)
+ def monitor_job(self, context: Context):
+ """
+ Monitor an AWS Batch job.
+
+ This can raise an exception or an AirflowTaskTimeout if the task was
+ created with ``execution_timeout``.
+ """
+ if not self.job_id:
+ raise AirflowException("AWS Batch job - job_id was not found")
+
+ # Persist job definition and queue links
+ self._persist_links(context)
+
+ if self.awslogs_enabled:
+ if self.waiters:
+ self.waiters.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
+ else:
+ self.hook.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
+ else:
+ if self.waiters:
+ self.waiters.wait_for_job(self.job_id)
+ else:
+ self.hook.wait_for_job(self.job_id)
+
+ # After job completes, fetch and persist CloudWatch logs once
+ try:
+ awslogs = self.hook.get_job_all_awslogs_info(self.job_id)
+ if awslogs:
+ self.log.info(
+ "AWS Batch job (%s) CloudWatch Events details found. Links to logs:", self.job_id
+ )
+ link_builder = CloudWatchEventsLink()
+ for log in awslogs:
+ self.log.info(link_builder.format_link(**log))
+
+ # Persist the first log stream as the UI link
+ CloudWatchEventsLink.persist(
+ context=context,
+ operator=self,
+ region_name=self.hook.conn_region_name,
+ aws_partition=self.hook.conn_partition,
+ **awslogs[0],
+ )
+ except AirflowException as ae:
+ self.log.warning("Cannot determine where to find the AWS logs for this Batch job: %s", ae)
+
Review Comment:
`monitor_job()` re-implements CloudWatch link persistence and uses a different warning message than `_persist_cloudwatch_link()`, so the two paths can diverge over time. Since `_persist_cloudwatch_link()` already encapsulates the persistence + logging behavior, consider reusing it here (and keeping `monitor_job()` focused on waiting + final success check).
```suggestion
self._persist_cloudwatch_link(context)
```
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -257,9 +263,13 @@ def execute_complete(self, context: Context, event: dict[str, Any] | None = None
self.job_id = validated_event["job_id"]
- # Fetch logs if awslogs_enabled
+ # Fetch logs if awslogs_enabled (links were already persisted before deferring)
if self.awslogs_enabled:
- self.monitor_job(context) # fetch logs, no need to return
+ # monitor_job() handles log fetching and success check
+ self.monitor_job(context)
+ else:
+ # Links already persisted before deferring, just check job success
+ self.hook.check_job_success(self.job_id)
Review Comment:
In deferrable mode, `execute_complete()` is invoked only after `BatchJobTrigger` has already waited for job completion. Calling `monitor_job()` here re-runs `wait_for_job()` (including potential delay/jitter + extra describe calls), which can unnecessarily slow the task and increase Batch API usage. Consider splitting `monitor_job()` into (a) a wait/stream-logs part for non-deferrable mode and (b) a post-completion part for deferrable mode (e.g., persist links / fetch CloudWatch link and then `check_job_success`).
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -257,9 +263,13 @@ def execute_complete(self, context: Context, event: dict[str, Any] | None = None
self.job_id = validated_event["job_id"]
- # Fetch logs if awslogs_enabled
+ # Fetch logs if awslogs_enabled (links were already persisted before deferring)
if self.awslogs_enabled:
- self.monitor_job(context) # fetch logs, no need to return
+ # monitor_job() handles log fetching and success check
+ self.monitor_job(context)
+ else:
+ # Links already persisted before deferring, just check job success
Review Comment:
`execute_complete()` now skips `monitor_job()` entirely when `awslogs_enabled=False`, which means in deferrable mode the CloudWatch operator link is never persisted after job completion (unlike the non-deferrable path where `monitor_job()` persists it after waiting). If the intent is that the UI link should be available regardless of whether logs are printed, consider persisting the CloudWatch link in this branch too (ideally guarded by `do_xcom_push`), or make the behavior consistent across deferrable/non-deferrable modes.
```suggestion
# Persist the CloudWatch link after deferred completion too, even when logs
# are not being fetched, to keep operator link behavior consistent.
if self.do_xcom_push:
self._persist_cloudwatch_link(context)
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3068197108
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -257,9 +259,14 @@ def execute_complete(self, context: Context, event: dict[str, Any] | None = None
self.job_id = validated_event["job_id"]
- # Fetch logs if awslogs_enabled
+ # Fetch logs and persist links if awslogs_enabled, otherwise just persist links
if self.awslogs_enabled:
- self.monitor_job(context) # fetch logs, no need to return
+ # monitor_job() handles link persistence, log fetching, and success check
+ self.monitor_job(context)
+ else:
+ # Persist operator links and check job success
+ self._persist_links(context)
+ self.hook.check_job_success(self.job_id)
Review Comment:
Removed redundant _persist_links() call in execute_complete() when awslogs_enabled=False (line 268).
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3068415985
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -257,9 +263,9 @@ def execute_complete(self, context: Context, event: dict[str, Any] | None = None
self.job_id = validated_event["job_id"]
Review Comment:
Set self.job_id before the status check (line 262)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on PR #64745:
URL: https://github.com/apache/airflow/pull/64745#issuecomment-4241227726
I see many reviews by copilot and also some comments from @kakatur saying some are addressed. @kakatur can you please resolve each individual copilot thread that you have resolved?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3253047899
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -252,14 +258,17 @@ def execute(self, context: Context) -> str | None:
def execute_complete(self, context: Context, event: dict[str, Any] | None = None) -> str:
validated_event = validate_execute_complete_event(event)
+ # Set job_id first so CloudWatch link can be persisted even on failure
+ self.job_id = validated_event["job_id"]
+
+ # Persist CloudWatch logs for both success and failure
+ self._persist_cloudwatch_link(context)
Review Comment:
@ferruzzi The method already returns early if `do_xcom_push`, `context`, or `job_id` are missing, which prevents unnecessary API calls in those cases.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3230319917
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -330,18 +339,25 @@ def submit_job(self, context: Context):
job_id=self.job_id,
)
- def monitor_job(self, context: Context):
+ def _persist_links(
+ self,
+ context: Context,
+ job_description: dict | None = None,
+ ) -> dict:
"""
- Monitor an AWS Batch job.
+ Persist job definition and queue links for UI display.
- This can raise an exception or an AirflowTaskTimeout if the task was
- created with ``execution_timeout``.
+ :param context: Task context
+ :param job_description: Optional pre-fetched job description to avoid redundant API calls
+ :return: Job description dict
"""
if not self.job_id:
raise AirflowException("AWS Batch job - job_id was not found")
Review Comment:
I know this is code you didn't touch, but I agree with you using ValueError in your new code and this feels like it should also be a ValueError. IIRC AirflowException is more for runtime issues, not "we didn't pass a value" errors.
Feel free to just resolve this if you don't want to handle it. Just a thought.
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -252,14 +258,17 @@ def execute(self, context: Context) -> str | None:
def execute_complete(self, context: Context, event: dict[str, Any] | None = None) -> str:
validated_event = validate_execute_complete_event(event)
+ # Set job_id first so CloudWatch link can be persisted even on failure
+ self.job_id = validated_event["job_id"]
+
+ # Persist CloudWatch logs for both success and failure
+ self._persist_cloudwatch_link(context)
Review Comment:
Consider wrapping this in an "if cloudwatch..." check to save the API call?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on PR #64745:
URL: https://github.com/apache/airflow/pull/64745#issuecomment-4271915595
> Could you also please add some context to the description so we know what's being fixed, how we can tell it is fixed, etc
**Fixes based on Copilot Comments:**
1. Reduced API calls - 67% reduction (3→1) before deferring by removing redundant wait_for_job() call in execute_complete() (commit 0a6cbadd5c)
2. Context validation - Added check for empty context (if not context or "ti" not in context) to skip CloudWatch API calls when context lacks task instance (commit 0058f4dc17)
3. Removed XCom placeholder - Added if not self.do_xcom_push: return to respect do_xcom_push setting instead of always attempting to push CloudWatch links (commit 3b78fc97d4)
4. Eliminated duplication in monitor_job - Extracted _persist_cloudwatch_link() helper method so CloudWatch info is fetched once instead of in multiple places (commit 3f17702130)
5. Bug fix - Reordered code in execute_complete() to set job_id before persisting CloudWatch link, ensuring links work on both success and failure paths (commit 3b78fc97d4)
6. Conditional logic based on awslogs_enabled - Initially gated CloudWatch links with if self.awslogs_enabled, then reverted to preserve UI links in all cases (commits 0058f4dc17, 4dc6b9d805)
**Fixes based on Test Failures:**
1. test_execute_without_failures - Updated to mock get_job_all_awslogs_info() instead of expecting multiple get_job_description() calls, enabled xcom_push, added proper dict context (commit e9b8d80dbf)
2. test_execute_complete_success_with_logs - Added proper mocking of hook methods (get_job_all_awslogs_info, check_job_success) and enabled xcom_push with proper context (commit e9b8d80dbf)
3. test_execute_complete_success_without_logs - Same mocking updates to handle CloudWatch link persistence in all scenarios (commit e9b8d80dbf)
4. mypy type check / prek hook (raise AirflowException) - Replaced AirflowException with ValueError for missing job_id, added type guard for job_id in _persist_cloudwatch_link() (commit 9b210efb34)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3068863969
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -404,6 +421,34 @@ def monitor_job(self, context: Context):
**awslogs[0],
)
+ def monitor_job(self, context: Context):
+ """
+ Monitor an AWS Batch job.
+
+ This can raise an exception or an AirflowTaskTimeout if the task was
+ created with ``execution_timeout``.
+ """
+ if not self.job_id:
+ raise AirflowException("AWS Batch job - job_id was not found")
+
+ # Persist job definition and queue links
+ self._persist_links(context)
+
+ if self.awslogs_enabled:
+ if self.waiters:
+ self.waiters.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
+ else:
+ self.hook.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
+ else:
+ if self.waiters:
+ self.waiters.wait_for_job(self.job_id)
+ else:
+ self.hook.wait_for_job(self.job_id)
+
+ # After job completes, persist CloudWatch logs when log integration is enabled
+ if self.awslogs_enabled:
+ self._persist_cloudwatch_link(context)
Review Comment:
Instead of writing the tests, restored the original behavior without impacting the goals of this PR.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3068199419
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -330,18 +337,38 @@ def submit_job(self, context: Context):
job_id=self.job_id,
)
- def monitor_job(self, context: Context):
+ def _persist_links(
+ self, context: Context, skip_cloudwatch: bool = False, job_description: dict | None = None
+ ) -> dict:
"""
- Monitor an AWS Batch job.
+ Persist operator links for UI display.
- This can raise an exception or an AirflowTaskTimeout if the task was
- created with ``execution_timeout``.
+ This method retrieves job details and persists the operator links
+ (job definition, job queue, CloudWatch logs) as XCom values so they
+ can be rendered in the Airflow UI.
+
+ :param context: Task context
+ :param skip_cloudwatch: If True, skip fetching CloudWatch logs (useful before deferring)
+ :param job_description: Optional pre-fetched job description to avoid redundant API calls
+ :return: Job description dict
"""
if not self.job_id:
raise AirflowException("AWS Batch job - job_id was not found")
+ # Fetch job description (needed for return value and link persistence)
+ try:
+ job_desc = job_description or self.hook.get_job_description(self.job_id)
Review Comment:
Changed to keyword argument job_id=self.job_id (line 358) for consistency
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3067338376
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -222,6 +222,9 @@ def execute(self, context: Context) -> str | None:
if not self.job_id:
raise AirflowException("AWS Batch job - job_id was not found")
+ # Persist operator links before deferring so they're available in the UI
+ self._persist_links(context)
+
job = self.hook.get_job_description(self.job_id)
job_status = job.get("status")
if job_status == self.hook.SUCCESS_STATE:
Review Comment:
I will consolidate these 3 calls into one.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3293396221
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -252,14 +258,17 @@ def execute(self, context: Context) -> str | None:
def execute_complete(self, context: Context, event: dict[str, Any] | None = None) -> str:
validated_event = validate_execute_complete_event(event)
+ # Set job_id first so CloudWatch link can be persisted even on failure
+ self.job_id = validated_event["job_id"]
+
+ # Persist CloudWatch logs for both success and failure
+ self._persist_cloudwatch_link(context)
+
if validated_event["status"] != "success":
raise AirflowException(f"Error while running job: {validated_event}")
- self.job_id = validated_event["job_id"]
-
- # Fetch logs if awslogs_enabled
- if self.awslogs_enabled:
- self.monitor_job(context) # fetch logs, no need to return
+ # Check job success (already know status is "success" from above)
+ self.hook.check_job_success(self.job_id)
self.log.info("Job completed successfully for job_id: %s", self.job_id)
return self.job_id
Review Comment:
Restore CloudWatch log fetching in deferrable mode when awslogs_enabled=True
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3068396130
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -257,9 +263,13 @@ def execute_complete(self, context: Context, event: dict[str, Any] | None = None
self.job_id = validated_event["job_id"]
- # Fetch logs if awslogs_enabled
+ # Fetch logs if awslogs_enabled (links were already persisted before deferring)
if self.awslogs_enabled:
- self.monitor_job(context) # fetch logs, no need to return
+ # monitor_job() handles log fetching and success check
+ self.monitor_job(context)
+ else:
+ # Links already persisted before deferring, just check job success
+ self.hook.check_job_success(self.job_id)
Review Comment:
Eliminated redundant monitor_job() call in execute_complete() to avoid re-running wait_for_job()
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3068394205
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -257,9 +263,13 @@ def execute_complete(self, context: Context, event: dict[str, Any] | None = None
self.job_id = validated_event["job_id"]
- # Fetch logs if awslogs_enabled
+ # Fetch logs if awslogs_enabled (links were already persisted before deferring)
if self.awslogs_enabled:
- self.monitor_job(context) # fetch logs, no need to return
+ # monitor_job() handles log fetching and success check
+ self.monitor_job(context)
+ else:
+ # Links already persisted before deferring, just check job success
Review Comment:
Added _persist_cloudwatch_link(context) in execute_complete() else branch (line 272)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3068402865
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -368,33 +381,28 @@ def monitor_job(self, context: Context):
job_queue_arn=job_queue_arn,
)
- if self.awslogs_enabled:
- if self.waiters:
- self.waiters.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
- else:
- self.hook.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
- else:
- if self.waiters:
- self.waiters.wait_for_job(self.job_id)
- else:
- self.hook.wait_for_job(self.job_id)
+ return job_desc
- awslogs = []
+ def _persist_cloudwatch_link(self, context: Context) -> None:
+ """
+ Persist CloudWatch logs link if available.
+
+ :param context: Task context
+ """
Review Comment:
`_persist_cloudwatch_link()` always calls `get_job_all_awslogs_info()` (which triggers a `describe_jobs`) even when `do_xcom_push=False`, but in that case `CloudWatchEventsLink.persist()` will no-op and the API call is wasted. Consider returning early in `_persist_cloudwatch_link()` when `not self.do_xcom_push` to avoid unnecessary AWS API usage.
```suggestion
"""
if not self.do_xcom_push:
return
```
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -257,9 +263,9 @@ def execute_complete(self, context: Context, event: dict[str, Any] | None = None
self.job_id = validated_event["job_id"]
Review Comment:
In `execute_complete()`, the early `status != "success"` check raises before `self.job_id` is set, so the operator cannot persist the CloudWatch operator link for deferred *failures*. This makes deferrable behavior inconsistent with `monitor_job()` (which persists the CloudWatch link and then raises on failure). Set `self.job_id` from the event before the status check, and persist the CloudWatch link (when possible) even on failure before raising the exception.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3463472619
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -220,14 +220,20 @@ def execute(self, context: Context) -> str | None:
if self.deferrable:
if not self.job_id:
- raise AirflowException("AWS Batch job - job_id was not found")
+ raise ValueError("AWS Batch job - job_id was not found")
Review Comment:
`execute()` now raises `ValueError` when `job_id` is missing in deferrable mode, but the docstring still claims only `AirflowException` is raised. Update the docstring so it accurately reflects possible exceptions.
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -368,33 +398,34 @@ def monitor_job(self, context: Context):
job_queue_arn=job_queue_arn,
)
- if self.awslogs_enabled:
- if self.waiters:
- self.waiters.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
- else:
- self.hook.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
- else:
- if self.waiters:
- self.waiters.wait_for_job(self.job_id)
- else:
- self.hook.wait_for_job(self.job_id)
+ return job_desc
+
+ def _persist_cloudwatch_link(self, context: Context) -> None:
+ """
+ Persist CloudWatch logs link if available.
+
+ :param context: Task context
+ """
+ if not context or "ti" not in context:
+ return
+
+ if not self.job_id:
+ return
Review Comment:
`BaseAwsLink.persist()` no-ops when `operator.do_xcom_push` is False (see `providers/amazon/.../links/base_aws.py:87-89`), but `_persist_cloudwatch_link()` still calls `get_job_all_awslogs_info()` first. This does extra AWS API work (and can log warnings) even though the link can never be persisted. Add an early return when `do_xcom_push` is disabled.
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -252,14 +258,31 @@ def execute(self, context: Context) -> str | None:
def execute_complete(self, context: Context, event: dict[str, Any] | None = None) -> str:
validated_event = validate_execute_complete_event(event)
- if validated_event["status"] != "success":
- raise AirflowException(f"Error while running job: {validated_event}")
-
+ # Set job_id first so CloudWatch link can be persisted even on failure
self.job_id = validated_event["job_id"]
# Fetch logs if awslogs_enabled
if self.awslogs_enabled:
- self.monitor_job(context) # fetch logs, no need to return
+ log_fetcher = self._get_batch_log_fetcher(self.job_id)
+ if log_fetcher:
+ self.log.info("Fetching logs for AWS Batch job: %s", self.job_id)
+ # Fetch up to 10,000 messages (CloudWatch limit)
+ log_messages = log_fetcher.get_last_log_messages(10000)
+ for message in log_messages:
+ self.log.info(message)
+ if log_messages:
+ self.log.info("Retrieved %d log messages from CloudWatch", len(log_messages))
+ else:
+ self.log.info("No logs found in CloudWatch for job: %s", self.job_id)
Review Comment:
`execute_complete()` calls `AwsTaskLogFetcher.get_last_log_messages()` directly, which can raise `botocore` exceptions (e.g. log stream not found / access denied). Unlike the background `run()` path (which catches these), this will fail the task just because log retrieval failed. Wrap the fetch in try/except and log a warning so deferred completion doesn’t error due to CloudWatch log issues.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3463447554
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -252,14 +258,17 @@ def execute(self, context: Context) -> str | None:
def execute_complete(self, context: Context, event: dict[str, Any] | None = None) -> str:
validated_event = validate_execute_complete_event(event)
+ # Set job_id first so CloudWatch link can be persisted even on failure
+ self.job_id = validated_event["job_id"]
+
+ # Persist CloudWatch logs for both success and failure
+ self._persist_cloudwatch_link(context)
+
if validated_event["status"] != "success":
raise AirflowException(f"Error while running job: {validated_event}")
- self.job_id = validated_event["job_id"]
-
- # Fetch logs if awslogs_enabled
- if self.awslogs_enabled:
- self.monitor_job(context) # fetch logs, no need to return
Review Comment:
@o-nikolas I've fixed execute_complete() to fetch and print CloudWatch logs when awslogs_enabled=True, matching non-deferrable behavior.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3463528305
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -252,14 +258,31 @@ def execute(self, context: Context) -> str | None:
def execute_complete(self, context: Context, event: dict[str, Any] | None = None) -> str:
validated_event = validate_execute_complete_event(event)
- if validated_event["status"] != "success":
- raise AirflowException(f"Error while running job: {validated_event}")
-
+ # Set job_id first so CloudWatch link can be persisted even on failure
self.job_id = validated_event["job_id"]
# Fetch logs if awslogs_enabled
if self.awslogs_enabled:
- self.monitor_job(context) # fetch logs, no need to return
+ log_fetcher = self._get_batch_log_fetcher(self.job_id)
+ if log_fetcher:
+ self.log.info("Fetching logs for AWS Batch job: %s", self.job_id)
+ # Fetch up to 10,000 messages (CloudWatch limit)
+ log_messages = log_fetcher.get_last_log_messages(10000)
+ for message in log_messages:
+ self.log.info(message)
+ if log_messages:
+ self.log.info("Retrieved %d log messages from CloudWatch", len(log_messages))
+ else:
+ self.log.info("No logs found in CloudWatch for job: %s", self.job_id)
Review Comment:
I've added try/except around get_last_log_messages(). Unlike the background thread's _get_log_events(), it doesn't handle botocore exceptions, so log retrieval failures would incorrectly fail the task.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #64745:
URL: https://github.com/apache/airflow/pull/64745#issuecomment-4794337511
Awesome work, congrats on your first merged pull request! You are invited to check our [Issue Tracker](https://github.com/apache/airflow/issues) for additional contributions.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on PR #64745:
URL: https://github.com/apache/airflow/pull/64745#issuecomment-4256755524
> I see many reviews by copilot and also some comments from @kakatur saying some are addressed. @kakatur can you please resolve each individual copilot thread that you have resolved?
@o-nikolas I left a response on each Copilot comment and marked them as resolved. I see that 8 tests have failed. I applied the fixes for those tests.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3068815416
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -368,33 +384,31 @@ def monitor_job(self, context: Context):
job_queue_arn=job_queue_arn,
)
- if self.awslogs_enabled:
- if self.waiters:
- self.waiters.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
- else:
- self.hook.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
- else:
- if self.waiters:
- self.waiters.wait_for_job(self.job_id)
- else:
- self.hook.wait_for_job(self.job_id)
+ return job_desc
+
+ def _persist_cloudwatch_link(self, context: Context) -> None:
+ """
+ Persist CloudWatch logs link if available.
+
+ :param context: Task context
+ """
+ if not self.do_xcom_push:
+ return
Review Comment:
Implemented: Check if do_xcom_push is False; Check if context is empty or missing "ti"
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3068820273
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -222,12 +222,18 @@ def execute(self, context: Context) -> str | None:
if not self.job_id:
raise AirflowException("AWS Batch job - job_id was not found")
- job = self.hook.get_job_description(self.job_id)
+ # Persist operator links before deferring so they're available in the UI
+ # Reuse job description to reduce API calls
+ job = self._persist_links(context)
job_status = job.get("status")
if job_status == self.hook.SUCCESS_STATE:
+ # Job already completed - persist CloudWatch logs
+ self._persist_cloudwatch_link(context)
self.log.info("Job completed.")
return self.job_id
if job_status == self.hook.FAILURE_STATE:
+ # Job already failed - persist CloudWatch logs
+ self._persist_cloudwatch_link(context)
Review Comment:
In deferrable mode, `execute()` now persists the CloudWatch link when the job is already in `SUCCEEDED`/`FAILED`, regardless of `awslogs_enabled`. This makes `awslogs_enabled` behave differently between deferrable vs non-deferrable paths (where `monitor_job()` gates CloudWatch work behind `awslogs_enabled`) and can add unexpected Batch/CloudWatch API calls when log integration is disabled. Consider gating `_persist_cloudwatch_link(context)` here behind `self.awslogs_enabled` (or clarifying the flag semantics if the link should always be persisted).
```suggestion
# Job already completed - persist CloudWatch logs when enabled
if self.awslogs_enabled:
self._persist_cloudwatch_link(context)
self.log.info("Job completed.")
return self.job_id
if job_status == self.hook.FAILURE_STATE:
# Job already failed - persist CloudWatch logs when enabled
if self.awslogs_enabled:
self._persist_cloudwatch_link(context)
```
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -404,6 +421,34 @@ def monitor_job(self, context: Context):
**awslogs[0],
)
+ def monitor_job(self, context: Context):
+ """
+ Monitor an AWS Batch job.
+
+ This can raise an exception or an AirflowTaskTimeout if the task was
+ created with ``execution_timeout``.
+ """
+ if not self.job_id:
+ raise AirflowException("AWS Batch job - job_id was not found")
+
+ # Persist job definition and queue links
+ self._persist_links(context)
+
+ if self.awslogs_enabled:
+ if self.waiters:
+ self.waiters.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
+ else:
+ self.hook.wait_for_job(self.job_id, get_batch_log_fetcher=self._get_batch_log_fetcher)
+ else:
+ if self.waiters:
+ self.waiters.wait_for_job(self.job_id)
+ else:
+ self.hook.wait_for_job(self.job_id)
+
+ # After job completes, persist CloudWatch logs when log integration is enabled
+ if self.awslogs_enabled:
+ self._persist_cloudwatch_link(context)
Review Comment:
`monitor_job()` now persists the CloudWatch link only when `awslogs_enabled=True`, which changes behavior vs previous versions and will require updating/adding unit tests (e.g. `test_execute_without_failures` currently asserts `get_job_description` is called twice due to unconditional CloudWatch discovery). Please adjust tests to cover both branches (no CloudWatch API calls when `awslogs_enabled=False`, and CloudWatch link persistence when enabled).
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3068863151
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -222,12 +222,18 @@ def execute(self, context: Context) -> str | None:
if not self.job_id:
raise AirflowException("AWS Batch job - job_id was not found")
- job = self.hook.get_job_description(self.job_id)
+ # Persist operator links before deferring so they're available in the UI
+ # Reuse job description to reduce API calls
+ job = self._persist_links(context)
job_status = job.get("status")
if job_status == self.hook.SUCCESS_STATE:
+ # Job already completed - persist CloudWatch logs
+ self._persist_cloudwatch_link(context)
self.log.info("Job completed.")
return self.job_id
if job_status == self.hook.FAILURE_STATE:
+ # Job already failed - persist CloudWatch logs
+ self._persist_cloudwatch_link(context)
Review Comment:
Updated code to let execute() terminal states respect awslogs_enabled
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3067495278
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -257,6 +260,9 @@ def execute_complete(self, context: Context, event: dict[str, Any] | None = None
self.job_id = validated_event["job_id"]
+ # Persist operator links for UI
+ self._persist_links(context)
Review Comment:
I will guard inside _persist_links() (lines 363-367) instead of only in execute_complete().
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3067495733
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -403,6 +395,50 @@ def monitor_job(self, context: Context):
aws_partition=self.hook.conn_partition,
**awslogs[0],
)
+ else:
+ # Persist placeholder to prevent "XCom not found" warnings
+ # CloudWatch logs will be updated when job completes
+ context["task_instance"].xcom_push(
+ key="cloudwatch_events",
+ value=None,
+ )
Review Comment:
I will remove the placeholder entirely
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #64745:
URL: https://github.com/apache/airflow/pull/64745#issuecomment-4189226035
Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide (https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:
- Pay attention to the quality of your code (ruff, mypy and type annotations). Our [prek-hooks]( https://github.com/apache/airflow/blob/main/contributing-docs/08_static_code_checks.rst#prerequisites-for-prek-hooks) will help you with that.
- In case of a new feature add useful documentation (in docstrings or in `docs/` directory). Adding a new operator? Check this short [guide](https://github.com/apache/airflow/blob/main/airflow-core/docs/howto/custom-operator.rst) Consider adding an example DAG that shows how users should use it.
- Consider using [Breeze environment](https://github.com/apache/airflow/blob/main/dev/breeze/doc/README.rst) for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
- Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
- Please follow [ASF Code of Conduct](https://www.apache.org/foundation/policies/conduct) for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
- Be sure to read the [Airflow Coding style]( https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#coding-style-and-best-practices).
- Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
Apache Airflow is a community-driven project and together we are making it better 🚀.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix batchoperator deferrable xcom links [airflow]
Posted by "kakatur (via GitHub)" <gi...@apache.org>.
kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3463530852
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -220,14 +220,20 @@ def execute(self, context: Context) -> str | None:
if self.deferrable:
if not self.job_id:
- raise AirflowException("AWS Batch job - job_id was not found")
+ raise ValueError("AWS Batch job - job_id was not found")
Review Comment:
Updated docstring to document both ValueError (missing job_id) and AirflowException (job submission/execution failures) for accuracy."
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org