Conditionally add a random UUID dedup ID if the queue isn't configured for content based deduplication (#799)

Fixes #796

Author: jwilmoth-nc

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/operations/SqsTemplate.java

@@ -22,6 +22,7 @@
 import io.awspring.cloud.sqs.listener.QueueAttributes;
 import io.awspring.cloud.sqs.listener.QueueNotFoundStrategy;
 import io.awspring.cloud.sqs.listener.SqsHeaders;
+import io.awspring.cloud.sqs.listener.SqsHeaders.MessageSystemAttributes;
 import io.awspring.cloud.sqs.listener.acknowledgement.AcknowledgementCallback;
 import io.awspring.cloud.sqs.support.converter.MessageAttributeDataTypes;
 import io.awspring.cloud.sqs.support.converter.MessageConversionContext;
@@ -32,8 +33,10 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
@@ -244,23 +247,35 @@ private Map<String, Object> addAdditionalReceiveHeaders(SqsReceiveOptionsImpl op
 	@Override
 	protected <T> org.springframework.messaging.Message<T> preProcessMessageForSend(String endpointToUse,
 			org.springframework.messaging.Message<T> message) {
-		return FifoUtils.isFifo(endpointToUse) ? addMissingFifoSendHeaders(message) : message;
+		return FifoUtils.isFifo(endpointToUse) ? addMissingFifoSendHeaders(endpointToUse, message) : message;
 	}
 
 	@Override
 	protected <T> Collection<org.springframework.messaging.Message<T>> preProcessMessagesForSend(String endpointToUse,
 			Collection<org.springframework.messaging.Message<T>> messages) {
 		return FifoUtils.isFifo(endpointToUse)
-				? messages.stream().map(this::addMissingFifoSendHeaders).collect(Collectors.toList())
+				? messages.stream().map(message -> addMissingFifoSendHeaders(endpointToUse, message)).toList()
 				: messages;
 	}
 
-	private <T> org.springframework.messaging.Message<T> addMissingFifoSendHeaders(
+	private <T> org.springframework.messaging.Message<T> addMissingFifoSendHeaders(String endpointName,
 			org.springframework.messaging.Message<T> message) {
-		return MessageHeaderUtils.addHeadersIfAbsent(message,
-				Map.of(SqsHeaders.MessageSystemAttributes.SQS_MESSAGE_GROUP_ID_HEADER, UUID.randomUUID().toString(),
-						SqsHeaders.MessageSystemAttributes.SQS_MESSAGE_DEDUPLICATION_ID_HEADER,
-						UUID.randomUUID().toString()));
+
+		Set<QueueAttributeName> additionalAttributes = Set.of(QueueAttributeName.CONTENT_BASED_DEDUPLICATION);
+		String contentBasedDedupQueueAttribute = getQueueAttributes(endpointName, additionalAttributes).join()
+				.getQueueAttribute(QueueAttributeName.CONTENT_BASED_DEDUPLICATION);
+
+		boolean isContentBasedDedup = Boolean.parseBoolean(contentBasedDedupQueueAttribute);
+		Map<String, Object> defaultHeaders;
+		if (isContentBasedDedup) {
+			defaultHeaders = Map.of(MessageSystemAttributes.SQS_MESSAGE_GROUP_ID_HEADER, UUID.randomUUID().toString());
+		}
+		else {
+			defaultHeaders = Map.of(MessageSystemAttributes.SQS_MESSAGE_GROUP_ID_HEADER, UUID.randomUUID().toString(),
+					MessageSystemAttributes.SQS_MESSAGE_DEDUPLICATION_ID_HEADER, UUID.randomUUID().toString());
+		}
+
+		return MessageHeaderUtils.addHeadersIfAbsent(message, defaultHeaders);
 	}
 
 	@Override
@@ -408,10 +423,21 @@ private boolean isSkipAttribute(MessageSystemAttributeName name) {
 	}
 
 	private CompletableFuture<QueueAttributes> getQueueAttributes(String endpointName) {
-		return this.queueAttributesCache.computeIfAbsent(endpointName,
-				newName -> QueueAttributesResolver.builder().sqsAsyncClient(this.sqsAsyncClient).queueName(newName)
-						.queueNotFoundStrategy(this.queueNotFoundStrategy).queueAttributeNames(this.queueAttributeNames)
-						.build().resolveQueueAttributes());
+		return getQueueAttributes(endpointName, Collections.emptySet());
+	}
+
+	private CompletableFuture<QueueAttributes> getQueueAttributes(String endpointName,
+			Set<QueueAttributeName> additionalAttributes) {
+		return this.queueAttributesCache.computeIfAbsent(endpointName, newName -> {
+			// ensure we have the content based dedupe config to determine default fifo send headers
+			Set<QueueAttributeName> namesToRequest = new HashSet<>(queueAttributeNames);
+			if (additionalAttributes != null && !additionalAttributes.isEmpty()) {
+				namesToRequest.addAll(additionalAttributes);
+			}
+			return QueueAttributesResolver.builder().sqsAsyncClient(this.sqsAsyncClient).queueName(newName)
+					.queueNotFoundStrategy(this.queueNotFoundStrategy).queueAttributeNames(namesToRequest).build()
+					.resolveQueueAttributes();
+		});
 	}
 
 	@Override