TimurRakhmatullin86 opened a new pull request, #69646: URL: https://github.com/apache/airflow/pull/69646
Kafka reports delivery outcomes asynchronously through the producer's delivery callback. Today `ProduceToTopicOperator`'s default callback only *logs* delivery errors, so a message rejected by the broker (e.g. `MSG_SIZE_TOO_LARGE`, unknown topic, ACL denied) is silently lost while the task is still marked successful. This PR adds an opt-in `raise_on_delivery_failure` parameter (default `False`, fully backwards compatible). When enabled, the operator collects errors passed to the delivery callback and, after the final `flush()` has resolved all outstanding deliveries, fails the task with an `AirflowException` reporting the number of failed messages and the first error. The configured `delivery_callback` is still invoked for every delivery report, so existing custom callbacks keep working unchanged. New unit tests cover: task failure on delivery error, success path with the flag enabled, unchanged default behaviour, and interaction with a custom `delivery_callback` (all with a mocked producer, no broker required). Existing tests are untouched. -- 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]
