allthingssecurity commented on code in PR #27672:
URL: https://github.com/apache/camel/pull/27672#discussion_r4240532652
##########
components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Consumer.java:
##########
@@ -124,23 +125,19 @@ protected int poll() throws Exception {
if (!doneFileCheckPasses(bucketName, doneFileName)) {
exchanges = new LinkedList<>();
} else if (fileName != null) {
- LOG.trace("Getting object in bucket [{}] with file name [{}]...",
bucketName, fileName);
-
- GetObjectRequest.Builder getRequest =
GetObjectRequest.builder().bucket(bucketName).key(fileName);
- if (getConfiguration().isUseCustomerKey()) {
- if
(ObjectHelper.isNotEmpty(getConfiguration().getCustomerKeyId())) {
-
getRequest.sseCustomerKey(getConfiguration().getCustomerKeyId());
- }
- if
(ObjectHelper.isNotEmpty(getConfiguration().getCustomerKeyMD5())) {
-
getRequest.sseCustomerKeyMD5(getConfiguration().getCustomerKeyMD5());
- }
- if
(ObjectHelper.isNotEmpty(getConfiguration().getCustomerAlgorithm())) {
-
getRequest.sseCustomerAlgorithm(getConfiguration().getCustomerAlgorithm());
+ // check if file is already in progress (add false = duplicate
file)
+ if (!getEndpoint().getInProgressRepository().add(fileName)) {
+ LOG.trace("Skipping as s3 object is already in progress: {}",
fileName);
+ exchanges = new LinkedList<>();
+ } else {
+ try {
+ exchanges = createExchanges(getObject(bucketName,
fileName), fileName);
+ } catch (Exception e) {
+ // remove in progress as we failed
+ getEndpoint().getInProgressRepository().remove(fileName);
+ throw e;
Review Comment:
Confirmed and fixed in 6f7e8404a8c7: the stream of the `fileName` poll is
kept in a variable and closed in the catch,
before the key is removed from the in-progress repository
(`AWS2S3Consumer.java:133-143`). One difference from the
suggestion: `getObject` stays inside the `try` (the variable starts as
`null`), as a failed get (`NoSuchKey`) must
still remove the key; `objectWithFileNameIsConsumedOnceItExists` covers
that. New test
`objectStreamIsClosedWhenTheExchangeOfAFileNamePollCannotBeCreated` calls
`poll()` with a client whose object stream
throws an `IOException` on read (`includeBody=true`); it fails on the
previous commit with
`the object stream of a.txt was not closed` and passes now. Module suite: 85
tests, 0 failures (68 integration tests
skipped, they need LocalStack). The listing path is not touched.
_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]
