fix(core): correct retry semantics for consumer maxAttempts configuration

The retry policy previously used maxRetries(properties.getMaxAttempts()), which caused an off-by-one error because maxRetries represents the number of retries while maxAttempts represents total delivery attempts.

Example:
maxAttempts=2 resulted in 3 executions (1 initial + 2 retries).

This change converts maxAttempts to retries by using:
maxRetries = maxAttempts - 1

A test (MaxAttemptsRetryTests) has been added to reproduce and verify the correct behavior.

Signed-off-by: vishwasio <[email protected]>

Author: vishwasio

core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java

@@ -209,7 +209,7 @@ protected RetryTemplate buildRetryTemplate(ConsumerProperties properties) {
 			}
 			RetryPolicy retryPolicy =
 				RetryPolicy.builder()
-					.maxRetries(properties.getMaxAttempts())
+					.maxRetries(Math.max(0, properties.getMaxAttempts() - 1))
 					.delay(Duration.ofMillis(properties.getBackOffInitialInterval()))
 					.multiplier(properties.getBackOffMultiplier())
 					.maxDelay(Duration.ofMillis(properties.getBackOffMaxInterval()))

core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/MaxAttemptsRetryTests.java

@@ -0,0 +1,49 @@
+package org.springframework.cloud.stream.binder;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.core.retry.RetryTemplate;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class MaxAttemptsRetryTests {
+
+	@Test
+	void maxAttemptsShouldNotExceedConfiguredAttempts() {
+
+		ConsumerProperties props = new ConsumerProperties();
+		props.setMaxAttempts(2);
+
+		TestBinder binder = new TestBinder();
+
+		RetryTemplate retryTemplate = binder.buildRetryTemplate(props);
+
+		AtomicInteger attempts = new AtomicInteger();
+
+		try {
+			retryTemplate.execute(() -> {
+				attempts.incrementAndGet();
+				throw new RuntimeException("fail");
+			});
+		} catch (Exception ignored) {
+		}
+
+		assertThat(attempts.get()).isEqualTo(2);
+	}
+
+	static class TestBinder extends AbstractBinder<Object, ConsumerProperties, ProducerProperties> {
+
+		@Override
+		protected Binding<Object> doBindConsumer(String name, String group, Object inboundBindTarget,
+												 ConsumerProperties consumerProperties) {
+			return null;
+		}
+
+		@Override
+		protected Binding<Object> doBindProducer(String name, Object outboundBindTarget,
+												 ProducerProperties producerProperties) {
+			return null;
+		}
+	}
+}