davsclaus commented on code in PR #27673:
URL: https://github.com/apache/camel/pull/27673#discussion_r4240436752
##########
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:
Nit: these exchanges are never routed, so they are never released back to
the exchange factory (matters with a pooled factory).
`GenericFileConsumer.removeExcessiveInProgressFiles` calls
`releaseExchange(exchange, true)` in the same situation.
##########
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:
Nit: the consumer's own on completion uses the key stored as exchange
property; reading the header here can go wrong if something rewrote headers.
For consistency with the aws2-s3 PR:
```suggestion
removeInProgress(((Exchange)
exchange).getProperty(IBMCOSConstants.KEY, String.class));
```
--
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]
