Add acknowledgement mode on @SqsListener annotation

docs/src/main/asciidoc/sqs.adoc

@@ -636,6 +636,8 @@ Note that if there are messages available the call may return earlier than this
 - `messageVisibilitySeconds` - Set the minimum visibility for the messages retrieved in a poll.
 Note that for `FIFO` single message listener methods, this visibility is applied to the whole batch before each message is sent to the listener.
 See <<FIFO Support>> for more information.
+- `acknowledgementMode` - Set the acknowledgement mode for the container. Default value is `SqsListenerAcknowledgementMode.DEFAULT`,
+meaning that it will use the acknowledgement mode defined for the container factory. See <<Acknowledgement Mode>> for more information.
 
 
 ===== Listener Method Arguments
@@ -1326,6 +1328,8 @@ NOTE: All options are available for both `single message` and `batch` message li
 - `ALWAYS` - Acknowledges a message or batch of messages after processing returns success or error.
 - `MANUAL` - The framework won't acknowledge messages automatically and `Acknowledgement` objects can be received in the listener method.
 
+The `Acknowledgement` strategy can be configured in the `SqsContainerOptions` or in the `@SqsListener` annotation.
+
 ==== Acknowledgement Batching
 
 The `acknowledgementInterval` and `acknowledgementThreshold` options enable acknowledgement batching.

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/annotation/AbstractListenerAnnotationBeanPostProcessor.java

@@ -22,6 +22,7 @@
 import io.awspring.cloud.sqs.config.HandlerMethodEndpoint;
 import io.awspring.cloud.sqs.config.SqsEndpoint;
 import io.awspring.cloud.sqs.config.SqsListenerConfigurer;
+import io.awspring.cloud.sqs.listener.acknowledgement.handler.AcknowledgementMode;
 import io.awspring.cloud.sqs.support.resolver.AcknowledgmentHandlerMethodArgumentResolver;
 import io.awspring.cloud.sqs.support.resolver.BatchAcknowledgmentArgumentResolver;
 import io.awspring.cloud.sqs.support.resolver.BatchPayloadMethodArgumentResolver;
@@ -219,6 +220,18 @@ protected Integer resolveAsInteger(String value, String propertyName) {
 		}
 	}
 
+	@Nullable
+	protected AcknowledgementMode resolveAcknowledgement(String value) {
+		try {
+			final String resolvedValue = resolveAsString(value, "acknowledgementMode");
+			return resolvedValue.equals(SqsListenerAcknowledgementMode.DEFAULT) ? null
+					: AcknowledgementMode.valueOf(resolvedValue);
+		}
+		catch (IllegalArgumentException e) {
+			throw new IllegalArgumentException("Cannot resolve " + value + " as AcknowledgementMode", e);
+		}
+	}
+
 	protected String getEndpointId(String id) {
 		if (StringUtils.hasText(id)) {
 			return resolveAsString(id, "id");

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/annotation/SqsListener.java

@@ -137,4 +137,10 @@
 	 */
 	String messageVisibilitySeconds() default "";
 
+	/**
+	 * The acknowledgement mode to be used for the provided queues. If not specified, the acknowledgement mode defined
+	 * for the container factory will be used.
+	 */
+	String acknowledgementMode() default SqsListenerAcknowledgementMode.DEFAULT;
+
 }

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/annotation/SqsListenerAcknowledgementMode.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2013-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.awspring.cloud.sqs.annotation;
+
+import io.awspring.cloud.sqs.listener.acknowledgement.handler.AlwaysAcknowledgementHandler;
+import io.awspring.cloud.sqs.listener.acknowledgement.handler.NeverAcknowledgementHandler;
+import io.awspring.cloud.sqs.listener.acknowledgement.handler.OnSuccessAcknowledgementHandler;
+
+/**
+ * Acknowledgement strategies supported by the {@link SqsListener} annotation.
+ *
+ * @author Joao Calassio
+ * @since 3.1
+ * @see OnSuccessAcknowledgementHandler
+ * @see AlwaysAcknowledgementHandler
+ * @see NeverAcknowledgementHandler
+ * @see io.awspring.cloud.sqs.listener.ContainerOptions
+ * @see SqsListener
+ */
+public class SqsListenerAcknowledgementMode {
+
+	/**
+	 * Use acknowledge mode defined by the container.
+	 */
+	public static final String DEFAULT = "DEFAULT";
+
+	/**
+	 * Messages will be acknowledged when message processing is successful.
+	 */
+	public static final String ON_SUCCESS = "ON_SUCCESS";
+
+	/**
+	 * Messages will be acknowledged whether processing was completed successfully or with an error.
+	 */
+	public static final String ALWAYS = "ALWAYS";
+
+	/**
+	 * Messages will not be acknowledged automatically by the container.
+	 * @see io.awspring.cloud.sqs.listener.acknowledgement.Acknowledgement
+	 */
+	public static final String MANUAL = "MANUAL";
+
+}

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/annotation/SqsListenerAnnotationBeanPostProcessor.java

@@ -51,7 +51,7 @@ protected Endpoint createEndpoint(SqsListener sqsListenerAnnotation) {
 						resolveAsInteger(sqsListenerAnnotation.maxConcurrentMessages(), "maxConcurrentMessages"))
 				.messageVisibility(
 						resolveAsInteger(sqsListenerAnnotation.messageVisibilitySeconds(), "messageVisibility"))
-				.build();
+				.acknowledgementMode(resolveAcknowledgement(sqsListenerAnnotation.acknowledgementMode())).build();
 	}
 
 	@Override

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/config/SqsEndpoint.java

