Add ConfirmationPublisher for async publisher confirmations

[lukebakken]

Traditional publisher confirms in the Java client require manual tracking of sequence numbers and manual 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 ConfirmationPublisher, a standalone AutoCloseable that provides automatic publisher confirmation tracking with a CompletableFuture-based API, optional bounded-outstanding backpressure, and a generic context parameter for message correlation. It owns a dedicated channel created from the supplied Connection and requires no modifications to Channel, ChannelN, or any other existing class.

API

// Unlimited outstanding confirmations
ConfirmationPublisher publisher = ConfirmationPublisher.create(connection);

// At most 100 unconfirmed messages; basicPublishAsync blocks at the limit
ConfirmationPublisher publisher = ConfirmationPublisher.create(connection, 100);

// Custom executor for future completion
ConfirmationPublisher publisher = ConfirmationPublisher.create(connection, 100, executor);

CompletableFuture<T> future =
    publisher.basicPublishAsync(exchange, routingKey, props, body, context);

The future completes with the caller-supplied context on basic.ack, or exceptionally with a PublishException on basic.nack or basic.return.

Design

The publisher owns a private channel, so no external publish can desynchronize the delivery-tag space. It keys its outstanding map by the channel's authoritative getNextPublishSeqNo(), read atomically with the publish under a lock, rather than maintaining a shadow counter. Futures are completed on a configurable executor (default ForkJoinPool.commonPool()), never on the connection's I/O thread.

A basic.return frame carries no delivery tag, so an x-seq-no header is injected only when mandatory = true (returns are impossible otherwise) to correlate the return back to a publish. Non-mandatory publishes avoid the header entirely. Note that when a mandatory message is routable, the broker echoes this header on delivery, so its consumers observe it; applications that must not expose it should publish such messages through their own channel.

Recovery and failure handling

• When the connection supports automatic recovery, the publisher keeps working across recoveries. A RecoveryListener fails all outstanding futures at the start of recovery, before confirm.select resets the delivery-tag sequence, so a reused tag cannot mis-correlate against a stale entry.
• A connection-level publish failure is left to the natural shutdown and recovery path. A publish failure that leaves the channel genuinely stuck open with an advanced sequence number aborts the private channel, since correlation could no longer be trusted.
• Publishing while a channel is transiently not in confirm mode (mid-recovery) fails the future rather than sending an untracked message.
• Completions fall back to the calling thread if the configured executor rejects them, so a saturated or shut-down user executor never leaves a future hanging.

PublishException

PublishException carries the publish sequence number, the user context, and, for a basic.return, the full Return (reply code and text, exchange, routing key, properties, body). Use isReturn() to distinguish a return from a nack; getReturned() is null for a nack.

[lukebakken]

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.

[acogoluegnes]

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.

[michaelklishin]

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

[lukebakken]

@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.

[lukebakken]

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

[michaelklishin]

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

[lukebakken]

Code review findings, tracked as inline comments so each can be resolved as it is addressed. Ranked roughly by severity in each comment's first line (F1 = most severe, F10 = least).

Note that F1, F2, F3, and part of F10 share one root cause: the wrapper keeps its own AtomicLong sequence counter parallel to the broker's delivery-tag space instead of using the channel's authoritative getNextPublishSeqNo() (or integrating tracking into ChannelN itself, where the counter and unconfirmedSet already live). Fixing that design point addresses all four together.

[sdewhitt]

This looks really handy!