From 649f748214fda2640aefd941b408e82b49d00fc9 Mon Sep 17 00:00:00 2001 From: Andrea Cosentino Date: Wed, 15 Jul 2026 15:33:29 +0200 Subject: [PATCH] CAMEL-24086: camel-aws2-sqs - give each FIFO batch entry a distinct MessageDeduplicationId When sending a batch to a FIFO queue, every entry is built from the same Exchange, so configureFifoAttributes derived the MessageDeduplicationId from that one Exchange via the configured strategy. The default ExchangeIdMessageDeduplicationIdStrategy returns exchange.getExchangeId(), identical for all entries, so SQS accepted the batch but silently dropped all but the first message within the deduplication window. The batch overload now appends the entry position to the strategy-provided id, giving each message a distinct, deterministic deduplication id (so a redelivered batch still deduplicates correctly). A null id (content-based deduplication via NullMessageDeduplicationIdStrategy) is left unset, and the group-id / dedup-id strategies are now null-guarded consistently with the single-message overload. Adds SqsProducerFifoBatchDeduplicationTest asserting the batch entries receive distinct deduplication ids. The single-message send path is unchanged. Co-Authored-By: Claude Fable 5 Signed-off-by: Andrea Cosentino --- .../component/aws2/sqs/Sqs2Producer.java | 34 ++++--- ...SqsProducerFifoBatchDeduplicationTest.java | 96 +++++++++++++++++++ 2 files changed, 119 insertions(+), 11 deletions(-) create mode 100644 components/camel-aws/camel-aws2-sqs/src/test/java/org/apache/camel/component/aws2/sqs/SqsProducerFifoBatchDeduplicationTest.java diff --git a/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Producer.java b/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Producer.java index e58ad11428b2b..0518d36791d7d 100644 --- a/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Producer.java +++ b/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Producer.java @@ -121,6 +121,7 @@ private void sendBatchMessage(SqsClient amazonSQS, Exchange exchange) { Collection entries = new ArrayList<>(); if (exchange.getIn().getBody() instanceof Iterable) { Iterable c = exchange.getIn().getBody(Iterable.class); + int index = 0; for (Object o : c) { String object = (String) o; SendMessageBatchRequestEntry.Builder entry = SendMessageBatchRequestEntry.builder(); @@ -128,7 +129,7 @@ private void sendBatchMessage(SqsClient amazonSQS, Exchange exchange) { entry.messageAttributes(translateAttributes(exchange.getIn().getHeaders(), exchange)); entry.messageBody(object); addDelay(entry, exchange); - configureFifoAttributes(entry, exchange); + configureFifoAttributes(entry, exchange, index++); entries.add(entry.build()); } request.entries(entries); @@ -142,13 +143,14 @@ private void sendBatchMessage(SqsClient amazonSQS, Exchange exchange) { } else if (exchange.getIn().getBody() instanceof String) { String c = exchange.getIn().getBody(String.class); String[] elements = c.split(getConfiguration().getBatchSeparator()); + int index = 0; for (String o : elements) { SendMessageBatchRequestEntry.Builder entry = SendMessageBatchRequestEntry.builder(); entry.id(UUID.randomUUID().toString()); entry.messageAttributes(translateAttributes(exchange.getIn().getHeaders(), exchange)); entry.messageBody(o); addDelay(entry, exchange); - configureFifoAttributes(entry, exchange); + configureFifoAttributes(entry, exchange, index++); entries.add(entry.build()); } request.entries(entries); @@ -240,17 +242,27 @@ private void configureFifoAttributes(SendMessageRequest.Builder request, Exchang } } - private void configureFifoAttributes(SendMessageBatchRequestEntry.Builder request, Exchange exchange) { + private void configureFifoAttributes(SendMessageBatchRequestEntry.Builder request, Exchange exchange, int index) { if (getEndpoint().getConfiguration().isFifoQueue()) { // use strategies - MessageGroupIdStrategy messageGroupIdStrategy = getEndpoint().getConfiguration().getMessageGroupIdStrategy(); - String messageGroupId = messageGroupIdStrategy.getMessageGroupId(exchange); - request.messageGroupId(messageGroupId); - - MessageDeduplicationIdStrategy messageDeduplicationIdStrategy - = getEndpoint().getConfiguration().getMessageDeduplicationIdStrategy(); - String messageDeduplicationId = messageDeduplicationIdStrategy.getMessageDeduplicationId(exchange); - request.messageDeduplicationId(messageDeduplicationId); + if (ObjectHelper.isNotEmpty(getEndpoint().getConfiguration().getMessageGroupIdStrategy())) { + MessageGroupIdStrategy messageGroupIdStrategy = getEndpoint().getConfiguration().getMessageGroupIdStrategy(); + String messageGroupId = messageGroupIdStrategy.getMessageGroupId(exchange); + request.messageGroupId(messageGroupId); + } + + if (ObjectHelper.isNotEmpty(getEndpoint().getConfiguration().getMessageDeduplicationIdStrategy())) { + MessageDeduplicationIdStrategy messageDeduplicationIdStrategy + = getEndpoint().getConfiguration().getMessageDeduplicationIdStrategy(); + String messageDeduplicationId = messageDeduplicationIdStrategy.getMessageDeduplicationId(exchange); + // Every entry of a batch is built from the same Exchange, so the strategy returns the + // same id for all of them. Append the entry position to give each message in the batch a + // distinct, deterministic deduplication id; otherwise a FIFO queue drops all but the + // first. A null id (e.g. content-based deduplication) is left unset. + if (messageDeduplicationId != null) { + request.messageDeduplicationId(messageDeduplicationId + "-" + index); + } + } } } diff --git a/components/camel-aws/camel-aws2-sqs/src/test/java/org/apache/camel/component/aws2/sqs/SqsProducerFifoBatchDeduplicationTest.java b/components/camel-aws/camel-aws2-sqs/src/test/java/org/apache/camel/component/aws2/sqs/SqsProducerFifoBatchDeduplicationTest.java new file mode 100644 index 0000000000000..c5d8eed9d5017 --- /dev/null +++ b/components/camel-aws/camel-aws2-sqs/src/test/java/org/apache/camel/component/aws2/sqs/SqsProducerFifoBatchDeduplicationTest.java @@ -0,0 +1,96 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 + * + * http://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 org.apache.camel.component.aws2.sqs; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +import org.apache.camel.BindToRegistry; +import org.apache.camel.EndpointInject; +import org.apache.camel.Exchange; +import org.apache.camel.Processor; +import org.apache.camel.ProducerTemplate; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.test.junit6.CamelTestSupport; +import org.junit.jupiter.api.Test; +import software.amazon.awssdk.services.sqs.model.SendMessageBatchRequest; +import software.amazon.awssdk.services.sqs.model.SendMessageBatchRequestEntry; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +public class SqsProducerFifoBatchDeduplicationTest extends CamelTestSupport { + + @BindToRegistry("client") + AmazonSQSClientMock mock = new AmazonSQSClientMock(); + + @EndpointInject("direct:start") + private ProducerTemplate template; + + @EndpointInject("mock:result") + private MockEndpoint result; + + @Test + public void eachBatchEntryGetsADistinctDeduplicationId() throws Exception { + result.expectedMessageCount(1); + + template.send("direct:start", new Processor() { + @Override + public void process(Exchange exchange) { + Collection c = new ArrayList<>(); + c.add("team1"); + c.add("team2"); + c.add("team3"); + c.add("team4"); + exchange.getIn().setBody(c); + } + }); + MockEndpoint.assertIsSatisfied(context); + + List requests = mock.getSendMessageBatchRequests(); + assertEquals(1, requests.size()); + List entries = requests.get(0).entries(); + assertEquals(4, entries.size()); + + // On a FIFO queue every entry must carry a distinct MessageDeduplicationId, otherwise SQS keeps + // only the first message of the batch. The group id may legitimately be shared across the batch. + Set dedupIds = new HashSet<>(); + for (SendMessageBatchRequestEntry entry : entries) { + assertNotNull(entry.messageDeduplicationId(), "each FIFO batch entry must set a deduplication id"); + assertNotNull(entry.messageGroupId(), "each FIFO batch entry must set a group id"); + dedupIds.add(entry.messageDeduplicationId()); + } + assertEquals(4, dedupIds.size(), "the four batch entries must have four distinct deduplication ids"); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("direct:start") + .to("aws2-sqs://camel-1.fifo?amazonSQSClient=#client&operation=sendBatchMessage" + + "&messageGroupIdStrategy=useExchangeId") + .to("mock:result"); + } + }; + } +}