shivaam commented on issue #8854: URL: https://github.com/apache/airflow/issues/8854#issuecomment-5081287865
Hi @mik-laj and @hussein-awala, I investigated this locally and wanted to check the preferred direction before proposing any code. The approach I prototyped was: 1. Reuse one Celery/Kombu producer for an Airflow workload batch. 2. Let Celery and Kombu prepare every message normally, preserving serialization, task IDs, routing, priorities and headers. 3. For the Redis broker only, buffer the final `LPUSH`/`PUBLISH` operations in a non-transactional redis-py pipeline. 4. Flush the pipeline once after the scheduler has processed the batch. 5. Preserve the existing publication path for non-Redis brokers. For a larger batch of tasks, I saw significant improvements in the latencies as we are avoiding multiple network trips. The main concern is the integration point. Kombu currently exposes [`Producer.publish()`](https://docs.celeryq.dev/projects/kombu/en/stable/userguide/producers.html) for individual publications, but no public batch context. The prototype therefore has to replace the Redis transport’s private [`_put()` and `_put_fanout()` methods](https://github.com/celery/kombu/blob/v5.6.2/kombu/transport/redis.py#L1034-L1047) temporarily so their final Redis commands go into a [redis-py pipeline](https://redis.readthedocs.io/en/stable/advanced_features.html#pipelines). That implementation is narrow and can be guarded by contract and real-Redis tests, but future changes to Kombu’s private methods would not automatically be inherited by Airflow. There are also failure semantics to agree on: - If the pipeline flush loses its response, Redis might have accepted none, some or all messages which can create batch failures for tasks and callbacks. I can open a draft PR as well if you align with the direction. -- 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]
