Kafka: Return consumed message results on request

[Vamsi-klu]

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?

• Yes — Codex (GPT-5)

Generated-by: Codex (GPT-5) following the guidelines