[
https://issues.apache.org/jira/browse/CAMEL-25521?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen updated CAMEL-25521:
--------------------------------
Fix Version/s: 4.23.0
> camel-aws2-s3 - consumer with fileName routes the object again while its
> exchange is in flight, and objects of a batch cut short by a route stop are
> never consumed after a restart
> -----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-25521
> URL: https://issues.apache.org/jira/browse/CAMEL-25521
> Project: Camel
> Issue Type: Bug
> Components: camel-aws2-s3
> Reporter: shashank
> Assignee: shashank
> Priority: Major
> Fix For: 4.23.0
>
>
> Since CAMEL-17110 the consumer adds each listed key to the endpoint's
> {{inProgressRepository}} ({{AWS2S3Consumer.createExchanges}}, line 245 at
> main 5af71ab258b8) and removes it in the exchange's on completion, after the
> delete or move (lines 330-357). Two paths miss it:
> * With {{fileName}} set, {{poll}} gets the object on every poll (lines
> 126-143) without looking at the repository. With an asynchronous route, for
> example {{to("seda:...")}} (seda hands the on completion over to its
> consumer), an asynchronous producer or {{threads()}}, {{processBatch}}
> returns before the exchange is done and the next poll (500 ms later by
> default) gets the object again and routes it a second time
> ({{deleteAfterRead=true}}, the default, deletes it only when the first
> exchange completes).
> * When the consumer stops while it processes a batch, {{isBatchAllowed()}}
> turns false and {{processBatch}} drops the remaining exchanges (line 318;
> their objects stay in the bucket, by design with the default
> {{ShutdownRunningTask.CompleteCurrentTaskOnly}}), but their keys stay in the
> repository. The default repository is a memory repository of 10000 entries
> that is only cleared when the endpoint stops, so after a route stop and start
> (route controller, supervising restart, JMX, a route policy) the next polls
> skip these objects as "already in progress": they are never consumed until
> the CamelContext is restarted.
> h3. Reproduction
> New {{AWS2S3ConsumerInProgressTest}} with a small in-process fake of the S3
> calls the consumer makes (JDK {{HttpServer}}, path-style, no container):
> * {{objectInFlightIsNotConsumedAgainWithFileName}}: {{fileName=a.txt}}, route
> to {{seda:hold}}, the seda route holds the exchange until the consumer has
> completed a second poll (counted with a {{pollStrategy}}). On main: {{a.txt
> was consumed again by a later poll while its first exchange was still in
> flight ==> expected: <1> but was: <2>}}.
> * {{objectsOfAStoppedBatchAreConsumedAfterRestart}}: three objects, the first
> exchange stops the route (from another thread) and waits until the consumer
> no longer processes the batch; after the route is stopped and started again,
> {{b.txt}} and {{c.txt}} must be consumed. On main: {{mock://result Received
> message count. Expected: <2> but was: <0>}}.
> * {{notProcessedExchangesReleaseTheirObjects}}: {{processBatch}} of a
> consumer that is not started (so it processes nothing) must remove the key
> and close the object stream of the exchange ({{includeBody=false}}). On main:
> {{b.txt is still in progress ==> expected: <false> but was: <true>}}; without
> only the close (mutation) the close assertion fails.
> h3. Proposed fix
> With {{fileName}}, the key is added to the in-progress repository before the
> get; the poll is skipped while it is already there, and the key is removed
> when the get fails (and, as before, when the exchange is done). After the
> batch loop, the keys of the exchanges a stopping consumer did not process are
> removed, and their bodies closed when they are the object streams
> ({{includeBody=false}}; with {{autocloseBody=true}} their on completion would
> have closed them, but it never runs). Both follow what the listing branch and
> the camel-minio fix (CAMEL-25512) do; no new option. 4.23 upgrade-guide note
> (with {{fileName}} and {{deleteAfterRead=false}} an object in flight is now
> consumed again only once its exchange is done). A further test checks that
> with {{fileName}} a failed get ({{NoSuchKey}}) does not leave the object in
> progress. camel-aws2-s3 unit tests: 83, 0 failures (the ITs need LocalStack
> and were not run).
> Found with TLA+ models of the consumer: with {{fileName}} and an asynchronous
> route, "with deleteAfterRead every object is routed once" is violated in 5
> steps (poll, dispatch, poll, dispatch); for the listing, "every object is
> routed by one of the polls" is violated in 12 steps (list, get o1 and o2, a
> stop drops the batch, then every later poll skips both). Both hold with the
> fix (also with three objects and four polls), and the controls (synchronous
> route, no stop) hold on main. 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-aws2-s3 since 2026-01-01
> (9 issues, none about the in-progress repository), text
> "inProgressRepository" (9; CAMEL-25512 is the minio fix, the others are
> camel-file/ftp), "aws2-s3" + "fileName" + "progress" (none); GitHub pull
> requests: no open PR touches {{AWS2S3Consumer}}.
> _Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
