allthingssecurity commented on code in PR #27672:
URL: https://github.com/apache/camel/pull/27672#discussion_r4240520092
##########
components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Consumer.java:
##########
@@ -361,6 +376,17 @@ public String toString() {
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 with includeBody=false their bodies are the
object streams, which nothing else closes
+ for (Object exchange : exchanges) {
+ String key = ((Exchange)
exchange).getProperty(AWS2S3Constants.KEY, String.class);
+ if (key != null) {
+ getEndpoint().getInProgressRepository().remove(key);
+ }
+ if (((Exchange) exchange).getIn().getBody() instanceof Closeable
body) {
Review Comment:
Confirmed and fixed in 79d32739480b. With `ignoreBody=true` the stream is
never the body, so no one can read or close
it; `createExchange` now closes it right away, as it already did for
`includeBody=true`
(`AWS2S3Consumer.java:533`). The headers come from `s3Object.response()` and
are not affected. New test
`ignoredBodyStreamIsClosed` (`includeBody=false&ignoreBody=true`) fails on
the previous commit with
`the object stream of b.txt was not closed`.
_Claude Code on behalf of allthingssecurity_
##########
components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Consumer.java:
##########
@@ -361,6 +376,17 @@ public String toString() {
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 with includeBody=false their bodies are the
object streams, which nothing else closes
+ for (Object exchange : exchanges) {
+ String key = ((Exchange)
exchange).getProperty(AWS2S3Constants.KEY, String.class);
+ if (key != null) {
+ getEndpoint().getInProgressRepository().remove(key);
+ }
+ if (((Exchange) exchange).getIn().getBody() instanceof Closeable
body) {
+ IOHelper.close(body);
Review Comment:
Done in 79d32739480b: the exchanges that are not routed are released with
`releaseExchange(exchange, true)`, after the
key was read and the body closed, since a pooled exchange is reset when it
is released (`AWS2S3Consumer.java:394`). The
exchanges already created when building the batch fails are released the
same way (line 309).
`notProcessedExchangesReleaseTheirObjects` now checks that the release
happens, and after the close; it fails on the
previous commit. Module suite: 84 tests, 0 failures (68 integration tests
skipped, they need LocalStack).
_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]
