This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch camel-4.14.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.14.x by this push:
new e54cc8af035e [backport camel-4.14.x] CAMEL-24343:
camel-google-calendar - stop the stream consumer from skipping events (#25385)
e54cc8af035e is described below
commit e54cc8af035e5d66d87a26d96bfe4aafb228f24b
Author: Claus Ibsen <[email protected]>
AuthorDate: Thu Aug 6 14:11:39 2026 +0200
[backport camel-4.14.x] CAMEL-24343: camel-google-calendar - stop the
stream consumer from skipping events (#25385)
CAMEL-24343: camel-google-calendar - stop the stream consumer from skipping
events
The stream consumer moved its updatedMin cursor to the local clock on an
empty
poll, skipping events modified in between, and one second past the newest
event
otherwise, dropping anything modified within that second. The cursor is now
the
update time of the newest event actually delivered, with the ids seen at
that
instant remembered so the inclusive filter does not deliver them twice. Also
replaces the unbounded self-recursion on an invalid sync token with a single
full re-sync, and guards a null item list.
Closes #25380
Co-authored-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 a951cdd9d7c6..a5358a61e845 100644
--- a/components/camel-google/camel-google-calendar/pom.xml
+++ b/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>
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();
+ }
+}
