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"); + } + }; + } +}