This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 12360f853d33 CAMEL-25759: camel-aws2-eventbridge - keep one target on
the rule for a queue given with queueUrl (#28074)
12360f853d33 is described below
commit 12360f853d332f6a996dc0a0d105876b1fd34059
Author: allthingssecurity <[email protected]>
AuthorDate: Mon Oct 19 14:50:31 2026 +0530
CAMEL-25759: camel-aws2-eventbridge - keep one target on the rule for a
queue given with queueUrl (#28074)
The consumer adds its SQS queue as a target of the EventBridge rule when it
starts, with a new random target id each time, and removes that target when
it
stops. With a queue given by the user (queueUrl) this leaves the rule with
two
targets for the same queue when two consumers share the queue (two routes,
or
two instances of the application), or when a consumer was not stopped
cleanly
(the JVM was killed) and starts again. EventBridge sends every matching
event
to every target, so every event is put on the queue twice, and routed twice.
The target of a user-provided queue now has an id derived from the queue
ARN,
so every start and every consumer of the queue updates the same target, and
the
target stays on the rule when the consumer stops: other consumers may still
use
the queue, and the queue keeps collecting the events of the rule while no
consumer runs. The auto-created queue (one per consumer, deleted on stop)
keeps
its random target id and its clean-up.
As suggested in the review: an instance on an earlier version and one on
this version add two different targets for the queue, so events are put
on the queue twice until the last instance on the earlier version stops.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../catalog/docs/aws2-eventbridge-component.adoc | 4 +-
.../src/main/docs/aws2-eventbridge-component.adoc | 4 +-
.../aws2/eventbridge/EventbridgeConsumer.java | 32 +++-
.../aws2/eventbridge/EventbridgeConsumerTest.java | 6 +-
.../EventbridgeConsumerUserQueueTargetTest.java | 161 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 11 ++
6 files changed, 207 insertions(+), 11 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws2-eventbridge-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws2-eventbridge-component.adoc
index 78a53b95aaec..ff540e28fcc4 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws2-eventbridge-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws2-eventbridge-component.adoc
@@ -612,7 +612,9 @@ This is the standard AWS pattern for programmatically
consuming EventBridge even
3. It adds the queue as a target of the EventBridge rule specified by
`ruleName`.
4. It polls the SQS queue on schedule, delivering each event into the Camel
route.
5. Successfully processed messages are deleted from the queue; failed messages
remain for retry.
-6. On shutdown the target is removed and the auto-created queue is deleted
(configurable).
+6. On shutdown the target of an auto-created queue is removed and the queue is
deleted (configurable).
+A queue given with `queueUrl` keeps its target, so it collects the events
while no consumer runs, and all the
+consumers that share it use the same target (its id is derived from the queue
ARN).
=== Consumer Configuration Options
diff --git
a/components/camel-aws/camel-aws2-eventbridge/src/main/docs/aws2-eventbridge-component.adoc
b/components/camel-aws/camel-aws2-eventbridge/src/main/docs/aws2-eventbridge-component.adoc
index 78a53b95aaec..ff540e28fcc4 100644
---
a/components/camel-aws/camel-aws2-eventbridge/src/main/docs/aws2-eventbridge-component.adoc
+++
b/components/camel-aws/camel-aws2-eventbridge/src/main/docs/aws2-eventbridge-component.adoc
@@ -612,7 +612,9 @@ This is the standard AWS pattern for programmatically
consuming EventBridge even
3. It adds the queue as a target of the EventBridge rule specified by
`ruleName`.
4. It polls the SQS queue on schedule, delivering each event into the Camel
route.
5. Successfully processed messages are deleted from the queue; failed messages
remain for retry.
-6. On shutdown the target is removed and the auto-created queue is deleted
(configurable).
+6. On shutdown the target of an auto-created queue is removed and the queue is
deleted (configurable).
+A queue given with `queueUrl` keeps its target, so it collects the events
while no consumer runs, and all the
+consumers that share it use the same target (its id is derived from the queue
ARN).
=== Consumer Configuration Options
diff --git
a/components/camel-aws/camel-aws2-eventbridge/src/main/java/org/apache/camel/component/aws2/eventbridge/EventbridgeConsumer.java
b/components/camel-aws/camel-aws2-eventbridge/src/main/java/org/apache/camel/component/aws2/eventbridge/EventbridgeConsumer.java
index abf4989734b2..54a3450a8dfb 100644
---
a/components/camel-aws/camel-aws2-eventbridge/src/main/java/org/apache/camel/component/aws2/eventbridge/EventbridgeConsumer.java
+++
b/components/camel-aws/camel-aws2-eventbridge/src/main/java/org/apache/camel/component/aws2/eventbridge/EventbridgeConsumer.java
@@ -16,6 +16,10 @@
*/
package org.apache.camel.component.aws2.eventbridge;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.HexFormat;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
@@ -61,7 +65,7 @@ import
software.amazon.awssdk.services.sqs.model.SetQueueAttributesRequest;
*
* <p>
* The consumer can auto-create an SQS queue and wire it to the EventBridge
rule, or use a user-provided queue. On
- * shutdown, auto-provisioned resources are cleaned up by default.
+ * shutdown, auto-provisioned resources are cleaned up by default. A
user-provided queue keeps its target on the rule.
* </p>
*/
public class EventbridgeConsumer extends ScheduledBatchPollingConsumer {
@@ -184,8 +188,12 @@ public class EventbridgeConsumer extends
ScheduledBatchPollingConsumer {
"Either queueUrl must be specified or autoCreateQueue must
be enabled for EventBridge consumer");
}
- // Add queue as EventBridge target
- targetId = "camel-eb-" + UUID.randomUUID().toString().substring(0, 8);
+ // Add queue as EventBridge target. The target of a user-provided
queue has an id derived from the queue, so
+ // every start and every consumer that shares the queue updates the
same target (putTargets replaces a target
+ // with the same id): two targets for one queue would each put a copy
of every event on the queue.
+ targetId = queueAutoCreated
+ ? "camel-eb-" + UUID.randomUUID().toString().substring(0, 8)
+ : targetIdForQueue(queueArn);
Target target = Target.builder()
.id(targetId)
.arn(queueArn)
@@ -211,8 +219,9 @@ public class EventbridgeConsumer extends
ScheduledBatchPollingConsumer {
EventbridgeConfiguration config = getEndpoint().getConfiguration();
try {
- // Remove target from EventBridge rule
- if (ObjectHelper.isNotEmpty(targetId)) {
+ // Remove the target of an auto-created queue. The target of a
user-provided queue is kept: other consumers
+ // can share the queue, and the queue keeps collecting the events
of the rule while this consumer is stopped.
+ if (queueAutoCreated && ObjectHelper.isNotEmpty(targetId)) {
EventBridgeClient ebClient =
getEndpoint().getEventbridgeClient();
ebClient.removeTargets(
RemoveTargetsRequest.builder()
@@ -265,6 +274,19 @@ public class EventbridgeConsumer extends
ScheduledBatchPollingConsumer {
}
}
+ /**
+ * The EventBridge target id for a user-provided queue: the same for every
start and every consumer of the queue.
+ */
+ static String targetIdForQueue(String queueArn) {
+ try {
+ byte[] digest =
MessageDigest.getInstance("SHA-256").digest(queueArn.getBytes(StandardCharsets.UTF_8));
+ // a target id has at most 64 characters of [.-_A-Za-z0-9]
+ return "camel-eb-" + HexFormat.of().formatHex(digest, 0, 16);
+ } catch (NoSuchAlgorithmException e) {
+ throw new IllegalStateException("SHA-256 is not available", e);
+ }
+ }
+
private String getQueueArn(String queueUrl) {
GetQueueAttributesResponse response = sqsClient.getQueueAttributes(
GetQueueAttributesRequest.builder()
diff --git
a/components/camel-aws/camel-aws2-eventbridge/src/test/java/org/apache/camel/component/aws2/eventbridge/EventbridgeConsumerTest.java
b/components/camel-aws/camel-aws2-eventbridge/src/test/java/org/apache/camel/component/aws2/eventbridge/EventbridgeConsumerTest.java
index e68b908a7042..32e254aaa1fd 100644
---
a/components/camel-aws/camel-aws2-eventbridge/src/test/java/org/apache/camel/component/aws2/eventbridge/EventbridgeConsumerTest.java
+++
b/components/camel-aws/camel-aws2-eventbridge/src/test/java/org/apache/camel/component/aws2/eventbridge/EventbridgeConsumerTest.java
@@ -371,15 +371,13 @@ public class EventbridgeConsumerTest extends
CamelTestSupport {
public void testUserProvidedQueueNotDeletedOnShutdown() throws Exception {
stubGetQueueAttributes(USER_QUEUE_URL, USER_QUEUE_ARN);
stubPutTargetsSuccess();
- when(eventBridgeClient.removeTargets(any(RemoveTargetsRequest.class)))
- .thenReturn(RemoveTargetsResponse.builder().build());
- // deleteQueueOnShutdown=true, but queue was user-provided so should
NOT be deleted
+ // deleteQueueOnShutdown=true, but queue was user-provided so should
NOT be deleted, and its target stays
EventbridgeConsumer consumer = createConsumer("test-rule",
USER_QUEUE_URL, true, true);
consumer.start();
consumer.stop();
-
verify(eventBridgeClient).removeTargets(any(RemoveTargetsRequest.class));
+ verify(eventBridgeClient,
never()).removeTargets(any(RemoveTargetsRequest.class));
verify(sqsClient, never()).deleteQueue(any(DeleteQueueRequest.class));
}
diff --git
a/components/camel-aws/camel-aws2-eventbridge/src/test/java/org/apache/camel/component/aws2/eventbridge/EventbridgeConsumerUserQueueTargetTest.java
b/components/camel-aws/camel-aws2-eventbridge/src/test/java/org/apache/camel/component/aws2/eventbridge/EventbridgeConsumerUserQueueTargetTest.java
new file mode 100644
index 000000000000..7c271a2f2824
--- /dev/null
+++
b/components/camel-aws/camel-aws2-eventbridge/src/test/java/org/apache/camel/component/aws2/eventbridge/EventbridgeConsumerUserQueueTargetTest.java
@@ -0,0 +1,161 @@
+/*
+ * 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.eventbridge;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+import software.amazon.awssdk.services.eventbridge.EventBridgeClient;
+import software.amazon.awssdk.services.eventbridge.model.PutTargetsRequest;
+import software.amazon.awssdk.services.eventbridge.model.PutTargetsResponse;
+import software.amazon.awssdk.services.eventbridge.model.RemoveTargetsRequest;
+import software.amazon.awssdk.services.eventbridge.model.RemoveTargetsResponse;
+import software.amazon.awssdk.services.eventbridge.model.Target;
+import software.amazon.awssdk.services.sqs.SqsClient;
+import software.amazon.awssdk.services.sqs.model.CreateQueueRequest;
+import software.amazon.awssdk.services.sqs.model.CreateQueueResponse;
+import software.amazon.awssdk.services.sqs.model.GetQueueAttributesRequest;
+import software.amazon.awssdk.services.sqs.model.GetQueueAttributesResponse;
+import software.amazon.awssdk.services.sqs.model.QueueAttributeName;
+import software.amazon.awssdk.services.sqs.model.SetQueueAttributesRequest;
+import software.amazon.awssdk.services.sqs.model.SetQueueAttributesResponse;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * The targets of the EventBridge rule when consumers use a user-provided
queue. EventBridge puts a copy of every
+ * matching event on the queue of every target of the rule, so the rule must
hold one target for the queue whatever the
+ * number of consumers that share it, and whether a consumer stopped cleanly
or not. The rule is a map of target id to
+ * target ARN kept by the mocked EventBridge client, with the semantics of
PutTargets (a target with the same id is
+ * replaced) and RemoveTargets.
+ */
+class EventbridgeConsumerUserQueueTargetTest extends CamelTestSupport {
+
+ private static final String USER_QUEUE_URL =
"https://sqs.us-east-1.amazonaws.com/123456789/user-queue";;
+ private static final String USER_QUEUE_ARN =
"arn:aws:sqs:us-east-1:123456789:user-queue";
+ private static final String AUTO_QUEUE_URL =
"https://sqs.us-east-1.amazonaws.com/123456789/camel-eventbridge-auto";;
+ private static final String AUTO_QUEUE_ARN =
"arn:aws:sqs:us-east-1:123456789:camel-eventbridge-auto";
+
+ private final Map<String, String> ruleTargets = new LinkedHashMap<>();
+ private SqsClient sqsClient;
+ private EventBridgeClient eventBridgeClient;
+
+ @BeforeEach
+ void setupClients() {
+ sqsClient = mock(SqsClient.class);
+ eventBridgeClient = mock(EventBridgeClient.class);
+
when(sqsClient.getQueueAttributes(any(GetQueueAttributesRequest.class))).thenAnswer(inv
-> {
+ GetQueueAttributesRequest req = inv.getArgument(0);
+ String arn = USER_QUEUE_URL.equals(req.queueUrl()) ?
USER_QUEUE_ARN : AUTO_QUEUE_ARN;
+ return
GetQueueAttributesResponse.builder().attributes(Map.of(QueueAttributeName.QUEUE_ARN,
arn)).build();
+ });
+ when(sqsClient.createQueue(any(CreateQueueRequest.class)))
+
.thenReturn(CreateQueueResponse.builder().queueUrl(AUTO_QUEUE_URL).build());
+
when(sqsClient.setQueueAttributes(any(SetQueueAttributesRequest.class)))
+ .thenReturn(SetQueueAttributesResponse.builder().build());
+
when(eventBridgeClient.putTargets(any(PutTargetsRequest.class))).thenAnswer(inv
-> {
+ PutTargetsRequest req = inv.getArgument(0);
+ for (Target target : req.targets()) {
+ ruleTargets.put(target.id(), target.arn());
+ }
+ return PutTargetsResponse.builder().failedEntryCount(0).build();
+ });
+
when(eventBridgeClient.removeTargets(any(RemoveTargetsRequest.class))).thenAnswer(inv
-> {
+ RemoveTargetsRequest req = inv.getArgument(0);
+ req.ids().forEach(ruleTargets::remove);
+ return RemoveTargetsResponse.builder().build();
+ });
+ }
+
+ @Test
+ void twoConsumersSharingAQueueAddOneTarget() throws Exception {
+ EventbridgeConsumer first = createConsumer(USER_QUEUE_URL);
+ EventbridgeConsumer second = createConsumer(USER_QUEUE_URL);
+ first.start();
+ second.start();
+
+ // two targets for the queue would put every event on the queue twice
+ assertEquals(1, targetsFor(USER_QUEUE_ARN), "targets of the rule: " +
ruleTargets);
+
+ second.stop();
+ // the consumer that still runs keeps receiving the events of the rule
+ assertEquals(1, targetsFor(USER_QUEUE_ARN), "targets of the rule: " +
ruleTargets);
+ first.stop();
+ }
+
+ @Test
+ void restartAfterACrashAddsNoSecondTarget() throws Exception {
+ // the first consumer is never stopped, as when the JVM is killed: its
target is left on the rule
+ EventbridgeConsumer crashed = createConsumer(USER_QUEUE_URL);
+ crashed.start();
+ EventbridgeConsumer restarted = createConsumer(USER_QUEUE_URL);
+ restarted.start();
+
+ assertEquals(1, targetsFor(USER_QUEUE_ARN), "targets of the rule: " +
ruleTargets);
+ restarted.stop();
+ }
+
+ @Test
+ void userQueueKeepsCollectingEventsWhileTheConsumerIsStopped() throws
Exception {
+ EventbridgeConsumer consumer = createConsumer(USER_QUEUE_URL);
+ consumer.start();
+ consumer.stop();
+
+ assertEquals(1, targetsFor(USER_QUEUE_ARN), "targets of the rule: " +
ruleTargets);
+ }
+
+ @Test
+ void autoCreatedQueueTargetIsStillRemovedOnStop() throws Exception {
+ EventbridgeConsumer consumer = createConsumer(null);
+ consumer.start();
+ assertEquals(1, targetsFor(AUTO_QUEUE_ARN), "targets of the rule: " +
ruleTargets);
+ consumer.stop();
+
+ assertTrue(ruleTargets.isEmpty(), "targets of the rule: " +
ruleTargets);
+ }
+
+ private long targetsFor(String arn) {
+ return ruleTargets.values().stream().filter(arn::equals).count();
+ }
+
+ private EventbridgeConsumer createConsumer(String queueUrl) throws
Exception {
+ EventbridgeComponent component =
context.getComponent("aws2-eventbridge", EventbridgeComponent.class);
+ String uri =
"aws2-eventbridge://default?accessKey=test&secretKey=test&ruleName=test-rule";
+ if (queueUrl != null) {
+ uri += "&queueUrl=" + queueUrl;
+ }
+ EventbridgeEndpoint endpoint = (EventbridgeEndpoint)
component.createEndpoint(uri);
+ endpoint.getConfiguration().setEventbridgeClient(eventBridgeClient);
+ EventbridgeEndpoint spyEndpoint = Mockito.spy(endpoint);
+ doReturn(sqsClient).when(spyEndpoint).getSqsClient();
+ doReturn(eventBridgeClient).when(spyEndpoint).getEventbridgeClient();
+
+ EventbridgeConsumer consumer = new EventbridgeConsumer(spyEndpoint,
exchange -> {
+ });
+ consumer.setStartScheduler(false);
+ return consumer;
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 5b1e7e99e39d..e5ad3f95769e 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -1439,6 +1439,17 @@ Previously it returned only the first chunk of the event
log (at most 500 events
any sign. A wide `after`/`before` window now returns every event in one list:
that is one request per 500 events plus
one that finds the end of the log, and more memory. Narrow the window when the
event log is large.
+=== camel-aws2-eventbridge - the consumer keeps one target for a queue given
with queueUrl
+
+A consumer that uses its own queue (`queueUrl`) added a target with a new
random id to the EventBridge rule at every
+start, and removed it on stop. Two consumers that share the queue, or a
restart after a consumer was not stopped
+cleanly, left two targets for the queue, and EventBridge put every event on
the queue twice. The target now has an id
+derived from the queue ARN, and stays on the rule when the consumer stops, so
the queue also collects the events of the
+rule while no consumer runs. Remove the target from the rule yourself when you
no longer use the queue, and remove
+targets with ids `camel-eb-` plus 8 characters that an earlier version left
for the queue. An auto-created queue is
+not affected. During a rolling upgrade from an earlier version, events can be
put on the queue twice until the last
+instance on the earlier version stops.
+
=== camel-chunk - message data printed with a direct tag is rendered literally
Previously the `chunk` component bound the message body, headers and variables
into the Chunk engine so that
