[Service Bus] Add batch delete and purge APIs

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<String>`) and `ServiceBusSessionReceiverClient` (returning `PagedIterable<String>`). 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.

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<ServiceBusReceivedMessage> 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.
+      *
+      * <p>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.</p>
+     *
+    * @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<DeleteMessagesResult> 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<DeleteMessagesResult> 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<PurgeMessagesResult> 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<PurgeMessagesResult> 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<PurgeMessagesResult> 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 <b>infinite</b> stream of {@link ServiceBusReceivedMessage messages} from the Service Bus entity.
      * This Flux continuously receives messages from a Service Bus entity until either:

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

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<Void> updateDisposition(String lockToken, DispositionStatus disposit
         })).then();
     }
 
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public Mono<DeleteMessagesResult> 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<String, Object> 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<String, Object> body = (Map<String, Object>) ((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}
      */

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";