From 41479947606d427ee7cd40d89cf82c5aed216880 Mon Sep 17 00:00:00 2001 From: Eldert Grootenboer Date: Tue, 1 Sep 2026 17:53:47 -0700 Subject: [PATCH] [Service Bus] Add batch delete and purge APIs --- .../azure-messaging-servicebus/CHANGELOG.md | 1 + .../ServiceBusReceiverAsyncClient.java | 118 +++++++++++++++ .../servicebus/ServiceBusReceiverClient.java | 67 +++++++++ .../implementation/ManagementChannel.java | 59 ++++++++ .../implementation/ManagementConstants.java | 2 + .../ServiceBusManagementNode.java | 14 ++ .../models/DeleteMessagesOptions.java | 39 +++++ .../models/DeleteMessagesResult.java | 29 ++++ .../models/PurgeMessagesOptions.java | 66 +++++++++ .../models/PurgeMessagesResult.java | 29 ++++ ...ceBusReceiverClientJavaDocCodeSamples.java | 98 ++++++++++++ .../ServiceBusReceiverAsyncClientTest.java | 139 ++++++++++++++++++ .../ManagementChannelTests.java | 136 +++++++++++++++++ 13 files changed, 797 insertions(+) create mode 100644 sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/models/DeleteMessagesOptions.java create mode 100644 sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/models/DeleteMessagesResult.java create mode 100644 sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/models/PurgeMessagesOptions.java create mode 100644 sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/models/PurgeMessagesResult.java diff --git a/sdk/servicebus/azure-messaging-servicebus/CHANGELOG.md b/sdk/servicebus/azure-messaging-servicebus/CHANGELOG.md index 2272155fb763..6157d8034bfc 100644 --- a/sdk/servicebus/azure-messaging-servicebus/CHANGELOG.md +++ b/sdk/servicebus/azure-messaging-servicebus/CHANGELOG.md @@ -4,6 +4,7 @@ ### Features Added +- Added batch delete and client-driven purge operations to `ServiceBusReceiverAsyncClient` and `ServiceBusReceiverClient`. Basic and Standard support up to 500 messages per request, Premium supports up to 4,000, and purge handles smaller batches caused by large messages. - Added `listSessions()` and `listSessions(OffsetDateTime sessionStateUpdatedAfter)` to `ServiceBusSessionReceiverAsyncClient` (returning `PagedFlux`) and `ServiceBusSessionReceiverClient` (returning `PagedIterable`). The no-arg overload returns sessions with active messages; the `sessionStateUpdatedAfter` overload returns sessions whose session state was updated after the given timestamp. Implements the `com.microsoft:get-message-sessions` AMQP management operation. ([#48956](https://github.com/Azure/azure-sdk-for-java/pull/48956)) - Added `getSqlFilterCount()` and `getCorrelationFilterCount()` to `TopicRuntimeProperties`, exposing the total number of SQL filters and correlation filters across all of a topic's subscriptions. - Added `ServiceBusServiceVersion.V2024_05` and made it the latest version. The administration client now uses `api-version=2024-05` by default, which is required for the topic filter counts above. diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClient.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClient.java index 0b3154ec33ef..6bfba49ed3cd 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClient.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClient.java @@ -31,6 +31,10 @@ import com.azure.messaging.servicebus.models.CompleteOptions; import com.azure.messaging.servicebus.models.DeadLetterOptions; import com.azure.messaging.servicebus.models.DeferOptions; +import com.azure.messaging.servicebus.models.DeleteMessagesOptions; +import com.azure.messaging.servicebus.models.DeleteMessagesResult; +import com.azure.messaging.servicebus.models.PurgeMessagesOptions; +import com.azure.messaging.servicebus.models.PurgeMessagesResult; import com.azure.messaging.servicebus.models.ServiceBusReceiveMode; import reactor.core.Disposable; import reactor.core.Disposables; @@ -906,6 +910,120 @@ Flux peekMessages(int maxMessages, long sequenceNumbe .onErrorMap(throwable -> mapError(throwable, ServiceBusErrorSource.RECEIVE))); } + /** + * Permanently deletes up to {@code maxMessages} eligible messages from the Service Bus entity or subqueue. + * Large messages can cause the service to delete fewer messages than requested. Locked, deferred, and scheduled + * messages are not eligible. Currently, batch delete is not supported when partitioning is enabled. + * + *

The SDK sends the destructive request once. If an error, cancellation, or timeout occurs, the deletion outcome + * is unknown and the request is not automatically dispatched again.

