CAMEL-25729: camel-aws-secrets-manager - read every CloudTrail event since the last check in the refresh task

components/camel-aws/camel-aws-secrets-manager/src/main/java/org/apache/camel/component/aws/secretsmanager/vault/CloudTrailReloadTriggerTask.java

@@ -18,6 +18,7 @@
 
 import java.net.URI;
 import java.time.Instant;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -99,13 +100,21 @@ public class CloudTrailReloadTriggerTask extends ServiceSupport implements Camel
 
     private static final String SECRETSMANAGER_UPDATE_SECRET_EVENT = "UpdateSecret";
 
+    // the largest page that LookupEvents accepts
+    private static final int LOOKUP_MAX_RESULTS = 50;
+    // the events that change the value of a secret, looked up one by one
+    private static final List<String> UPDATE_EVENT_NAMES
+            = List.of(SECRETSMANAGER_UPDATE_EVENT, SECRETSMANAGER_UPDATE_SECRET_EVENT);
+
     private CamelContext camelContext;
     private boolean reloadEnabled = true;
     private String secrets;
     private CloudTrailClient cloudTrailClient;
     private SqsClient sqsClient;
     private SecretsManagerPropertiesFunction propertiesFunction;
     private volatile Instant lastTime;
+    // the ids of the events seen at exactly lastTime, which the next lookup returns again (its startTime is inclusive)
+    private final Set<String> lastTimeEventIds = new HashSet<>();
     private volatile Instant lastCheckTime;
     private volatile Instant lastReloadTime;
     private final Map<String, Instant> updates = new HashMap<>();
@@ -220,6 +229,10 @@ protected void doStart() throws Exception {
             }
             cloudTrailClient = isOverrideEndpoint
                     ? clientBuilder.endpointOverride(URI.create(uriEndpointOverride)).build() : clientBuilder.build();
