Add transport-aware batch publishing with Redis pipelines

[shivaam]

Summary

Add a public, transport-neutral Producer.batch() context and implement it for
the Redis transport with client.pipeline(transaction=False).

Normal Producer.publish() calls still own serialization, message preparation,
declarations, routing, queue priority selection, queue expiration, and fanout.
Only the final Redis LPUSH, PEXPIRE, and PUBLISH commands are deferred.
Transports that do not implement batching retain immediate publication.

This is motivated by Apache Airflow #8854.
Airflow may publish hundreds of Celery tasks in one scheduler iteration. Its
local prototype demonstrated the performance benefit, but had to replace
Kombu's private _put() and _put_fanout() methods.

Public API

with producer.batch(max_size=500) as batch:
    producer.publish(message_1, ...)
    producer.publish(message_2, ...)
    batch.flush()  # optional; the context remains usable
    producer.publish(message_3, ...)

• A successful outermost exit flushes pending operations.
• Nested contexts share the outer batch and its max_size.
• An exception leaving any nested context aborts the batch and discards
  operations that have not already been flushed.
• Empty contexts create no transport session.
• Unsupported transports publish immediately.
• producer.supports_batch_publish advertises actual transport support.

Design decision

The selected design combines public Producer.batch() ergonomics with a
transport-owned channel session created by create_publish_batch().

This keeps the API usable by Celery, which already passes a Kombu producer into
Task.apply_async(), while leaving broker command ownership with the transport.

Alternatives considered:

1. A producer-only implementation was rejected because the producer does not
   own Redis queue names, priority keys, wire encoding, or fanout topics.
2. Exposing only a channel/transport context was rejected because application
   callers would still need transport knowledge and Celery publishes through
   the producer API.
3. publish_many(requests) was rejected because it would duplicate the large
   Producer.publish() argument surface and force callers to construct a full
   list instead of using their existing publication loop.

Failure semantics

Redis pipeline execution is non-transactional. If the response to
pipeline.execute() is lost, Redis may have accepted none, some, or all of the
commands.

Kombu raises BatchPublishError and does not automatically replay the batch,
including when an individual publication used retry=True. Applications may
start a new batch and retry explicitly when at-least-once delivery and possible
duplicates are acceptable.

max_size bounds buffered Redis commands. Long-running contexts can use
batch.flush() for an elapsed-time boundary, and Redis socket_timeout bounds
network blocking.

Validation

• Full unit suite: 1611 passed, 178 skipped
• Real Redis 8.6.1 integration suite: 64 passed
• New real-service coverage runs with and without global_keyprefix and
  exercises direct routing, topic routing, priorities/FIFO, and fanout Pub/Sub.
• Pre-commit: all hooks passed, including flake8, isort, codespell, and mypy.
• Celery 5.6.3 smoke test: three Task.apply_async(..., producer=producer)
  calls remained deferred inside the context and all three appeared in Redis
  on successful exit.
• Non-strict Sphinx HTML build passed. The existing strict -W build still
  fails on unrelated warnings already present on main.

The included benchmark publishes 100 messages with 10 ms simulated per-request
latency over seven alternating rounds:

Mode	Median
Ordinary publication	1.2290 s
Redis batch	0.0164 s

Observed speedup: 75.1x.

Compatibility

• Existing publication is unchanged unless Producer.batch() is used.
• Redis TLS and Sentinel share the same batch-capable channel code; their
  inheritance/capability paths have unit coverage.
• Standard Redis is covered with a real server.
• TLS and Sentinel were not tested against real services.
• Redis Cluster is not currently an upstream Kombu transport and is not
  supported or tested here.

[Copilot]

Pull request overview

This PR introduces a new, public, transport-neutral batching API on kombu.Producer (Producer.batch()), and adds a Redis transport implementation that defers the final Redis commands using a non-transactional redis-py pipeline to reduce round trips while preserving existing message preparation/routing behavior.

Changes:

• Add Producer.batch() context manager with nested/abort/flush semantics and Producer.supports_batch_publish capability detection.
• Implement Redis-side batching via a PublishBatch that buffers LPUSH/PEXPIRE/PUBLISH into client.pipeline(transaction=False) and flushes on exit / size thresholds.
• Add BatchPublishError, documentation, changelog entry, unit + integration coverage, and a benchmark script.

Reviewed changes

Copilot reviewed 10 out of 10 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
kombu/messaging.py	Adds the public Producer.batch() API, batching state machine, and publish-path integration.
kombu/transport/redis.py	Implements Redis PublishBatch and hooks _put/_put_fanout to buffer final Redis commands when batching.
kombu/transport/base.py	Adds the batch_publish capability flag to default transport capabilities.
kombu/exceptions.py	Introduces BatchPublishError for uncertain batch delivery outcomes.
t/unit/test_messaging.py	Adds unit tests for batch context semantics (nesting, abort, flush, scoping).
t/unit/transport/test_redis.py	Adds unit tests validating Redis pipeline usage and command buffering behavior.
t/integration/test_redis.py	Adds real-Redis integration tests for batched direct/topic/fanout behavior.
docs/userguide/producers.rst	Documents batch publishing API, semantics, and Redis behavior/limitations.
Changelog.rst	Adds a “Next release” changelog entry for the new API.
benchmarks/redis_publish_batch.py	Adds a benchmark script to compare ordinary vs batched Redis publication under simulated latency.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[codecov]

Codecov Report

❌ Patch coverage is 93.24324% with 15 lines in your changes missing coverage. Please review.
✅ Project coverage is 83.15%. Comparing base (b47b680) to head (fd947e2).
⚠️ Report is 1 commits behind head on main.
✅ All tests successful. No failed tests found.

Files with missing lines	Patch %	Lines
kombu/messaging.py	91.52%	6 Missing and 4 partials ⚠️
kombu/transport/redis.py	95.14%	2 Missing and 3 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2572      +/-   ##
==========================================
+ Coverage   82.91%   83.15%   +0.23%     
==========================================
  Files          79       79              
  Lines       10408    10626     +218     
  Branches     1199     1241      +42     
==========================================
+ Hits         8630     8836     +206     
- Misses       1577     1584       +7     
- Partials      201      206       +5     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.