[backport camel-4.18.x] CAMEL-24343: camel-google-calendar - stop the stream consumer from skipping events

components/camel-google/camel-google-calendar/pom.xml

@@ -136,6 +136,11 @@
             <artifactId>camel-test-junit5</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>

components/camel-google/camel-google-calendar/src/main/java/org/apache/camel/component/google/calendar/stream/GoogleCalendarStreamConsumer.java

@@ -18,9 +18,11 @@
 
 import java.util.ArrayList;
 import java.util.Date;
+import java.util.HashSet;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Queue;
+import java.util.Set;
 
 import com.google.api.client.googleapis.json.GoogleJsonResponseException;
 import com.google.api.client.util.DateTime;
@@ -46,6 +48,9 @@ public class GoogleCalendarStreamConsumer extends ScheduledBatchPollingConsumer
     private static final Logger LOG = LoggerFactory.getLogger(GoogleCalendarStreamConsumer.class);
 
     private DateTime lastUpdate;
+    // the updatedMin filter is inclusive, so the events carrying exactly the lastUpdate timestamp are
+    // returned again on the next poll: remember them to not deliver them twice
+    private final Set<String> lastUpdateEventIds = new HashSet<>();
 
     // sync and page tokens for synchronization flow
     // see https://developers.google.com/calendar/v3/sync
@@ -90,7 +95,6 @@ protected int poll() throws Exception {
         }
 
         Queue<Exchange> answer = new LinkedList<>();
-        List<Date> dateList = new ArrayList<>();
 
         Events c;
 
