[
https://issues.apache.org/jira/browse/CAMEL-25522?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
shashank updated CAMEL-25522:
-----------------------------
Description:
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. A further test checks
that with {{fileName}} a failed get does not leave the object in progress.
camel-ibm-cos unit tests: 6, 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._
was:
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._
> 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. A further test checks
> that with {{fileName}} a failed get does not leave the object in progress.
> camel-ibm-cos unit tests: 6, 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)