+            if (lastTime == null) {
+                // look for the updates made from now on, not in the event history of the last 90 days
+                lastTime = Instant.now();
+            }
         } else {
             SqsClientBuilder clientBuilder = SqsClient.builder();
             if (ObjectHelper.isNotEmpty(accessKey) && ObjectHelper.isNotEmpty(secretKey) && ObjectHelper.isNotEmpty(region)) {
@@ -263,6 +276,31 @@ protected void doShutdown() throws Exception {
         }
 
         updates.clear();
+        lastTime = null;
+        lastTimeEventIds.clear();
+    }
+
+    /**
+     * Adds every event with the given event name since the last check to the list, reading all pages.
+     */
+    private void lookupEvents(String eventName, List<Event> events) {
+        String nextToken = null;
+        do {
+            LookupEventsRequest.Builder eventsRequestBuilder = LookupEventsRequest.builder()
+                    .maxResults(LOOKUP_MAX_RESULTS)
+                    .lookupAttributes(LookupAttribute.builder().attributeKey(LookupAttributeKey.EVENT_NAME)
+                            .attributeValue(eventName).build());
+            if (lastTime != null) {
+                // inclusive, so an event in the same second as the newest event seen is not skipped
+                eventsRequestBuilder.startTime(lastTime);
+            }
+            if (nextToken != null) {
+                eventsRequestBuilder.nextToken(nextToken);
+            }
+            LookupEventsResponse response = cloudTrailClient.lookupEvents(eventsRequestBuilder.build());
+            events.addAll(response.events());
+            nextToken = response.nextToken();
+        } while (ObjectHelper.isNotEmpty(nextToken));
     }
 
     @Override
@@ -273,26 +311,30 @@ public void run() {
         if (!useSqsNotification) {
 
             try {
-                LookupEventsRequest.Builder eventsRequestBuilder = LookupEventsRequest.builder()
-                        .maxResults(100)
-                        .lookupAttributes(LookupAttribute.builder().attributeKey(LookupAttributeKey.EVENT_SOURCE)
-                                .attributeValue(SECRETSMANAGER_AMAZONAWS_COM).build());
-
-                if (lastTime != null) {
-                    eventsRequestBuilder.startTime(lastTime.plusMillis(1000));
-                }
-
-                LookupEventsRequest lookupEventsRequest = eventsRequestBuilder.build();
-
-                LookupEventsResponse response = cloudTrailClient.lookupEvents(lookupEventsRequest);
-                List<Event> events = response.events();
-
-                if (!events.isEmpty()) {
-                    lastTime = events.get(0).eventTime();
+                // read every page of the update events since the last check. LookupEvents takes a single lookup
+                // attribute, so there is one lookup per event name: filtering on the event source instead would also
+                // return every read (GetSecretValue, DescribeSecret), and the pages would grow with the read traffic
+                List<Event> events = new ArrayList<>();
+                for (String eventName : UPDATE_EVENT_NAMES) {
+                    lookupEvents(eventName, events);
                 }
 
                 LOG.debug("Found {} events", events.size());
+                Instant newest = lastTime;
+                Set<String> newestEventIds = new HashSet<>();
                 for (Event event : events) {
+                    if (lastTimeEventIds.contains(event.eventId())) {
+                        // already seen by the previous check
+                        continue;
+                    }
+                    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());
+                    }
                     if (event.eventSource().equalsIgnoreCase(SECRETSMANAGER_AMAZONAWS_COM)) {
                         if (event.eventName().equalsIgnoreCase(SECRETSMANAGER_UPDATE_EVENT)
                                 || event.eventName().equalsIgnoreCase(SECRETSMANAGER_UPDATE_SECRET_EVENT)) {
@@ -311,6 +353,13 @@ public void run() {
                         }
                     }
                 }
+                if (newest != null && newest.equals(lastTime)) {
+                    // no newer event: the events at lastTime seen before are still returned by the next lookup
+                    newestEventIds.addAll(lastTimeEventIds);
+                }
+                lastTime = newest;
+                lastTimeEventIds.clear();
+                lastTimeEventIds.addAll(newestEventIds);
             } catch (Exception e) {
                 LOG.warn(
                         "Error during AWS Secrets Refresh Task due to {}. This exception is ignored. Will try again on next run.",

components/camel-aws/camel-aws-secrets-manager/src/test/java/org/apache/camel/component/aws/secretsmanager/vault/CloudTrailReloadTriggerTaskLookupTest.java

@@ -0,0 +1,230 @@
+/*
+ * 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.secretsmanager.vault;
+
+import java.lang.reflect.Field;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.spi.ContextReloadStrategy;
+import org.apache.camel.vault.AwsVaultConfiguration;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.services.cloudtrail.CloudTrailClient;
+import software.amazon.awssdk.services.cloudtrail.model.Event;
+import software.amazon.awssdk.services.cloudtrail.model.LookupAttribute;
+import software.amazon.awssdk.services.cloudtrail.model.LookupAttributeKey;
+import software.amazon.awssdk.services.cloudtrail.model.LookupEventsRequest;
+import software.amazon.awssdk.services.cloudtrail.model.LookupEventsResponse;
+import software.amazon.awssdk.services.cloudtrail.model.Resource;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests the CloudTrail lookups of the refresh task against a fake {@code LookupEvents} that answers like the service:
+ * only the events that match the single lookup attribute and have an event time at or after {@code startTime}, newest
+ * first, at most 50 per page, and a next token while events are left.
+ */
+class CloudTrailReloadTriggerTaskLookupTest {
+
+    private static final String SOURCE = "secretsmanager.amazonaws.com";
+    private static final int MAX_PAGE_SIZE = 50;
+
+    private final List<Event> visibleEvents = new ArrayList<>();
+    private final List<LookupEventsRequest> requests = new ArrayList<>();
+    private final CloudTrailClient client = mock(CloudTrailClient.class);
+    private final ContextReloadStrategy reload = mock(ContextReloadStrategy.class);
+    private DefaultCamelContext context;
+    private int eventIds;
+
+    @BeforeEach
+    void setUp() throws Exception {
+        when(client.lookupEvents(any(LookupEventsRequest.class))).thenAnswer(inv -> lookup(inv.getArgument(0)));
+        context = new DefaultCamelContext();
+        context.addService(reload);
+        context.start();
+    }
+
+    @AfterEach
+    void tearDown() {
+        context.stop();
+    }
+
+    private LookupEventsResponse lookup(LookupEventsRequest request) {
+        requests.add(request);
+        // the service takes at most one lookup attribute
+        assertTrue(request.lookupAttributes().size() <= 1, "LookupEvents takes at most one lookup attribute");
+        LookupAttribute attribute = request.lookupAttributes().isEmpty() ? null : request.lookupAttributes().get(0);
+        List<Event> matching = visibleEvents.stream()
+                .filter(e -> attribute == null || matches(attribute, e))
+                .filter(e -> request.startTime() == null || !e.eventTime().isBefore(request.startTime()))
+                .sorted(Comparator.comparing(Event::eventTime).reversed())
+                .toList();
+        int pageSize = request.maxResults() == null ? MAX_PAGE_SIZE : Math.min(request.maxResults(), MAX_PAGE_SIZE);
+        int from = request.nextToken() == null ? 0 : Integer.parseInt(request.nextToken());
+        int to = Math.min(from + pageSize, matching.size());
+        LookupEventsResponse.Builder response = LookupEventsResponse.builder().events(matching.subList(from, to));
+        if (to < matching.size()) {
+            response.nextToken(Integer.toString(to));
+        }
+        return response.build();
+    }
+
+    private static boolean matches(LookupAttribute attribute, Event event) {
+        return switch (attribute.attributeKey()) {
+            case EVENT_NAME -> attribute.attributeValue().equals(event.eventName());
+            case EVENT_SOURCE -> attribute.attributeValue().equals(event.eventSource());
+            default -> throw new IllegalArgumentException("Unexpected lookup attribute " + attribute);
+        };
+    }
+
+    private void addEvent(String name, String secret, Instant time) {
+        visibleEvents.add(Event.builder().eventId("id-" + eventIds++).eventSource(SOURCE).eventName(name)
+                .eventTime(time).resources(Resource.builder().resourceName(secret).build()).build());
+    }
+
+    private CloudTrailReloadTriggerTask task(Instant lastTime) throws Exception {
+        CloudTrailReloadTriggerTask task = new CloudTrailReloadTriggerTask();
+        task.setCamelContext(context);
+        setField(task, "cloudTrailClient", client);
+        setField(task, "secrets", "tracked-secret");
+        setField(task, "lastTime", lastTime);
+        return task;
+    }
+
+    private static void setField(Object target, String name, Object value) throws Exception {
+        Field f = CloudTrailReloadTriggerTask.class.getDeclaredField(name);
+        f.setAccessible(true);
+        f.set(target, value);
+    }
+
+    @Test
+    void updateBehindMoreThanOnePageOfOtherEventsReloadsTheContext() throws Exception {
+        Instant t0 = Instant.parse("2026-10-17T10:00:00Z");
+        CloudTrailReloadTriggerTask task = task(t0);
+        // the secret is updated, then other secrets are updated 60 times before the next check
+        addEvent("PutSecretValue", "tracked-secret", t0.plusSeconds(1));
+        for (int i = 0; i < 60; i++) {
+            addEvent("PutSecretValue", "other-secret", t0.plusSeconds(2 + i % 20));
+        }
+
+        task.run();
+
+        assertTrue(task.getUpdates().containsKey("tracked-secret"), "the update on the second page must be seen");
+        verify(reload, times(1)).onReload(task);
+        assertTrue(requests.stream().allMatch(r -> r.maxResults() != null && r.maxResults() <= MAX_PAGE_SIZE),
+                "LookupEvents accepts at most 50 results per page");
+    }
+
+    @Test
+    void updateInTheSameSecondAsTheNewestEventSeenIsNotSkipped() throws Exception {
+        Instant t0 = Instant.parse("2026-10-17T10:00:00Z");
+        CloudTrailReloadTriggerTask task = task(t0);
+        Instant t1 = t0.plusSeconds(10);
+        addEvent("UpdateSecret", "other-secret", t1);
+
+        task.run();
+        assertFalse(task.getUpdates().containsKey("tracked-secret"));
+
+        // CloudTrail event times are in seconds: an update in the same second becomes visible after the first check
+        addEvent("PutSecretValue", "tracked-secret", t1);
+        task.run();
+
+        assertTrue(task.getUpdates().containsKey("tracked-secret"), "the update in the same second must be seen");
+        verify(reload, times(1)).onReload(task);
+
+        // the events already seen at that second do not reload the context again
+        task.run();
+        verify(reload, times(1)).onReload(task);
+    }
+
+    @Test
+    void firstCheckLooksUpFromTheStartOfTheTask() throws Exception {
+        AwsVaultConfiguration aws = context.getVaultConfiguration().aws();
+        aws.setAccessKey("key");
+        aws.setSecretKey("secret");
+        aws.setRegion("eu-west-1");
+        aws.setSecrets("tracked-secret");
+        CloudTrailReloadTriggerTask task = new CloudTrailReloadTriggerTask();
+        task.setCamelContext(context);
+        Instant before = Instant.now();
+        task.start();
+        try {
+            Field f = CloudTrailReloadTriggerTask.class.getDeclaredField("cloudTrailClient");
+            f.setAccessible(true);
+            ((CloudTrailClient) f.get(task)).close();
+            f.set(task, client);
+            // an update made long before the task started
+            addEvent("PutSecretValue", "tracked-secret", before.minusSeconds(86400));
+
+            task.run();
+
+            assertEquals(2, requests.size(), "one lookup per update event name");
+            for (LookupEventsRequest request : requests) {
+                assertNotNull(request.startTime(), "the first lookup must not read the whole event history");
+                assertFalse(request.startTime().isBefore(before.minusSeconds(1)));
+            }
+            assertFalse(task.getUpdates().containsKey("tracked-secret"));
+            verify(reload, times(0)).onReload(task);
+        } finally {
+            setField(task, "cloudTrailClient", null);
+            task.stop();
+        }
+    }
+
+    @Test
+    void readsOfSecretsDoNotAddLookupRequests() throws Exception {
+        Instant t0 = Instant.parse("2026-10-17T10:00:00Z");
+        CloudTrailReloadTriggerTask task = task(t0);
+        // a busy account: other applications read secrets 200 times between two checks
+        for (int i = 0; i < 200; i++) {
+            addEvent(i % 2 == 0 ? "GetSecretValue" : "DescribeSecret", "other-secret", t0.plusSeconds(1 + i % 30));
+        }
+        addEvent("UpdateSecret", "tracked-secret", t0.plusSeconds(5));
+
+        task.run();
+
+        assertTrue(task.getUpdates().containsKey("tracked-secret"), "an UpdateSecret event must be seen");
+        verify(reload, times(1)).onReload(task);
+        assertEquals(2, requests.size(), "one request per update event name, whatever the number of reads");
+        assertEquals(List.of("PutSecretValue", "UpdateSecret"),
+                requests.stream().map(r -> {
+                    LookupAttribute a = r.lookupAttributes().get(0);
+                    assertEquals(LookupAttributeKey.EVENT_NAME, a.attributeKey());
+                    return a.attributeValue();
+                }).toList());
+
+        // the next check looks up from the newest update seen and does not reload again
+        requests.clear();
+        task.run();
+        verify(reload, times(1)).onReload(task);
+        assertEquals(2, requests.size());
+        assertTrue(requests.stream().allMatch(r -> t0.plusSeconds(5).equals(r.startTime())));
+    }
+}

docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc

@@ -1118,6 +1118,16 @@ MINA does not pass a failed handshake on as an error: the producer fails with a
 `org.apache.mina.filter.ssl.SslHandler`. A producer that starts timing out after the upgrade should be checked for a
 certificate that does not list the host.
 
+=== camel-aws-secrets-manager - the CloudTrail refresh task reads every event since the last check
+
+With `camel.vault.aws.refreshEnabled=true` and without SQS notification, each check now reads every page of the
+`PutSecretValue` and `UpdateSecret` events since the previous check, including the events in the same second as the
+newest event seen before, so an update is no longer missed when more events than one page happened in between. The
+first check looks for the updates made after the task started: an update made before the start no longer reloads the
+context. A check sends two `LookupEvents` requests, one per event name, and one more per further 50 updates (before:
+always one request, for all Secrets Manager events); reads of secrets no longer add requests. LookupEvents allows two
+requests per second per account and Region.
+
 === camel-aws2-kinesis - the consumer continues after the last record when a shard iterator expired
 
 When a cached shard iterator has expired (an iterator is valid for 5 minutes, so after a route was suspended for