davsclaus commented on code in PR #27697:
URL: https://github.com/apache/camel/pull/27697#discussion_r4245538867
##########
components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Consumer.java:
##########
@@ -283,23 +284,36 @@ protected Queue<Exchange> createExchanges(List<S3Object>
s3ObjectSummaries) {
getRequest.sseCustomerAlgorithm(getConfiguration().getCustomerAlgorithm());
}
}
- ResponseInputStream<GetObjectResponse> s3Object
- = getAmazonS3Client().getObject(getRequest.build(),
ResponseTransformer.toInputStream());
-
- if (includeS3Object(s3Object)) {
- s3Objects.add(s3Object);
- Exchange exchange = createExchange(s3Object, key);
- answer.add(exchange);
- } else {
- // If includeFolders != true and the object is not
included, it is safe to close the object here.
- // If includeFolders == true, the exchange will close the
object.
- IOHelper.close(s3Object);
- // remove in progress
- getEndpoint().getInProgressRepository().remove(key);
+ // a failure to get one listed object, or to create its
exchange, does not fail the poll: the object is
+ // skipped, the other listed objects are consumed, and a later
poll tries it again
+ ResponseInputStream<GetObjectResponse> s3Object = null;
+ try {
+ s3Object =
getAmazonS3Client().getObject(getRequest.build(),
ResponseTransformer.toInputStream());
+ if (includeS3Object(s3Object)) {
+ Exchange exchange = createExchange(s3Object, key);
+ s3Objects.add(s3Object);
+ answer.add(exchange);
+ } else {
+ // If includeFolders != true and the object is not
included, it is safe to close the object
+ // here. If includeFolders == true, the exchange will
close the object.
+ IOHelper.close(s3Object);
+ // remove in progress
+ getEndpoint().getInProgressRepository().remove(key);
+ }
+ } catch (NoSuchKeyException e) {
+ // deleted after it was listed (by another consumer, a
lifecycle rule, or the exchange of an
+ // earlier poll that completed meanwhile)
+ LOG.debug("Skipping as s3 object no longer exists: {}",
key);
+ skipS3Object(s3Object, key);
+ } catch (Exception e) {
+ // no stack trace: an object that cannot be got (archived,
access denied) is logged at every poll
+ LOG.warn("Error getting S3Object {} from bucket {} due:
{}. Skipping it for this poll.", key,
Review Comment:
This consumer already has a skip-and-continue pattern in `processCommit`:
`getExceptionHandler().handleException("... This exception is ignored.", e)`.
Using it here keeps the skip, but lets users who route consumer errors
(`bridgeErrorHandler`, a custom `exceptionHandler`) see the failure. With the
default `LoggingExceptionHandler` the cost is a WARN with a stack trace, as
before.
--
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]
