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 ab77bb5ee23b CAMEL-24343: camel-google-calendar - stop the stream
consumer from skipping events (#25352)
ab77bb5ee23b is described below
commit ab77bb5ee23beabce32d6c9588ea2e66fd6d834f
Author: Andrea Cosentino <[email protected]>
AuthorDate: Wed Aug 5 14:28:01 2026 +0200
CAMEL-24343: camel-google-calendar - stop the stream consumer from skipping
events (#25352)
With considerLastUpdate the consumer moved its updatedMin cursor to the
local
clock whenever a poll returned nothing, so every event modified between the
two
polls was filtered out and never delivered. On a non-empty poll it moved the
cursor one second past the newest event instead, which dropped everything
modified within that same second.
The cursor is now the update time of the newest event actually delivered,
and
the ids seen at that exact instant are remembered so the inclusive
updatedMin
filter does not deliver them twice.
Also replaces the unbounded self-recursion on an invalid sync token (HTTP
410)
with a single full re-sync, and guards the item list of the response, which
is
null when the calendar returns no events.
Signed-off-by: Andrea Cosentino <[email protected]>
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
---
.../camel-google/camel-google-calendar/pom.xml | 5 +
.../stream/GoogleCalendarStreamConsumer.java | 111 +++++++++++++++------
.../GoogleCalendarStreamConsumerCursorTest.java | 110 ++++++++++++++++++++
3 files changed, 194 insertions(+), 32 deletions(-)
diff --git a/components/camel-google/camel-google-calendar/pom.xml
b/components/camel-google/camel-google-calendar/pom.xml
index 29ccb173c300..647f432398cb 100644
--- a/components/camel-google/camel-google-calendar/pom.xml
+++ b/components/camel-google/camel-google-calendar/pom.xml
@@ -141,6 +141,11 @@
<artifactId>camel-test-junit6</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.assertj</groupId>
+ <artifactId>assertj-core</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
diff --git
a/components/camel-google/camel-google-calendar/src/main/java/org/apache/camel/component/google/calendar/stream/GoogleCalendarStreamConsumer.java
b/components/camel-google/camel-google-calendar/src/main/java/org/apache/camel/component/google/calendar/stream/GoogleCalendarStreamConsumer.java
index 37014154f48a..77c58243bee8 100644
---
a/components/camel-google/camel-google-calendar/src/main/java/org/apache/camel/component/google/calendar/stream/GoogleCalendarStreamConsumer.java
+++
b/components/camel-google/camel-google-calendar/src/main/java/org/apache/camel/component/google/calendar/stream/GoogleCalendarStreamConsumer.java
@@ -18,9 +18,11 @@ package org.apache.camel.component.google.calendar.stream;
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 @@ public class GoogleCalendarStreamConsumer extends
ScheduledBatchPollingConsumer
}
Queue<Exchange> answer = new LinkedList<>();
- List<Date> dateList = new ArrayList<>();
Events c;
@@ -109,18 +113,23 @@ public class GoogleCalendarStreamConsumer extends
ScheduledBatchPollingConsumer
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 @@ public class GoogleCalendarStreamConsumer extends
ScheduledBatchPollingConsumer
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
diff --git
a/components/camel-google/camel-google-calendar/src/test/java/org/apache/camel/component/google/calendar/stream/GoogleCalendarStreamConsumerCursorTest.java
b/components/camel-google/camel-google-calendar/src/test/java/org/apache/camel/component/google/calendar/stream/GoogleCalendarStreamConsumerCursorTest.java
new file mode 100644
index 000000000000..28c84afc0789
--- /dev/null
+++
b/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();
+ }
+}
