allthingssecurity commented on code in PR #27673:
URL: https://github.com/apache/camel/pull/27673#discussion_r4240520218
##########
components/camel-ibm/camel-ibm-cos/src/main/java/org/apache/camel/component/ibm/cos/IBMCOSConsumer.java:
##########
@@ -136,6 +149,15 @@ public int processBatch(Queue<Object> exchanges) throws
Exception {
AsyncCallback cb = defaultConsumerCallback(exchange, true);
getAsyncProcessor().process(exchange, cb);
}
+ // the remaining exchanges are not processed as the consumer is
stopping: a later poll must consume their
+ // objects again, and their on completions, which close the content
stream, never run
+ for (Object exchange : exchanges) {
+ Message message = ((Exchange) exchange).getIn();
+ removeInProgress(message.getHeader(IBMCOSConstants.KEY,
String.class));
Review Comment:
Applied in 1fadc8296e0a: the key now comes from the exchange property that
`createExchange` stores
(`IBMCOSConsumer.java:157`), with a null check as in camel-aws2-s3. One
detail: the consumer's own on completion uses
the key captured when the exchange is created rather than the property, and
these exchanges were never routed so
nothing can have rewritten their headers; the change is for consistency with
aws2-s3 rather than a fix.
`notProcessedExchangesCloseTheirContent` now sets a different key in the
header and fails on the previous commit with
`b.txt is still in progress`.
_Claude Code on behalf of allthingssecurity_
##########
components/camel-ibm/camel-ibm-cos/src/main/java/org/apache/camel/component/ibm/cos/IBMCOSConsumer.java:
##########
@@ -136,6 +149,15 @@ public int processBatch(Queue<Object> exchanges) throws
Exception {
AsyncCallback cb = defaultConsumerCallback(exchange, true);
getAsyncProcessor().process(exchange, cb);
}
+ // the remaining exchanges are not processed as the consumer is
stopping: a later poll must consume their
+ // objects again, and their on completions, which close the content
stream, never run
+ for (Object exchange : exchanges) {
+ Message message = ((Exchange) exchange).getIn();
+ removeInProgress(message.getHeader(IBMCOSConstants.KEY,
String.class));
+ if (message.getBody() instanceof Closeable body) {
+ IOHelper.close(body);
Review Comment:
Done in 1fadc8296e0a: `releaseExchange(exchange, true)` after the key was
read and the content stream closed, since a
pooled exchange is reset when it is released (`IBMCOSConsumer.java:166`).
The test checks that the release happens,
and after the close. Module suite: 6 tests, 0 failures (15 integration tests
skipped, they need an IBM Cloud account).
_Claude Code on behalf of allthingssecurity_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
