[
https://issues.apache.org/jira/browse/CAMEL-25520?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
shashank updated CAMEL-25520:
-----------------------------
Description:
{{GoogleMailStreamConsumer.processBatch}} hands each exchange to the route with
{{getAsyncProcessor().process(exchange, EmptyAsyncCallback.get())}} (line 136
at main 5af71ab258b8) and returns without waiting; the message is marked as
read ({{markAsRead=true}}, the default since CAMEL-16329) only by the
exchange's on completion ({{processCommit}}, lines 120-133 and 145-166). When
the route continues asynchronously, for example {{to("seda:...")}} (seda hands
the on completion over to its consumer), an asynchronous producer such as
kafka, or {{threads()}}, the poll thread returns, and the next poll (500 ms
later by default) lists the message again (it is still unread, and the default
query is {{is:unread}}), gets it again and routes it a second time, and so on
for as long as the first exchange runs.
When one of these duplicates fails, its rollback ({{processRollback}}, lines
170-188) adds the UNREAD label back, also when another exchange of the same
message has already completed and marked it read, so a message that was
processed successfully is consumed once more by a later poll.
h3. Reproduction
New {{GoogleMailStreamConsumerInProgressTest}} with a Gmail client whose
transport answers from an in-memory mailbox ({{FakeGmailClientFactory}},
google-http-client {{MockHttpTransport}}: labels, list of unread messages, get,
modify): one unread message, route
{{google-mail-stream://index?clientId=id&clientSecret=secret&delay=10}} to
{{seda:hold}}; the seda route holds the exchange until the consumer has
completed a second poll (counted with a {{pollStrategy}}). On main:
{noformat}
m1 was consumed again by a later poll while its first exchange was still in
flight ==> expected: <1> but was: <2>
{noformat}
h3. Proposed fix
The consumer keeps the ids of the messages whose exchanges are in progress (a
set in the consumer, as the in-progress repository of aws2-s3): a poll skips
them; an id is added before the message is fetched and released when its
exchange completes or fails (in the on completion, after marking read or
unread), and when the poll does not process it (get failure, consumer
stopping). No new option; the set is not cleared when the route is stopped (ids
of exchanges still in flight are released when they complete), and an exchange
that never completes keeps its message reserved. With {{markAsRead=false}} a
message is now consumed again only after its previous exchange completed
(before, every poll routed it again, also while it was in flight); 4.23
upgrade-guide note. Two more tests check that a message whose exchange failed
is consumed again and then marked as read ({{failedMessageIsConsumedAgain}}),
and that a message whose exchange a stopping consumer did not process is not
left in progress ({{notProcessedMessageIsConsumedAgain}}); both pass on main
too, which has no in-progress set, and guard the releases on failure and on
stop (the second one fails when the release on stop is removed). A further test
checks that a message whose get failed is consumed by a later poll.
camel-google-mail unit tests: 48, 0 failures.
Found with a TLA+ model of the poll (list unread, get, dispatch, stop of the
batch) and the asynchronous completion (mark read) or failure (mark unread): "a
message is routed again only after an exchange for it failed" is violated in 10
steps (poll, get, dispatch, poll again, get, dispatch again); with failures, "a
message processed successfully is not routed again" is violated too (a
duplicate fails after the first completed). With a synchronous route all
properties hold; with the fix they hold also with failures, and no id stays in
progress when nothing is in flight. Then confirmed with the real consumer as
above.
Affected: main, camel-4.22.x, camel-4.18.x, camel-4.14.x (same code).
Duplicate check (2026-10-10): JIRA component camel-google-mail-stream (5
issues: raw body, markAsRead default, verifier; none about duplicates), text
"GoogleMailStreamConsumer" (4, none related), component camel-google-mail since
2023 (2, unrelated); GitHub pull requests "google-mail stream": 4 merged, none
about in-flight messages; no open PR touches camel-google-mail.
_Filed with Claude Code on behalf of allthingssecurity._
was:
{{GoogleMailStreamConsumer.processBatch}} hands each exchange to the route with
{{getAsyncProcessor().process(exchange, EmptyAsyncCallback.get())}} (line 136
at main 5af71ab258b8) and returns without waiting; the message is marked as
read ({{markAsRead=true}}, the default since CAMEL-16329) only by the
exchange's on completion ({{processCommit}}, lines 120-133 and 145-166). When
the route continues asynchronously, for example {{to("seda:...")}} (seda hands
the on completion over to its consumer), an asynchronous producer such as
kafka, or {{threads()}}, the poll thread returns, and the next poll (500 ms
later by default) lists the message again (it is still unread, and the default
query is {{is:unread}}), gets it again and routes it a second time, and so on
for as long as the first exchange runs.
When one of these duplicates fails, its rollback ({{processRollback}}, lines
170-188) adds the UNREAD label back, also when another exchange of the same
message has already completed and marked it read, so a message that was
processed successfully is consumed once more by a later poll.
h3. Reproduction
New {{GoogleMailStreamConsumerInProgressTest}} with a Gmail client whose
transport answers from an in-memory mailbox ({{FakeGmailClientFactory}},
google-http-client {{MockHttpTransport}}: labels, list of unread messages, get,
modify): one unread message, route
{{google-mail-stream://index?clientId=id&clientSecret=secret&delay=10}} to
{{seda:hold}}; the seda route holds the exchange until the consumer has
completed a second poll (counted with a {{pollStrategy}}). On main:
{noformat}
m1 was consumed again by a later poll while its first exchange was still in
flight ==> expected: <1> but was: <2>
{noformat}
h3. Proposed fix
The consumer keeps the ids of the messages whose exchanges are in progress (a
set in the consumer, as the in-progress repository of aws2-s3): a poll skips
them; an id is added before the message is fetched and released when its
exchange completes or fails (in the on completion, after marking read or
unread), and when the poll does not process it (get failure, consumer
stopping). No new option; the set is not cleared when the route is stopped (ids
of exchanges still in flight are released when they complete), and an exchange
that never completes keeps its message reserved. With {{markAsRead=false}} a
message is now consumed again only after its previous exchange completed
(before, every poll routed it again, also while it was in flight); 4.23
upgrade-guide note. Two more tests check that a message whose exchange failed
is consumed again and then marked as read ({{failedMessageIsConsumedAgain}}),
and that a message whose exchange a stopping consumer did not process is not
left in progress ({{notProcessedMessageIsConsumedAgain}}); both pass on main
too, which has no in-progress set, and guard the releases on failure and on
stop (the second one fails when the release on stop is removed).
camel-google-mail unit tests: 47, 0 failures.
Found with a TLA+ model of the poll (list unread, get, dispatch, stop of the
batch) and the asynchronous completion (mark read) or failure (mark unread): "a
message is routed again only after an exchange for it failed" is violated in 10
steps (poll, get, dispatch, poll again, get, dispatch again); with failures, "a
message processed successfully is not routed again" is violated too (a
duplicate fails after the first completed). With a synchronous route all
properties hold; with the fix they hold also with failures, and no id stays in
progress when nothing is in flight. Then confirmed with the real consumer as
above.
Affected: main, camel-4.22.x, camel-4.18.x, camel-4.14.x (same code).
Duplicate check (2026-10-10): JIRA component camel-google-mail-stream (5
issues: raw body, markAsRead default, verifier; none about duplicates), text
"GoogleMailStreamConsumer" (4, none related), component camel-google-mail since
2023 (2, unrelated); GitHub pull requests "google-mail stream": 4 merged, none
about in-flight messages; no open PR touches camel-google-mail.
_Filed with Claude Code on behalf of allthingssecurity._
> camel-google-mail - google-mail-stream consumer routes a message again while
> its first exchange is still in flight (asynchronous routes)
> ----------------------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-25520
> URL: https://issues.apache.org/jira/browse/CAMEL-25520
> Project: Camel
> Issue Type: Bug
> Components: camel-google-mail-stream
> Reporter: shashank
> Assignee: shashank
> Priority: Major
>
> {{GoogleMailStreamConsumer.processBatch}} hands each exchange to the route
> with {{getAsyncProcessor().process(exchange, EmptyAsyncCallback.get())}}
> (line 136 at main 5af71ab258b8) and returns without waiting; the message is
> marked as read ({{markAsRead=true}}, the default since CAMEL-16329) only by
> the exchange's on completion ({{processCommit}}, lines 120-133 and 145-166).
> When the route continues asynchronously, for example {{to("seda:...")}} (seda
> hands the on completion over to its consumer), an asynchronous producer such
> as kafka, or {{threads()}}, the poll thread returns, and the next poll (500
> ms later by default) lists the message again (it is still unread, and the
> default query is {{is:unread}}), gets it again and routes it a second time,
> and so on for as long as the first exchange runs.
> When one of these duplicates fails, its rollback ({{processRollback}}, lines
> 170-188) adds the UNREAD label back, also when another exchange of the same
> message has already completed and marked it read, so a message that was
> processed successfully is consumed once more by a later poll.
> h3. Reproduction
> New {{GoogleMailStreamConsumerInProgressTest}} with a Gmail client whose
> transport answers from an in-memory mailbox ({{FakeGmailClientFactory}},
> google-http-client {{MockHttpTransport}}: labels, list of unread messages,
> get, modify): one unread message, route
> {{google-mail-stream://index?clientId=id&clientSecret=secret&delay=10}} to
> {{seda:hold}}; the seda route holds the exchange until the consumer has
> completed a second poll (counted with a {{pollStrategy}}). On main:
> {noformat}
> m1 was consumed again by a later poll while its first exchange was still in
> flight ==> expected: <1> but was: <2>
> {noformat}
> h3. Proposed fix
> The consumer keeps the ids of the messages whose exchanges are in progress (a
> set in the consumer, as the in-progress repository of aws2-s3): a poll skips
> them; an id is added before the message is fetched and released when its
> exchange completes or fails (in the on completion, after marking read or
> unread), and when the poll does not process it (get failure, consumer
> stopping). No new option; the set is not cleared when the route is stopped
> (ids of exchanges still in flight are released when they complete), and an
> exchange that never completes keeps its message reserved. With
> {{markAsRead=false}} a message is now consumed again only after its previous
> exchange completed (before, every poll routed it again, also while it was in
> flight); 4.23 upgrade-guide note. Two more tests check that a message whose
> exchange failed is consumed again and then marked as read
> ({{failedMessageIsConsumedAgain}}), and that a message whose exchange a
> stopping consumer did not process is not left in progress
> ({{notProcessedMessageIsConsumedAgain}}); both pass on main too, which has no
> in-progress set, and guard the releases on failure and on stop (the second
> one fails when the release on stop is removed). A further test checks that a
> message whose get failed is consumed by a later poll. camel-google-mail unit
> tests: 48, 0 failures.
> Found with a TLA+ model of the poll (list unread, get, dispatch, stop of the
> batch) and the asynchronous completion (mark read) or failure (mark unread):
> "a message is routed again only after an exchange for it failed" is violated
> in 10 steps (poll, get, dispatch, poll again, get, dispatch again); with
> failures, "a message processed successfully is not routed again" is violated
> too (a duplicate fails after the first completed). With a synchronous route
> all properties hold; with the fix they hold also with failures, and no id
> stays in progress when nothing is in flight. Then confirmed with the real
> consumer as above.
> Affected: main, camel-4.22.x, camel-4.18.x, camel-4.14.x (same code).
> Duplicate check (2026-10-10): JIRA component camel-google-mail-stream (5
> issues: raw body, markAsRead default, verifier; none about duplicates), text
> "GoogleMailStreamConsumer" (4, none related), component camel-google-mail
> since 2023 (2, unrelated); GitHub pull requests "google-mail stream": 4
> merged, none about in-flight messages; no open PR touches camel-google-mail.
> _Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
