Add `ConfirmationChannel` for async publisher confirmations #1824

lukebakken · 2025-12-08T20:11:16Z

Traditional publisher confirms in the Java client require manual
tracking of sequence numbers and correlation of Basic.Return messages.
This makes per-message error handling complex and provides no built-in
async pattern, backpressure mechanism, or message correlation support.

This change introduces ConfirmationChannel, a wrapper that provides
automatic publisher confirmation tracking with a
CompletableFuture-based API, optional throttling, and generic context
parameter for message correlation. The implementation uses
listener-based integration with existing Channel instances, requiring
no modifications to ChannelN.

New API components:

ConfirmationChannel interface - Extends Channel and adds
basicPublishAsync() methods that return CompletableFuture<T>
ConfirmationChannelN implementation - Wraps any Channel instance
and tracks confirmations via return/confirm/shutdown listeners
PublishException - Exception thrown when message is nack'd or
returned, with sequence number, routing details, and user context

The wrapper maintains independent sequence numbers using AtomicLong
and stores confirmation state in a ConcurrentHashMap. Each entry holds
the future, rate limiter permit, and user-provided context. Messages
include an x-seq-no header for correlating Basic.Return responses.

Rate limiting is optional via RateLimiter parameter. The
ThrottlingRateLimiter implementation uses progressive delays
(0-1000ms) based on capacity usage, applying backpressure when available
permits fall below a threshold (default 50%).

The basicPublish() and waitForConfirms() methods throw
UnsupportedOperationException on ConfirmationChannel to prevent
mixing synchronous and asynchronous patterns. All other Channel
methods delegate to the wrapped instance.

lukebakken · 2025-12-08T20:16:22Z

Hey @acogoluegnes!

Here's a proposal for implementing "automatic" publisher confirmation tracking in a similar manner to the .NET client. As you're the expert here, please suggest better names, a better implementation, etc, as I'm just barely familiar enough with Java and this project to implement this feature with the help of a genie. I did, of course, review the code.

If you like the way this is going, I thought I'd also add rate-throttling in a similar manner to the .NET client, i.e. as the outstanding confirmation window closes, increase delay between publishes to (hopefully) allow the broker to catch up.

src/main/java/com/rabbitmq/client/impl/ChannelN.java

src/main/java/com/rabbitmq/client/ChannelOptions.java

acogoluegnes · 2025-12-09T08:28:40Z

Thanks for this contributon @lukebakken. I think it is on the right track. I added some comments in the code. Just a few more remarks:

the new methods in the public interfaces should have a default implementation if we back port the PR to 5.x. This is for backward compatibility. Throwing an exception is good enough.
synchronized blocks and Object-based synchronization is considered old-style Java concurrency. We could see if there are more modern utilities in the Java concurrency toolkit to handle what we need in ChannelN. I'm not super opinionated on this topic though and this is an implementation details we can polish later.
the bulk of the PR is in ChannelN, it is not rocket science but add some non-trivial logic to a somewhat already complex class. I was wondering if it could be possible to externalize all the logic in a PublishConfirmState class used from ChannelN. This state class would register regular return and confirm listeners on the channel. This could even be an interface with a no-op implementation when the feature is not activated, minimalizing the impact on ChannelN. We can discuss this design when all the features are added.

src/main/java/com/rabbitmq/client/ChannelOptions.java

michaelklishin · 2025-12-10T21:27:06Z

@lukebakken I have sent you an external contributor invite for this repo.

lukebakken · 2025-12-11T04:04:46Z

@michaelklishin @acogoluegnes I took this comment to heart and re-implemented this feature as a ConfirmationChannel class that wraps ChannelN and does its own confirmation tracking. That way, no modifications to the existing Channel interface or ChannelN class are necessary.

The new `ConfirmationChannel` API introduced in rabbitmq/rabbitmq-java-client#1824 provides asynchronous publisher confirmation tracking with a `CompletableFuture`-based API, rate limiting, and message correlation support. This change adds `PublisherConfirmsAsync.java` to demonstrate the `ConfirmationChannel` API. The tutorial shows how to create a `ConfirmationChannel` wrapper with rate limiting, publish messages asynchronously with correlation context, and wait for all confirmations using `CompletableFuture.allOf()`. Depends on rabbitmq/rabbitmq-java-client#1824

lukebakken · 2025-12-11T18:01:13Z

Alrighty gang, all set I think. Small tutorial here - rabbitmq/rabbitmq-tutorials#707

Traditional publisher confirms in the Java client require manual tracking of sequence numbers and correlation of Basic.Return messages. This makes per-message error handling complex and provides no built-in async pattern, backpressure mechanism, or message correlation support. This change introduces `ConfirmationChannel`, a wrapper that provides automatic publisher confirmation tracking with a `CompletableFuture`-based API, optional throttling, and generic context parameter for message correlation. The implementation uses listener-based integration with existing `Channel` instances, requiring no modifications to `ChannelN`. New API components: - `ConfirmationChannel` interface - Extends `Channel` and adds `basicPublishAsync()` methods that return `CompletableFuture<T>` - `ConfirmationChannelN` implementation - Wraps any `Channel` instance and tracks confirmations via return/confirm/shutdown listeners - `PublishException` - Exception thrown when message is nack'd or returned, with sequence number, routing details, and user context The wrapper maintains independent sequence numbers using `AtomicLong` and stores confirmation state in a `ConcurrentHashMap`. Each entry holds the future, rate limiter permit, and user-provided context. Messages include an `x-seq-no` header for correlating Basic.Return responses. Rate limiting is optional via `RateLimiter` parameter. The `ThrottlingRateLimiter` implementation uses progressive delays (0-1000ms) based on capacity usage, applying backpressure when available permits fall below a threshold (default 50%). The `basicPublish()` and `waitForConfirms()` methods throw `UnsupportedOperationException` on `ConfirmationChannel` to prevent mixing synchronous and asynchronous patterns. All other `Channel` methods delegate to the wrapped instance. Tests include 9 unit tests for `ThrottlingRateLimiter` and 24 integration tests for publisher confirmation tracking with context parameter verification, rate limiting scenarios, and error handling.

michaelklishin · 2025-12-11T20:42:33Z

@lukebakken since this is a feature by most definitions, we now would have to go through a special approval process on our end :(

lukebakken force-pushed the lukebakken/publisher-confirm-tracking branch 2 times, most recently from 6a31a46 to 714370e Compare December 8, 2025 23:53