+ * + * @param maxMessages The positive maximum number of messages to delete. The service limit is 500 for Basic and + * Standard and 4,000 for Premium. + * @return The result containing the number of messages actually deleted by the service. + * @throws IllegalArgumentException if {@code maxMessages} is not positive. + * @throws IllegalStateException if the receiver is already disposed. + * @throws ServiceBusException if the request fails. + */ + public Mono deleteMessages(int maxMessages) { + return deleteMessages(maxMessages, new DeleteMessagesOptions()); + } + + /** + * Permanently deletes up to {@code maxMessages} eligible messages from the Service Bus entity or subqueue. + * The operation is best effort and may return a short positive count. Locked, deferred, and scheduled messages are + * not eligible. A dispatched request is not automatically retried and an error can leave an unknown outcome. + * + * @param maxMessages The positive maximum number of messages to delete. The service limit is 500 for Basic and + * Standard and 4,000 for Premium. + * @param options Options that configure the delete operation. + * @return The result containing the number of messages actually deleted by the service. + * @throws NullPointerException if {@code options} is null. + * @throws IllegalArgumentException if {@code maxMessages} is not positive. + * @throws IllegalStateException if the receiver is already disposed. + * @throws ServiceBusException if the request fails. + */ + public Mono deleteMessages(int maxMessages, DeleteMessagesOptions options) { + if (isDisposed.get()) { + return monoError(LOGGER, + new IllegalStateException(String.format(INVALID_OPERATION_DISPOSED_RECEIVER, "deleteMessages"))); + } + if (maxMessages < 1) { + return monoError(LOGGER, + new IllegalArgumentException("'maxMessages' must be positive; got " + maxMessages + ".")); + } + if (options == null) { + return monoError(LOGGER, new NullPointerException("'options' cannot be null.")); + } + + final OffsetDateTime configuredCutoff = options.getEnqueueTimeUtcOlderThan(); + final String sessionId = receiverOptions.getSessionId(); + return Mono.defer(() -> { + final OffsetDateTime cutoff = configuredCutoff == null ? OffsetDateTime.now() : configuredCutoff; + return tracer.traceMono("ServiceBus.deleteMessages", + connectionProcessor.flatMap(connection -> connection.getManagementNode(entityPath, entityType)) + .flatMap(node -> node.deleteMessages(maxMessages, cutoff, sessionId, getLinkName(sessionId))) + .onErrorMap(throwable -> mapError(throwable, ServiceBusErrorSource.RECEIVE))); + }); + } + + /** + * Permanently purges eligible messages enqueued before the purge started. The purge start time stays unchanged + * for every request, so newer messages remain. Large messages can produce smaller batches, which purge continues + * processing. Locked, deferred, and scheduled messages remain. Currently, purge is not supported when partitioning + * is enabled. + * If an error, cancellation, or timeout occurs after dispatch, the purge can be partial and its exact deletion + * outcome is unknown. + * + * @return The result containing the total number of messages deleted by the service. + * @throws IllegalStateException if the receiver is already disposed. + * @throws ServiceBusException if any request fails. + */ + public Mono purgeMessages() { + return purgeMessages(new PurgeMessagesOptions()); + } + + /** + * Permanently purges eligible messages enqueued before the configured time, using that same time and request size + * for every request. Large messages can produce smaller batches, which purge continues processing. Locked, + * deferred, and scheduled messages remain. Currently, purge is not supported when partitioning is enabled. + * If an error, cancellation, or timeout occurs after dispatch, the purge can be partial and its exact deletion + * outcome is unknown. + * + * @param options Options that configure the purge operation. + * @return The result containing the total number of messages deleted by the service. + * @throws NullPointerException if {@code options} is null. + * @throws IllegalStateException if the receiver is already disposed. + * @throws ServiceBusException if any request fails. + */ + public Mono purgeMessages(PurgeMessagesOptions options) { + if (isDisposed.get()) { + return monoError(LOGGER, + new IllegalStateException(String.format(INVALID_OPERATION_DISPOSED_RECEIVER, "purgeMessages"))); + } + if (options == null) { + return monoError(LOGGER, new NullPointerException("'options' cannot be null.")); + } + + final OffsetDateTime configuredCutoff = options.getEnqueueTimeUtcOlderThan(); + final int maxMessagesPerBatch = options.getMaxMessagesPerBatch(); + return Mono.defer(() -> { + final OffsetDateTime cutoff = configuredCutoff == null ? OffsetDateTime.now() : configuredCutoff; + return purgeMessages(cutoff, maxMessagesPerBatch, 0); + }); + } + + private Mono purgeMessages(OffsetDateTime cutoff, int maxMessagesPerBatch, long deletedCount) { + return deleteMessages(maxMessagesPerBatch, new DeleteMessagesOptions().setEnqueueTimeUtcOlderThan(cutoff)) + .flatMap(result -> { + if (result.getDeletedCount() == 0) { + return Mono.just(new PurgeMessagesResult(deletedCount)); + } + return purgeMessages(cutoff, maxMessagesPerBatch, + Math.addExact(deletedCount, result.getDeletedCount())); + }); + } + /** * Receives an infinite stream of {@link ServiceBusReceivedMessage messages} from the Service Bus entity. * This Flux continuously receives messages from a Service Bus entity until either: diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusReceiverClient.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusReceiverClient.java index 6edf07d2f898..31b7133e58fb 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusReceiverClient.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusReceiverClient.java @@ -13,6 +13,10 @@ import com.azure.messaging.servicebus.models.CompleteOptions; import com.azure.messaging.servicebus.models.DeadLetterOptions; import com.azure.messaging.servicebus.models.DeferOptions; +import com.azure.messaging.servicebus.models.DeleteMessagesOptions; +import com.azure.messaging.servicebus.models.DeleteMessagesResult; +import com.azure.messaging.servicebus.models.PurgeMessagesOptions; +import com.azure.messaging.servicebus.models.PurgeMessagesResult; import com.azure.messaging.servicebus.models.ServiceBusReceiveMode; import reactor.core.publisher.Flux; import reactor.core.publisher.Sinks; @@ -332,6 +336,69 @@ public byte[] getSessionState() { return asyncClient.getSessionState().block(operationTimeout); } + /** + * Deletes up to {@code maxMessages} messages from the Service Bus entity. + * The SDK sends the destructive request once and surfaces any failure without automatically dispatching it again. + * + * @param maxMessages The maximum number of messages to delete. The service limit is 500 for Basic and Standard and + * 4,000 for Premium. Currently, batch delete is not supported when partitioning is enabled. + * @return The result containing the number of messages actually deleted by the service. + * @throws IllegalArgumentException if {@code maxMessages} is not positive. + * @throws IllegalStateException if the receiver is already disposed. + * @throws ServiceBusException if the request fails. + */ + public DeleteMessagesResult deleteMessages(int maxMessages) { + return asyncClient.deleteMessages(maxMessages).block(operationTimeout); + } + + /** + * Deletes up to {@code maxMessages} messages from the Service Bus entity. + * The SDK sends the destructive request once and surfaces any failure without automatically dispatching it again. + * + * @param maxMessages The maximum number of messages to delete. The service limit is 500 for Basic and Standard and + * 4,000 for Premium. Currently, batch delete is not supported when partitioning is enabled. + * @param options Options that configure the delete operation. + * @return The result containing the number of messages actually deleted by the service. + * @throws NullPointerException if {@code options} is null. + * @throws IllegalArgumentException if {@code maxMessages} is not positive. + * @throws IllegalStateException if the receiver is already disposed. + * @throws ServiceBusException if the request fails. + */ + public DeleteMessagesResult deleteMessages(int maxMessages, DeleteMessagesOptions options) { + return asyncClient.deleteMessages(maxMessages, options).block(operationTimeout); + } + + /** + * Purges messages enqueued before the purge started. The purge start time stays unchanged for every request, so + * newer messages remain. Large messages can produce smaller batches, which purge continues processing. + * Currently, purge is not supported when partitioning is enabled. + * If an error, cancellation, or timeout occurs after dispatch, the purge can be partial and its exact deletion + * outcome is unknown. + * + * @return The result containing the total number of messages deleted by the service. + * @throws IllegalStateException if the receiver is already disposed. + * @throws ServiceBusException if any request fails. + */ + public PurgeMessagesResult purgeMessages() { + return asyncClient.purgeMessages().block(operationTimeout); + } + + /** + * Purges messages enqueued before the configured time. That time stays unchanged for every request, so newer + * messages remain. Large messages can produce smaller batches, which purge continues processing. Currently, + * purge is not supported when partitioning is enabled. If an error, cancellation, or timeout occurs after + * dispatch, the purge can be partial and its exact deletion outcome is unknown. + * + * @param options Options that configure the purge operation. + * @return The result containing the total number of messages deleted by the service. + * @throws NullPointerException if {@code options} is null. + * @throws IllegalStateException if the receiver is already disposed. + * @throws ServiceBusException if any request fails. + */ + public PurgeMessagesResult purgeMessages(PurgeMessagesOptions options) { + return asyncClient.purgeMessages(options).block(operationTimeout); + } + /** * Reads the next active message without changing the state of the receiver or the message source. The first call to * {@code peekMessage()} fetches the first active message for this receiver. Each subsequent call fetches the diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ManagementChannel.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ManagementChannel.java index 9331097ec5d7..f92182422bae 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ManagementChannel.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ManagementChannel.java @@ -23,6 +23,7 @@ import com.azure.messaging.servicebus.ServiceBusTransactionContext; import com.azure.messaging.servicebus.administration.models.CreateRuleOptions; import com.azure.messaging.servicebus.administration.models.RuleProperties; +import com.azure.messaging.servicebus.models.DeleteMessagesResult; import com.azure.messaging.servicebus.models.ServiceBusReceiveMode; import org.apache.qpid.proton.Proton; import org.apache.qpid.proton.amqp.Binary; @@ -57,6 +58,7 @@ import static com.azure.core.util.FluxUtil.fluxError; import static com.azure.core.util.FluxUtil.monoError; import static com.azure.messaging.servicebus.implementation.ManagementConstants.OPERATION_ADD_RULE; +import static com.azure.messaging.servicebus.implementation.ManagementConstants.OPERATION_BATCH_DELETE_MESSAGES; import static com.azure.messaging.servicebus.implementation.ManagementConstants.OPERATION_GET_MESSAGE_SESSIONS; import static com.azure.messaging.servicebus.implementation.ManagementConstants.OPERATION_GET_RULES; import static com.azure.messaging.servicebus.implementation.ManagementConstants.OPERATION_GET_SESSION_STATE; @@ -470,6 +472,63 @@ public Mono updateDisposition(String lockToken, DispositionStatus disposit })).then(); } + /** + * {@inheritDoc} + */ + @Override + public Mono deleteMessages(int maxMessages, OffsetDateTime enqueueTimeUtcOlderThan, + String sessionId, String associatedLinkName) { + if (maxMessages < 1) { + return monoError(logger, + new IllegalArgumentException("'maxMessages' must be positive; got " + maxMessages + ".")); + } + if (enqueueTimeUtcOlderThan == null) { + return monoError(logger, new NullPointerException("'enqueueTimeUtcOlderThan' cannot be null.")); + } + + return isAuthorized(OPERATION_BATCH_DELETE_MESSAGES).then(channelCache.get().flatMap(channel -> { + final Message message = createManagementMessage(OPERATION_BATCH_DELETE_MESSAGES, associatedLinkName); + final Map body = new HashMap<>(); + body.put(ManagementConstants.MESSAGE_COUNT_KEY, maxMessages); + body.put(ManagementConstants.ENQUEUED_TIME_UTC, Date.from(enqueueTimeUtcOlderThan.toInstant())); + if (!CoreUtils.isNullOrEmpty(sessionId)) { + body.put(ManagementConstants.SESSION_ID, sessionId); + } + message.setBody(new AmqpValue(body)); + + return sendWithVerify(channel, message, null); + })).flatMap(response -> { + final AmqpResponseCode statusCode = RequestResponseUtils.getStatusCode(response); + if (statusCode == AmqpResponseCode.NO_CONTENT) { + return Mono.just(new DeleteMessagesResult(0)); + } + final String errorCondition = RequestResponseUtils.getErrorCondition(response); + final boolean messageNotFound = statusCode == AmqpResponseCode.NOT_FOUND + && AmqpErrorCondition.MESSAGE_NOT_FOUND.getErrorCondition().equals(errorCondition); + if (statusCode != AmqpResponseCode.OK && !messageNotFound) { + return monoError(logger, + new IllegalStateException("Batch delete returned unexpected status code: " + statusCode + ".")); + } + + final Object responseBody = response.getBody(); + if (!(responseBody instanceof AmqpValue) || !(((AmqpValue) responseBody).getValue() instanceof Map)) { + return monoError(logger, + new IllegalStateException("Batch delete returned a successful response with an invalid body.")); + } + + @SuppressWarnings("unchecked") + final Map body = (Map) ((AmqpValue) responseBody).getValue(); + final Object deletedCount = body.get(ManagementConstants.MESSAGE_COUNT_KEY); + if (!(deletedCount instanceof Integer) + || (Integer) deletedCount < 0 + || (Integer) deletedCount > maxMessages) { + return monoError(logger, + new IllegalStateException("Batch delete response did not contain a valid message-count.")); + } + return Mono.just(new DeleteMessagesResult((Integer) deletedCount)); + }); + } + /** * {@inheritDoc} */ diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ManagementConstants.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ManagementConstants.java index 728e9efb8a16..da387d5cf072 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ManagementConstants.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ManagementConstants.java @@ -20,6 +20,7 @@ public class ManagementConstants { public static final String LOCK_TOKEN_KEY = "lock-token"; public static final String LOCK_TOKENS_KEY = "lock-tokens"; public static final String MESSAGE_COUNT_KEY = "message-count"; + public static final String ENQUEUED_TIME_UTC = "enqueued-time-utc"; public static final String MESSAGE = "message"; public static final String MESSAGES = "messages"; public static final String MESSAGE_ID = "message-id"; @@ -83,6 +84,7 @@ public class ManagementConstants { static final String OPERATION_SET_SESSION_STATE = AmqpConstants.VENDOR + ":set-session-state"; static final String OPERATION_UPDATE_DISPOSITION = AmqpConstants.VENDOR + ":update-disposition"; static final String OPERATION_ADD_RULE = AmqpConstants.VENDOR + ":add-rule"; + static final String OPERATION_BATCH_DELETE_MESSAGES = AmqpConstants.VENDOR + ":batch-delete-messages"; static final String OPERATION_REMOVE_RULE = AmqpConstants.VENDOR + ":remove-rule"; static final String OPERATION_GET_RULES = AmqpConstants.VENDOR + ":enumerate-rules"; static final String OPERATION_GET_MESSAGE_SESSIONS = AmqpConstants.VENDOR + ":get-message-sessions"; diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ServiceBusManagementNode.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ServiceBusManagementNode.java index e05facd030f0..6f9044990858 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ServiceBusManagementNode.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ServiceBusManagementNode.java @@ -8,6 +8,7 @@ import com.azure.messaging.servicebus.ServiceBusTransactionContext; import com.azure.messaging.servicebus.administration.models.CreateRuleOptions; import com.azure.messaging.servicebus.administration.models.RuleProperties; +import com.azure.messaging.servicebus.models.DeleteMessagesResult; import com.azure.messaging.servicebus.models.ServiceBusReceiveMode; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -124,6 +125,19 @@ Mono updateDisposition(String lockToken, DispositionStatus dispositionStat String deadLetterErrorDescription, Map propertiesToModify, String sessionId, String associatedLinkName, ServiceBusTransactionContext transactionContext); + /** + * Deletes up to {@code maxMessages} messages enqueued before the given cutoff. + * + * @param maxMessages The maximum number of messages to delete. The service limit is 500 for Basic and Standard + * and 4,000 for Premium. + * @param enqueueTimeUtcOlderThan Only messages enqueued before this time are deleted. + * @param sessionId The session identifier, or {@code null} for a non-session entity. + * @param associatedLinkName The associated receive-link name, or {@code null} if no link is open. + * @return The number of messages actually deleted by the service. + */ + Mono deleteMessages(int maxMessages, OffsetDateTime enqueueTimeUtcOlderThan, String sessionId, + String associatedLinkName); + /** * Create a rule with the {@link CreateRuleOptions} for Service Bus subscription. * diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/models/DeleteMessagesOptions.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/models/DeleteMessagesOptions.java new file mode 100644 index 000000000000..7e9736ed5b6e --- /dev/null +++ b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/models/DeleteMessagesOptions.java @@ -0,0 +1,39 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.messaging.servicebus.models; + +import java.time.OffsetDateTime; + +/** + * Options for permanently deleting eligible messages from a Service Bus entity or subqueue. + */ +public final class DeleteMessagesOptions { + private OffsetDateTime enqueueTimeUtcOlderThan; + + /** + * Creates an instance of {@link DeleteMessagesOptions}. + */ + public DeleteMessagesOptions() { + } + + /** + * Gets the enqueue-time cutoff. Only messages enqueued before this time are eligible for deletion. + * + * @return The enqueue-time cutoff, or {@code null} to use the time the operation starts. + */ + public OffsetDateTime getEnqueueTimeUtcOlderThan() { + return enqueueTimeUtcOlderThan; + } + + /** + * Sets the enqueue-time cutoff. Only messages enqueued before this time are eligible for deletion. + * + * @param enqueueTimeUtcOlderThan The enqueue-time cutoff. + * @return The updated {@link DeleteMessagesOptions}. + */ + public DeleteMessagesOptions setEnqueueTimeUtcOlderThan(OffsetDateTime enqueueTimeUtcOlderThan) { + this.enqueueTimeUtcOlderThan = enqueueTimeUtcOlderThan; + return this; + } +} diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/models/DeleteMessagesResult.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/models/DeleteMessagesResult.java new file mode 100644 index 000000000000..4ac8b491986b --- /dev/null +++ b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/models/DeleteMessagesResult.java @@ -0,0 +1,29 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.messaging.servicebus.models; + +/** + * The result of deleting a batch of messages from a Service Bus entity. + */ +public final class DeleteMessagesResult { + private final long deletedCount; + + /** + * Creates an instance of {@link DeleteMessagesResult}. + * + * @param deletedCount The number of messages deleted by the service. + */ + public DeleteMessagesResult(long deletedCount) { + this.deletedCount = deletedCount; + } + + /** + * Gets the number of messages deleted by the service. + * + * @return The number of deleted messages. + */ + public long getDeletedCount() { + return deletedCount; + } +} diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/models/PurgeMessagesOptions.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/models/PurgeMessagesOptions.java new file mode 100644 index 000000000000..b2761dc6d7d3 --- /dev/null +++ b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/models/PurgeMessagesOptions.java @@ -0,0 +1,66 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.messaging.servicebus.models; + +import java.time.OffsetDateTime; + +/** + * Options for permanently purging eligible messages from a Service Bus entity or subqueue. + */ +public final class PurgeMessagesOptions { + private OffsetDateTime enqueueTimeUtcOlderThan; + private int maxMessagesPerBatch = 500; + + /** + * Creates an instance of {@link PurgeMessagesOptions}. + */ + public PurgeMessagesOptions() { + } + + /** + * Gets the enqueue-time threshold that stays unchanged for every purge request. + * + * @return The enqueue-time threshold, or {@code null} to use the time the purge starts. + */ + public OffsetDateTime getEnqueueTimeUtcOlderThan() { + return enqueueTimeUtcOlderThan; + } + + /** + * Sets the enqueue-time threshold. Only messages enqueued before this time can be deleted, and the value stays + * unchanged for every purge request. + * + * @param enqueueTimeUtcOlderThan The enqueue-time threshold. + * @return The updated {@link PurgeMessagesOptions}. + */ + public PurgeMessagesOptions setEnqueueTimeUtcOlderThan(OffsetDateTime enqueueTimeUtcOlderThan) { + this.enqueueTimeUtcOlderThan = enqueueTimeUtcOlderThan; + return this; + } + + /** + * Gets the maximum number of messages requested in each batch-delete call. + * + * @return The maximum messages per batch. The default is 500. + */ + public int getMaxMessagesPerBatch() { + return maxMessagesPerBatch; + } + + /** + * Sets the maximum number of messages requested in each batch-delete call. The service limit is 500 for Basic and + * Standard and 4,000 for Premium. + * + * @param maxMessagesPerBatch The positive maximum number of messages per batch. + * @return The updated {@link PurgeMessagesOptions}. + * @throws IllegalArgumentException if {@code maxMessagesPerBatch} is less than one. + */ + public PurgeMessagesOptions setMaxMessagesPerBatch(int maxMessagesPerBatch) { + if (maxMessagesPerBatch < 1) { + throw new IllegalArgumentException("'maxMessagesPerBatch' must be greater than 0."); + } + this.maxMessagesPerBatch = maxMessagesPerBatch; + return this; + } +} diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/models/PurgeMessagesResult.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/models/PurgeMessagesResult.java new file mode 100644 index 000000000000..e9e38a14261e --- /dev/null +++ b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/models/PurgeMessagesResult.java @@ -0,0 +1,29 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.messaging.servicebus.models; + +/** + * The result of purging messages from a Service Bus entity. + */ +public final class PurgeMessagesResult { + private final long deletedCount; + + /** + * Creates an instance of {@link PurgeMessagesResult}. + * + * @param deletedCount The total number of messages deleted by the service. + */ + public PurgeMessagesResult(long deletedCount) { + this.deletedCount = deletedCount; + } + + /** + * Gets the total number of messages deleted by the service. + * + * @return The total number of deleted messages. + */ + public long getDeletedCount() { + return deletedCount; + } +} diff --git a/sdk/servicebus/azure-messaging-servicebus/src/samples/java/com/azure/messaging/servicebus/ServiceBusReceiverClientJavaDocCodeSamples.java b/sdk/servicebus/azure-messaging-servicebus/src/samples/java/com/azure/messaging/servicebus/ServiceBusReceiverClientJavaDocCodeSamples.java index d5a1a2ee300f..6e85ef9f8e77 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/samples/java/com/azure/messaging/servicebus/ServiceBusReceiverClientJavaDocCodeSamples.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/samples/java/com/azure/messaging/servicebus/ServiceBusReceiverClientJavaDocCodeSamples.java @@ -9,6 +9,9 @@ import com.azure.identity.DefaultAzureCredentialBuilder; import com.azure.messaging.servicebus.models.AbandonOptions; import com.azure.messaging.servicebus.models.CompleteOptions; +import com.azure.messaging.servicebus.models.DeleteMessagesResult; +import com.azure.messaging.servicebus.models.PurgeMessagesOptions; +import com.azure.messaging.servicebus.models.PurgeMessagesResult; import com.azure.messaging.servicebus.models.ServiceBusReceiveMode; import com.azure.messaging.servicebus.models.SubQueue; import org.junit.jupiter.api.Test; @@ -20,6 +23,7 @@ import java.security.SecureRandom; import java.time.Duration; +import java.time.OffsetDateTime; import java.util.Iterator; import java.util.concurrent.atomic.AtomicInteger; @@ -597,6 +601,100 @@ public void transactionsSnippetAsync() { } } + /** + * Deletes a batch and then purges the remaining messages synchronously. + */ + @Test + public void deleteAndPurgeMessages() { + ServiceBusReceiverClient receiver = new ServiceBusClientBuilder() + .credential(fullyQualifiedNamespace, new DefaultAzureCredentialBuilder().build()) + .receiver() + .queueName(queueName) + .buildClient(); + + // BEGIN: com.azure.messaging.servicebus.servicebusreceiverclient.deleteAndPurgeMessages + int requestedCount = 100; + DeleteMessagesResult deleteResult = receiver.deleteMessages(requestedCount); + // Any request can delete fewer messages than requested, especially when messages are large. + System.out.printf("Requested %d; the service deleted %d.%n", requestedCount, deleteResult.getDeletedCount()); + + PurgeMessagesResult purgeResult = receiver.purgeMessages(); + System.out.printf("The service purged %d remaining messages.%n", purgeResult.getDeletedCount()); + // END: com.azure.messaging.servicebus.servicebusreceiverclient.deleteAndPurgeMessages + + receiver.close(); + } + + /** + * Purges messages with a fixed cutoff and a Premium batch size. + */ + @Test + public void purgeMessagesWithPremiumBatchSize() { + ServiceBusReceiverClient receiver = new ServiceBusClientBuilder() + .credential(fullyQualifiedNamespace, new DefaultAzureCredentialBuilder().build()) + .receiver() + .queueName(queueName) + .buildClient(); + + // BEGIN: com.azure.messaging.servicebus.servicebusreceiverclient.purgeMessagesWithPremiumBatchSize + OffsetDateTime enqueueTimeThreshold = OffsetDateTime.now(); + PurgeMessagesOptions options = new PurgeMessagesOptions() + .setEnqueueTimeUtcOlderThan(enqueueTimeThreshold) + // Premium supports up to 4,000 messages per request. + .setMaxMessagesPerBatch(4000); + + PurgeMessagesResult result = receiver.purgeMessages(options); + System.out.printf("Purged %d messages enqueued before %s.%n", result.getDeletedCount(), enqueueTimeThreshold); + // END: com.azure.messaging.servicebus.servicebusreceiverclient.purgeMessagesWithPremiumBatchSize + + receiver.close(); + } + + /** + * Purges messages from one named session. + */ + @Test + public void purgeMessagesFromSession() { + String sessionId = "session-1"; + ServiceBusSessionReceiverClient sessionClient = new ServiceBusClientBuilder() + .credential(fullyQualifiedNamespace, new DefaultAzureCredentialBuilder().build()) + .sessionReceiver() + .queueName(sessionEnabledQueueName) + .buildClient(); + ServiceBusReceiverClient sessionReceiver = sessionClient.acceptSession(sessionId); + + // BEGIN: com.azure.messaging.servicebus.servicebusreceiverclient.purgeMessagesFromSession + PurgeMessagesResult result = sessionReceiver.purgeMessages(); + System.out.printf("Removed %d messages from session %s.%n", result.getDeletedCount(), sessionId); + // END: com.azure.messaging.servicebus.servicebusreceiverclient.purgeMessagesFromSession + + sessionReceiver.close(); + sessionClient.close(); + } + + /** + * Deletes a batch and then purges the remaining messages asynchronously. + */ + @Test + public void deleteAndPurgeMessagesAsync() { + ServiceBusReceiverAsyncClient receiver = new ServiceBusClientBuilder() + .credential(fullyQualifiedNamespace, new DefaultAzureCredentialBuilder().build()) + .receiver() + .queueName(queueName) + .buildAsyncClient(); + + // BEGIN: com.azure.messaging.servicebus.servicebusreceiverasyncclient.deleteAndPurgeMessages + receiver.deleteMessages(100) + .doOnNext(result -> System.out.printf("Deleted %d messages from the first batch.%n", + result.getDeletedCount())) + .then(receiver.purgeMessages()) + .doOnNext(result -> System.out.printf("Purged %d remaining messages.%n", result.getDeletedCount())) + .block(); + // END: com.azure.messaging.servicebus.servicebusreceiverasyncclient.deleteAndPurgeMessages + + receiver.close(); + } + @Test public void connectionSharingAcrossClients() { // BEGIN: com.azure.messaging.servicebus.connection.sharing diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClientTest.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClientTest.java index 8469c428b985..24910ce27f80 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClientTest.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClientTest.java @@ -45,6 +45,9 @@ import com.azure.messaging.servicebus.models.CompleteOptions; import com.azure.messaging.servicebus.models.DeadLetterOptions; import com.azure.messaging.servicebus.models.DeferOptions; +import com.azure.messaging.servicebus.models.DeleteMessagesOptions; +import com.azure.messaging.servicebus.models.DeleteMessagesResult; +import com.azure.messaging.servicebus.models.PurgeMessagesOptions; import com.azure.messaging.servicebus.models.ServiceBusReceiveMode; import org.apache.qpid.proton.amqp.messaging.Accepted; import org.apache.qpid.proton.amqp.messaging.Rejected; @@ -111,6 +114,7 @@ import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; import static org.mockito.Mockito.when; class ServiceBusReceiverAsyncClientTest { @@ -1648,6 +1652,141 @@ void receiveMessageNegativeLagReportsMetricsSyncInstr() { assertCommonMetricAttributes(attributes, null); } + @Test + void deleteMessagesReturnsActualCount() { + final OffsetDateTime cutoff = OffsetDateTime.of(2026, 8, 27, 12, 30, 0, 0, ZoneOffset.UTC); + when(managementNode.deleteMessages(4000, cutoff, null, null)) + .thenReturn(Mono.just(new DeleteMessagesResult(7))); + + StepVerifier + .create(receiver.deleteMessages(4000, new DeleteMessagesOptions().setEnqueueTimeUtcOlderThan(cutoff))) + .assertNext(result -> assertEquals(7, result.getDeletedCount())) + .expectComplete() + .verify(DEFAULT_TIMEOUT); + } + + @Test + void deleteMessagesCapturesDefaultCutoffOnSubscription() { + final ArgumentCaptor cutoffCaptor = ArgumentCaptor.forClass(OffsetDateTime.class); + when(managementNode.deleteMessages(eq(10), any(OffsetDateTime.class), isNull(), isNull())) + .thenReturn(Mono.just(new DeleteMessagesResult(0))); + + final Mono operation = receiver.deleteMessages(10); + final OffsetDateTime operationStart = OffsetDateTime.now(); + + StepVerifier.create(operation).expectNextCount(1).expectComplete().verify(DEFAULT_TIMEOUT); + + verify(managementNode).deleteMessages(eq(10), cutoffCaptor.capture(), isNull(), isNull()); + Assertions.assertFalse(cutoffCaptor.getValue().isBefore(operationStart)); + } + + @ParameterizedTest + @ValueSource(ints = { -1, 0 }) + void deleteMessagesRejectsCountOutsideServiceRange(int maxMessages) { + StepVerifier.create(receiver.deleteMessages(maxMessages)) + .expectError(IllegalArgumentException.class) + .verify(DEFAULT_TIMEOUT); + + verifyNoInteractions(managementNode); + } + + @Test + void purgeMessagesAccumulatesShortPositiveBatchAndUsesOneCutoff() { + final ArgumentCaptor cutoffCaptor = ArgumentCaptor.forClass(OffsetDateTime.class); + when(managementNode.deleteMessages(eq(500), any(OffsetDateTime.class), isNull(), isNull())) + .thenReturn(Mono.just(new DeleteMessagesResult(500))) + .thenReturn(Mono.just(new DeleteMessagesResult(2))) + .thenReturn(Mono.just(new DeleteMessagesResult(0))); + + StepVerifier.create(receiver.purgeMessages()) + .assertNext(result -> assertEquals(502, result.getDeletedCount())) + .expectComplete() + .verify(DEFAULT_TIMEOUT); + + verify(managementNode, times(3)).deleteMessages(eq(500), cutoffCaptor.capture(), isNull(), isNull()); + assertEquals(3, cutoffCaptor.getAllValues().size()); + assertTrue(cutoffCaptor.getAllValues().stream().allMatch(cutoffCaptor.getValue()::equals)); + } + + @Test + void purgeMessagesSupportsPremiumBatchSize() { + final ArgumentCaptor cutoffCaptor = ArgumentCaptor.forClass(OffsetDateTime.class); + when(managementNode.deleteMessages(eq(4000), any(OffsetDateTime.class), isNull(), isNull())) + .thenReturn(Mono.just(new DeleteMessagesResult(4000))) + .thenReturn(Mono.just(new DeleteMessagesResult(2))) + .thenReturn(Mono.just(new DeleteMessagesResult(0))); + + final PurgeMessagesOptions options = new PurgeMessagesOptions().setMaxMessagesPerBatch(4000); + StepVerifier.create(receiver.purgeMessages(options)) + .assertNext(result -> assertEquals(4002, result.getDeletedCount())) + .expectComplete() + .verify(DEFAULT_TIMEOUT); + + verify(managementNode, times(3)).deleteMessages(eq(4000), cutoffCaptor.capture(), isNull(), isNull()); + assertTrue(cutoffCaptor.getAllValues().stream().allMatch(cutoffCaptor.getValue()::equals)); + } + + @ParameterizedTest + @ValueSource(ints = { -1, 0 }) + void purgeMessagesRejectsInvalidBatchSize(int maxMessagesPerBatch) { + Assertions.assertThrows(IllegalArgumentException.class, + () -> new PurgeMessagesOptions().setMaxMessagesPerBatch(maxMessagesPerBatch)); + verifyNoInteractions(managementNode); + } + + @Test + void purgeMessagesAllowsServiceToEnforceBatchSize() { + when(managementNode.deleteMessages(eq(4001), any(OffsetDateTime.class), isNull(), isNull())) + .thenReturn(Mono.just(new DeleteMessagesResult(0))); + + final PurgeMessagesOptions options = new PurgeMessagesOptions().setMaxMessagesPerBatch(4001); + StepVerifier.create(receiver.purgeMessages(options)) + .assertNext(result -> assertEquals(0, result.getDeletedCount())) + .expectComplete() + .verify(DEFAULT_TIMEOUT); + + verify(managementNode).deleteMessages(eq(4001), any(OffsetDateTime.class), isNull(), isNull()); + } + + @Test + void purgeMessagesDispatchesOnceWhenRequestFails() { + when(managementNode.deleteMessages(eq(500), any(OffsetDateTime.class), isNull(), isNull())) + .thenReturn(Mono.error(new IllegalStateException("request failed"))); + + StepVerifier.create(receiver.purgeMessages(new PurgeMessagesOptions())) + .expectError(ServiceBusException.class) + .verify(DEFAULT_TIMEOUT); + + verify(managementNode).deleteMessages(eq(500), any(OffsetDateTime.class), isNull(), isNull()); + } + + @Test + void deleteMessagesForSessionForwardsSessionAndLinkName() { + final String linkName = "session-link"; + final OffsetDateTime cutoff = OffsetDateTime.of(2026, 8, 27, 12, 30, 0, 0, ZoneOffset.UTC); + final ServiceBusSessionManager sessionManager = mock(ServiceBusSessionManager.class); + when(sessionManager.getLinkName(SESSION_ID)).thenReturn(linkName); + when(managementNode.deleteMessages(5, cutoff, SESSION_ID, linkName)) + .thenReturn(Mono.just(new DeleteMessagesResult(3))); + + final ServiceBusReceiverAsyncClient client + = new ServiceBusReceiverAsyncClient(NAMESPACE, ENTITY_PATH, MessagingEntityType.QUEUE, + createNamedSessionOptions(ServiceBusReceiveMode.PEEK_LOCK, PREFETCH, null, false, SESSION_ID), + connectionCacheWrapper, CLEANUP_INTERVAL, instrumentation, messageSerializer, onClientClose, + sessionManager); + try { + StepVerifier + .create(client.deleteMessages(5, new DeleteMessagesOptions().setEnqueueTimeUtcOlderThan(cutoff))) + .assertNext(result -> assertEquals(3, result.getDeletedCount())) + .expectComplete() + .verify(DEFAULT_TIMEOUT); + } finally { + client.close(); + } + + verify(managementNode).deleteMessages(5, cutoff, SESSION_ID, linkName); + } + private ServiceBusReceivedMessage mockReceivedMessage(Instant enqueuedTime) { ServiceBusReceivedMessage receivedMessage = mock(ServiceBusReceivedMessage.class); when(receivedMessage.getLockedUntil()).thenReturn(OffsetDateTime.now()); diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/implementation/ManagementChannelTests.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/implementation/ManagementChannelTests.java index 6026ba0a218e..e43e25528dea 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/implementation/ManagementChannelTests.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/implementation/ManagementChannelTests.java @@ -67,6 +67,7 @@ import static com.azure.messaging.servicebus.implementation.ManagementConstants.MANAGEMENT_OPERATION_KEY; import static com.azure.messaging.servicebus.implementation.ManagementConstants.OPERATION_GET_SESSION_STATE; import static com.azure.messaging.servicebus.implementation.ManagementConstants.OPERATION_GET_MESSAGE_SESSIONS; +import static com.azure.messaging.servicebus.implementation.ManagementConstants.OPERATION_BATCH_DELETE_MESSAGES; import static com.azure.messaging.servicebus.implementation.ManagementConstants.OPERATION_RENEW_SESSION_LOCK; import static com.azure.messaging.servicebus.implementation.ManagementConstants.OPERATION_SET_SESSION_STATE; import static com.azure.messaging.servicebus.implementation.ManagementConstants.OPERATION_UPDATE_DISPOSITION; @@ -1485,4 +1486,139 @@ void getMessageSessionsParsesIterableSessionIds() { .expectComplete() .verify(TIMEOUT); } + + @Test + void deleteMessagesReturnsActualCountAndSetsWireFields() { + final OffsetDateTime cutoff = OffsetDateTime.of(2026, 8, 27, 12, 30, 0, 0, ZoneOffset.UTC); + final Map responseBody = new HashMap<>(); + responseBody.put(ManagementConstants.MESSAGE_COUNT_KEY, 7); + responseMessage.setBody(new AmqpValue(responseBody)); + + StepVerifier.create(managementChannel.deleteMessages(10, cutoff, "session-id", LINK_NAME)) + .assertNext(result -> assertEquals(7, result.getDeletedCount())) + .expectComplete() + .verify(TIMEOUT); + + verify(requestResponseChannel).sendWithAck(messageCaptor.capture(), isNull()); + final Message sentMessage = messageCaptor.getValue(); + final Map appProperties = sentMessage.getApplicationProperties().getValue(); + assertEquals("com.microsoft:batch-delete-messages", appProperties.get(MANAGEMENT_OPERATION_KEY)); + assertEquals(OPERATION_BATCH_DELETE_MESSAGES, appProperties.get(MANAGEMENT_OPERATION_KEY)); + assertTrue(appProperties.get(ManagementConstants.SERVER_TIMEOUT) instanceof Long); + assertEquals(LINK_NAME, appProperties.get(ASSOCIATED_LINK_NAME_KEY)); + + @SuppressWarnings("unchecked") + final Map body = (Map) ((AmqpValue) sentMessage.getBody()).getValue(); + assertEquals(10, body.get(ManagementConstants.MESSAGE_COUNT_KEY)); + assertEquals(Date.from(cutoff.toInstant()), body.get(ManagementConstants.ENQUEUED_TIME_UTC)); + assertEquals("session-id", body.get(ManagementConstants.SESSION_ID)); + } + + @Test + void deleteMessagesSupportsPremiumCount() { + final Map responseBody = new HashMap<>(); + responseBody.put(ManagementConstants.MESSAGE_COUNT_KEY, 4000); + responseMessage.setBody(new AmqpValue(responseBody)); + + StepVerifier.create(managementChannel.deleteMessages(4000, OffsetDateTime.now(), null, null)) + .assertNext(result -> assertEquals(4000, result.getDeletedCount())) + .expectComplete() + .verify(TIMEOUT); + } + + @Test + void deleteMessagesReturnsZeroForNoContent() { + final Map responseApplicationProperties = new HashMap<>(applicationProperties); + responseApplicationProperties.put(STATUS_CODE_KEY, AmqpResponseCode.NO_CONTENT.getValue()); + responseMessage.setApplicationProperties(new ApplicationProperties(responseApplicationProperties)); + responseMessage.setBody(null); + + StepVerifier.create(managementChannel.deleteMessages(500, OffsetDateTime.now(), null, null)) + .assertNext(result -> assertEquals(0, result.getDeletedCount())) + .expectComplete() + .verify(TIMEOUT); + } + + @ParameterizedTest + @MethodSource("unexpectedBatchDeleteResponses") + void deleteMessagesRejectsUnexpectedSuccessfulResponses(AmqpResponseCode statusCode, + AmqpErrorCondition errorCondition) { + final Map responseApplicationProperties = new HashMap<>(applicationProperties); + responseApplicationProperties.put(STATUS_CODE_KEY, statusCode.getValue()); + if (errorCondition != null) { + responseApplicationProperties.put("error-condition", errorCondition.getErrorCondition()); + } + responseMessage.setApplicationProperties(new ApplicationProperties(responseApplicationProperties)); + responseMessage.setBody(null); + + StepVerifier.create(managementChannel.deleteMessages(500, OffsetDateTime.now(), null, null)) + .expectErrorMatches(error -> error instanceof IllegalStateException + && error.getMessage().contains("unexpected status code")) + .verify(TIMEOUT); + } + + private static Stream unexpectedBatchDeleteResponses() { + return Stream.of(Arguments.of(AmqpResponseCode.ACCEPTED, null), + Arguments.of(AmqpResponseCode.NOT_FOUND, AmqpErrorCondition.SESSION_NOT_FOUND)); + } + + @Test + void deleteMessagesMapsMessageNotFoundToAggregateCount() { + final Map responseApplicationProperties = new HashMap<>(applicationProperties); + responseApplicationProperties.put(STATUS_CODE_KEY, AmqpResponseCode.NOT_FOUND.getValue()); + responseApplicationProperties.put("error-condition", AmqpErrorCondition.MESSAGE_NOT_FOUND.getErrorCondition()); + responseMessage.setApplicationProperties(new ApplicationProperties(responseApplicationProperties)); + final Map responseBody = new HashMap<>(); + responseBody.put(ManagementConstants.MESSAGE_COUNT_KEY, 2); + responseMessage.setBody(new AmqpValue(responseBody)); + + StepVerifier.create(managementChannel.deleteMessages(10, OffsetDateTime.now(), null, null)) + .assertNext(result -> assertEquals(2, result.getDeletedCount())) + .expectComplete() + .verify(TIMEOUT); + } + + @Test + void deleteMessagesRejectsMessageNotFoundWithoutCount() { + final Map responseApplicationProperties = new HashMap<>(applicationProperties); + responseApplicationProperties.put(STATUS_CODE_KEY, AmqpResponseCode.NOT_FOUND.getValue()); + responseApplicationProperties.put("error-condition", AmqpErrorCondition.MESSAGE_NOT_FOUND.getErrorCondition()); + responseMessage.setApplicationProperties(new ApplicationProperties(responseApplicationProperties)); + responseMessage.setBody(null); + + StepVerifier.create(managementChannel.deleteMessages(10, OffsetDateTime.now(), null, null)) + .expectErrorMatches( + error -> error instanceof IllegalStateException && error.getMessage().contains("invalid body")) + .verify(TIMEOUT); + } + + @ParameterizedTest + @MethodSource("invalidBatchDeleteCounts") + void deleteMessagesRejectsInvalidResponseCounts(Object deletedCount) { + final Map responseBody = new HashMap<>(); + responseBody.put(ManagementConstants.MESSAGE_COUNT_KEY, deletedCount); + responseMessage.setBody(new AmqpValue(responseBody)); + + StepVerifier.create(managementChannel.deleteMessages(10, OffsetDateTime.now(), null, null)) + .expectErrorMatches( + error -> error instanceof IllegalStateException && error.getMessage().contains("valid message-count")) + .verify(TIMEOUT); + } + + private static Stream invalidBatchDeleteCounts() { + return Stream.of(Arguments.of(-1), Arguments.of(11), Arguments.of(1L), Arguments.of(1.5)); + } + + @Test + void deleteMessagesDispatchesOnceWhenRequestFails() { + when(requestResponseChannel.sendWithAck(any(Message.class), isNull())) + .thenReturn(Mono.error(new IllegalStateException("request failed"))); + + StepVerifier.create(managementChannel.deleteMessages(500, OffsetDateTime.now(), null, null)) + .expectErrorMatches( + error -> error instanceof IllegalStateException && "request failed".equals(error.getMessage())) + .verify(TIMEOUT); + + verify(requestResponseChannel).sendWithAck(any(Message.class), isNull()); + } }