This is an automated email from the ASF dual-hosted git repository.
oscerd pushed a commit to branch camel-4.14.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.14.x by this push:
new 9419451ec032 [backport camel-4.14.x] CAMEL-24086: camel-aws2-sqs -
give each FIFO batch entry a distinct MessageDeduplicationId (#24786)
9419451ec032 is described below
commit 9419451ec032bf0112800ff09374670f4089db57
Author: Andrea Cosentino <[email protected]>
AuthorDate: Thu Jul 16 15:12:12 2026 +0200
[backport camel-4.14.x] CAMEL-24086: camel-aws2-sqs - give each FIFO batch
entry a distinct MessageDeduplicationId (#24786)
CAMEL-24086: camel-aws2-sqs - give each FIFO batch entry a distinct
MessageDeduplicationId (#24728)
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.
Signed-off-by: Andrea Cosentino <[email protected]>
Co-authored-by: Claude Fable 5 <[email protected]>
---
.../camel/component/aws2/sqs/Sqs2Producer.java | 34 +++++---
.../sqs/SqsProducerFifoBatchDeduplicationTest.java | 96 ++++++++++++++++++++++
2 files changed, 119 insertions(+), 11 deletions(-)
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 5a006a5f36c7..7ebd81c5a35b 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
@@ -120,6 +120,7 @@ public class Sqs2Producer extends DefaultProducer {
Collection<SendMessageBatchRequestEntry> 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();
@@ -127,7 +128,7 @@ public class Sqs2Producer extends DefaultProducer {
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);
@@ -137,13 +138,14 @@ public class Sqs2Producer extends DefaultProducer {
} 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);
@@ -216,17 +218,27 @@ public class Sqs2Producer extends DefaultProducer {
}
}
- 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 000000000000..9b19b2296281
--- /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.junit5.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<String> c = new ArrayList<>();
+ c.add("team1");
+ c.add("team2");
+ c.add("team3");
+ c.add("team4");
+ exchange.getIn().setBody(c);
+ }
+ });
+ MockEndpoint.assertIsSatisfied(context);
+
+ List<SendMessageBatchRequest> requests =
mock.getSendMessageBatchRequests();
+ assertEquals(1, requests.size());
+ List<SendMessageBatchRequestEntry> 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<String> 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");
+ }
+ };
+ }
+}
