You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "Vamsi-klu (via GitHub)" <gi...@apache.org> on 2026/07/11 04:47:44 UTC
[PR] Kafka: Return consumed message results on request [airflow]
Vamsi-klu opened a new pull request, #69740:
URL: https://github.com/apache/airflow/pull/69740
## Problem
`ConsumeFromTopicOperator` always returned `None`, even when its per-message `apply_function` produced values. That made it awkward to use the operator in downstream task pipelines that expect the consumed and processed values as the task return value.
closes: #56494
## What Changed
This adds an opt-in `return_apply_function_results` flag. The default behavior remains unchanged: the operator returns `None`.
When the flag is enabled with per-message `apply_function`, the operator collects non-`None` return values in consume order and returns them as a list after commits and consumer close handling. `apply_function_batch` behavior is intentionally unchanged.
## Impact
Kafka consume tasks can now feed processed message results directly into downstream tasks when the caller opts in. Existing users are unaffected because the new behavior is disabled by default, and batch processing keeps its current return behavior.
## Testing
- `uv run ruff format providers/apache/kafka/src/airflow/providers/apache/kafka/operators/consume.py providers/apache/kafka/tests/unit/apache/kafka/operators/test_consume.py`
- `uv run ruff check --fix providers/apache/kafka/src/airflow/providers/apache/kafka/operators/consume.py providers/apache/kafka/tests/unit/apache/kafka/operators/test_consume.py`
- `UV_CACHE_DIR=/tmp/uv-cache uv run --project providers/apache/kafka pytest providers/apache/kafka/tests/unit/apache/kafka/operators/test_consume.py -xvs`
- Commit hooks for provider files
- `git diff --check`
The tests cover the problem statement expectations: default `None` return, opt-in result collection, filtering `None`, preserving consume order across batches, and leaving `apply_function_batch` return behavior unchanged.
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes — Codex (GPT-5)
Generated-by: Codex (GPT-5) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-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] Kafka: Return consumed message results on request [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk merged PR #69740:
URL: https://github.com/apache/airflow/pull/69740
--
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] Kafka: Return consumed message results on request [airflow]
Posted by "Vamsi-klu (via GitHub)" <gi...@apache.org>.
Vamsi-klu commented on PR #69740:
URL: https://github.com/apache/airflow/pull/69740#issuecomment-5153075966
Thanks for the careful review. Both caveats are now in the parameter docstring as of b90bc955: the XCom warning ("Results are returned through normal task return handling and may be stored in XCom, so avoid enabling this for large result sets") and an explicit note that None results are dropped, so the returned list does not align positionally with the consumed messages. I kept the wording in line with the operator guide so the two stay consistent.
Drafted-by: Claude Code (Fable 5); reviewed by @Vamsi-klu before posting
--
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] Kafka: Return consumed message results on request [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #69740:
URL: https://github.com/apache/airflow/pull/69740#issuecomment-5151596451
Nicely scoped. Defaulting to `False` means existing Dags are untouched, gating the return on both the flag and `apply_function` avoids surprising anyone using `apply_function_batch`, and the tests cover the cases that matter — default-returns-`None`, plus a combined test that filters `None` results and preserves consume order against a mock consumer that drains progressively rather than handing everything back at once.
I went looking for the memory and XCom hazard here, since this operator can run with `read_to_end=True` and no `max_messages`, and you had already covered it in the docs:
> Returned values use normal task return handling and may be stored in XCom, so avoid returning large payloads.
One thing before merge: that warning only exists in `docs/operators/index.rst`, not in the parameter docstring:
```
:param return_apply_function_results: Whether to collect non-None return values from the per-message
``apply_function`` and return them as a list. This option does not apply to ``apply_function_batch``.
```
The API reference is where someone looks when deciding whether to switch a flag on, and that is exactly the moment the size caveat matters. Could you add a clause there — something like "Results are returned through normal task return handling and may be stored in XCom, so avoid enabling this for large result sets"?
While you are in there, it would be worth noting in the same docstring that `None` results are dropped, so the returned list does not line up positionally with the consumed messages. It is implied by "non-None return values", but a caller trying to map results back to offsets could reasonably miss it.
Happy to merge once that's in.
---
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
--
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] Kafka: Return consumed message results on request [airflow]
Posted by "Vamsi-klu (via GitHub)" <gi...@apache.org>.
Vamsi-klu commented on PR #69740:
URL: https://github.com/apache/airflow/pull/69740#issuecomment-5235537421
Both requested docstring caveats are present on the current head `bbfce564`: the XCom size warning and the note that `None` results are dropped, so the returned list does not align positionally with consumed messages. The wording matches the operator guide, all checks are successful, neutral, or skipped, and GitHub reports the branch mergeable. Ready for another look whenever convenient.
---
Drafted-by: Codex (GPT-5); reviewed by @Vamsi-klu before posting
--
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