@@ -16,6 +16,7 @@
 package io.awspring.cloud.sqs.config;
 
 import io.awspring.cloud.sqs.annotation.SqsListener;
+import io.awspring.cloud.sqs.listener.acknowledgement.handler.AcknowledgementMode;
 import java.time.Duration;
 import java.util.Collection;
 import org.springframework.lang.Nullable;
@@ -38,12 +39,16 @@ public class SqsEndpoint extends AbstractEndpoint {
 
 	private final Integer maxMessagesPerPoll;
 
+	@Nullable
+	private final AcknowledgementMode acknowledgementMode;
+
 	protected SqsEndpoint(SqsEndpointBuilder builder) {
 		super(builder.queueNames, builder.factoryName, builder.id);
 		this.maxConcurrentMessages = builder.maxConcurrentMessages;
 		this.pollTimeoutSeconds = builder.pollTimeoutSeconds;
 		this.messageVisibility = builder.messageVisibility;
 		this.maxMessagesPerPoll = builder.maxMessagesPerPoll;
+		this.acknowledgementMode = builder.acknowledgementMode;
 	}
 
 	/**
@@ -91,6 +96,15 @@ public Duration getMessageVisibility() {
 		return this.messageVisibility != null ? Duration.ofSeconds(this.messageVisibility) : null;
 	}
 
+	/**
+	 * Returns the acknowledgement mode configured for this endpoint.
+	 * @return the acknowledgement mode.
+	 */
+	@Nullable
+	public AcknowledgementMode getAcknowledgementMode() {
+		return this.acknowledgementMode;
+	}
+
 	public static class SqsEndpointBuilder {
 
 		private Collection<String> queueNames;
@@ -107,6 +121,9 @@ public static class SqsEndpointBuilder {
 
 		private Integer maxMessagesPerPoll;
 
+		@Nullable
+		private AcknowledgementMode acknowledgementMode;
+
 		public SqsEndpointBuilder queueNames(Collection<String> queueNames) {
 			this.queueNames = queueNames;
 			return this;
@@ -142,6 +159,11 @@ public SqsEndpointBuilder id(String id) {
 			return this;
 		}
 
+		public SqsEndpointBuilder acknowledgementMode(@Nullable AcknowledgementMode acknowledgementMode) {
+			this.acknowledgementMode = acknowledgementMode;
+			return this;
+		}
+
 		public SqsEndpoint build() {
 			return new SqsEndpoint(this);
 		}

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/config/SqsMessageListenerContainerFactory.java

@@ -161,7 +161,8 @@ private void configureFromSqsEndpoint(SqsEndpoint sqsEndpoint, SqsContainerOptio
 		ConfigUtils.INSTANCE.acceptIfNotNull(sqsEndpoint.getMaxConcurrentMessages(), options::maxConcurrentMessages)
 				.acceptIfNotNull(sqsEndpoint.getMaxMessagesPerPoll(), options::maxMessagesPerPoll)
 				.acceptIfNotNull(sqsEndpoint.getPollTimeout(), options::pollTimeout)
-				.acceptIfNotNull(sqsEndpoint.getMessageVisibility(), options::messageVisibility);
+				.acceptIfNotNull(sqsEndpoint.getMessageVisibility(), options::messageVisibility)
+				.acceptIfNotNull(sqsEndpoint.getAcknowledgementMode(), options::acknowledgementMode);
 	}
 
 	/**

spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/annotation/SqsListenerAcknowledgementModeTests.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2013-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.awspring.cloud.sqs.annotation;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import io.awspring.cloud.sqs.listener.acknowledgement.handler.AcknowledgementMode;
+import java.lang.reflect.Field;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+/**
+ * Tests for {@link SqsListenerAcknowledgementMode} enum values
+ *
+ * @author Joao Calassio
+ */
+class SqsListenerAcknowledgementModeTests {
+
+	@ParameterizedTest
+	@EnumSource(AcknowledgementMode.class)
+	void shouldHaveAllValuesOfAcknowledgementModeEnum(final AcknowledgementMode acknowledgementMode)
+			throws NoSuchFieldException, IllegalAccessException {
+		Class<SqsListenerAcknowledgementMode> clz = SqsListenerAcknowledgementMode.class;
+		Field correspondingValue = clz.getDeclaredField(acknowledgementMode.name());
+		assertEquals(acknowledgementMode.name(), correspondingValue.get(clz));
+	}
+
+}