diff --git a/RUNNING_TESTS.md b/RUNNING_TESTS.md index f948127ba1..c57c8f702a 100644 --- a/RUNNING_TESTS.md +++ b/RUNNING_TESTS.md @@ -88,6 +88,18 @@ top-level directory of the source tree: -Dit.test=DeadLetterExchange ``` +* To run a single test class: + +``` +./mvnw verify -Dit.test=Confirm +``` + +* To run a specific test method within a test class: + +``` +./mvnw verify -Dit.test=Confirm#publisherBasicConfirmation +``` + Test reports can be found in `target/failsafe-reports`. ## Running Against a Broker in a Docker Container @@ -95,7 +107,7 @@ Test reports can be found in `target/failsafe-reports`. Run the broker: ``` -docker run -it --rm --name rabbitmq -p 5672:5672 rabbitmq:3.8 +docker run --pull always --rm --tty --interactive --name rabbitmq --publish 5672:5672 rabbitmq:latest ``` Launch the tests: diff --git a/doc/publisher-confirmations/README.md b/doc/publisher-confirmations/README.md new file mode 100644 index 0000000000..6acb08c905 --- /dev/null +++ b/doc/publisher-confirmations/README.md @@ -0,0 +1,212 @@ +# ConfirmationPublisher - Asynchronous Publisher Confirmations + +## Overview + +`ConfirmationPublisher` provides asynchronous 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`, requiring no modifications to +the existing `Channel` or `ChannelN` classes. + +## Motivation + +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. + +`ConfirmationPublisher` addresses these limitations: + +- Automatic confirmation tracking via `CompletableFuture` API +- Generic context parameter for message correlation +- Optional bounded-outstanding-confirms backpressure (simple `Semaphore`) +- Futures completed on a configurable executor, never on the I/O thread +- Clean recovery semantics: outstanding futures fail, then publishing resumes + +## Architecture + +``` +Connection + | + +-- createChannel() (private to the publisher) + | + ConfirmationPublisher + - ConcurrentSkipListMap + - publishLock (serializes seqNo read + publish) + - Semaphore (optional backpressure) + - Confirm/Return/Shutdown listeners +``` + +The publisher keys its outstanding map by `channel.getNextPublishSeqNo()`, +read atomically with the publish under a `ReentrantLock`. Because the channel +is private, no external publish can desynchronize the counter. This guarantees +that the map key is the broker's delivery tag and correlation is correct. + +### Confirmation Tracking + +**Lifecycle:** + +1. Acquire a semaphore permit (if bounded) +2. Lock, read `channel.getNextPublishSeqNo()`, insert entry, publish, unlock +3. If publish fails, remove entry, release permit, abort channel if tag-space + skewed +4. Broker sends `basic.ack`/`basic.nack`/`basic.return` +5. Entry claimed via `ConcurrentSkipListMap.remove()`, permit released via + `AtomicBoolean.compareAndSet`, future completed on the configured executor + +**Multiple-ack handling:** + +```java +NavigableMap confirmed = + outstanding.headMap(deliveryTag, true); +for (Long seqNo : confirmed.keySet()) { + completeOne(seqNo, nack); +} +``` + +This is O(k log n) on the confirmed entries, matching `ChannelN`'s own +`unconfirmedSet.headSet(seqNo + 1).clear()` pattern. + +### Return Correlation + +A `basic.return` does not carry a delivery tag, so the publisher injects an +`x-seq-no` header **only when `mandatory = true`** (returns are impossible +otherwise). On the return path, the header is read back to find the +corresponding entry. Non-mandatory publishes avoid the header and the +`BasicProperties` copy entirely. + +The broker echoes this header on normal delivery as well as on return, so when +a mandatory message is routable its consumers receive it with the `x-seq-no` +header present; it cannot be stripped on the routable path. Applications that +must not expose the header should publish such messages through their own +channel rather than this publisher. + +The publisher rejects publishes where the caller already set an `x-seq-no` +header to prevent silent clobbering. + +### Connection Recovery + +When the wrapped channel is an `AutorecoveringChannel`, connection loss +triggers the shutdown listener, which fails all outstanding futures +exceptionally (their outcome is unknown). After recovery the channel's +`getNextPublishSeqNo()` returns the reset counter (verified experimentally), +so new publishes key correctly with no reset hook needed. + +## API + +### Creating a Publisher + +```java +// Unlimited outstanding +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, myExecutor); +``` + +### Publishing + +```java + CompletableFuture basicPublishAsync(String exchange, String routingKey, + AMQP.BasicProperties props, byte[] body, T context) + + CompletableFuture basicPublishAsync(String exchange, String routingKey, + boolean mandatory, AMQP.BasicProperties props, byte[] body, T context) +``` + +The context parameter is returned in the completed future on success and +available via `PublishException.getContext()` on failure. + +### Closing + +```java +publisher.close(); +``` + +Outstanding futures complete exceptionally with `ShutdownSignalException`. + +## Error Handling + +### PublishException + +```java +public class PublishException extends IOException { + public long getSequenceNumber(); + public boolean isReturn(); + public Return getReturned(); // null for nacks + public Object getContext(); +} +``` + +- **basic.nack**: `isReturn() == false`, `getReturned() == null` +- **basic.return**: `isReturn() == true`, `getReturned()` carries the full + `Return` (reply code, text, exchange, routing key, properties, body) +- **Shutdown/close**: future completes with the `ShutdownSignalException` + +## Usage Examples + +### Basic + +```java +Connection connection = factory.newConnection(); +ConfirmationPublisher publisher = ConfirmationPublisher.create(connection, 100); + +publisher.basicPublishAsync("exchange", "routing.key", props, body, "msg-123") + .thenAccept(msgId -> System.out.println("Confirmed: " + msgId)) + .exceptionally(ex -> { + System.err.println("Failed: " + ex.getCause().getMessage()); + return null; + }); +``` + +### With Context Objects + +```java +class OrderContext { + final String orderId; + final Instant sent; + OrderContext(String orderId) { + this.orderId = orderId; + this.sent = Instant.now(); + } +} + +OrderContext ctx = new OrderContext("order-12345"); +publisher.basicPublishAsync("orders", "new", props, body, ctx) + .thenAccept(c -> { + Duration latency = Duration.between(c.sent, Instant.now()); + System.out.println(c.orderId + " confirmed in " + latency.toMillis() + "ms"); + }); +``` + +### Handling Returns + +```java +publisher.basicPublishAsync("exchange", "key", true, props, body, "msg-1") + .exceptionally(ex -> { + if (ex.getCause() instanceof PublishException) { + PublishException pe = (PublishException) ex.getCause(); + if (pe.isReturn()) { + Return r = pe.getReturned(); + System.err.println("Unroutable: " + r.getReplyText()); + } + } + return null; + }); +``` + +## Design Decisions + +| Decision | Rationale | +|----------|-----------| +| Owns the channel | Prevents external publishes from skewing the tag space | +| Keys by `getNextPublishSeqNo()` | Uses the channel's authoritative counter; no shadow counter to diverge | +| `ReentrantLock` around seqNo + publish | Makes them atomic; contention negligible since `ChannelN` already serializes `transmit` | +| `ConcurrentSkipListMap` | O(k log n) `headMap` for multiple-acks vs O(n) keyset scan | +| `AtomicBoolean` on slot release | Prevents double-release race between ack and shutdown paths | +| Completes futures on an executor | Never blocks the connection's frame-reader thread | +| Header only on `mandatory=true` | Non-mandatory publishes avoid the wire overhead entirely | +| Aborts channel on publish failure after tag consumed | Safe because the channel is private; guarantees tag-space consistency | diff --git a/src/main/java/com/rabbitmq/client/ConfirmationPublisher.java b/src/main/java/com/rabbitmq/client/ConfirmationPublisher.java new file mode 100644 index 0000000000..30b883e76d --- /dev/null +++ b/src/main/java/com/rabbitmq/client/ConfirmationPublisher.java @@ -0,0 +1,459 @@ +// Copyright (c) 2007-2026 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries. +// +// This software, the RabbitMQ Java client library, is triple-licensed under the +// Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2 +// ("GPL") and the Apache License version 2 ("ASL"). For the MPL, please see +// LICENSE-MPL-RabbitMQ. For the GPL, please see LICENSE-GPL2. For the ASL, +// please see LICENSE-APACHE2. +// +// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND, +// either express or implied. See the LICENSE file for specific language governing +// rights and limitations of this software. +// +// If you have any questions regarding licensing, please contact us at +// info@rabbitmq.com. + +package com.rabbitmq.client; + +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentSkipListMap; +import java.util.concurrent.Executor; +import java.util.concurrent.ForkJoinPool; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.ReentrantLock; +import java.util.NavigableMap; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A publisher with asynchronous publisher confirmation tracking. + *

