potiuk commented on PR #69646: URL: https://github.com/apache/airflow/pull/69646#issuecomment-5153937129
This is worth having. A message the broker rejects — too large, unknown topic, ACL denied — is currently dropped while the task reports success, and data loss that looks like success is the worst failure mode to leave as the only option. The implementation reads well. Defaulting to `False` keeps existing Dags untouched, and raising after the final `flush()` is the only correct point, since that is what resolves outstanding deliveries — raising earlier would miss failures still in flight. Wrapping the user's `delivery_callback` rather than replacing it is the detail most likely to go wrong in a change like this, and `test_operator_raise_on_delivery_failure_custom_callback` pins it directly with `_custom_delivery_callback_calls == [error]`. Defining `KafkaMessageDeliveryError` rather than raising `AirflowException` directly is also the right call. One thing I would like documented before this merges. Enabling the flag changes retry semantics in a way that is not obvious from its name: the task fails *after* part of the batch has already been delivered, so a retry re-produces the whole batch and duplicates every message that succeeded the first time. That is inherent to at-least-once delivery rather than a flaw here, but someone switching on "raise on delivery failure" will not necessarily expect duplicates as the price. A sentence in the `:param raise_on_delivery_failure:` docstring would cover it — something like "On failure the task is retried from the beginning, so messages that were delivered successfully will be produced again." Minor, take it or leave it: the wrapper collects errors even when the flag is off. It is harmless, since only failures are appended and a run with many of those has larger problems, but the collection serves no purpose in that path. Separately — this branch has drifted a long way behind `main` (608 commits) and is now conflicting, so it will need a rebase before it can merge regardless of the above. --- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
