This is an automated email from the ASF dual-hosted git repository.
oscerd pushed a commit to branch camel-4.18.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.18.x by this push:
new 52df8d8b5a0f [backport camel-4.18.x] CAMEL-24082: camel-aws-cloudtrail
- fix consumer event loss (static cursor, no pagination, window skip) (#24798)
52df8d8b5a0f is described below
commit 52df8d8b5a0f161d8633ed4871ac215d25b9d5df
Author: Andrea Cosentino <[email protected]>
AuthorDate: Thu Jul 16 15:28:34 2026 +0200
[backport camel-4.18.x] CAMEL-24082: camel-aws-cloudtrail - fix consumer
event loss (static cursor, no pagination, window skip) (#24798)
Backport of #24726.
Co-authored-by: Claude Fable 5 <[email protected]>
---
.../camel/catalog/components/aws-cloudtrail.json | 4 +-
components/camel-aws/camel-aws-cloudtrail/pom.xml | 6 ++
.../component/aws/cloudtrail/aws-cloudtrail.json | 4 +-
.../aws/cloudtrail/CloudtrailConfiguration.java | 8 +-
.../aws/cloudtrail/CloudtrailConsumer.java | 87 ++++++++++++-----
.../aws/cloudtrail/CloudtrailConsumerTest.java | 106 +++++++++++++++++++++
.../dsl/AwsCloudtrailComponentBuilderFactory.java | 6 +-
.../dsl/CloudtrailEndpointBuilderFactory.java | 12 ++-
8 files changed, 198 insertions(+), 35 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws-cloudtrail.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws-cloudtrail.json
index f8fddd7e50a9..af4a06d45445 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws-cloudtrail.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws-cloudtrail.json
@@ -27,7 +27,7 @@
"bridgeErrorHandler": { "index": 0, "kind": "property", "displayName":
"Bridge Error Handler", "group": "consumer", "label": "consumer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false, "description":
"Allows for bridging the consumer to the Camel routing Error Handler, which
mean any exceptions (if possible) occurred while the Camel consumer is trying
to pickup incoming messages, or the like [...]
"configuration": { "index": 1, "kind": "property", "displayName":
"Configuration", "group": "consumer", "label": "", "required": false, "type":
"object", "javaType":
"org.apache.camel.component.aws.cloudtrail.CloudtrailConfiguration",
"deprecated": false, "autowired": false, "secret": false, "description":
"Component configuration" },
"eventSource": { "index": 2, "kind": "property", "displayName": "Event
Source", "group": "consumer", "label": "", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": false, "configurationClass":
"org.apache.camel.component.aws.cloudtrail.CloudtrailConfiguration",
"configurationField": "configuration", "description": "Specify an event source
to select events" },
- "maxResults": { "index": 3, "kind": "property", "displayName": "Max
Results", "group": "consumer", "label": "", "required": false, "type":
"integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 1, "configurationClass":
"org.apache.camel.component.aws.cloudtrail.CloudtrailConfiguration",
"configurationField": "configuration", "description": "Maximum number of
records that will be fetched in each poll" },
+ "maxResults": { "index": 3, "kind": "property", "displayName": "Max
Results", "group": "consumer", "label": "", "required": false, "type":
"integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 50, "configurationClass":
"org.apache.camel.component.aws.cloudtrail.CloudtrailConfiguration",
"configurationField": "configuration", "description": "Maximum number of
records that will be fetched in each lookup page. Each poll drains all pages,
[...]
"overrideEndpoint": { "index": 4, "kind": "property", "displayName":
"Override Endpoint", "group": "consumer", "label": "", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.aws.cloudtrail.CloudtrailConfiguration",
"configurationField": "configuration", "description": "Set the need for
overriding the endpoint. This option needs to be used in co [...]
"region": { "index": 5, "kind": "property", "displayName": "Region",
"group": "consumer", "label": "", "required": false, "type": "enum",
"javaType": "java.lang.String", "enum": [ "ap-south-2", "ap-south-1",
"eu-south-1", "eu-south-2", "us-gov-east-1", "me-central-1", "il-central-1",
"ca-central-1", "eu-central-1", "us-iso-west-1", "eu-central-2",
"eu-isoe-west-1", "us-west-1", "us-west-2", "af-south-1", "eu-north-1",
"eu-west-3", "eu-west-2", "eu-west-1", "ap-northeast-3", "ap-north [...]
"uriEndpointOverride": { "index": 6, "kind": "property", "displayName":
"Uri Endpoint Override", "group": "consumer", "label": "", "required": false,
"type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.aws.cloudtrail.CloudtrailConfiguration",
"configurationField": "configuration", "description": "Set the overriding uri
endpoint. This option needs to be used in combination with ov [...]
@@ -56,7 +56,7 @@
"properties": {
"label": { "index": 0, "kind": "path", "displayName": "Label", "group":
"consumer", "label": "", "required": true, "type": "string", "javaType":
"java.lang.String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.aws.cloudtrail.CloudtrailConfiguration",
"configurationField": "configuration", "description": "A label for indexing
cloudtrail endpoints" },
"eventSource": { "index": 1, "kind": "parameter", "displayName": "Event
Source", "group": "consumer", "label": "", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": false, "configurationClass":
"org.apache.camel.component.aws.cloudtrail.CloudtrailConfiguration",
"configurationField": "configuration", "description": "Specify an event source
to select events" },
- "maxResults": { "index": 2, "kind": "parameter", "displayName": "Max
Results", "group": "consumer", "label": "", "required": false, "type":
"integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 1, "configurationClass":
"org.apache.camel.component.aws.cloudtrail.CloudtrailConfiguration",
"configurationField": "configuration", "description": "Maximum number of
records that will be fetched in each poll" },
+ "maxResults": { "index": 2, "kind": "parameter", "displayName": "Max
Results", "group": "consumer", "label": "", "required": false, "type":
"integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 50, "configurationClass":
"org.apache.camel.component.aws.cloudtrail.CloudtrailConfiguration",
"configurationField": "configuration", "description": "Maximum number of
records that will be fetched in each lookup page. Each poll drains all pages
[...]
"overrideEndpoint": { "index": 3, "kind": "parameter", "displayName":
"Override Endpoint", "group": "consumer", "label": "", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.aws.cloudtrail.CloudtrailConfiguration",
"configurationField": "configuration", "description": "Set the need for
overriding the endpoint. This option needs to be used in c [...]
"region": { "index": 4, "kind": "parameter", "displayName": "Region",
"group": "consumer", "label": "", "required": false, "type": "enum",
"javaType": "java.lang.String", "enum": [ "ap-south-2", "ap-south-1",
"eu-south-1", "eu-south-2", "us-gov-east-1", "me-central-1", "il-central-1",
"ca-central-1", "eu-central-1", "us-iso-west-1", "eu-central-2",
"eu-isoe-west-1", "us-west-1", "us-west-2", "af-south-1", "eu-north-1",
"eu-west-3", "eu-west-2", "eu-west-1", "ap-northeast-3", "ap-nort [...]
"sendEmptyMessageWhenIdle": { "index": 5, "kind": "parameter",
"displayName": "Send Empty Message When Idle", "group": "consumer", "label":
"consumer", "required": false, "type": "boolean", "javaType": "boolean",
"deprecated": false, "autowired": false, "secret": false, "defaultValue":
false, "description": "If the polling consumer did not poll any files, you can
enable this option to send an empty message (no body) instead." },
diff --git a/components/camel-aws/camel-aws-cloudtrail/pom.xml
b/components/camel-aws/camel-aws-cloudtrail/pom.xml
index e8e1905c6ad2..f91039d46eb2 100644
--- a/components/camel-aws/camel-aws-cloudtrail/pom.xml
+++ b/components/camel-aws/camel-aws-cloudtrail/pom.xml
@@ -70,6 +70,12 @@
<artifactId>camel-test-spring-junit5</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-junit-jupiter</artifactId>
+ <version>${mockito-version}</version>
+ <scope>test</scope>
+ </dependency>
<!-- test infra -->
<dependency>
diff --git
a/components/camel-aws/camel-aws-cloudtrail/src/generated/resources/META-INF/org/apache/camel/component/aws/cloudtrail/aws-cloudtrail.json
b/components/camel-aws/camel-aws-cloudtrail/src/generated/resources/META-INF/org/apache/camel/component/aws/cloudtrail/aws-cloudtrail.json
index f8fddd7e50a9..af4a06d45445 100644
---
a/components/camel-aws/camel-aws-cloudtrail/src/generated/resources/META-INF/org/apache/camel/component/aws/cloudtrail/aws-cloudtrail.json
+++
b/components/camel-aws/camel-aws-cloudtrail/src/generated/resources/META-INF/org/apache/camel/component/aws/cloudtrail/aws-cloudtrail.json
@@ -27,7 +27,7 @@
"bridgeErrorHandler": { "index": 0, "kind": "property", "displayName":
"Bridge Error Handler", "group": "consumer", "label": "consumer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false, "description":
"Allows for bridging the consumer to the Camel routing Error Handler, which
mean any exceptions (if possible) occurred while the Camel consumer is trying
to pickup incoming messages, or the like [...]
"configuration": { "index": 1, "kind": "property", "displayName":
"Configuration", "group": "consumer", "label": "", "required": false, "type":
"object", "javaType":
"org.apache.camel.component.aws.cloudtrail.CloudtrailConfiguration",
"deprecated": false, "autowired": false, "secret": false, "description":
"Component configuration" },
"eventSource": { "index": 2, "kind": "property", "displayName": "Event
Source", "group": "consumer", "label": "", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": false, "configurationClass":
"org.apache.camel.component.aws.cloudtrail.CloudtrailConfiguration",
"configurationField": "configuration", "description": "Specify an event source
to select events" },
- "maxResults": { "index": 3, "kind": "property", "displayName": "Max
Results", "group": "consumer", "label": "", "required": false, "type":
"integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 1, "configurationClass":
"org.apache.camel.component.aws.cloudtrail.CloudtrailConfiguration",
"configurationField": "configuration", "description": "Maximum number of
records that will be fetched in each poll" },
+ "maxResults": { "index": 3, "kind": "property", "displayName": "Max
Results", "group": "consumer", "label": "", "required": false, "type":
"integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 50, "configurationClass":
"org.apache.camel.component.aws.cloudtrail.CloudtrailConfiguration",
"configurationField": "configuration", "description": "Maximum number of
records that will be fetched in each lookup page. Each poll drains all pages,
[...]
"overrideEndpoint": { "index": 4, "kind": "property", "displayName":
"Override Endpoint", "group": "consumer", "label": "", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.aws.cloudtrail.CloudtrailConfiguration",
"configurationField": "configuration", "description": "Set the need for
overriding the endpoint. This option needs to be used in co [...]
"region": { "index": 5, "kind": "property", "displayName": "Region",
"group": "consumer", "label": "", "required": false, "type": "enum",
"javaType": "java.lang.String", "enum": [ "ap-south-2", "ap-south-1",
"eu-south-1", "eu-south-2", "us-gov-east-1", "me-central-1", "il-central-1",
"ca-central-1", "eu-central-1", "us-iso-west-1", "eu-central-2",
"eu-isoe-west-1", "us-west-1", "us-west-2", "af-south-1", "eu-north-1",
"eu-west-3", "eu-west-2", "eu-west-1", "ap-northeast-3", "ap-north [...]
"uriEndpointOverride": { "index": 6, "kind": "property", "displayName":
"Uri Endpoint Override", "group": "consumer", "label": "", "required": false,
"type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.aws.cloudtrail.CloudtrailConfiguration",
"configurationField": "configuration", "description": "Set the overriding uri
endpoint. This option needs to be used in combination with ov [...]
@@ -56,7 +56,7 @@
"properties": {
"label": { "index": 0, "kind": "path", "displayName": "Label", "group":
"consumer", "label": "", "required": true, "type": "string", "javaType":
"java.lang.String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.aws.cloudtrail.CloudtrailConfiguration",
"configurationField": "configuration", "description": "A label for indexing
cloudtrail endpoints" },
"eventSource": { "index": 1, "kind": "parameter", "displayName": "Event
Source", "group": "consumer", "label": "", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": false, "configurationClass":
"org.apache.camel.component.aws.cloudtrail.CloudtrailConfiguration",
"configurationField": "configuration", "description": "Specify an event source
to select events" },
- "maxResults": { "index": 2, "kind": "parameter", "displayName": "Max
Results", "group": "consumer", "label": "", "required": false, "type":
"integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 1, "configurationClass":
"org.apache.camel.component.aws.cloudtrail.CloudtrailConfiguration",
"configurationField": "configuration", "description": "Maximum number of
records that will be fetched in each poll" },
+ "maxResults": { "index": 2, "kind": "parameter", "displayName": "Max
Results", "group": "consumer", "label": "", "required": false, "type":
"integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 50, "configurationClass":
"org.apache.camel.component.aws.cloudtrail.CloudtrailConfiguration",
"configurationField": "configuration", "description": "Maximum number of
records that will be fetched in each lookup page. Each poll drains all pages
[...]
"overrideEndpoint": { "index": 3, "kind": "parameter", "displayName":
"Override Endpoint", "group": "consumer", "label": "", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.aws.cloudtrail.CloudtrailConfiguration",
"configurationField": "configuration", "description": "Set the need for
overriding the endpoint. This option needs to be used in c [...]
"region": { "index": 4, "kind": "parameter", "displayName": "Region",
"group": "consumer", "label": "", "required": false, "type": "enum",
"javaType": "java.lang.String", "enum": [ "ap-south-2", "ap-south-1",
"eu-south-1", "eu-south-2", "us-gov-east-1", "me-central-1", "il-central-1",
"ca-central-1", "eu-central-1", "us-iso-west-1", "eu-central-2",
"eu-isoe-west-1", "us-west-1", "us-west-2", "af-south-1", "eu-north-1",
"eu-west-3", "eu-west-2", "eu-west-1", "ap-northeast-3", "ap-nort [...]
"sendEmptyMessageWhenIdle": { "index": 5, "kind": "parameter",
"displayName": "Send Empty Message When Idle", "group": "consumer", "label":
"consumer", "required": false, "type": "boolean", "javaType": "boolean",
"deprecated": false, "autowired": false, "secret": false, "defaultValue":
false, "description": "If the polling consumer did not poll any files, you can
enable this option to send an empty message (no body) instead." },
diff --git
a/components/camel-aws/camel-aws-cloudtrail/src/main/java/org/apache/camel/component/aws/cloudtrail/CloudtrailConfiguration.java
b/components/camel-aws/camel-aws-cloudtrail/src/main/java/org/apache/camel/component/aws/cloudtrail/CloudtrailConfiguration.java
index b705b743379b..5d02bcb3fb6c 100644
---
a/components/camel-aws/camel-aws-cloudtrail/src/main/java/org/apache/camel/component/aws/cloudtrail/CloudtrailConfiguration.java
+++
b/components/camel-aws/camel-aws-cloudtrail/src/main/java/org/apache/camel/component/aws/cloudtrail/CloudtrailConfiguration.java
@@ -47,9 +47,11 @@ public class CloudtrailConfiguration implements Cloneable,
AwsCommonConfiguratio
@UriParam(label = "advanced", description = "Amazon Cloudtrail client to
use for all requests for this endpoint")
@Metadata(autowired = true)
private CloudTrailClient cloudTrailClient;
- @UriParam(description = "Maximum number of records that will be fetched in
each poll",
- defaultValue = "1")
- private int maxResults = 1;
+ @UriParam(description = "Maximum number of records that will be fetched in
each lookup page. Each poll drains"
+ + " all pages, so this controls the page size
rather than the total per poll. AWS allows"
+ + " up to 50.",
+ defaultValue = "50")
+ private int maxResults = 50;
@UriParam(description = "Specify an event source to select events")
private String eventSource;
@UriParam(label = "proxy", enums = "HTTP,HTTPS", defaultValue = "HTTPS",
diff --git
a/components/camel-aws/camel-aws-cloudtrail/src/main/java/org/apache/camel/component/aws/cloudtrail/CloudtrailConsumer.java
b/components/camel-aws/camel-aws-cloudtrail/src/main/java/org/apache/camel/component/aws/cloudtrail/CloudtrailConsumer.java
index 2bd5ba58cfbb..5516d85787be 100644
---
a/components/camel-aws/camel-aws-cloudtrail/src/main/java/org/apache/camel/component/aws/cloudtrail/CloudtrailConsumer.java
+++
b/components/camel-aws/camel-aws-cloudtrail/src/main/java/org/apache/camel/component/aws/cloudtrail/CloudtrailConsumer.java
@@ -20,8 +20,10 @@ import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.util.ArrayDeque;
import java.util.ArrayList;
+import java.util.HashSet;
import java.util.List;
import java.util.Queue;
+import java.util.Set;
import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
@@ -33,40 +35,89 @@ import
software.amazon.awssdk.services.cloudtrail.CloudTrailClient;
import software.amazon.awssdk.services.cloudtrail.model.*;
public class CloudtrailConsumer extends ScheduledBatchPollingConsumer {
- private static Instant lastTime;
+
+ // Cursor for the next lookup window. Kept per-consumer (not static) so
multiple cloudtrail
+ // routes in the same JVM do not clobber each other's position.
+ private Instant lastTime;
+ // Event ids already delivered at exactly lastTime. startTime is
inclusive, so these events are
+ // returned again on the next poll and must be filtered to avoid duplicate
delivery.
+ private final Set<String> lastProcessedEventIds = new HashSet<>();
public CloudtrailConsumer(CloudtrailEndpoint endpoint, Processor
processor) {
super(endpoint, processor);
}
@Override
- protected int poll() throws Exception {
- LookupEventsRequest.Builder eventsRequestBuilder
- =
LookupEventsRequest.builder().maxResults(getEndpoint().getConfiguration().getMaxResults());
+ protected void doStart() throws Exception {
+ super.doStart();
+ if (lastTime == null) {
+ // Start tailing from the consumer start time rather than
replaying CloudTrail history.
+ lastTime = Instant.now();
+ }
+ }
+ @Override
+ protected int poll() throws Exception {
List<LookupAttribute> attributes = new ArrayList<>();
if
(ObjectHelper.isNotEmpty(getEndpoint().getConfiguration().getEventSource())) {
LookupAttribute eventSource =
LookupAttribute.builder().attributeKey(LookupAttributeKey.EVENT_SOURCE)
.attributeValue(getEndpoint().getConfiguration().getEventSource()).build();
attributes.add(eventSource);
}
- if (!attributes.isEmpty()) {
- eventsRequestBuilder.lookupAttributes(attributes);
- }
- if (ObjectHelper.isNotEmpty(lastTime)) {
- eventsRequestBuilder.startTime(lastTime.plusMillis(1000));
- }
- LookupEventsResponse response =
getClient().lookupEvents(eventsRequestBuilder.build());
+ // Drain every page of the current window so no event is silently left
behind.
+ List<Event> events = new ArrayList<>();
+ String nextToken = null;
+ do {
+ LookupEventsRequest.Builder eventsRequestBuilder
+ =
LookupEventsRequest.builder().maxResults(getEndpoint().getConfiguration().getMaxResults());
+ if (!attributes.isEmpty()) {
+ eventsRequestBuilder.lookupAttributes(attributes);
+ }
+ if (ObjectHelper.isNotEmpty(lastTime)) {
+ eventsRequestBuilder.startTime(lastTime);
+ }
+ if (nextToken != null) {
+ eventsRequestBuilder.nextToken(nextToken);
+ }
+
+ LookupEventsResponse response =
getClient().lookupEvents(eventsRequestBuilder.build());
+ events.addAll(response.events());
+ nextToken = response.nextToken();
+ } while (ObjectHelper.isNotEmpty(nextToken));
// okay we have some response from aws so lets mark the consumer as
ready
forceConsumerAsReady();
- if (!response.events().isEmpty()) {
- lastTime = response.events().get(0).eventTime();
+ // CloudTrail returns events newest-first. Advance the cursor to the
newest event time and
+ // track the ids seen at that instant, skipping any event already
delivered on a prior poll.
+ Instant newest = lastTime;
+ Set<String> newestEventIds = new HashSet<>();
+ Queue<Exchange> exchanges = new ArrayDeque<>();
+ for (Event event : events) {
+ if (lastProcessedEventIds.contains(event.eventId())) {
+ continue;
+ }
+ exchanges.add(createExchange(event));
+
+ Instant eventTime = event.eventTime();
+ if (newest == null || eventTime.isAfter(newest)) {
+ newest = eventTime;
+ newestEventIds.clear();
+ newestEventIds.add(event.eventId());
+ } else if (eventTime.equals(newest)) {
+ newestEventIds.add(event.eventId());
+ }
}
- Queue<Exchange> exchanges = createExchanges(response.events());
+ if (newest != null && newest.equals(lastTime)) {
+ // No event newer than the previous boundary; keep accumulating
ids at that instant.
+ newestEventIds.addAll(lastProcessedEventIds);
+ }
+ lastTime = newest;
+ lastProcessedEventIds.clear();
+ lastProcessedEventIds.addAll(newestEventIds);
+
return processBatch(CastUtils.cast(exchanges));
}
@@ -93,14 +144,6 @@ public class CloudtrailConsumer extends
ScheduledBatchPollingConsumer {
return (CloudtrailEndpoint) super.getEndpoint();
}
- private Queue<Exchange> createExchanges(List<Event> events) {
- Queue<Exchange> exchanges = new ArrayDeque<>();
- for (Event event : events) {
- exchanges.add(createExchange(event));
- }
- return exchanges;
- }
-
protected Exchange createExchange(Event event) {
Exchange exchange = createExchange(true);
exchange.getMessage().setBody(event.cloudTrailEvent().getBytes(StandardCharsets.UTF_8));
diff --git
a/components/camel-aws/camel-aws-cloudtrail/src/test/java/org/apache/camel/component/aws/cloudtrail/CloudtrailConsumerTest.java
b/components/camel-aws/camel-aws-cloudtrail/src/test/java/org/apache/camel/component/aws/cloudtrail/CloudtrailConsumerTest.java
new file mode 100644
index 000000000000..414573afe44f
--- /dev/null
+++
b/components/camel-aws/camel-aws-cloudtrail/src/test/java/org/apache/camel/component/aws/cloudtrail/CloudtrailConsumerTest.java
@@ -0,0 +1,106 @@
+/*
+ * 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.aws.cloudtrail;
+
+import java.time.Instant;
+
+import org.apache.camel.AsyncProcessor;
+import org.apache.camel.CamelContext;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
+import software.amazon.awssdk.services.cloudtrail.CloudTrailClient;
+import software.amazon.awssdk.services.cloudtrail.model.Event;
+import software.amazon.awssdk.services.cloudtrail.model.LookupEventsRequest;
+import software.amazon.awssdk.services.cloudtrail.model.LookupEventsResponse;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.LENIENT)
+public class CloudtrailConsumerTest {
+
+ @Mock
+ private CloudTrailClient cloudTrailClient;
+ @Mock
+ private AsyncProcessor processor;
+
+ private final CamelContext context = new DefaultCamelContext();
+ private final CloudtrailComponent component = new
CloudtrailComponent(context);
+
+ private CloudtrailConsumer underTest;
+
+ private static Event event(String id, Instant time) {
+ return
Event.builder().eventId(id).eventName("name").eventSource("src").username("user")
+ .eventTime(time).cloudTrailEvent("{}").build();
+ }
+
+ @BeforeEach
+ public void setup() throws Exception {
+ component.start();
+ CloudtrailConfiguration configuration = new CloudtrailConfiguration();
+ configuration.setCloudTrailClient(cloudTrailClient);
+ CloudtrailEndpoint endpoint = new
CloudtrailEndpoint("aws-cloudtrail://test", configuration, component);
+ endpoint.start();
+ underTest = new CloudtrailConsumer(endpoint, processor);
+ underTest.start();
+ }
+
+ @Test
+ public void pollDrainsEveryPage() throws Exception {
+ Instant t = Instant.parse("2099-01-01T00:00:00Z");
+ // Page 1 carries a nextToken; page 2 does not. Both pages must be
delivered in a single poll.
+ var page1 = LookupEventsResponse.builder()
+ .events(event("e1", t.plusSeconds(3)), event("e2",
t.plusSeconds(2)))
+ .nextToken("token")
+ .build();
+ var page2 = LookupEventsResponse.builder()
+ .events(event("e3", t.plusSeconds(1)))
+ .build();
+
when(cloudTrailClient.lookupEvents(any(LookupEventsRequest.class))).thenReturn(page1,
page2);
+
+ int processed = underTest.poll();
+
+ Assertions.assertEquals(3, processed, "poll() must deliver events from
every page of the window");
+ verify(cloudTrailClient,
times(2)).lookupEvents(any(LookupEventsRequest.class));
+ }
+
+ @Test
+ public void pollDoesNotRedeliverBoundaryEventsAcrossPolls() throws
Exception {
+ Instant t = Instant.parse("2099-01-01T00:00:00Z");
+ // startTime is inclusive, so the newest event of poll #1 is returned
again in poll #2 and must be skipped.
+ var firstPoll = LookupEventsResponse.builder()
+ .events(event("e2", t.plusSeconds(2)), event("e1",
t.plusSeconds(1)))
+ .build();
+ var secondPoll = LookupEventsResponse.builder()
+ .events(event("e3", t.plusSeconds(3)), event("e2",
t.plusSeconds(2)))
+ .build();
+
when(cloudTrailClient.lookupEvents(any(LookupEventsRequest.class))).thenReturn(firstPoll,
secondPoll);
+
+ Assertions.assertEquals(2, underTest.poll(), "first poll delivers both
new events");
+ Assertions.assertEquals(1, underTest.poll(), "second poll must skip
the already-delivered boundary event e2");
+ }
+}
diff --git
a/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/AwsCloudtrailComponentBuilderFactory.java
b/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/AwsCloudtrailComponentBuilderFactory.java
index 3da9fff636ec..4f097f796cfb 100644
---
a/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/AwsCloudtrailComponentBuilderFactory.java
+++
b/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/AwsCloudtrailComponentBuilderFactory.java
@@ -111,11 +111,13 @@ public interface AwsCloudtrailComponentBuilderFactory {
/**
- * Maximum number of records that will be fetched in each poll.
+ * Maximum number of records that will be fetched in each lookup page.
+ * Each poll drains all pages, so this controls the page size rather
+ * than the total per poll. AWS allows up to 50.
*
* The option is a: <code>int</code> type.
*
- * Default: 1
+ * Default: 50
* Group: consumer
*
* @param maxResults the value to set
diff --git
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/CloudtrailEndpointBuilderFactory.java
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/CloudtrailEndpointBuilderFactory.java
index 5cc1666b3d63..3635fe4af48a 100644
---
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/CloudtrailEndpointBuilderFactory.java
+++
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/CloudtrailEndpointBuilderFactory.java
@@ -59,11 +59,13 @@ public interface CloudtrailEndpointBuilderFactory {
return this;
}
/**
- * Maximum number of records that will be fetched in each poll.
+ * Maximum number of records that will be fetched in each lookup page.
+ * Each poll drains all pages, so this controls the page size rather
+ * than the total per poll. AWS allows up to 50.
*
* The option is a: <code>int</code> type.
*
- * Default: 1
+ * Default: 50
* Group: consumer
*
* @param maxResults the value to set
@@ -74,11 +76,13 @@ public interface CloudtrailEndpointBuilderFactory {
return this;
}
/**
- * Maximum number of records that will be fetched in each poll.
+ * Maximum number of records that will be fetched in each lookup page.
+ * Each poll drains all pages, so this controls the page size rather
+ * than the total per poll. AWS allows up to 50.
*
* The option will be converted to a <code>int</code> type.
*
- * Default: 1
+ * Default: 50
* Group: consumer
*
* @param maxResults the value to set