@@ -109,18 +113,23 @@ protected int poll() throws Exception {
             try {
                 c = request.execute();
             } catch (GoogleJsonResponseException e) {
-                if (e.getStatusCode() == 410) {
-                    // A 410 status code, "Gone", indicates that the sync token is invalid.
-                    LOG.info("Invalid sync token, clearing sync and page tokens and re-syncing.");
-                    syncToken = null;
-                    pageToken = null;
-                    return poll();
-                } else {
+                if (e.getStatusCode() != 410) {
                     throw e;
                 }
+                // A 410 status code, "Gone", indicates that the sync token is invalid. Drop the tokens and
+                // perform a single full sync instead of recursing into poll() again
+                LOG.info("Invalid sync token, clearing sync and page tokens and re-syncing.");
+                syncToken = null;
+                pageToken = null;
+                request.setSyncToken(null);
+                request.setPageToken(null);
+                if (getConfiguration().isConsumeFromNow()) {
+                    request.setTimeMin(new DateTime(new Date()));
+                }
+                c = request.execute();
             }
 
-            if (c.getItems().isEmpty()) {
+            if (c.getItems() == null || c.getItems().isEmpty()) {
                 LOG.info("No new events to sync.");
             }
 
@@ -134,36 +143,74 @@ protected int poll() throws Exception {
             c = request.setOrderBy("updated").execute();
         }
 
-        if (c != null) {
-            List<Event> list = c.getItems();
-
-            for (Event event : list) {
-                Exchange exchange = getEndpoint().createExchange(getEndpoint().getExchangePattern(), event);
-                answer.add(exchange);
-                DateTime updated = event.getUpdated();
-                if (updated != null) {
-                    dateList.add(new Date(updated.getValue()));
-                }
+        if (c != null && c.getItems() != null) {
+            for (Event event : selectUndeliveredAndMoveCursor(c.getItems())) {
+                answer.add(getEndpoint().createExchange(getEndpoint().getExchangePattern(), event));
             }
         }
 
-        lastUpdate = retrieveLastUpdateDate(dateList);
         return processBatch(CastUtils.cast(answer));
     }
 
-    private DateTime retrieveLastUpdateDate(List<Date> dateList) {
-        Date finalLastUpdate;
-        if (!dateList.isEmpty()) {
-            dateList.sort(Date::compareTo);
-            Date lastUpdateDate = dateList.get(dateList.size() - 1);
-            java.util.Calendar calendar = java.util.Calendar.getInstance();
-            calendar.setTime(lastUpdateDate);
-            calendar.add(java.util.Calendar.SECOND, 1);
-            finalLastUpdate = calendar.getTime();
-        } else {
-            finalLastUpdate = new Date();
+    /**
+     * Returns the events of this poll that a previous poll did not already deliver, and moves the update cursor to the
+     * newest event actually returned.
+     * <p>
+     * The cursor is only moved when something was delivered: a poll that returned nothing must not advance it, or the
+     * events modified between the two polls would never be seen.
+     */
+    List<Event> selectUndeliveredAndMoveCursor(List<Event> events) {
+        List<Event> answer = new ArrayList<>(events.size());
+        long newestUpdate = lastUpdate != null ? lastUpdate.getValue() : Long.MIN_VALUE;
+        Set<String> newestEventIds = new HashSet<>();
+
+        for (Event event : events) {
+            DateTime updated = event.getUpdated();
+            if (alreadyDelivered(event, updated)) {
+                continue;
+            }
+
+            answer.add(event);
+
+            if (updated != null) {
+                if (updated.getValue() > newestUpdate) {
+                    newestUpdate = updated.getValue();
+                    newestEventIds.clear();
+                    newestEventIds.add(event.getId());
+                } else if (updated.getValue() == newestUpdate) {
+                    newestEventIds.add(event.getId());
+                }
+            }
         }
-        return new DateTime(finalLastUpdate);
+
+        if (!newestEventIds.isEmpty()) {
+            if (lastUpdate != null && newestUpdate == lastUpdate.getValue()) {
+                // still the same instant: the events remembered for it have to be kept, or they would be
+                // delivered again by the next poll
+                lastUpdateEventIds.addAll(newestEventIds);
+            } else {
+                lastUpdate = new DateTime(newestUpdate);
+                lastUpdateEventIds.clear();
+                lastUpdateEventIds.addAll(newestEventIds);
+            }
+        }
+
+        return answer;
+    }
+
+    DateTime getLastUpdate() {
+        return lastUpdate;
+    }
+
+    /**
+     * The updatedMin filter is inclusive, so the events carrying exactly the lastUpdate timestamp come back on the next
+     * poll. They were already delivered, and the cursor cannot be moved past them without risking events modified
+     * within the same instant.
+     */
+    private boolean alreadyDelivered(Event event, DateTime updated) {
+        return updated != null && lastUpdate != null
+                && updated.getValue() == lastUpdate.getValue()
+                && lastUpdateEventIds.contains(event.getId());
     }
 
     @Override

components/camel-google/camel-google-calendar/src/test/java/org/apache/camel/component/google/calendar/stream/GoogleCalendarStreamConsumerCursorTest.java

@@ -0,0 +1,110 @@
+/*
+ * 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.google.calendar.stream;
+
+import java.util.List;
+
+import com.google.api.client.util.DateTime;
+import com.google.api.services.calendar.model.Event;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Verifies how the stream consumer moves its {@code updatedMin} cursor between polls. The calendar filter is inclusive,
+ * so the boundary events come back on the next poll and have to be recognised as already delivered, without skipping
+ * anything modified within the same instant.
+ */
+class GoogleCalendarStreamConsumerCursorTest {
+
+    private static final long T1 = 1_700_000_000_000L;
+    private static final long T2 = T1 + 5_000L;
+
+    private DefaultCamelContext context;
+    private GoogleCalendarStreamConsumer consumer;
+
+    @BeforeEach
+    void setUp() throws Exception {
+        context = new DefaultCamelContext();
+        context.start();
+        GoogleCalendarStreamEndpoint endpoint = context.getEndpoint(
+                "google-calendar-stream://events?considerLastUpdate=true&clientId=id&clientSecret=secret",
+                GoogleCalendarStreamEndpoint.class);
+        consumer = new GoogleCalendarStreamConsumer(endpoint, exchange -> {
+        });
+    }
+
+    @AfterEach
+    void tearDown() {
+        context.stop();
+    }
+
+    private static Event event(String id, long updated) {
+        return new Event().setId(id).setUpdated(new DateTime(updated));
+    }
+
+    private static List<String> ids(List<Event> events) {
+        return events.stream().map(Event::getId).toList();
+    }
+
+    @Test
+    void theBoundaryEventIsNotDeliveredTwice() {
+        assertThat(ids(consumer.selectUndeliveredAndMoveCursor(List.of(event("a", T1), event("b", T2)))))
+                .containsExactly("a", "b");
+        assertThat(consumer.getLastUpdate().getValue()).isEqualTo(T2);
+
+        // updatedMin is inclusive, so the next poll gets the newest event back
+        assertThat(consumer.selectUndeliveredAndMoveCursor(List.of(event("b", T2)))).isEmpty();
+    }
+
+    @Test
+    void anEventModifiedInTheSameInstantIsStillDelivered() {
+        assertThat(ids(consumer.selectUndeliveredAndMoveCursor(List.of(event("a", T2))))).containsExactly("a");
+
+        // "b" carries exactly the cursor timestamp but was never delivered: bumping the cursor past that
+        // instant, as adding a second did, would drop it
+        assertThat(ids(consumer.selectUndeliveredAndMoveCursor(List.of(event("a", T2), event("b", T2)))))
+                .containsExactly("b");
+        assertThat(consumer.getLastUpdate().getValue()).isEqualTo(T2);
+
+        assertThat(consumer.selectUndeliveredAndMoveCursor(List.of(event("a", T2), event("b", T2)))).isEmpty();
+    }
+
+    @Test
+    void anEmptyPollDoesNotMoveTheCursor() {
+        consumer.selectUndeliveredAndMoveCursor(List.of(event("a", T1)));
+        assertThat(consumer.getLastUpdate().getValue()).isEqualTo(T1);
+
+        // nothing new to report: the cursor has to stay where it was, moving it to "now" would skip
+        // everything modified in between
+        assertThat(consumer.selectUndeliveredAndMoveCursor(List.of())).isEmpty();
+        assertThat(consumer.getLastUpdate().getValue()).isEqualTo(T1);
+
+        assertThat(ids(consumer.selectUndeliveredAndMoveCursor(List.of(event("b", T2))))).containsExactly("b");
+        assertThat(consumer.getLastUpdate().getValue()).isEqualTo(T2);
+    }
+
+    @Test
+    void eventsWithoutAnUpdateTimeDoNotMoveTheCursor() {
+        assertThat(ids(consumer.selectUndeliveredAndMoveCursor(List.of(new Event().setId("a")))))
+                .containsExactly("a");
+        assertThat(consumer.getLastUpdate()).isNull();
+    }
+}