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]

Reply via email to