Option to disable autocomplete for processor (Azure#17190)

Author: srnagar

sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusClientBuilder.java

@@ -771,6 +771,20 @@ public ServiceBusSessionProcessorClientBuilder maxConcurrentCalls(int maxConcurr
             return this;
         }
 
+        /**
+         * Disables auto-complete and auto-abandon of received messages. By default, a successfully processed message is
+         * {@link ServiceBusReceivedMessageContext#complete() completed}. If an error happens when
+         * the message is processed, it is {@link ServiceBusReceivedMessageContext#abandon()
+         * abandoned}.
+         *
+         * @return The modified {@link ServiceBusSessionProcessorClientBuilder} object.
+         */
+        public ServiceBusSessionProcessorClientBuilder disableAutoComplete() {
+            sessionReceiverClientBuilder.disableAutoComplete();
+            processorClientOptions.setDisableAutoComplete(true);
+            return this;
+        }
+
         /**
          * Creates a <b>session-aware</b> Service Bus processor responsible for reading
          * {@link ServiceBusReceivedMessage messages} from a specific queue or topic.
@@ -1191,6 +1205,20 @@ public ServiceBusProcessorClientBuilder maxConcurrentCalls(int maxConcurrentCall
             return this;
         }
 
+        /**
+         * Disables auto-complete and auto-abandon of received messages. By default, a successfully processed message is
+         * {@link ServiceBusReceivedMessageContext#complete() completed}. If an error happens when
+         * the message is processed, it is {@link ServiceBusReceivedMessageContext#abandon()
+         * abandoned}.
+         *
+         * @return The modified {@link ServiceBusProcessorClientBuilder} object.
+         */
+        public ServiceBusProcessorClientBuilder disableAutoComplete() {
+            serviceBusReceiverClientBuilder.disableAutoComplete();
+            processorClientOptions.setDisableAutoComplete(true);
+            return this;
+        }
+
         /**
          * Creates Service Bus message processor responsible for reading {@link ServiceBusReceivedMessage
          * messages} from a specific queue or topic.

sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusProcessorClient.java

@@ -169,8 +169,10 @@ public void onNext(ServiceBusMessageContext serviceBusMessageContext) {
                             processMessage.accept(serviceBusReceivedMessageContext);
                         } catch (Exception ex) {
                             handleError(new ServiceBusReceiverException(ex, ServiceBusErrorSource.USER_CALLBACK));
-                            logger.warning("Error when processing message. Abandoning message.", ex);
-                            abandonMessage(serviceBusMessageContext, receiverClient);
+                            if (!processorOptions.isDisableAutoComplete()) {
+                                logger.warning("Error when processing message. Abandoning message.", ex);
+                                abandonMessage(serviceBusMessageContext, receiverClient);
+                            }
                         }
                     }
                     if (isRunning.get()) {

sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/models/ServiceBusProcessorClientOptions.java

@@ -13,6 +13,24 @@
 public final class ServiceBusProcessorClientOptions {
 
     private int maxConcurrentCalls = 1;
+    private boolean disableAutoComplete;
+
+    /**
+     * Returns true if the auto-complete and auto-abandon feature is disabled.
+     * @return true if the auto-complete and auto-abandon feature is disabled.
+     */
+    public boolean isDisableAutoComplete() {
+        return disableAutoComplete;
+    }
+
+    /**
+     * Disables auto-complete and auto-abandon feature if this is set to {@code true}.
+     * @param disableAutoComplete Disables auto-complete and auto-abandon feature if this is set to {@code true}.
+     */
+    public ServiceBusProcessorClientOptions setDisableAutoComplete(boolean disableAutoComplete) {
+        this.disableAutoComplete = disableAutoComplete;
+        return this;
+    }
 
     /**
      * The max concurrent messages that should be processed by the processor.

sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusProcessorTest.java

@@ -24,6 +24,7 @@
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -271,6 +272,53 @@ public void testUserMessageHandlerError() throws InterruptedException {
         verify(asyncClient, times(5)).abandon(any(ServiceBusReceivedMessage.class));
     }
 
+    @Test
+    public void testUserMessageHandlerErrorWithAutoCompleteDisabled() throws InterruptedException {
+
+        Flux<ServiceBusMessageContext> messageFlux =
+            Flux.create(emitter -> {
+                for (int i = 0; i < 5; i++) {
+                    ServiceBusReceivedMessage serviceBusReceivedMessage =
+                        new ServiceBusReceivedMessage(BinaryData.fromString("hello"));
+                    serviceBusReceivedMessage.setMessageId(String.valueOf(i));
+                    ServiceBusMessageContext serviceBusMessageContext =
+                        new ServiceBusMessageContext(serviceBusReceivedMessage);
+                    emitter.next(serviceBusMessageContext);
+                }
+            });
+
+        ServiceBusClientBuilder.ServiceBusReceiverClientBuilder receiverBuilder =
+            mock(ServiceBusClientBuilder.ServiceBusReceiverClientBuilder.class);
+
+        ServiceBusReceiverAsyncClient asyncClient = mock(ServiceBusReceiverAsyncClient.class);
+        when(receiverBuilder.buildAsyncClient()).thenReturn(asyncClient);
+        when(asyncClient.receiveMessagesWithContext()).thenReturn(messageFlux);
+        when(asyncClient.isConnectionClosed()).thenReturn(false);
+        doNothing().when(asyncClient).close();
+
+        AtomicInteger messageId = new AtomicInteger();
+        CountDownLatch countDownLatch = new CountDownLatch(5);
+        ServiceBusProcessorClient serviceBusProcessorClient = new ServiceBusProcessorClient(receiverBuilder,
+            messageContext -> {
+                assertEquals(String.valueOf(messageId.getAndIncrement()), messageContext.getMessage().getMessageId());
+                throw new IllegalStateException(); // throw error from user handler
+            },
+            error -> {
+                assertTrue(error instanceof ServiceBusReceiverException);
+                ServiceBusReceiverException exception = (ServiceBusReceiverException) error;
+                assertTrue(exception.getErrorSource() == ServiceBusErrorSource.USER_CALLBACK);
+                countDownLatch.countDown();
+            },
+            new ServiceBusProcessorClientOptions().setMaxConcurrentCalls(1).setDisableAutoComplete(true));
+
+        serviceBusProcessorClient.start();
+        boolean success = countDownLatch.await(5, TimeUnit.SECONDS);
+        serviceBusProcessorClient.close();
+        assertTrue(success, "Failed to receive all expected messages");
+
+        verify(asyncClient, never()).abandon(any(ServiceBusReceivedMessage.class));
+    }
+
     private ServiceBusClientBuilder.ServiceBusReceiverClientBuilder getBuilder(
         Flux<ServiceBusMessageContext> messageFlux) {