This is an automated email from the ASF dual-hosted git repository.
davsclaus 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 2540e4f928cc CAMEL-25520: camel-google-mail - Do not consume a message
again while its exchange is in flight (#27671)
2540e4f928cc is described below
commit 2540e4f928cc4f4633bbaaf946e61d2684f6b4db
Author: allthingssecurity <[email protected]>
AuthorDate: Sun Oct 11 13:30:45 2026 +0530
CAMEL-25520: camel-google-mail - Do not consume a message again while its
exchange is in flight (#27671)
The google-mail-stream consumer routes each exchange asynchronously and
marks the
message read only in its on completion, so with an asynchronous route the
next poll
listed the still unread message again and routed it a second time; a failed
duplicate could even mark an already processed message unread again.
The consumer now keeps the ids of messages whose exchanges are in flight
and skips
them on the next poll, including messages that were in flight when the poll
started
(snapshot taken before listing). An id is released when its exchange
completes or
fails, when getting the message fails, or when the consumer stops before
routing it.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../mail/stream/GoogleMailStreamConsumer.java | 61 +++++-
.../google/mail/stream/FakeGmailClientFactory.java | 187 +++++++++++++++++
.../GoogleMailStreamConsumerInProgressTest.java | 226 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 7 +
4 files changed, 474 insertions(+), 7 deletions(-)
diff --git
a/components/camel-google/camel-google-mail/src/main/java/org/apache/camel/component/google/mail/stream/GoogleMailStreamConsumer.java
b/components/camel-google/camel-google-mail/src/main/java/org/apache/camel/component/google/mail/stream/GoogleMailStreamConsumer.java
index 182f2c6674b0..6d11d3af72b9 100644
---
a/components/camel-google/camel-google-mail/src/main/java/org/apache/camel/component/google/mail/stream/GoogleMailStreamConsumer.java
+++
b/components/camel-google/camel-google-mail/src/main/java/org/apache/camel/component/google/mail/stream/GoogleMailStreamConsumer.java
@@ -21,6 +21,8 @@ import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import com.google.api.client.util.Base64;
import com.google.api.services.gmail.Gmail;
@@ -48,6 +50,8 @@ import org.slf4j.LoggerFactory;
public class GoogleMailStreamConsumer extends ScheduledBatchPollingConsumer {
private static final Logger LOG =
LoggerFactory.getLogger(GoogleMailStreamConsumer.class);
+ // messages whose exchanges are being processed, which a later poll must
not consume again
+ private final Set<String> inProgress = ConcurrentHashMap.newKeySet();
private String unreadLabelId;
private List<String> labelsIds;
@@ -85,23 +89,55 @@ public class GoogleMailStreamConsumer extends
ScheduledBatchPollingConsumer {
Queue<Exchange> answer = new LinkedList<>();
+ // taken before the messages are listed: a message that is in flight
now can complete, and be marked as read
+ // and released, after it was listed and before the loop below looks
at it, and must then not be consumed again
+ // from the stale listing
+ Set<String> inProgressAtPollStart = Set.copyOf(inProgress);
+
ListMessagesResponse c = request.execute();
// okay we have some response from Google so lets mark the consumer as
ready
forceConsumerAsReady();
if (c.getMessages() != null) {
- for (Message message : c.getMessages()) {
- Message mess
- = getClient().users().messages().get("me",
message.getId()).setFormat(messageFormat()).execute();
- Exchange exchange =
createExchange(getEndpoint().getExchangePattern(), mess);
- answer.add(exchange);
+ try {
+ for (Message message : c.getMessages()) {
+ addExchange(message.getId(), answer,
inProgressAtPollStart);
+ }
+ } catch (Exception e) {
+ // none of the messages of this poll is processed
+ releaseInProgress(answer);
+ throw e;
}
}
return processBatch(CastUtils.cast(answer));
}
+ private void addExchange(String id, Queue<Exchange> answer, Set<String>
inProgressAtPollStart) throws Exception {
+ // skip the message if an exchange of a previous poll is still
processing it, or was when this poll started
+ if (inProgressAtPollStart.contains(id) || !inProgress.add(id)) {
+ LOG.trace("Skipping message {} as it is already in progress", id);
+ return;
+ }
+ try {
+ Message mess = getClient().users().messages().get("me",
id).setFormat(messageFormat()).execute();
+ answer.add(createExchange(getEndpoint().getExchangePattern(),
mess));
+ } catch (Exception e) {
+ inProgress.remove(id);
+ throw e;
+ }
+ }
+
+ private void releaseInProgress(Queue<?> notProcessed) {
+ for (Object notRouted : notProcessed) {
+ Exchange exchange = (Exchange) notRouted;
+
inProgress.remove(exchange.getIn().getHeader(GoogleMailStreamConstants.MAIL_ID,
String.class));
+ // the exchange is never routed, so nothing else releases it to
the exchange factory
+ releaseExchange(exchange, false);
+ }
+ }
+
@Override
public int processBatch(Queue<Object> exchanges) {
int total = exchanges.size();
@@ -117,14 +153,23 @@ public class GoogleMailStreamConsumer extends
ScheduledBatchPollingConsumer {
// update pending number of exchanges
pendingExchanges = total - index - 1;
+ final String id =
exchange.getIn().getHeader(GoogleMailStreamConstants.MAIL_ID, String.class);
// add on completion to handle after work when the exchange is done
exchange.getExchangeExtension().addOnCompletion(new
Synchronization() {
public void onComplete(Exchange exchange) {
- processCommit(exchange, unreadLabelId);
+ try {
+ processCommit(exchange, unreadLabelId);
+ } finally {
+ inProgress.remove(id);
+ }
}
public void onFailure(Exchange exchange) {
- processRollback(exchange, unreadLabelId);
+ try {
+ processRollback(exchange, unreadLabelId);
+ } finally {
+ inProgress.remove(id);
+ }
}
@Override
@@ -135,6 +180,8 @@ public class GoogleMailStreamConsumer extends
ScheduledBatchPollingConsumer {
getAsyncProcessor().process(exchange, EmptyAsyncCallback.get());
}
+ // the remaining exchanges are not processed as the consumer is
stopping
+ releaseInProgress(exchanges);
return total;
}
diff --git
a/components/camel-google/camel-google-mail/src/test/java/org/apache/camel/component/google/mail/stream/FakeGmailClientFactory.java
b/components/camel-google/camel-google-mail/src/test/java/org/apache/camel/component/google/mail/stream/FakeGmailClientFactory.java
new file mode 100644
index 000000000000..da0e9937d9a3
--- /dev/null
+++
b/components/camel-google/camel-google-mail/src/test/java/org/apache/camel/component/google/mail/stream/FakeGmailClientFactory.java
@@ -0,0 +1,187 @@
+/*
+ * 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.mail.stream;
+
+import java.io.IOException;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import
com.google.api.client.googleapis.testing.auth.oauth2.MockGoogleCredential;
+import com.google.api.client.http.LowLevelHttpRequest;
+import com.google.api.client.http.LowLevelHttpResponse;
+import com.google.api.client.json.gson.GsonFactory;
+import com.google.api.client.testing.http.MockHttpTransport;
+import com.google.api.client.testing.http.MockLowLevelHttpRequest;
+import com.google.api.client.testing.http.MockLowLevelHttpResponse;
+import com.google.api.client.util.Base64;
+import com.google.api.services.gmail.Gmail;
+import org.apache.camel.CamelContext;
+import org.apache.camel.component.google.mail.GoogleMailClientFactory;
+
+/**
+ * A Gmail client whose transport answers the calls of the stream consumer
from an in-memory mailbox: listing the
+ * labels, listing the messages with the UNREAD label (the default query
{@code is:unread}), getting a message, and
+ * removing or adding the UNREAD label. A get can be made to fail once, and a
task can be run between a listing and its
+ * response. Any other request fails with status 500 and is recorded.
+ */
+class FakeGmailClientFactory implements GoogleMailClientFactory {
+
+ private static final String MESSAGES = "/gmail/v1/users/me/messages";
+
+ private final Map<String, String> bodies = new ConcurrentHashMap<>();
+ private final Set<String> unread = ConcurrentHashMap.newKeySet();
+ private final Map<String, AtomicInteger> gets = new ConcurrentHashMap<>();
+ private final List<String> unexpected = new CopyOnWriteArrayList<>();
+ private final Set<String> failNextGet = ConcurrentHashMap.newKeySet();
+ private final CountDownLatch markedAsRead = new CountDownLatch(1);
+ private final AtomicReference<Runnable> afterNextList = new
AtomicReference<>();
+
+ void addUnread(String id, String body) {
+ bodies.put(id, body);
+ unread.add(id);
+ }
+
+ /**
+ * Makes the next get of the message fail with status 500.
+ */
+ void failNextGet(String id) {
+ failNextGet.add(id);
+ }
+
+ /**
+ * Runs the task once, after the next listing has read the unread messages
and before its response is returned.
+ */
+ void afterNextList(Runnable task) {
+ afterNextList.set(task);
+ }
+
+ boolean isUnread(String id) {
+ return unread.contains(id);
+ }
+
+ /**
+ * Waits until a message has been marked as read (the UNREAD label
removed).
+ */
+ boolean awaitMarkedAsRead(long timeout, TimeUnit unit) throws
InterruptedException {
+ return markedAsRead.await(timeout, unit);
+ }
+
+ int gets(String id) {
+ AtomicInteger count = gets.get(id);
+ return count == null ? 0 : count.get();
+ }
+
+ List<String> unexpectedRequests() {
+ return unexpected;
+ }
+
+ @Override
+ public Gmail makeClient(
+ String clientId, String clientSecret, Collection<String> scopes,
String applicationName,
+ String refreshToken, String accessToken) {
+ return makeClient();
+ }
+
+ @Override
+ public Gmail makeClient(
+ CamelContext camelContext, String serviceAccountKey,
Collection<String> scopes, String applicationName,
+ String delegate) {
+ return makeClient();
+ }
+
+ private Gmail makeClient() {
+ MockHttpTransport transport = new MockHttpTransport() {
+ @Override
+ public LowLevelHttpRequest buildRequest(String method, String url)
{
+ return new MockLowLevelHttpRequest(url) {
+ @Override
+ public LowLevelHttpResponse execute() throws IOException {
+ return answer(method, URI.create(url).getPath(),
getContentAsString());
+ }
+ };
+ }
+ };
+ return new Gmail.Builder(transport, GsonFactory.getDefaultInstance(),
new MockGoogleCredential.Builder().build())
+ .setApplicationName("fake").build();
+ }
+
+ private LowLevelHttpResponse answer(String method, String path, String
content) {
+ if ("GET".equals(method) && path.endsWith("/users/me/labels")) {
+ return json("{\"labels\": [{\"id\": \"UNREAD\", \"name\":
\"UNREAD\"}]}");
+ }
+ if ("GET".equals(method) && path.endsWith("/users/me/labels/UNREAD")) {
+ return json("{\"id\": \"UNREAD\", \"name\": \"UNREAD\"}");
+ }
+ if ("GET".equals(method) && path.endsWith(MESSAGES)) {
+ StringBuilder messages = new StringBuilder();
+ for (String id : unread) {
+ messages.append(messages.isEmpty() ? "" : ",
").append("{\"id\": \"").append(id).append("\"}");
+ }
+ Runnable task = afterNextList.getAndSet(null);
+ if (task != null) {
+ task.run();
+ }
+ return json("{\"messages\": [" + messages + "]}");
+ }
+ int idx = path.indexOf(MESSAGES + "/");
+ if (idx >= 0) {
+ String rest = path.substring(idx + MESSAGES.length() + 1);
+ if ("GET".equals(method) && failNextGet.remove(rest)) {
+ MockLowLevelHttpResponse response = new
MockLowLevelHttpResponse();
+ response.setStatusCode(500);
+ return response;
+ }
+ if ("GET".equals(method) && bodies.containsKey(rest)) {
+ gets.computeIfAbsent(rest, k -> new
AtomicInteger()).incrementAndGet();
+ String data =
Base64.encodeBase64URLSafeString(bodies.get(rest).getBytes(StandardCharsets.UTF_8));
+ return json("{\"id\": \"" + rest + "\", \"payload\":
{\"mimeType\": \"text/plain\", \"body\": {\"data\": \""
+ + data + "\"}}}");
+ }
+ if ("POST".equals(method) && rest.endsWith("/modify")) {
+ String id = rest.substring(0, rest.length() -
"/modify".length());
+ if (content.contains("removeLabelIds")) {
+ unread.remove(id);
+ markedAsRead.countDown();
+ } else if (content.contains("addLabelIds")) {
+ unread.add(id);
+ }
+ return json("{\"id\": \"" + id + "\"}");
+ }
+ }
+ unexpected.add(method + " " + path);
+ MockLowLevelHttpResponse response = new MockLowLevelHttpResponse();
+ response.setStatusCode(500);
+ return response;
+ }
+
+ private static LowLevelHttpResponse json(String content) {
+ MockLowLevelHttpResponse response = new MockLowLevelHttpResponse();
+ response.setContentType("application/json");
+ response.setContent(content);
+ return response;
+ }
+}
diff --git
a/components/camel-google/camel-google-mail/src/test/java/org/apache/camel/component/google/mail/stream/GoogleMailStreamConsumerInProgressTest.java
b/components/camel-google/camel-google-mail/src/test/java/org/apache/camel/component/google/mail/stream/GoogleMailStreamConsumerInProgressTest.java
new file mode 100644
index 000000000000..7101afddb634
--- /dev/null
+++
b/components/camel-google/camel-google-mail/src/test/java/org/apache/camel/component/google/mail/stream/GoogleMailStreamConsumerInProgressTest.java
@@ -0,0 +1,226 @@
+/*
+ * 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.mail.stream;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Consumer;
+import org.apache.camel.Endpoint;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.spi.CamelEvent;
+import org.apache.camel.spi.CamelEvent.ExchangeCompletedEvent;
+import org.apache.camel.spi.PollingConsumerPollStrategy;
+import org.apache.camel.support.EventNotifierSupport;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A message whose exchange is still being routed asynchronously must not be
consumed again by the next poll, nor by a
+ * poll that listed it just before its exchange completed, and a message whose
exchange failed, whose get failed, or
+ * whose exchange a stopping consumer did not process, is consumed again.
+ */
+class GoogleMailStreamConsumerInProgressTest extends CamelTestSupport {
+
+ private final FakeGmailClientFactory gmail = new FakeGmailClientFactory();
+ private final CountDownLatch inRoute = new CountDownLatch(1);
+ private final CountDownLatch twoPolls = new CountDownLatch(2);
+ private final CountDownLatch failedPoll = new CountDownLatch(1);
+ private final CountDownLatch release = new CountDownLatch(1);
+ private final CountDownLatch completed = new CountDownLatch(1);
+ private final CountDownLatch windowPollDone = new CountDownLatch(1);
+ private final AtomicInteger attempts = new AtomicInteger();
+ private volatile boolean windowOpened;
+ private volatile String heldExchangeId;
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ gmail.addUnread("m1", "hello");
+ CamelContext context = super.createCamelContext();
+ context.getComponent("google-mail-stream",
GoogleMailStreamComponent.class).setClientFactory(gmail);
+ // the completed event of an exchange is sent after all its on
completions have run
+ context.getManagementStrategy().addEventNotifier(new
EventNotifierSupport() {
+ @Override
+ public void notify(CamelEvent event) {
+ if (event instanceof ExchangeCompletedEvent done
+ &&
done.getExchange().getExchangeId().equals(heldExchangeId)) {
+ completed.countDown();
+ }
+ }
+ });
+ // counts the polls that completed (the exchanges of the poll were
handed to the route)
+ context.getRegistry().bind("countPolls", new
PollingConsumerPollStrategy() {
+ @Override
+ public boolean begin(Consumer consumer, Endpoint endpoint) {
+ return true;
+ }
+
+ @Override
+ public void commit(Consumer consumer, Endpoint endpoint, int
polledMessages) {
+ twoPolls.countDown();
+ if (windowOpened) {
+ windowPollDone.countDown();
+ }
+ }
+
+ @Override
+ public boolean rollback(Consumer consumer, Endpoint endpoint, int
retryCounter, Exception e) {
+ failedPoll.countDown();
+ return false;
+ }
+ });
+ return context;
+ }
+
+ @AfterEach
+ void releaseRoute() {
+ release.countDown();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ // seda hands the on completions (mark as read) over to these
routes
+ from("seda:hold")
+ .process(e -> {
+ heldExchangeId = e.getExchangeId();
+ inRoute.countDown();
+ release.await(20, TimeUnit.SECONDS);
+ })
+ .to("mock:result");
+
+ from("seda:failFirst")
+ .process(e -> {
+ if (attempts.incrementAndGet() == 1) {
+ throw new IllegalStateException("first attempt
fails");
+ }
+ })
+ .to("mock:result");
+ }
+ };
+ }
+
+ private void consumeTo(String uri) throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("google-mail-stream://index?clientId=id&clientSecret=secret&delay=10&pollStrategy=#countPolls")
+ .to(uri);
+ }
+ });
+ }
+
+ @Test
+ void messageInFlightIsNotConsumedAgain() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("hello");
+
+ consumeTo("seda:hold");
+
+ assertTrue(inRoute.await(20, TimeUnit.SECONDS), "the message was not
consumed");
+ // the second poll has completed while the first exchange is still
held in the seda route
+ assertTrue(twoPolls.await(20, TimeUnit.SECONDS), "the consumer did not
poll again");
+ int gets = gmail.gets("m1");
+ release.countDown();
+
+ assertEquals(1, gets, "m1 was consumed again by a later poll while its
first exchange was still in flight");
+ mock.assertIsSatisfied();
+ assertTrue(gmail.unexpectedRequests().isEmpty(), () -> "unexpected
requests " + gmail.unexpectedRequests());
+ }
+
+ @Test
+ void messageCompletedAfterTheListingIsNotConsumedAgain() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("hello");
+
+ consumeTo("seda:hold");
+
+ assertTrue(inRoute.await(20, TimeUnit.SECONDS), "the message was not
consumed");
+ AtomicBoolean completedInWindow = new AtomicBoolean();
+ // the next poll lists m1 while its exchange is in flight; before the
consumer looks at the listed messages,
+ // the exchange completes: m1 is marked as read and no longer in
progress
+ gmail.afterNextList(() -> {
+ release.countDown();
+ try {
+ completedInWindow.set(completed.await(20, TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ windowOpened = true;
+ });
+
+ assertTrue(windowPollDone.await(20, TimeUnit.SECONDS), "the poll that
listed the stale m1 did not end");
+ assertTrue(completedInWindow.get(), "the exchange of m1 did not
complete during the poll");
+ assertFalse(gmail.isUnread("m1"), "m1 was not marked as read");
+ assertEquals(1, gmail.gets("m1"), "m1 was consumed again from a
listing made while its exchange was in flight");
+ mock.assertIsSatisfied();
+ assertTrue(gmail.unexpectedRequests().isEmpty(), () -> "unexpected
requests " + gmail.unexpectedRequests());
+ }
+
+ @Test
+ void failedMessageIsConsumedAgain() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("hello");
+
+ consumeTo("seda:failFirst");
+
+ mock.assertIsSatisfied();
+ assertEquals(2, attempts.get());
+ assertTrue(gmail.awaitMarkedAsRead(20, TimeUnit.SECONDS), "the message
was not marked as read");
+ assertFalse(gmail.isUnread("m1"), "the message is unread again");
+ assertTrue(gmail.unexpectedRequests().isEmpty(), () -> "unexpected
requests " + gmail.unexpectedRequests());
+ }
+
+ @Test
+ void messageIsConsumedAfterAFailedGet() throws Exception {
+ // the get of the first poll fails, which must not leave the message
in progress
+ gmail.failNextGet("m1");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("hello");
+
+ consumeTo("mock:result");
+
+ mock.assertIsSatisfied();
+ assertEquals(0, failedPoll.getCount(), "the get of the first poll did
not fail");
+ assertTrue(gmail.unexpectedRequests().isEmpty(), () -> "unexpected
requests " + gmail.unexpectedRequests());
+ }
+
+ @Test
+ void notProcessedMessageIsConsumedAgain() throws Exception {
+ GoogleMailStreamEndpoint endpoint = context.getEndpoint(
+ "google-mail-stream://index?clientId=id&clientSecret=secret",
GoogleMailStreamEndpoint.class);
+ // never started, so the consumer does not process any exchange of the
batch
+ GoogleMailStreamConsumer consumer = (GoogleMailStreamConsumer)
endpoint.createConsumer(exchange -> {
+ });
+
+ consumer.poll();
+ consumer.poll();
+
+ assertEquals(2, gmail.gets("m1"), "m1 stayed in progress although its
exchange was not processed");
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
index e451ae156109..943563d8329e 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
@@ -3738,6 +3738,13 @@ example deleted by another consumer after the listing)
is now skipped instead of
The `camel-aws2-s3-vectors` consumer with `deleteAfterRead=true` no longer
consumes a vector again while its exchange is
still in flight. A vector whose exchange failed is still consumed again by a
later poll.
+=== camel-google-mail - messages whose exchange is in flight
+
+The `google-mail-stream` consumer no longer consumes a message again while an
exchange of an earlier poll is still
+processing it, which happened with an asynchronous route (for example
`to("seda:...")`) as the message is only marked
+as read when its exchange is done. The next polls now skip such a message
(with `markAsRead=false` it is consumed again
+once its exchange is done).
+
=== camel-google-sheets - custom column names of a range that does not start
at column A
The `google-sheets:application-x-struct` data type now maps the custom
`columnNames` to the columns of the range by
