[jira] [Updated] (CAMEL-25521) 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

["shashank (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/CAMEL-25521?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

shashank updated CAMEL-25521:
-----------------------------
    Description: 
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._


  was:
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). camel-aws2-s3 unit tests: 82, 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._



> 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
>
> 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)