This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch backport/CAMEL-24343-4.14.x in repository https://gitbox.apache.org/repos/asf/camel.git
commit f027c2d9824a23eac121516c6ac06f5223b75491 Author: Andrea Cosentino <[email protected]> AuthorDate: Thu Aug 6 13:46:38 2026 +0200 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: 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(); + } +}
