This is an automated email from the ASF dual-hosted git repository.
oscerd 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 8fddde698057 CAMEL-24082: camel-aws-cloudtrail - fix consumer event
loss (static cursor, no pagination, window skip) (#24726)
8fddde698057 is described below
commit 8fddde698057e84e1941777dec925754373ea2b7
Author: Andrea Cosentino <[email protected]>
AuthorDate: Thu Jul 16 11:13:29 2026 +0200
CAMEL-24082: camel-aws-cloudtrail - fix consumer event loss (static cursor,
no pagination, window skip) (#24726)
The CloudTrail consumer could silently drop events:
- The lookup cursor was a private static field, shared by every cloudtrail
consumer in the JVM, so two routes clobbered each other's position.
- poll() issued a single non-paginated lookupEvents call and never followed
response.nextToken(); combined with the default maxResults=1 only one
event
per poll was ever retrieved.
- The cursor was advanced to the newest event time plus one second, so
events
arriving within that window (or sharing the newest timestamp) were
skipped,
with no event-id de-duplication.
The cursor is now a per-consumer instance field initialized to the consumer
start time (tail from startup rather than replay history). poll() paginates
through every page of the window, advances the cursor using the newest event
time as an inclusive startTime, and de-duplicates the boundary events by id.
The default maxResults is raised to 50 (the AWS maximum); with pagination it
is a page-size hint rather than a per-poll cap.
Adds CloudtrailConsumerTest (pagination drain + cross-poll boundary de-dup);
the module previously had no unit tests, so a test-scoped mockito dependency
is added, consistent with the sibling aws2 modules. Documents the behavior
and default change in the 4.22 upgrade guide.
Signed-off-by: Andrea Cosentino <[email protected]>
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 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_22.adoc | 15 +++
.../dsl/AwsCloudtrailComponentBuilderFactory.java | 6 +-
.../dsl/CloudtrailEndpointBuilderFactory.java | 12 ++-
9 files changed, 213 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 5c4eb77404da..aeb70e751b95 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 917f3cfcc79d..faa09399c801 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-junit6</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 5c4eb77404da..aeb70e751b95 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 c6dbde89dae1..67d379500774 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/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
index 66473459e499..80753018c84a 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
@@ -499,3 +499,18 @@ account for `OptimisticLockingException`.
The `TemplateParser` now catches `TokenMgrError` (a JavaCC lexer error) and
wraps it in
`ParseRuntimeException`. Previously a malformed stored-procedure template with
characters
outside the token alphabet would propagate as a raw `java.lang.Error`.
+
+=== camel-aws-cloudtrail - consumer event delivery fixed
+
+The CloudTrail consumer previously kept its lookup cursor in a `static` field
(shared across all
+cloudtrail consumers in the JVM), issued a single non-paginated `LookupEvents`
call with a default
+`maxResults` of `1`, and advanced the cursor by the newest event time plus one
second. As a result
+it could silently drop events. The consumer now keeps the cursor per-endpoint,
paginates through
+every page of the window, and de-duplicates events by id at the window
boundary.
+
+Two behavior changes follow:
+
+* The default `maxResults` is now `50` (the AWS maximum). With pagination this
is a page-size hint,
+ not a per-poll cap; set it lower to reduce page sizes.
+* On startup the consumer tails events from the moment it starts, rather than
first replaying the
+ single most-recent historical event.
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 688bf9049c13..51963ff94e1a 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 4440ae526a1c..f6817cbe5fa8 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