+ * A {@code ConfirmationPublisher} owns a dedicated channel created from the + * supplied {@link Connection}. The channel is put in confirm mode and every + * message published through {@link #basicPublishAsync} is tracked: the + * returned {@link CompletableFuture} completes with the caller-supplied + * context when the broker confirms the message, or exceptionally with + * {@link PublishException} when the broker rejects ({@code basic.nack}) or + * returns ({@code basic.return}) it. + *

+ * The publisher's channel is private to it: topology operations and ordinary + * publishing should use the application's own channels. + *

+ * When the connection supports automatic recovery, the publisher keeps + * working across recoveries: futures outstanding at the time of a connection + * failure complete exceptionally (their outcome is unknown; callers decide + * whether to republish) and publishing can resume once recovery completes. + *

+ * Example usage: + * 

{@code
+ * ConfirmationPublisher publisher = ConfirmationPublisher.create(connection, 100);
+ * publisher.basicPublishAsync("exchange", "routing.key", props, body, "msg-123")
+ *     .thenAccept(msgId -> System.out.println("Confirmed: " + msgId))
+ *     .exceptionally(ex -> {
+ *         System.err.println("Failed: " + ex.getCause());
+ *         return null;
+ *     });
+ * }
+ * + * @see PublishException + */ +public class ConfirmationPublisher implements AutoCloseable { + + private static final Logger LOGGER = LoggerFactory.getLogger(ConfirmationPublisher.class); + + /** + * Header used to correlate {@code basic.return} with a publish. Only added + * to messages published with {@code mandatory = true}, since a + * {@code basic.return} frame carries no delivery tag and this header is the + * only way to match it back to a publish. + *

+ * The broker echoes this header both in the return frame and on normal + * delivery, so when a mandatory message is routable its consumers + * receive it with this header present. It cannot be stripped on the routable + * path. Applications that must not expose it should publish such messages + * through their own channel rather than this publisher. + */ + public static final String SEQUENCE_NUMBER_HEADER = "x-seq-no"; + + private final Channel channel; + private final Semaphore outstandingLimit; + private final Executor confirmationExecutor; + private final ReentrantLock publishLock = new ReentrantLock(); + private final ConcurrentSkipListMap outstanding = + new ConcurrentSkipListMap(); + + /** + * Creates a publisher with no bound on outstanding confirmations. + * + * @param connection the connection to create the publisher's channel from + * @return a new publisher + * @throws IOException if channel creation or {@code confirm.select} fails + */ + public static ConfirmationPublisher create(Connection connection) throws IOException { + return new ConfirmationPublisher(connection, 0, ForkJoinPool.commonPool()); + } + + /** + * Creates a publisher that allows at most {@code maxOutstandingConfirms} + * unconfirmed messages; {@link #basicPublishAsync} blocks when the limit + * is reached. + * + * @param connection the connection to create the publisher's channel from + * @param maxOutstandingConfirms maximum unconfirmed messages, must be positive + * @return a new publisher + * @throws IOException if channel creation or {@code confirm.select} fails + */ + public static ConfirmationPublisher create(Connection connection, int maxOutstandingConfirms) + throws IOException { + if (maxOutstandingConfirms <= 0) { + throw new IllegalArgumentException("maxOutstandingConfirms must be positive: " + maxOutstandingConfirms); + } + return new ConfirmationPublisher(connection, maxOutstandingConfirms, ForkJoinPool.commonPool()); + } + + /** + * Creates a publisher with a bound on outstanding confirmations and a + * custom executor for completing futures. + * + * @param connection the connection to create the publisher's channel from + * @param maxOutstandingConfirms maximum unconfirmed messages, must be positive + * @param confirmationExecutor executor used to complete the returned futures + * (and thus run non-async continuations); never the connection's I/O thread + * @return a new publisher + * @throws IOException if channel creation or {@code confirm.select} fails + */ + public static ConfirmationPublisher create(Connection connection, int maxOutstandingConfirms, + Executor confirmationExecutor) throws IOException { + if (maxOutstandingConfirms <= 0) { + throw new IllegalArgumentException("maxOutstandingConfirms must be positive: " + maxOutstandingConfirms); + } + if (confirmationExecutor == null) { + throw new IllegalArgumentException("confirmationExecutor must not be null"); + } + return new ConfirmationPublisher(connection, maxOutstandingConfirms, confirmationExecutor); + } + + private ConfirmationPublisher(Connection connection, int maxOutstandingConfirms, + Executor confirmationExecutor) throws IOException { + this.channel = connection.createChannel(); + this.outstandingLimit = maxOutstandingConfirms > 0 ? new Semaphore(maxOutstandingConfirms) : null; + this.confirmationExecutor = confirmationExecutor; + this.channel.confirmSelect(); + this.channel.addConfirmListener(this::handleAck, this::handleNack); + this.channel.addReturnListener(this::handleReturn); + this.channel.addShutdownListener(this::handleShutdown); + if (this.channel instanceof Recoverable) { + ((Recoverable) this.channel).addRecoveryListener(new RecoveryListener() { + @Override + public void handleRecoveryStarted(Recoverable recoverable) { + onRecoveryStarted(); + } + + @Override + public void handleRecovery(Recoverable recoverable) { + } + }); + } + } + + /** + * Asynchronously publish a message with publisher confirmation tracking. + *

+ * Thread-safe. If the publisher was created with a bound on outstanding + * confirmations, this method blocks while the bound is reached. + * + * @param exchange the exchange to publish to + * @param routingKey the routing key + * @param props message properties (null for default) + * @param body message body + * @param context user-provided context object for correlation (can be null) + * @param the type of the context parameter + * @return a future completing with the context on {@code basic.ack}, or + * exceptionally with {@link PublishException} on {@code basic.nack}, + * or with the underlying exception if the publish itself fails + */ + public CompletableFuture basicPublishAsync(String exchange, String routingKey, + AMQP.BasicProperties props, byte[] body, T context) { + return basicPublishAsync(exchange, routingKey, false, props, body, context); + } + + /** + * Asynchronously publish a message with the mandatory flag and publisher + * confirmation tracking. + *

+ * When {@code mandatory} is true a {@value #SEQUENCE_NUMBER_HEADER} header + * is added to the message so that a {@code basic.return} can be correlated + * with this publish; the future then completes exceptionally with a + * {@link PublishException} carrying the {@link Return}. + * + * @param exchange the exchange to publish to + * @param routingKey the routing key + * @param mandatory true if the message must be routable to at least one queue + * @param props message properties (null for default); must not already + * contain a {@value #SEQUENCE_NUMBER_HEADER} header when mandatory + * @param body message body + * @param context user-provided context object for correlation (can be null) + * @param the type of the context parameter + * @return a future completing with the context on {@code basic.ack}, or + * exceptionally with {@link PublishException} on {@code basic.nack} + * or {@code basic.return}, or with the underlying exception if the + * publish itself fails + * @throws IllegalArgumentException if {@code mandatory} is true and + * {@code props} already contains a {@value #SEQUENCE_NUMBER_HEADER} + * header; this is a caller programming error and is reported + * synchronously rather than through the returned future + */ + public CompletableFuture basicPublishAsync(String exchange, String routingKey, + boolean mandatory, AMQP.BasicProperties props, byte[] body, T context) { + if (mandatory && props != null && props.getHeaders() != null + && props.getHeaders().containsKey(SEQUENCE_NUMBER_HEADER)) { + throw new IllegalArgumentException( + "message properties must not contain a " + SEQUENCE_NUMBER_HEADER + " header"); + } + CompletableFuture future = new CompletableFuture(); + OutstandingConfirmation confirmation = + new OutstandingConfirmation(future, context, this.outstandingLimit); + try { + if (this.outstandingLimit != null) { + this.outstandingLimit.acquire(); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + future.completeExceptionally(e); + return future; + } + this.publishLock.lock(); + try { + long seqNo = this.channel.getNextPublishSeqNo(); + if (seqNo == 0) { + // The channel is transiently not in confirm mode: an + // autorecovering channel has swapped in a fresh delegate whose + // confirm.select has not been re-issued yet. Publishing now + // would send an untracked message the broker never confirms. + confirmation.releaseSlot(); + future.completeExceptionally(new IOException( + "channel is not in confirm mode (recovery in progress)")); + return future; + } + AMQP.BasicProperties publishProps = + mandatory ? withSequenceNumberHeader(props, seqNo) : props; + this.outstanding.put(seqNo, confirmation); + try { + this.channel.basicPublish(exchange, routingKey, mandatory, publishProps, body); + } catch (Exception e) { + this.outstanding.remove(seqNo); + confirmation.releaseSlot(); + abortChannelIfTagSpaceStuck(e); + future.completeExceptionally(e); + } + } finally { + this.publishLock.unlock(); + } + return future; + } + + /** + * @return the number of publishes awaiting a broker confirmation; this is + * not a constant-time operation and is intended for monitoring, not + * for use on a hot path + */ + public int getOutstandingCount() { + return this.outstanding.size(); + } + + /** + * Closes the publisher and its channel with a graceful handshake. + * Confirmations that arrive before the channel finishes closing complete + * their futures normally; any futures still outstanding once the channel is + * closed complete exceptionally with a {@link ShutdownSignalException} + * (their messages are unconfirmed and their outcome unknown). + */ + @Override + public void close() throws IOException { + try { + this.channel.close(); + } catch (TimeoutException e) { + throw new IOException(e); + } catch (AlreadyClosedException e) { + // already closed by shutdown or a previous call + } + } + + private static AMQP.BasicProperties withSequenceNumberHeader(AMQP.BasicProperties props, long seqNo) { + Map headers = new HashMap(); + if (props != null && props.getHeaders() != null) { + headers.putAll(props.getHeaders()); + } + headers.put(SEQUENCE_NUMBER_HEADER, seqNo); + AMQP.BasicProperties.Builder builder = + props == null ? new AMQP.BasicProperties.Builder() : props.builder(); + return builder.headers(headers).build(); + } + + /** + * The delegate channel increments its publish sequence number before the + * frame reaches the wire, so a failed publish leaves the channel's counter + * ahead of the broker's delivery-tag sequence and every later confirmation + * would be correlated incorrectly. + *

+ * An {@link IOException} means the connection is failing: the imminent + * shutdown fails all outstanding futures and, if the connection recovers, + * the counter resets to 1 (the broker restarts delivery tags), so the skew + * self-heals and the channel must be left alone to recover. Any other + * exception (for example from a misbehaving {@code ObservationCollector}) + * leaves the channel open with a genuinely skewed counter and no shutdown + * pending; the channel is private to this publisher, so the safe reaction + * is to abort it, which fails all outstanding futures via the shutdown + * listener. + */ + private void abortChannelIfTagSpaceStuck(Exception failure) { + if (failure instanceof IOException) { + return; + } + if (this.channel.isOpen()) { + LOGGER.warn("Aborting confirmation publisher channel: publish failed " + + "after a sequence number was consumed", failure); + this.channel.abort(); + } + } + + private void handleAck(long deliveryTag, boolean multiple) { + completeConfirmations(deliveryTag, multiple, false); + } + + private void handleNack(long deliveryTag, boolean multiple) { + completeConfirmations(deliveryTag, multiple, true); + } + + private void completeConfirmations(long deliveryTag, boolean multiple, boolean nack) { + if (multiple) { + // Confirm/return handlers all run on the single connection thread and + // concurrent publishes only insert higher sequence numbers, so this + // head view is stable: complete each entry, then clear it in one pass. + NavigableMap confirmed = + this.outstanding.headMap(deliveryTag, true); + for (Map.Entry entry : confirmed.entrySet()) { + complete(entry.getKey(), entry.getValue(), nack); + } + confirmed.clear(); + } else { + OutstandingConfirmation confirmation = this.outstanding.remove(deliveryTag); + if (confirmation != null) { + complete(deliveryTag, confirmation, nack); + } + } + } + + private void complete(long seqNo, OutstandingConfirmation confirmation, boolean nack) { + confirmation.releaseSlot(); + if (nack) { + final PublishException e = new PublishException(seqNo, confirmation.context); + dispatch(() -> confirmation.completeExceptionally(e)); + } else { + dispatch(confirmation::complete); + } + } + + private void handleReturn(Return returned) { + Long seqNo = extractSequenceNumber(returned.getProperties()); + if (seqNo == null) { + // not published through this publisher, or header lost: the + // subsequent basic.ack still completes the future successfully + LOGGER.debug("Ignoring basic.return without a usable {} header", SEQUENCE_NUMBER_HEADER); + return; + } + final OutstandingConfirmation confirmation = this.outstanding.remove(seqNo); + if (confirmation == null) { + return; + } + confirmation.releaseSlot(); + final PublishException e = new PublishException(seqNo, returned, confirmation.context); + dispatch(() -> confirmation.completeExceptionally(e)); + } + + private static Long extractSequenceNumber(AMQP.BasicProperties props) { + if (props == null || props.getHeaders() == null) { + return null; + } + Object value = props.getHeaders().get(SEQUENCE_NUMBER_HEADER); + return value instanceof Long ? (Long) value : null; + } + + private void handleShutdown(ShutdownSignalException cause) { + failAllOutstanding(cause); + } + + /** + * Fails every outstanding future exceptionally and drains the map. Called + * both on channel shutdown and, for an autorecovering channel, at the start + * of recovery: recovery re-issues {@code confirm.select}, which resets the + * delivery-tag sequence to 1, so any entry that outlived the shutdown drain + * must be failed before the reused tag space could mis-correlate it. + */ + private void failAllOutstanding(Throwable cause) { + Map.Entry entry; + while ((entry = this.outstanding.pollFirstEntry()) != null) { + final OutstandingConfirmation confirmation = entry.getValue(); + confirmation.releaseSlot(); + dispatch(() -> confirmation.completeExceptionally(cause)); + } + } + + private void onRecoveryStarted() { + // The new delegate is already installed and open by this point, so the + // outcome of any still-outstanding publish is unknown; fail it before + // recovery resets the delivery-tag sequence. + failAllOutstanding(new IOException( + "connection recovery started; outcome of outstanding publishes is unknown")); + } + + /** + * Runs a completion task on the configured executor, falling back to the + * calling thread if the executor rejects it (for example a user-supplied + * executor that is saturated or shut down). Completing inline still runs off + * the future's own continuations but guarantees the future never hangs. + */ + private void dispatch(Runnable completion) { + try { + this.confirmationExecutor.execute(completion); + } catch (RejectedExecutionException e) { + LOGGER.warn("Confirmation executor rejected a completion task; " + + "completing on the calling thread", e); + completion.run(); + } + } + + private static final class OutstandingConfirmation { + private final CompletableFuture future; + private final Object context; + private final Semaphore limit; + private final AtomicBoolean slotReleased = new AtomicBoolean(false); + + private OutstandingConfirmation(CompletableFuture future, Object context, Semaphore limit) { + this.future = future; + this.context = context; + this.limit = limit; + } + + @SuppressWarnings("unchecked") + private void complete() { + ((CompletableFuture) this.future).complete(this.context); + } + + private void completeExceptionally(Throwable t) { + this.future.completeExceptionally(t); + } + + private void releaseSlot() { + if (this.limit != null && this.slotReleased.compareAndSet(false, true)) { + this.limit.release(); + } + } + } +} diff --git a/src/main/java/com/rabbitmq/client/Connection.java b/src/main/java/com/rabbitmq/client/Connection.java index 634d0ae7df..949bb71f92 100644 --- a/src/main/java/com/rabbitmq/client/Connection.java +++ b/src/main/java/com/rabbitmq/client/Connection.java @@ -201,23 +201,23 @@ default Optional openChannel(int channelNumber) throws IOException { * Close this connection and all its channels * with the {@link com.rabbitmq.client.AMQP#REPLY_SUCCESS} close code * and message 'OK'. - * + * * This method behaves in a similar way as {@link #close()}, with the only difference * that it waits with a provided timeout for all the close operations to * complete. When timeout is reached the socket is forced to close. - * + * * @param timeout timeout (in milliseconds) for completing all the close-related * operations, use -1 for infinity * @throws IOException if an I/O problem is encountered */ void close(int timeout) throws IOException; - + /** * Close this connection and all its channels. * * Waits with the given timeout for all the close operations to complete. * When timeout is reached the socket is forced to close. - * + * * @param closeCode the close code (See under "Reply Codes" in the AMQP specification) * @param closeMessage a message indicating the reason for closing the connection * @param timeout timeout (in milliseconds) for completing all the close-related @@ -235,18 +235,18 @@ default Optional openChannel(int channelNumber) throws IOException { * Any encountered exceptions in the close operations are silently discarded. */ void abort(); - + /** * Abort this connection and all its channels. * * Forces the connection to close and waits for all the close operations to complete. * Any encountered exceptions in the close operations are silently discarded. - * + * * @param closeCode the close code (See under "Reply Codes" in the AMQP specification) * @param closeMessage a message indicating the reason for closing the connection */ void abort(int closeCode, String closeMessage); - + /** * Abort this connection and all its channels * with the {@link com.rabbitmq.client.AMQP#REPLY_SUCCESS} close code diff --git a/src/main/java/com/rabbitmq/client/PublishException.java b/src/main/java/com/rabbitmq/client/PublishException.java new file mode 100644 index 0000000000..59c6ec56c3 --- /dev/null +++ b/src/main/java/com/rabbitmq/client/PublishException.java @@ -0,0 +1,95 @@ +// Copyright (c) 2007-2026 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries. +// +// This software, the RabbitMQ Java client library, is triple-licensed under the +// Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2 +// ("GPL") and the Apache License version 2 ("ASL"). For the MPL, please see +// LICENSE-MPL-RabbitMQ. For the GPL, please see LICENSE-GPL2. For the ASL, +// please see LICENSE-APACHE2. +// +// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND, +// either express or implied. See the LICENSE file for specific language governing +// rights and limitations of this software. +// +// If you have any questions regarding licensing, please contact us at +// info@rabbitmq.com. + +package com.rabbitmq.client; + +import java.io.IOException; + +/** + * Exception a {@link ConfirmationPublisher} future completes with when the + * broker rejects a message ({@code basic.nack}) or returns it as unroutable + * ({@code basic.return}). + *

+ * Use {@link #isReturn()} to distinguish the two cases. For a return, + * {@link #getReturned()} carries the full {@link Return} (reply code and text, + * exchange, routing key, properties, body); for a nack it is null. + * + * @see ConfirmationPublisher#basicPublishAsync(String, String, boolean, com.rabbitmq.client.AMQP.BasicProperties, byte[], Object) + */ +public class PublishException extends IOException { + + private final long sequenceNumber; + private final Return returned; + private final Object context; + + /** + * Constructor for the {@code basic.nack} case. + * + * @param sequenceNumber the publish sequence number + * @param context the user-provided context object + */ + public PublishException(long sequenceNumber, Object context) { + super(String.format("Message %d nack'd", sequenceNumber)); + this.sequenceNumber = sequenceNumber; + this.returned = null; + this.context = context; + } + + /** + * Constructor for the {@code basic.return} case. + * + * @param sequenceNumber the publish sequence number + * @param returned the broker's return + * @param context the user-provided context object + */ + public PublishException(long sequenceNumber, Return returned, Object context) { + super(String.format("Message %d returned: %s (%d)", sequenceNumber, + returned.getReplyText(), returned.getReplyCode())); + this.sequenceNumber = sequenceNumber; + this.returned = returned; + this.context = context; + } + + /** + * @return the publish sequence number of the failed message + */ + public long getSequenceNumber() { + return sequenceNumber; + } + + /** + * @return true if caused by {@code basic.return} (unroutable message), + * false if caused by {@code basic.nack} (broker rejection) + */ + public boolean isReturn() { + return returned != null; + } + + /** + * @return the broker's {@link Return} for the unroutable message, or null + * for a nack + */ + public Return getReturned() { + return returned; + } + + /** + * @return the user-provided context object passed to {@code basicPublishAsync}, + * or null if none was provided + */ + public Object getContext() { + return context; + } +} diff --git a/src/main/java/com/rabbitmq/client/impl/ChannelN.java b/src/main/java/com/rabbitmq/client/impl/ChannelN.java index ea2df2d717..ddf3c27ed3 100644 --- a/src/main/java/com/rabbitmq/client/impl/ChannelN.java +++ b/src/main/java/com/rabbitmq/client/impl/ChannelN.java @@ -118,6 +118,7 @@ public ChannelN(AMQConnection connection, int channelNumber, * @param channelNumber The channel number to be associated with this channel * @param workService service for managing this channel's consumer callbacks * @param metricsCollector service for managing metrics + * @param observationCollector service for managing observations */ public ChannelN(AMQConnection connection, int channelNumber, ConsumerWorkService workService, diff --git a/src/main/java/com/rabbitmq/client/impl/recovery/RecoveryAwareChannelN.java b/src/main/java/com/rabbitmq/client/impl/recovery/RecoveryAwareChannelN.java index 87e90f5f1b..c238e804a6 100644 --- a/src/main/java/com/rabbitmq/client/impl/recovery/RecoveryAwareChannelN.java +++ b/src/main/java/com/rabbitmq/client/impl/recovery/RecoveryAwareChannelN.java @@ -70,6 +70,7 @@ public RecoveryAwareChannelN(AMQConnection connection, int channelNumber, Consum * @param channelNumber The channel number to be associated with this channel * @param workService service for managing this channel's consumer callbacks * @param metricsCollector service for managing metrics + * @param observationCollector service for managing observations */ public RecoveryAwareChannelN(AMQConnection connection, int channelNumber, ConsumerWorkService workService, MetricsCollector metricsCollector, ObservationCollector observationCollector) { diff --git a/src/test/java/com/rabbitmq/client/test/functional/Confirm.java b/src/test/java/com/rabbitmq/client/test/functional/Confirm.java index 62392d6da9..745f1700e9 100644 --- a/src/test/java/com/rabbitmq/client/test/functional/Confirm.java +++ b/src/test/java/com/rabbitmq/client/test/functional/Confirm.java @@ -23,19 +23,26 @@ import com.rabbitmq.client.AMQP; +import com.rabbitmq.client.AlreadyClosedException; import com.rabbitmq.client.Channel; +import com.rabbitmq.client.ConfirmationPublisher; import com.rabbitmq.client.ConfirmListener; import com.rabbitmq.client.DefaultConsumer; import com.rabbitmq.client.GetResponse; import com.rabbitmq.client.MessageProperties; +import com.rabbitmq.client.PublishException; import com.rabbitmq.client.ShutdownSignalException; import com.rabbitmq.client.test.BrokerTestCase; import java.io.IOException; +import java.util.ArrayList; import java.util.Collections; +import java.util.List; import java.util.Map; import java.util.SortedSet; import java.util.TreeSet; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; import java.util.concurrent.TimeoutException; import org.junit.jupiter.api.TestInfo; @@ -316,4 +323,185 @@ protected void publish(String exchangeName, String queueName, : MessageProperties.BASIC, "nop".getBytes()); } + + @Test public void publisherBasicConfirmation() throws Exception { + ConfirmationPublisher publisher = ConfirmationPublisher.create(connection); + String queue = channel.queueDeclare().getQueue(); + + int messageCount = 100; + List> futures = new ArrayList>(); + + for (int i = 0; i < messageCount; i++) { + futures.add(publisher.basicPublishAsync("", queue, null, ("msg" + i).getBytes(), i)); + } + + for (int i = 0; i < messageCount; i++) { + assertEquals(Integer.valueOf(i), futures.get(i).join()); + } + + assertEquals(messageCount, channel.messageCount(queue)); + publisher.close(); + } + + @Test public void publisherMandatoryReturn() throws Exception { + ConfirmationPublisher publisher = ConfirmationPublisher.create(connection); + + CompletableFuture future = publisher.basicPublishAsync( + "", "nonexistent-queue", true, null, "test".getBytes(), null); + + try { + future.join(); + fail("Expected PublishException"); + } catch (CompletionException e) { + assertTrue(e.getCause() instanceof PublishException); + PublishException pe = (PublishException) e.getCause(); + assertTrue(pe.isReturn()); + assertEquals(AMQP.NO_ROUTE, pe.getReturned().getReplyCode()); + } + + publisher.close(); + } + + @Test public void publisherMaxOutstanding() throws Exception { + ConfirmationPublisher publisher = ConfirmationPublisher.create(connection, 10); + String queue = channel.queueDeclare().getQueue(); + + List> futures = new ArrayList>(); + for (int i = 0; i < 50; i++) { + futures.add(publisher.basicPublishAsync("", queue, null, ("msg" + i).getBytes(), i)); + } + + for (int i = 0; i < 50; i++) { + assertEquals(Integer.valueOf(i), futures.get(i).join()); + } + + assertEquals(50, channel.messageCount(queue)); + publisher.close(); + } + + @Test public void publisherCloseFailsOutstanding() throws Exception { + ConfirmationPublisher publisher = ConfirmationPublisher.create(connection); + String queue = channel.queueDeclare().getQueue(); + + List> futures = new ArrayList>(); + for (int i = 0; i < 10; i++) { + futures.add(publisher.basicPublishAsync("", queue, null, ("msg" + i).getBytes(), null)); + } + + publisher.close(); + + // Every future must settle after close: a graceful close completes + // confirmations that already arrived normally, and fails the rest with + // a ShutdownSignalException. None may hang. + for (CompletableFuture future : futures) { + try { + future.get(60, java.util.concurrent.TimeUnit.SECONDS); + } catch (java.util.concurrent.ExecutionException e) { + assertTrue(e.getCause() instanceof ShutdownSignalException); + } + } + } + + @Test public void publisherContextCorrelation() throws Exception { + ConfirmationPublisher publisher = ConfirmationPublisher.create(connection); + String queue = channel.queueDeclare().getQueue(); + + int messageCount = 10; + List> futures = new ArrayList>(); + + for (int i = 0; i < messageCount; i++) { + String correlationId = "msg-" + i; + futures.add(publisher.basicPublishAsync("", queue, null, + ("message" + i).getBytes(), correlationId)); + } + + for (int i = 0; i < messageCount; i++) { + assertEquals("msg-" + i, futures.get(i).join()); + } + + assertEquals(messageCount, channel.messageCount(queue)); + publisher.close(); + } + + @Test public void publisherContextInException() throws Exception { + ConfirmationPublisher publisher = ConfirmationPublisher.create(connection); + + String messageId = "unroutable-msg-456"; + CompletableFuture future = publisher.basicPublishAsync( + "", "nonexistent-queue", true, null, "test".getBytes(), messageId); + + try { + future.join(); + fail("Expected PublishException"); + } catch (CompletionException e) { + assertTrue(e.getCause() instanceof PublishException); + PublishException pe = (PublishException) e.getCause(); + assertTrue(pe.isReturn()); + assertEquals(AMQP.NO_ROUTE, pe.getReturned().getReplyCode()); + assertEquals(messageId, pe.getContext()); + } + + publisher.close(); + } + + @Test public void publisherSurvivesConnectionRecovery() throws Exception { + ConfirmationPublisher publisher = ConfirmationPublisher.create(connection); + String queue = channel.queueDeclare().getQueue(); + + for (int i = 0; i < 10; i++) { + publisher.basicPublishAsync("", queue, null, ("before" + i).getBytes(), i).join(); + } + + com.rabbitmq.client.test.TestUtils.closeAndWaitForRecovery( + (com.rabbitmq.client.RecoverableConnection) connection); + + int confirmed = 0; + for (int i = 0; i < 10; i++) { + publisher.basicPublishAsync("", queue, null, ("after" + i).getBytes(), i).join(); + confirmed++; + } + + assertEquals(10, confirmed); + publisher.close(); + } + + @Test public void publisherCompletesWhenExecutorRejects() throws Exception { + // An executor that rejects every task: confirmations must still complete + // (inline, on the connection thread) rather than hang forever. + java.util.concurrent.Executor rejecting = command -> { + throw new java.util.concurrent.RejectedExecutionException("always rejects"); + }; + ConfirmationPublisher publisher = ConfirmationPublisher.create(connection, 100, rejecting); + String queue = channel.queueDeclare().getQueue(); + + int messageCount = 20; + List> futures = new ArrayList>(); + for (int i = 0; i < messageCount; i++) { + futures.add(publisher.basicPublishAsync("", queue, null, ("msg" + i).getBytes(), i)); + } + + for (int i = 0; i < messageCount; i++) { + assertEquals(Integer.valueOf(i), + futures.get(i).get(60, java.util.concurrent.TimeUnit.SECONDS)); + } + + publisher.close(); + } + + @Test public void publisherUnlimited() throws Exception { + ConfirmationPublisher publisher = ConfirmationPublisher.create(connection); + String queue = channel.queueDeclare().getQueue(); + + int messageCount = 1000; + List> futures = new ArrayList>(); + + for (int i = 0; i < messageCount; i++) { + futures.add(publisher.basicPublishAsync("", queue, null, ("msg" + i).getBytes(), null)); + } + + CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join(); + + assertEquals(messageCount, channel.messageCount(queue)); + publisher.close(); + } }