Add Spring Cloud Stream retry support to ServiceBus Binder

sdk/spring/CHANGELOG.md

@@ -9,6 +9,17 @@ This section includes changes in `spring-cloud-azure-autoconfigure` module.
 
 - Fixed JDBC/Azure Database and Redis passwordless connection scope defaulting using the wrong `azure.scopes` value for Azure China and Azure US Government when `spring.cloud.azure.profile.cloud-type` is set to `azure_china` or `azure_us_government`. The scopes are now correctly derived from the merged cloud type. ([#47096](https://github.com/Azure/azure-sdk-for-java/issues/47096))
 
+### Spring Cloud Azure Stream Binder Service Bus
+This section includes changes in `spring-cloud-azure-stream-binder-servicebus` module.
+
+#### Features Added
+
+- Add support for Spring Cloud Stream consumer retry properties (`maxAttempts`, `backOffInitialInterval`,
+  `backOffMaxInterval`, `backOffMultiplier`) to enable retry with exponential backoff for message processing
+  failures. [#47135](https://github.com/Azure/azure-sdk-for-java/issues/47135).
+- Add support for injecting a custom `RetryTemplate` from Spring context for advanced retry scenarios.
+  [#47135](https://github.com/Azure/azure-sdk-for-java/issues/47135).
+
 ## 7.2.0 (2026-04-17)
 - This release is compatible with Spring Boot 4.0.0-4.0.5. (Note: 4.0.x (x>5) should be supported, but they aren't tested with this release.)
 - This release is compatible with Spring Cloud 2025.1.0-2025.1.1. (Note: 2025.1.x (x>1) should be supported, but they aren't tested with this release.)

sdk/spring/spring-cloud-azure-stream-binder-servicebus/pom.xml

@@ -59,6 +59,11 @@
       <version>4.0.5</version> <!-- {x-version-update;springboot4_org.springframework.boot:spring-boot-starter-actuator;external_dependency} -->
       <optional>true</optional>
     </dependency>
+    <dependency>
+      <groupId>org.springframework.retry</groupId>
+      <artifactId>spring-retry</artifactId>
+      <version>2.0.12</version> <!-- {x-version-update;springboot4_org.springframework.retry:spring-retry;external_dependency} -->
+    </dependency>
     <!-- Added this dependency to include necessary annotations used by reactor core.
         Without this dependency, javadoc throws a warning as it cannot find enum When.MAYBE
         which is used in @Nullable annotation in reactor core classes -->
@@ -156,6 +161,7 @@
             <bannedDependencies>
               <includes>
                 <include>org.springframework.boot:spring-boot-starter-actuator:[4.0.5]</include> <!-- {x-include-update;springboot4_org.springframework.boot:spring-boot-starter-actuator;external_dependency} -->
+                <include>org.springframework.retry:spring-retry:[2.0.12]</include> <!-- {x-include-update;springboot4_org.springframework.retry:spring-retry;external_dependency} -->
               </includes>
             </bannedDependencies>
           </rules>

sdk/spring/spring-cloud-azure-stream-binder-servicebus/src/main/java/com/azure/spring/cloud/stream/binder/servicebus/implementation/ServiceBusMessageChannelBinder.java

@@ -53,6 +53,9 @@
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.support.ErrorMessage;
+import org.springframework.retry.backoff.ExponentialBackOffPolicy;
+import org.springframework.retry.policy.SimpleRetryPolicy;
+import org.springframework.retry.support.RetryTemplate;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
@@ -78,7 +81,6 @@ public class ServiceBusMessageChannelBinder extends
     private static final Logger LOGGER = LoggerFactory.getLogger(ServiceBusMessageChannelBinder.class);
     private static final DefaultErrorMessageStrategy DEFAULT_ERROR_MESSAGE_STRATEGY = new DefaultErrorMessageStrategy();
     private static final String EXCEPTION_MESSAGE = "exception-message";
-
     private ServiceBusExtendedBindingProperties bindingProperties = new ServiceBusExtendedBindingProperties();
     private NamespaceProperties namespaceProperties;
     private ServiceBusTemplate serviceBusTemplate;
@@ -91,6 +93,7 @@ public class ServiceBusMessageChannelBinder extends
 
     private final List<ServiceBusProducerFactoryCustomizer> producerFactoryCustomizers = new ArrayList<>();
     private final List<ServiceBusProcessorFactoryCustomizer> processorFactoryCustomizers = new ArrayList<>();
+    private RetryTemplate retryTemplate;
 
     /**
      * Construct a {@link ServiceBusMessageChannelBinder} with the specified headersToEmbed and {@link ServiceBusChannelProvisioner}.
@@ -148,6 +151,16 @@ protected MessageProducer createConsumerEndpoint(ConsumerDestination destination
         inboundAdapter.setInstrumentationId(instrumentationId);
         inboundAdapter.setErrorChannel(errorInfrastructure.getErrorChannel());
         inboundAdapter.setMessageConverter(messageConverter);
+
+        // Configure retry only when retry is enabled by the consumer properties.
+        if (shouldConfigureRetry(properties)) {
+            // Once retry is enabled, use the injected RetryTemplate if available; otherwise create one from properties.
+            RetryTemplate retryTemplateToUse = this.retryTemplate != null
+                ? this.retryTemplate
+                : createRetryTemplate(properties);
+            inboundAdapter.setRetryTemplate(retryTemplateToUse);
+        }
+
         return inboundAdapter;
     }
 
@@ -377,4 +390,63 @@ public void addProcessorFactoryCustomizer(ServiceBusProcessorFactoryCustomizer p
         }
     }
 
+    /**
+     * Set a custom retry template for message processing retries.
+     * If not set, a retry template will be created automatically based on consumer properties when maxAttempts > 1.
+     *
+     * @param retryTemplate the retry template to use
+     */
+    public void setRetryTemplate(RetryTemplate retryTemplate) {
+        this.retryTemplate = retryTemplate;
+    }
+
+    /**
+     * Get the retry template configured for this binder.
+     *
+     * @return the retry template, or {@code null} if none has been set
+     */
+    public RetryTemplate getRetryTemplate() {
+        return this.retryTemplate;
+    }
+
+    private boolean shouldConfigureRetry(ExtendedConsumerProperties<ServiceBusConsumerProperties> properties) {
+        return properties.getMaxAttempts() > 1;
+    }
+
+    /**
+     * Create a RetryTemplate based on the consumer properties.
+     *
+     * @param properties the extended consumer properties
+     * @return the configured RetryTemplate
+     */
+    private RetryTemplate createRetryTemplate(ExtendedConsumerProperties<ServiceBusConsumerProperties> properties) {
+        RetryTemplate retryTemplate = new RetryTemplate();
+
+        // Configure retry policy
+        SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
+        retryPolicy.setMaxAttempts(properties.getMaxAttempts());
+        retryTemplate.setRetryPolicy(retryPolicy);
+
+        // Configure backoff policy
+        retryTemplate.setBackOffPolicy(createExponentialBackOffPolicy(properties));
+
+        return retryTemplate;
+    }
+
+    /**
+     * Create an {@link ExponentialBackOffPolicy} from the consumer properties.
+     * Package-private to allow direct verification in tests without reflective access to RetryTemplate internals.
+     *
+     * @param properties the extended consumer properties
+     * @return the configured ExponentialBackOffPolicy
+     */
+    ExponentialBackOffPolicy createExponentialBackOffPolicy(
+        ExtendedConsumerProperties<ServiceBusConsumerProperties> properties) {
+        ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
+        backOffPolicy.setInitialInterval(properties.getBackOffInitialInterval());
+        backOffPolicy.setMultiplier(properties.getBackOffMultiplier());
+        backOffPolicy.setMaxInterval(properties.getBackOffMaxInterval());
+        return backOffPolicy;
+    }
+
 }

sdk/spring/spring-cloud-azure-stream-binder-servicebus/src/main/java/com/azure/spring/cloud/stream/binder/servicebus/implementation/config/ServiceBusBinderConfiguration.java

@@ -37,6 +37,7 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
+import org.springframework.retry.support.RetryTemplate;
 
 import static com.azure.spring.cloud.autoconfigure.implementation.context.AzureContextUtils.DEFAULT_TOKEN_CREDENTIAL_BEAN_NAME;
 
@@ -96,6 +97,7 @@ ServiceBusChannelProvisioner serviceBusChannelProvisioner() {
      * @param messageConverter the message converter.
      * @param producerFactoryCustomizers customizers to customize producer factories.
      * @param processorFactoryCustomizers customizers to customize processor factories.
+     * @param retryTemplate optional custom retry template for message processing retries; must be a bean named {@code serviceBusRetryTemplate}.
      *
      * @return the {@link ServiceBusMessageChannelBinder} bean.
      */
@@ -106,14 +108,16 @@ ServiceBusMessageChannelBinder serviceBusBinder(ServiceBusChannelProvisioner cha
                                                            ObjectProvider<NamespaceProperties> namespaceProperties,
                                                            ObjectProvider<ServiceBusMessageConverter> messageConverter,
                                                            ObjectProvider<ServiceBusProducerFactoryCustomizer> producerFactoryCustomizers,
-                                                           ObjectProvider<ServiceBusProcessorFactoryCustomizer> processorFactoryCustomizers) {
+                                                           ObjectProvider<ServiceBusProcessorFactoryCustomizer> processorFactoryCustomizers,
+                                                           @Qualifier("serviceBusRetryTemplate") ObjectProvider<RetryTemplate> retryTemplate) {
 
         ServiceBusMessageChannelBinder binder = new ServiceBusMessageChannelBinder(null, channelProvisioner);
         binder.setBindingProperties(bindingProperties);
         binder.setNamespaceProperties(namespaceProperties.getIfAvailable());
         binder.setMessageConverter(messageConverter.getIfAvailable());
         producerFactoryCustomizers.orderedStream().forEach(binder::addProducerFactoryCustomizer);
         processorFactoryCustomizers.orderedStream().forEach(binder::addProcessorFactoryCustomizer);
+        retryTemplate.ifAvailable(binder::setRetryTemplate);
         return binder;
     }
 

sdk/spring/spring-cloud-azure-stream-binder-servicebus/src/test/java/com/azure/spring/cloud/stream/binder/servicebus/implementation/ServiceBusRetryTest.java

@@ -0,0 +1,207 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+package com.azure.spring.cloud.stream.binder.servicebus.implementation;
+
+import com.azure.spring.cloud.service.servicebus.properties.ServiceBusEntityType;
+import com.azure.spring.cloud.stream.binder.servicebus.core.implementation.provisioning.ServiceBusChannelProvisioner;
+import com.azure.spring.cloud.stream.binder.servicebus.core.properties.ServiceBusBindingProperties;
+import com.azure.spring.cloud.stream.binder.servicebus.core.properties.ServiceBusConsumerProperties;
+import com.azure.spring.cloud.stream.binder.servicebus.core.properties.ServiceBusExtendedBindingProperties;
+import com.azure.spring.integration.servicebus.inbound.ServiceBusInboundChannelAdapter;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.cloud.stream.binder.BinderHeaders;
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.HeaderMode;
+import org.springframework.cloud.stream.provisioning.ConsumerDestination;
+import org.springframework.context.support.GenericApplicationContext;
+import org.springframework.integration.core.MessageProducer;
+import org.springframework.retry.backoff.ExponentialBackOffPolicy;
+import org.springframework.retry.support.RetryTemplate;
+
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for retry functionality in ServiceBusMessageChannelBinder.
+ */
+@ExtendWith(MockitoExtension.class)
+class ServiceBusRetryTest {
+
+    @Mock
+    private ConsumerDestination consumerDestination;
+
+    private final ServiceBusExtendedBindingProperties extendedBindingProperties =
+        new ServiceBusExtendedBindingProperties();
+
+    private ExtendedConsumerProperties<ServiceBusConsumerProperties> consumerProperties;
+
+    private final ServiceBusConsumerProperties serviceBusConsumerProperties = new ServiceBusConsumerProperties();
+
+    private ServiceBusMessageChannelTestBinder binder;
+
+    private GenericApplicationContext applicationContext;
+
+    private static final String ENTITY_NAME = "test-entity";
+    private static final String GROUP = "test";
+    private static final String NAMESPACE_NAME = "test-namespace";
+
+    @BeforeEach
+    void init() {
+        binder = new ServiceBusMessageChannelTestBinder(
+            BinderHeaders.STANDARD_HEADERS, new ServiceBusChannelProvisioner());
+        applicationContext = new GenericApplicationContext();
+        applicationContext.refresh();
+        binder.setApplicationContext(applicationContext);
+    }
+
+    @AfterEach
+    void tearDown() {
+        if (applicationContext != null) {
+            applicationContext.close();
+        }
+    }
+
+    @Test
+    void testRetryTemplateConfiguredWhenMaxAttemptsGreaterThanOne() {
+        // Arrange
+        prepareConsumerProperties();
+        consumerProperties.setMaxAttempts(3);
+        consumerProperties.setBackOffInitialInterval(1000);
+        consumerProperties.setBackOffMultiplier(2.0);
+        consumerProperties.setBackOffMaxInterval(5000);
+        when(consumerDestination.getName()).thenReturn(ENTITY_NAME);
+
+        // Act
+        MessageProducer producer = binder.createConsumerEndpoint(consumerDestination, GROUP, consumerProperties);
+
+        // Assert
+        assertThat(producer).isInstanceOf(ServiceBusInboundChannelAdapter.class);
+        ServiceBusInboundChannelAdapter adapter = (ServiceBusInboundChannelAdapter) producer;
+        RetryTemplate retryTemplate = adapter.getRetryTemplate();
+        assertThat(retryTemplate).isNotNull();
+
+        // Verify maxAttempts=3 by executing the template and counting actual attempts
+        AtomicInteger callCount = new AtomicInteger(0);
+        assertThatThrownBy(() -> retryTemplate.execute(ctx -> {
+            callCount.incrementAndGet();
+            throw new RuntimeException("test");
+        })).isInstanceOf(RuntimeException.class);
+        assertThat(callCount.get()).isEqualTo(3);
+
+        // Verify backoff policy configuration via the binder's factory method (no reflection needed)
+        ExponentialBackOffPolicy backOffPolicy = binder.createExponentialBackOffPolicy(consumerProperties);
+        assertThat(backOffPolicy.getInitialInterval()).isEqualTo(1000L);
+        assertThat(backOffPolicy.getMultiplier()).isEqualTo(2.0);
+        assertThat(backOffPolicy.getMaxInterval()).isEqualTo(5000L);
+    }
+
+    @Test
+    void testRetryTemplateNotConfiguredWhenMaxAttemptsIsOne() {
+        // Arrange
+        prepareConsumerProperties();
+        consumerProperties.setMaxAttempts(1);
+        when(consumerDestination.getName()).thenReturn(ENTITY_NAME);
+
+        // Act
+        MessageProducer producer = binder.createConsumerEndpoint(consumerDestination, GROUP, consumerProperties);
+
+        // Assert
+        assertThat(producer).isInstanceOf(ServiceBusInboundChannelAdapter.class);
+        assertThat(((ServiceBusInboundChannelAdapter) producer).getRetryTemplate()).isNull();
+    }
+
+    @Test
+    void testRetryTemplateConfiguredWithDefaultSettings() {
+        // Arrange
+        prepareConsumerProperties();
+        // Spring Cloud Stream default maxAttempts is 3 (> 1), so a RetryTemplate should be created.
+        when(consumerDestination.getName()).thenReturn(ENTITY_NAME);
+
+        // Act
+        MessageProducer producer = binder.createConsumerEndpoint(consumerDestination, GROUP, consumerProperties);
+
+        // Assert
+        assertThat(producer).isInstanceOf(ServiceBusInboundChannelAdapter.class);
+        ServiceBusInboundChannelAdapter adapter = (ServiceBusInboundChannelAdapter) producer;
+        RetryTemplate retryTemplate = adapter.getRetryTemplate();
+        assertThat(retryTemplate).isNotNull();
+
+        // Verify maxAttempts matches Spring Cloud Stream's default via observable behavior
+        int expectedMaxAttempts = new ExtendedConsumerProperties<>(new ServiceBusConsumerProperties()).getMaxAttempts();
+        AtomicInteger callCount = new AtomicInteger(0);
+        assertThatThrownBy(() -> retryTemplate.execute(ctx -> {
+            callCount.incrementAndGet();
+            throw new RuntimeException("test");
+        })).isInstanceOf(RuntimeException.class);
+        assertThat(callCount.get()).isEqualTo(expectedMaxAttempts);
+    }
+
+    @Test
+    void testCustomRetryTemplateIsUsed() {
+        // Arrange
+        prepareConsumerProperties();
+        consumerProperties.setMaxAttempts(3);
+        when(consumerDestination.getName()).thenReturn(ENTITY_NAME);
+
+        // Create a custom RetryTemplate
+        RetryTemplate customRetryTemplate = new RetryTemplate();
+        binder.setRetryTemplate(customRetryTemplate);
+
+        // Act
+        MessageProducer producer = binder.createConsumerEndpoint(consumerDestination, GROUP, consumerProperties);
+
+        // Assert
+        assertThat(producer).isInstanceOf(ServiceBusInboundChannelAdapter.class);
+        ServiceBusInboundChannelAdapter adapter = (ServiceBusInboundChannelAdapter) producer;
+        assertThat(adapter.getRetryTemplate()).isNotNull();
+        assertThat(adapter.getRetryTemplate()).isSameAs(customRetryTemplate);
+    }
+
+    @Test
+    void testCustomRetryTemplateNotAppliedWhenMaxAttemptsIsOne() {
+        // Arrange: maxAttempts=1 disables retry even when a custom RetryTemplate bean is injected
+        prepareConsumerProperties();
+        consumerProperties.setMaxAttempts(1);
+        when(consumerDestination.getName()).thenReturn(ENTITY_NAME);
+
+        RetryTemplate customRetryTemplate = new RetryTemplate();
+        binder.setRetryTemplate(customRetryTemplate);
+
+        // Act
+        MessageProducer producer = binder.createConsumerEndpoint(consumerDestination, GROUP, consumerProperties);
+
+        // Assert
+        assertThat(producer).isInstanceOf(ServiceBusInboundChannelAdapter.class);
+        assertThat(((ServiceBusInboundChannelAdapter) producer).getRetryTemplate()).isNull();
+    }
+
+    private void prepareConsumerProperties() {
+        serviceBusConsumerProperties.setEntityName(ENTITY_NAME);
+        serviceBusConsumerProperties.setSubscriptionName(GROUP);
+        serviceBusConsumerProperties.setEntityType(ServiceBusEntityType.TOPIC);
+        serviceBusConsumerProperties.setNamespace(NAMESPACE_NAME);
+        serviceBusConsumerProperties.getRetry().setTryTimeout(Duration.ofMinutes(5));
+        serviceBusConsumerProperties.setAutoComplete(false);
+        ServiceBusBindingProperties bindingProperties = new ServiceBusBindingProperties();
+        bindingProperties.setConsumer(serviceBusConsumerProperties);
+
+        Map<String, ServiceBusBindingProperties> bindings = new HashMap<>();
+        bindings.put(ENTITY_NAME, bindingProperties);
+        extendedBindingProperties.setBindings(bindings);
+        binder.setBindingProperties(extendedBindingProperties);
+
+        consumerProperties = new ExtendedConsumerProperties<>(serviceBusConsumerProperties);
+        consumerProperties.setHeaderMode(HeaderMode.embeddedHeaders);
+    }
+}