This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-24157 in repository https://gitbox.apache.org/repos/asf/camel.git
commit e0ae269c028955c903d4a107dfbd99015bc52701 Author: Claus Ibsen <[email protected]> AuthorDate: Fri Jul 17 18:03:07 2026 +0200 CAMEL-24157: camel-aws2-s3 - Fix pagination, SSE-C reads, prefix move, presigner, and deprecate maxConnections Fix ListObjectsV2 pagination fallback to use startAfter instead of continuationToken when nextContinuationToken is absent (S3-compatible stores). Apply SSE-C customer key headers on getObject, getObjectRange, and consumer fileName path (read operations were missing SSE-C unlike write operations). Fix removePrefixOnMove to use substring instead of regex replaceFirst which caused PatternSyntaxException on prefixes with special characters and skipped deleteAfterRead on failure. Fix presigner to support session credentials, profile credentials, and null region. Deprecate maxConnections option that was never wired to the S3 client. Co-Authored-By: Claude Opus 4.6 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../camel/component/aws2/s3/AWS2S3Consumer.java | 38 +++++++++++++---- .../camel/component/aws2/s3/AWS2S3Endpoint.java | 13 +++++- .../camel/component/aws2/s3/AWS2S3Producer.java | 47 +++++++++++++++++++--- 3 files changed, 83 insertions(+), 15 deletions(-) diff --git a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Consumer.java b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Consumer.java index eebd38a2777b..a0e5fce60195 100644 --- a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Consumer.java +++ b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Consumer.java @@ -68,6 +68,7 @@ public class AWS2S3Consumer extends ScheduledBatchPollingConsumer { private static final Logger LOG = LoggerFactory.getLogger(AWS2S3Consumer.class); private String marker; + private boolean markerIsContinuationToken = true; private transient String s3ConsumerToString; public AWS2S3Consumer(AWS2S3Endpoint endpoint, Processor processor) { @@ -125,8 +126,20 @@ public class AWS2S3Consumer extends ScheduledBatchPollingConsumer { } 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()); + } + } ResponseInputStream<GetObjectResponse> s3Object - = getAmazonS3Client().getObject(GetObjectRequest.builder().bucket(bucketName).key(fileName).build()); + = getAmazonS3Client().getObject(getRequest.build()); exchanges = createExchanges(s3Object, fileName); } else { LOG.trace("Queueing objects in bucket [{}]...", bucketName); @@ -147,26 +160,35 @@ public class AWS2S3Consumer extends ScheduledBatchPollingConsumer { // continue from where we left last time if (marker != null) { LOG.trace("Resuming from marker: {}", marker); - listObjectsRequest.continuationToken(marker); + if (markerIsContinuationToken) { + listObjectsRequest.continuationToken(marker); + } else { + listObjectsRequest.startAfter(marker); + } } ListObjectsV2Response listObjects = getAmazonS3Client().listObjectsV2(listObjectsRequest.build()); if (Boolean.TRUE.equals(listObjects.isTruncated())) { String next = listObjects.nextContinuationToken(); - if (next == null && listObjects.hasContents() && ObjectHelper.isEmpty(listObjects.prefix())) { - // fallback to use last key from the returned list of objects + if (next != null) { + markerIsContinuationToken = true; + } else if (listObjects.hasContents()) { + // S3-compatible stores may omit nextContinuationToken; + // fall back to startAfter with the last returned key int size = listObjects.contents().size(); if (size > 0) { S3Object last = listObjects.contents().get(size - 1); next = last.key(); } + markerIsContinuationToken = false; } marker = next; - LOG.trace("Returned list is truncated, so setting next continuation token: {}", marker); + LOG.trace("Returned list is truncated, so setting next marker: {}", marker); } else { // no more data so clear marker marker = null; + markerIsContinuationToken = true; } if (LOG.isTraceEnabled()) { LOG.trace("Found {} objects in bucket [{}]...", listObjects.contents().size(), bucketName); @@ -364,8 +386,10 @@ public class AWS2S3Consumer extends ScheduledBatchPollingConsumer { builder.append(AWS2S3Utils.evaluateDestinationBucketSuffix(exchange, getConfiguration())); String destinationKey = builder.toString(); - if (getConfiguration().isRemovePrefixOnMove()) { - destinationKey = destinationKey.replaceFirst(getConfiguration().getPrefix(), ""); + if (getConfiguration().isRemovePrefixOnMove() + && ObjectHelper.isNotEmpty(getConfiguration().getPrefix()) + && destinationKey.startsWith(getConfiguration().getPrefix())) { + destinationKey = destinationKey.substring(getConfiguration().getPrefix().length()); } getAmazonS3Client().copyObject(CopyObjectRequest.builder().destinationKey(destinationKey) diff --git a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Endpoint.java b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Endpoint.java index e2db0faba804..7350a7bebd8f 100644 --- a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Endpoint.java +++ b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Endpoint.java @@ -63,7 +63,11 @@ public class AWS2S3Endpoint extends ScheduledPollEndpoint implements EndpointSer private AWS2S3Configuration configuration; @UriParam(label = "consumer", defaultValue = "10") private int maxMessagesPerPoll = 10; - @UriParam(label = "consumer", defaultValue = "60") + @Deprecated + @UriParam(label = "consumer", defaultValue = "60", + description = "Set the maxConnections parameter in the S3 client configuration. This option is" + + " deprecated and has no effect. Use the httpClientBuilder option on the S3 client" + + " configuration to customize the HTTP client connection pool.") private int maxConnections = 50 + maxMessagesPerPoll; @UriParam(label = "consumer,advanced", description = "A pluggable in-progress repository " + "org.apache.camel.spi.IdempotentRepository. The in-progress repository is used to account the current in " @@ -229,13 +233,18 @@ public class AWS2S3Endpoint extends ScheduledPollEndpoint implements EndpointSer this.maxMessagesPerPoll = maxMessagesPerPoll; } + /** + * @deprecated This option has no effect and will be removed in a future release. + */ + @Deprecated public int getMaxConnections() { return maxConnections; } /** - * Set the maxConnections parameter in the S3 client configuration + * @deprecated This option has no effect and will be removed in a future release. */ + @Deprecated public void setMaxConnections(int maxConnections) { this.maxConnections = maxConnections; } diff --git a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Producer.java b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Producer.java index 728cc5d946ef..eecb6949cf74 100644 --- a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Producer.java +++ b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Producer.java @@ -42,7 +42,9 @@ import org.apache.camel.util.ObjectHelper; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; +import software.amazon.awssdk.auth.credentials.AwsSessionCredentials; import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider; +import software.amazon.awssdk.auth.credentials.ProfileCredentialsProvider; import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; import software.amazon.awssdk.awscore.AwsResponse; import software.amazon.awssdk.core.ResponseInputStream; @@ -642,6 +644,17 @@ public class AWS2S3Producer extends DefaultProducer { if (ObjectHelper.isNotEmpty(ifUnmodifiedSince)) { req.ifUnmodifiedSince(ifUnmodifiedSince); } + if (getConfiguration().isUseCustomerKey()) { + if (ObjectHelper.isNotEmpty(getConfiguration().getCustomerKeyId())) { + req.sseCustomerKey(getConfiguration().getCustomerKeyId()); + } + if (ObjectHelper.isNotEmpty(getConfiguration().getCustomerKeyMD5())) { + req.sseCustomerKeyMD5(getConfiguration().getCustomerKeyMD5()); + } + if (ObjectHelper.isNotEmpty(getConfiguration().getCustomerAlgorithm())) { + req.sseCustomerAlgorithm(getConfiguration().getCustomerAlgorithm()); + } + } ResponseInputStream<GetObjectResponse> res = s3Client.getObject(req.build(), ResponseTransformer.toInputStream()); Message message = getMessageForResponse(exchange); @@ -677,6 +690,17 @@ public class AWS2S3Producer extends DefaultProducer { GetObjectRequest.Builder req = GetObjectRequest.builder().bucket(bucketName).key(keyName) .range("bytes=" + Long.parseLong(rangeStart) + "-" + Long.parseLong(rangeEnd)); + if (getConfiguration().isUseCustomerKey()) { + if (ObjectHelper.isNotEmpty(getConfiguration().getCustomerKeyId())) { + req.sseCustomerKey(getConfiguration().getCustomerKeyId()); + } + if (ObjectHelper.isNotEmpty(getConfiguration().getCustomerKeyMD5())) { + req.sseCustomerKeyMD5(getConfiguration().getCustomerKeyMD5()); + } + if (ObjectHelper.isNotEmpty(getConfiguration().getCustomerAlgorithm())) { + req.sseCustomerAlgorithm(getConfiguration().getCustomerAlgorithm()); + } + } ResponseInputStream<GetObjectResponse> res = s3Client.getObject(req.build(), ResponseTransformer.toInputStream()); Message message = getMessageForResponse(exchange); @@ -1089,12 +1113,23 @@ public class AWS2S3Producer extends DefaultProducer { } S3Presigner.Builder builder = S3Presigner.builder(); - builder.credentialsProvider( - getConfiguration().isUseDefaultCredentialsProvider() - ? DefaultCredentialsProvider.create() : StaticCredentialsProvider.create( - AwsBasicCredentials.create(getConfiguration().getAccessKey(), - getConfiguration().getSecretKey()))) - .region(Region.of(getConfiguration().getRegion())); + if (getConfiguration().isUseDefaultCredentialsProvider()) { + builder.credentialsProvider(DefaultCredentialsProvider.create()); + } else if (getConfiguration().isUseProfileCredentialsProvider()) { + builder.credentialsProvider( + ProfileCredentialsProvider.create(getConfiguration().getProfileCredentialsName())); + } else if (getConfiguration().isUseSessionCredentials()) { + builder.credentialsProvider(StaticCredentialsProvider.create( + AwsSessionCredentials.create(getConfiguration().getAccessKey(), + getConfiguration().getSecretKey(), getConfiguration().getSessionToken()))); + } else { + builder.credentialsProvider(StaticCredentialsProvider.create( + AwsBasicCredentials.create(getConfiguration().getAccessKey(), + getConfiguration().getSecretKey()))); + } + if (ObjectHelper.isNotEmpty(getConfiguration().getRegion())) { + builder.region(Region.of(getConfiguration().getRegion())); + } if (getConfiguration().isForcePathStyle()) { builder.serviceConfiguration(S3Configuration.builder().pathStyleAccessEnabled(true).build());
