Re: [PR] Fix multiple_outputs no-op on deferrable KubernetesPodOperator [airflow]
jscheffl merged PR #67226: URL: https://github.com/apache/airflow/pull/67226 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Fix multiple_outputs no-op on deferrable KubernetesPodOperator [airflow]
jscheffl commented on PR #67226: URL: https://github.com/apache/airflow/pull/67226#issuecomment-4512289103 Looks almost good. One final (hopefully final) finding. Can you close the comments resolved? If static checks are fixed alongside one comment from me then I think it is good to merge -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Fix multiple_outputs no-op on deferrable KubernetesPodOperator [airflow]
jscheffl commented on code in PR #67226: URL: https://github.com/apache/airflow/pull/67226#discussion_r3283917687 ## providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py: ## @@ -1109,6 +,31 @@ def _clean(self, event: dict[str, Any], result: dict | None, context: Context) - result=result, ) +def _push_xcom_with_fan_out(self, ti: Any, value: Any) -> None: +"""Push ``return_value`` and, when ``multiple_outputs`` is set, also fan a dict out per key. + +Mirrors the task runner's ``_push_xcom_if_needed`` so the failure-path manual pushes +in ``cleanup`` (sync) and ``trigger_reentry`` (async) honour ``multiple_outputs`` — +previously they pushed only ``return_value``, silently dropping per-key fan-out. +On success both paths return the value and let the runner perform the push instead. +""" +if not value: Review Comment: If the value evaluates to `False` would not persist. Can you change to explicit NULL check? ```suggestion if value is not None: ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Fix multiple_outputs no-op on deferrable KubernetesPodOperator [airflow]
paultmathew commented on PR #67226: URL: https://github.com/apache/airflow/pull/67226#issuecomment-4512090401 Thanks for the review @jscheffl. Pushed a fix with an operator-local _push_xcom_with_fan_out helper used by both sync (post_complete_action) and async (trigger_reentry) failure-path pushes — so all four code paths now honour multiple_outputs consistently. Added matching failure-path tests for both modes. I considered promoting _push_xcom_if_needed to a public task-SDK helper as you mentioned. Manual XCOM_RETURN_KEY pushes from operator code are essentially KPO-specific today (grep across all providers only turns up one other place), so I kept the helper local to keep this PR focused — happy to file a follow-up issue for the wider task-SDK change if you'd prefer. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Fix multiple_outputs no-op on deferrable KubernetesPodOperator [airflow]
Copilot commented on code in PR #67226:
URL: https://github.com/apache/airflow/pull/67226#discussion_r3277060913
##
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py:
##
@@ -3516,6 +3523,54 @@ def
test_async_kpo_wait_termination_before_cleanup_on_failure(
post_complete_action.assert_called_once()
+@patch(KUB_OP_PATH.format("extract_xcom"))
+@patch(KUB_OP_PATH.format("post_complete_action"))
+@patch(HOOK_CLASS)
+def test_async_trigger_reentry_returns_sidecar_output_for_multiple_outputs(
+mocked_hook, post_complete_action, mock_extract_xcom
+):
+"""``trigger_reentry`` must return the sidecar output so the task runner
+runs ``_push_xcom_if_needed``, which honors ``multiple_outputs`` by fanning
+out the returned dict into per-key XComs. Before #67224 the operator pushed
+``return_value`` manually inside the ``finally`` block and returned
+``None``, which silently bypassed the runner's fan-out — making
+``multiple_outputs=True`` a no-op on deferrable KPO. The sync path's
+``execute_sync`` already returns ``result`` (line 760 in pod.py); this
+test pins the deferrable path to the same contract.
+"""
+metadata = {"metadata.name": TEST_NAME, "metadata.namespace":
TEST_NAMESPACE}
+succeeded_state = mock.MagicMock(**metadata, **{"status.phase":
"Succeeded"})
+mocked_hook.return_value.get_pod.return_value = succeeded_state
+# ``return_value`` (not ``side_effect``) so the pod-await poll loop can
+# call this any number of times without exhausting a fixed list.
+mocked_hook.return_value.core_v1_client.read_namespaced_pod.return_value =
succeeded_state
+
+sidecar_output = {"export_arn": "arn:aws:dynamodb:::export/x", "s3_uri":
"s3://b/p"}
+mock_extract_xcom.return_value = sidecar_output
+
+k = KubernetesPodOperator(task_id="task", deferrable=True,
do_xcom_push=True, multiple_outputs=True)
+context = create_context(k)
+context["ti"].xcom_push = MagicMock()
+
+success_event = {
+"status": "success",
+"message": TEST_SUCCESS_MESSAGE,
+"name": TEST_NAME,
+"namespace": TEST_NAMESPACE,
+}
+
+result = k.trigger_reentry(context, success_event)
+
+# The dict is returned — this is the fix. The task runner's
+# ``_push_xcom_if_needed`` (in
``task-sdk/.../execution_time/task_runner.py``)
+# then handles both ``return_value`` push and the ``multiple_outputs``
+# per-key fan-out, exercised end-to-end in ``test-sdk/`` unit tests.
+assert result is sidecar_output
Review Comment:
This assertion uses `is` to compare dicts (object identity). Prefer `==` so
the test validates the returned content and won’t break if `trigger_reentry`
later returns an equivalent copy instead of the same dict instance.
##
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py:
##
@@ -3516,6 +3523,54 @@ def
test_async_kpo_wait_termination_before_cleanup_on_failure(
post_complete_action.assert_called_once()
+@patch(KUB_OP_PATH.format("extract_xcom"))
+@patch(KUB_OP_PATH.format("post_complete_action"))
+@patch(HOOK_CLASS)
+def test_async_trigger_reentry_returns_sidecar_output_for_multiple_outputs(
+mocked_hook, post_complete_action, mock_extract_xcom
+):
+"""``trigger_reentry`` must return the sidecar output so the task runner
+runs ``_push_xcom_if_needed``, which honors ``multiple_outputs`` by fanning
+out the returned dict into per-key XComs. Before #67224 the operator pushed
+``return_value`` manually inside the ``finally`` block and returned
+``None``, which silently bypassed the runner's fan-out — making
+``multiple_outputs=True`` a no-op on deferrable KPO. The sync path's
+``execute_sync`` already returns ``result`` (line 760 in pod.py); this
+test pins the deferrable path to the same contract.
+"""
Review Comment:
The new test docstring includes an issue reference (#67224) and a hard-coded
source line reference (“line 760 in pod.py”). Airflow’s test docstrings should
describe behavior rather than track tickets, and line numbers tend to drift as
code changes. Consider shortening the docstring to just the behavioral contract
and move historical context/issue refs to a regular comment (or the PR
description), and avoid mentioning specific line numbers.
##
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py:
##
@@ -3451,13 +3451,17 @@ def
test_async_kpo_wait_termination_before_cleanup_on_success(
# check if it gets the pod
mocked_hook.return_value.get_pod.assert_called_once_with(TEST_NAME,
TEST_NAMESPACE)
-# assert that the xcom are extracted/not extracted
+# On the success path, ``trigger_reentry`` returns the sidecar output and
+# leaves the XCom push to the task runner's ``_push_xcom_if_needed`` —
+# see #67224. The operator no longer pushes ``return_value`` manually her
Re: [PR] Fix multiple_outputs no-op on deferrable KubernetesPodOperator [airflow]
jscheffl commented on code in PR #67226:
URL: https://github.com/apache/airflow/pull/67226#discussion_r3277028430
##
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py:
##
@@ -1074,12 +1074,19 @@ def trigger_reentry(self, context: Context, event:
dict[str, Any]) -> Any:
"Trigger emitted an %s event, failing the task: %s",
event["status"], event["message"]
)
message = event.get("stack_trace", event["message"])
+# Push manually before the raise — matches the sync-path
+# failure-push in cleanup ("Ensure that existing XCom is
+# pushed even in case of failure").
+if self.do_xcom_push and xcom_sidecar_output:
+context["ti"].xcom_push(XCOM_RETURN_KEY,
xcom_sidecar_output)
Review Comment:
Thanks for adding for consistency. But now you are repeating the same "bug"
like it was before and as you're trying to fix? In the case of failure in async
it is the same (bad) behavior like in the sync call that multiple outputs is
not handled.
Can you add a utility that is in both ends consistently used that minics the
task runner handling for multiple outputs or use a task runner public method
for this?
##
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py:
##
@@ -3451,13 +3451,17 @@ def
test_async_kpo_wait_termination_before_cleanup_on_success(
# check if it gets the pod
mocked_hook.return_value.get_pod.assert_called_once_with(TEST_NAME,
TEST_NAMESPACE)
-# assert that the xcom are extracted/not extracted
+# On the success path, ``trigger_reentry`` returns the sidecar output and
+# leaves the XCom push to the task runner's ``_push_xcom_if_needed`` —
+# see #67224. The operator no longer pushes ``return_value`` manually here.
Review Comment:
I think no comment needed on bugifx
```suggestion
```
##
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py:
##
@@ -1074,12 +1074,19 @@ def trigger_reentry(self, context: Context, event:
dict[str, Any]) -> Any:
"Trigger emitted an %s event, failing the task: %s",
event["status"], event["message"]
)
message = event.get("stack_trace", event["message"])
+# Push manually before the raise — matches the sync-path
+# failure-push in cleanup ("Ensure that existing XCom is
+# pushed even in case of failure").
+if self.do_xcom_push and xcom_sidecar_output:
+context["ti"].xcom_push(XCOM_RETURN_KEY,
xcom_sidecar_output)
raise AirflowException(message)
finally:
self._clean(event=event, context=context,
result=xcom_sidecar_output)
-if self.do_xcom_push and xcom_sidecar_output:
-context["ti"].xcom_push(XCOM_RETURN_KEY, xcom_sidecar_output)
+# Return on success so the task runner's _push_xcom_if_needed handles
+# return_value and multiple_outputs fan-out, matching execute_sync.
Review Comment:
I think comment not needed as it represents the default.
```suggestion
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
[PR] Fix multiple_outputs no-op on deferrable KubernetesPodOperator [airflow]
paultmathew opened a new pull request, #67226: URL: https://github.com/apache/airflow/pull/67226 ## Summary `KubernetesPodOperator(do_xcom_push=True, multiple_outputs=True, deferrable=True)` silently failed to fan out the sidecar's `return.json` dict into per-key XComs — only `return_value` was published. Downstream tasks subscripting a key got `None` at runtime with no error. Root cause: `trigger_reentry` pushed `return_value` manually inside a `finally` block and never returned the value to the task runner, so the runner's `_push_xcom_if_needed` (the code that honors `multiple_outputs` and fans the dict out) was bypassed. The sync path's `execute_sync` already returns `result` (`pod.py:760`). This aligns the deferrable path with the same contract. ## Behaviour change - **Success path** (new): `trigger_reentry` returns `xcom_sidecar_output`. The task runner pushes `return_value` and, when `multiple_outputs=True`, also fans the dict out into per-key XComs. - **Failure path** (preserved): the manual `xcom_push(XCOM_RETURN_KEY, ...)` moves into the `event["status"] != "success"` branch, above the `raise`. Partial sidecar output is still surfaced in XCom, and the push now happens even when `_clean` subsequently raises (strict improvement — previously the in-`finally` push was unreachable in that case). ## Closes Fixes #67224 ## Tests Updated `test_async_kpo_wait_termination_before_cleanup_on_success` and `test_async_kpo_wait_termination_before_cleanup_on_failure` to reflect the new contract. Added `test_async_trigger_reentry_returns_sidecar_output_for_multiple_outputs` to lock in the success-path return value with `multiple_outputs=True`. The end-to-end `_push_xcom_if_needed` fan-out behavior is already covered by task-SDK unit 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
