[
https://issues.apache.org/jira/browse/CAMEL-25522?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Work on CAMEL-25522 started by shashank.
----------------------------------------
> camel-ibm-cos - consumer with fileName routes the object again while its
> exchange is in flight, and a batch cut short by a route stop leaves its
> objects in progress and their streams open
> -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-25522
> URL: https://issues.apache.org/jira/browse/CAMEL-25522
> Project: Camel
> Issue Type: Bug
> Reporter: shashank
> Assignee: shashank
> Priority: Major
>
> The consumer adds each listed key to the endpoint's {{inProgressRepository}}
> ({{IBMCOSConsumer.createExchanges}}, line 163 at main 5af71ab258b8) and
> removes it in the on completion that {{createExchange}} adds (lines 230-240),
> which also closes the object's content stream ({{includeBody=true}} and
> {{autocloseBody=true}}, the defaults, lines 190 and 216-222). Two paths miss
> it:
> * With {{fileName}} set, {{poll}} gets the object on every poll (lines 68-72)
> 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 124;
> their objects stay in the bucket, by design with the default
> {{ShutdownRunningTask.CompleteCurrentTaskOnly}}). Their on completions never
> run: their keys stay in the repository (a memory repository of 10000 entries,
> only cleared when the endpoint stops), so after a route stop and start the
> next polls skip these objects as "already in progress" and they are never
> consumed until the CamelContext is restarted; and their content streams,
> opened by {{getObject}} for every listed object before the batch is routed,
> are never closed, so each such stop keeps up to {{maxMessagesPerPoll}} - 1
> connections of the client's pool (50 by default).
> h3. Reproduction
> New {{IBMCOSConsumerInProgressTest}}, with a small in-process fake of the S3
> calls the consumer makes (JDK {{HttpServer}}, path-style) and an IBM COS
> client with static HMAC credentials pointing at it ({{cosClient=#cos}}):
> * {{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>}}.
> * {{notProcessedExchangesCloseTheirContent}}: {{processBatch}} of a consumer
> that is not started (so it processes nothing) must release the key and close
> the body stream of the exchange. On main: {{b.txt is still in progress ==>
> expected: <false> but was: <true>}}; with the key fix but without the close
> (mutation): the close assertion fails.
> h3. Proposed fix
> With {{fileName}}, the key is added to the in-progress repository before the
> get (when a repository is configured, as in the listing); 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 body
> streams closed. No new option. 4.23 upgrade-guide note. camel-ibm-cos unit
> tests: 5, 0 failures (the ITs need an IBM Cloud account and were not run).
> Found with TLA+ models of the consumer (the same models as the camel-aws2-s3
> ticket, whose consumer has the same steps): with {{fileName}} and an
> asynchronous route, "with deleteAfterRead every object is routed once" is
> violated in 5 steps; for the listing, "every object is routed by one of the
> polls" is violated in 12 steps (a stop drops the batch, then every later poll
> skips its objects). Both hold with the fix. The open streams are not
> modelled; they were found by reading the code next to the model and are
> covered by the third test.
> Affected: main, camel-4.22.x, camel-4.18.x (the component is not in 4.14.x).
> Duplicate check (2026-10-10): JIRA text "ibm-cos" (10 issues; CAMEL-24491
> fixed the listing's in-progress check, the others are producer and header
> fixes), "inProgressRepository" (9, none about ibm-cos); GitHub pull requests:
> no open PR touches {{IBMCOSConsumer}}.
> _Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
