You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2025/08/03 09:06:40 UTC
(camel) branch s3u created (now 17067330d7e)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a change to branch s3u
in repository https://gitbox.apache.org/repos/asf/camel.git
at 17067330d7e CAMEL-16871: camel-aws-s3 - multipartUpload to support generic input stream also
This branch includes the following new commits:
new 17067330d7e CAMEL-16871: camel-aws-s3 - multipartUpload to support generic input stream also
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(camel) 01/01: CAMEL-16871: camel-aws-s3 - multipartUpload to support generic input stream also
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch s3u
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 17067330d7e1907dc5a449ff22bc466b914acd22
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Sun Aug 3 11:05:55 2025 +0200
CAMEL-16871: camel-aws-s3 - multipartUpload to support generic input stream also
---
.../apache/camel/catalog/components/aws2-s3.json | 4 +-
.../apache/camel/component/aws2/s3/aws2-s3.json | 4 +-
.../component/aws2/s3/AWS2S3Configuration.java | 4 +-
.../camel/component/aws2/s3/AWS2S3Producer.java | 99 +++++++++++----------
.../camel/component/aws2/s3/utils/AWS2S3Utils.java | 3 +
.../s3/integration/S3StreamUploadMultipartIT.java | 10 +--
...ltipartIT.java => S3UploadFileMultipartIT.java} | 35 ++++----
...IT.java => S3UploadInputStreamMultipartIT.java} | 37 ++++----
...UploadInputStreamMultipartNoStreamCacheIT.java} | 39 ++++----
.../camel-aws2-s3/src/test/resources/empty-big.bin | Bin 0 -> 10485760 bytes
.../endpoint/dsl/AWS2S3EndpointBuilderFactory.java | 14 +--
11 files changed, 125 insertions(+), 124 deletions(-)
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-s3.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-s3.json
index 71939689b15..2c7d968d35f 100644
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-s3.json
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-s3.json
@@ -59,7 +59,7 @@
"multiPartUpload": { "index": 32, "kind": "property", "displayName": "Multi Part Upload", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "If it is true, camel will upload the file with multipart format. The part size [...]
"namingStrategy": { "index": 33, "kind": "property", "displayName": "Naming Strategy", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "org.apache.camel.component.aws2.s3.stream.AWSS3NamingStrategyEnum", "enum": [ "progressive", "random" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "progressive", "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuratio [...]
"operation": { "index": 34, "kind": "property", "displayName": "Operation", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "org.apache.camel.component.aws2.s3.AWS2S3Operations", "enum": [ "copyObject", "listObjects", "deleteObject", "deleteBucket", "listBuckets", "getObject", "getObjectRange", "createDownloadLink", "headBucket", "headObject" ], "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel [...]
- "partSize": { "index": 35, "kind": "property", "displayName": "Part Size", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 26214400, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "Set up the partSize which is used in multipart upload, the default size is 25M. Camel will onl [...]
+ "partSize": { "index": 35, "kind": "property", "displayName": "Part Size", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 26214400, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "Set up the partSize which is used in multipart upload, the default size is 25 MB. The minimum [...]
"restartingPolicy": { "index": 36, "kind": "property", "displayName": "Restarting Policy", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "org.apache.camel.component.aws2.s3.stream.AWSS3RestartingPolicyEnum", "enum": [ "override", "lastPart" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "override", "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configurat [...]
"storageClass": { "index": 37, "kind": "property", "displayName": "Storage Class", "group": "producer", "label": "producer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "The storage class to set in the com.amazonaws.services.s3.model.PutObjectRequest request." },
"streamingUploadMode": { "index": 38, "kind": "property", "displayName": "Streaming Upload Mode", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "When stream mode is true, the upload to bucket will be done in streaming" },
@@ -168,7 +168,7 @@
"multiPartUpload": { "index": 38, "kind": "parameter", "displayName": "Multi Part Upload", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "If it is true, camel will upload the file with multipart format. The part size [...]
"namingStrategy": { "index": 39, "kind": "parameter", "displayName": "Naming Strategy", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "org.apache.camel.component.aws2.s3.stream.AWSS3NamingStrategyEnum", "enum": [ "progressive", "random" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "progressive", "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configurati [...]
"operation": { "index": 40, "kind": "parameter", "displayName": "Operation", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "org.apache.camel.component.aws2.s3.AWS2S3Operations", "enum": [ "copyObject", "listObjects", "deleteObject", "deleteBucket", "listBuckets", "getObject", "getObjectRange", "createDownloadLink", "headBucket", "headObject" ], "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.came [...]
- "partSize": { "index": 41, "kind": "parameter", "displayName": "Part Size", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 26214400, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "Set up the partSize which is used in multipart upload, the default size is 25M. Camel will on [...]
+ "partSize": { "index": 41, "kind": "parameter", "displayName": "Part Size", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 26214400, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "Set up the partSize which is used in multipart upload, the default size is 25 MB. The minimum [...]
"restartingPolicy": { "index": 42, "kind": "parameter", "displayName": "Restarting Policy", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "org.apache.camel.component.aws2.s3.stream.AWSS3RestartingPolicyEnum", "enum": [ "override", "lastPart" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "override", "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configura [...]
"storageClass": { "index": 43, "kind": "parameter", "displayName": "Storage Class", "group": "producer", "label": "producer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "The storage class to set in the com.amazonaws.services.s3.model.PutObjectRequest request." },
"streamingUploadMode": { "index": 44, "kind": "parameter", "displayName": "Streaming Upload Mode", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "When stream mode is true, the upload to bucket will be done in streaming" },
diff --git a/components/camel-aws/camel-aws2-s3/src/generated/resources/META-INF/org/apache/camel/component/aws2/s3/aws2-s3.json b/components/camel-aws/camel-aws2-s3/src/generated/resources/META-INF/org/apache/camel/component/aws2/s3/aws2-s3.json
index 71939689b15..2c7d968d35f 100644
--- a/components/camel-aws/camel-aws2-s3/src/generated/resources/META-INF/org/apache/camel/component/aws2/s3/aws2-s3.json
+++ b/components/camel-aws/camel-aws2-s3/src/generated/resources/META-INF/org/apache/camel/component/aws2/s3/aws2-s3.json
@@ -59,7 +59,7 @@
"multiPartUpload": { "index": 32, "kind": "property", "displayName": "Multi Part Upload", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "If it is true, camel will upload the file with multipart format. The part size [...]
"namingStrategy": { "index": 33, "kind": "property", "displayName": "Naming Strategy", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "org.apache.camel.component.aws2.s3.stream.AWSS3NamingStrategyEnum", "enum": [ "progressive", "random" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "progressive", "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuratio [...]
"operation": { "index": 34, "kind": "property", "displayName": "Operation", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "org.apache.camel.component.aws2.s3.AWS2S3Operations", "enum": [ "copyObject", "listObjects", "deleteObject", "deleteBucket", "listBuckets", "getObject", "getObjectRange", "createDownloadLink", "headBucket", "headObject" ], "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel [...]
- "partSize": { "index": 35, "kind": "property", "displayName": "Part Size", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 26214400, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "Set up the partSize which is used in multipart upload, the default size is 25M. Camel will onl [...]
+ "partSize": { "index": 35, "kind": "property", "displayName": "Part Size", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 26214400, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "Set up the partSize which is used in multipart upload, the default size is 25 MB. The minimum [...]
"restartingPolicy": { "index": 36, "kind": "property", "displayName": "Restarting Policy", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "org.apache.camel.component.aws2.s3.stream.AWSS3RestartingPolicyEnum", "enum": [ "override", "lastPart" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "override", "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configurat [...]
"storageClass": { "index": 37, "kind": "property", "displayName": "Storage Class", "group": "producer", "label": "producer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "The storage class to set in the com.amazonaws.services.s3.model.PutObjectRequest request." },
"streamingUploadMode": { "index": 38, "kind": "property", "displayName": "Streaming Upload Mode", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "When stream mode is true, the upload to bucket will be done in streaming" },
@@ -168,7 +168,7 @@
"multiPartUpload": { "index": 38, "kind": "parameter", "displayName": "Multi Part Upload", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "If it is true, camel will upload the file with multipart format. The part size [...]
"namingStrategy": { "index": 39, "kind": "parameter", "displayName": "Naming Strategy", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "org.apache.camel.component.aws2.s3.stream.AWSS3NamingStrategyEnum", "enum": [ "progressive", "random" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "progressive", "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configurati [...]
"operation": { "index": 40, "kind": "parameter", "displayName": "Operation", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "org.apache.camel.component.aws2.s3.AWS2S3Operations", "enum": [ "copyObject", "listObjects", "deleteObject", "deleteBucket", "listBuckets", "getObject", "getObjectRange", "createDownloadLink", "headBucket", "headObject" ], "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.came [...]
- "partSize": { "index": 41, "kind": "parameter", "displayName": "Part Size", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 26214400, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "Set up the partSize which is used in multipart upload, the default size is 25M. Camel will on [...]
+ "partSize": { "index": 41, "kind": "parameter", "displayName": "Part Size", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 26214400, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "Set up the partSize which is used in multipart upload, the default size is 25 MB. The minimum [...]
"restartingPolicy": { "index": 42, "kind": "parameter", "displayName": "Restarting Policy", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "org.apache.camel.component.aws2.s3.stream.AWSS3RestartingPolicyEnum", "enum": [ "override", "lastPart" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "override", "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configura [...]
"storageClass": { "index": 43, "kind": "parameter", "displayName": "Storage Class", "group": "producer", "label": "producer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "The storage class to set in the com.amazonaws.services.s3.model.PutObjectRequest request." },
"streamingUploadMode": { "index": 44, "kind": "parameter", "displayName": "Streaming Upload Mode", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "When stream mode is true, the upload to bucket will be done in streaming" },
diff --git a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Configuration.java b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Configuration.java
index 243261095a9..f84228d05db 100644
--- a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Configuration.java
+++ b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Configuration.java
@@ -149,7 +149,9 @@ public class AWS2S3Configuration implements Cloneable {
}
/**
- * Set up the partSize which is used in multipart upload, the default size is 25M.
+ * Set up the partSize which is used in multipart upload, the default size is 25 MB.
+ *
+ * The minimum size in AWS is 5 MB.
*
* Camel will only do multipart uploads for files that are larger than the part-size thresholds. Files that are
* smaller will be uploaded in a single operation.
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 68b7f85fc26..7e6180d577b 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
@@ -114,36 +114,51 @@ public class AWS2S3Producer extends DefaultProducer {
}
public void processMultiPart(final Exchange exchange) throws Exception {
- File filePayload;
- Object obj = exchange.getIn().getMandatoryBody();
+ Object obj = exchange.getIn().getBody();
+ InputStream inputStream;
+ File filePayload = null;
+
+ // the content-length may already be known
+ long contentLength = exchange.getIn().getHeader(AWS2S3Constants.CONTENT_LENGTH, -1, Long.class);
+
// Need to check if the message body is WrappedFile
- if (obj instanceof WrappedFile) {
- obj = ((WrappedFile<?>) obj).getFile();
+ if (obj instanceof WrappedFile<?> wf) {
+ obj = wf.getFile();
}
- if (obj instanceof File) {
- filePayload = (File) obj;
+ if (obj instanceof File f) {
+ filePayload = f;
+ inputStream = new FileInputStream(f);
+ contentLength = f.length();
} else {
- throw new IllegalArgumentException("aws2-s3: MultiPart upload requires a File input.");
+ // okay we use input stream
+ inputStream = exchange.getIn().getMandatoryBody(InputStream.class);
+ if (contentLength <= 0) {
+ contentLength = AWS2S3Utils.determineLengthInputStream(inputStream);
+ if (contentLength == -1) {
+ // fallback to read into memory to calculate length
+ LOG.debug(
+ "The content length is not defined. It needs to be determined by reading the data into memory");
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ IOHelper.copyAndCloseInput(inputStream, baos);
+ byte[] arr = baos.toByteArray();
+ contentLength = arr.length;
+ inputStream = new ByteArrayInputStream(arr);
+ }
+ }
}
Map<String, String> objectMetadata = determineMetadata(exchange);
- Long contentLength = exchange.getIn().getHeader(AWS2S3Constants.CONTENT_LENGTH, Long.class);
- if (contentLength == null || contentLength == 0) {
- contentLength = filePayload.length();
- }
-
long partSize = getConfiguration().getPartSize();
if (contentLength == 0 || contentLength < partSize) {
// optimize to do a single op if content length is known and < part size
- LOG.debug("File size < partSize. Uploading file in single operation: {}", filePayload);
- processSingleOp(exchange);
+ LOG.debug("Payload size < partSize ({} > {}). Uploading payload in single operation", contentLength, partSize);
+ doPutObject(exchange, objectMetadata, filePayload, inputStream, contentLength);
return;
}
- LOG.debug("File size >= partSize. Uploading file using multi-part operation: {}", filePayload);
-
- objectMetadata.put("Content-Length", contentLength.toString());
+ LOG.debug("Payload size >= partSize ({} > {}). Uploading payload using multi-part operation", contentLength, partSize);
+ objectMetadata.put("Content-Length", "" + contentLength);
final String keyName = AWS2S3Utils.determineKey(exchange, getConfiguration());
final String bucketName = AWS2S3Utils.determineBucketName(exchange, getConfiguration());
@@ -198,36 +213,29 @@ public class AWS2S3Producer extends DefaultProducer {
List<CompletedPart> completedParts = new ArrayList<CompletedPart>();
CompleteMultipartUploadResponse uploadResult;
- long filePosition = 0;
+ long position = 0;
try {
- for (int part = 1; filePosition < contentLength; part++) {
- partSize = Math.min(partSize, contentLength - filePosition);
+ for (int part = 1; position < contentLength; part++) {
+ partSize = Math.min(partSize, contentLength - position);
UploadPartRequest uploadRequest = UploadPartRequest.builder().bucket(getConfiguration().getBucketName())
.key(keyName).uploadId(initResponse.uploadId())
.partNumber(part).build();
- LOG.trace("Uploading part [{}] for {}", part, keyName);
- try (InputStream fileInputStream = new FileInputStream(filePayload)) {
- if (filePosition > 0) {
- long skipped = fileInputStream.skip(filePosition);
- if (skipped == 0) {
- LOG.warn("While trying to upload the file {} file, 0 bytes were skipped", keyName);
- }
- }
+ LOG.debug("Uploading multi-part [{}] at position: [{}] for {}", part, position, keyName);
- String etag = getEndpoint().getS3Client()
- .uploadPart(uploadRequest, RequestBody.fromInputStream(fileInputStream, partSize)).eTag();
- CompletedPart partUpload = CompletedPart.builder().partNumber(part).eTag(etag).build();
- completedParts.add(partUpload);
- filePosition += partSize;
- }
+ String etag = getEndpoint().getS3Client()
+ .uploadPart(uploadRequest, RequestBody.fromInputStream(inputStream, partSize)).eTag();
+ CompletedPart partUpload = CompletedPart.builder().partNumber(part).eTag(etag).build();
+ completedParts.add(partUpload);
+ position += partSize;
}
- CompletedMultipartUpload completeMultipartUpload = CompletedMultipartUpload.builder().parts(completedParts).build();
- CompleteMultipartUploadRequest.Builder compRequestBuilder;
- compRequestBuilder = CompleteMultipartUploadRequest.builder().multipartUpload(completeMultipartUpload)
- .bucket(getConfiguration().getBucketName()).key(keyName).uploadId(initResponse.uploadId());
+ LOG.debug("Completing multi-part upload for {}", keyName);
+ CompletedMultipartUpload completeMultipartUpload = CompletedMultipartUpload.builder().parts(completedParts).build();
+ CompleteMultipartUploadRequest.Builder compRequestBuilder = CompleteMultipartUploadRequest.builder()
+ .multipartUpload(completeMultipartUpload).bucket(getConfiguration().getBucketName()).key(keyName)
+ .uploadId(initResponse.uploadId());
if (getConfiguration().isConditionalWritesEnabled()) {
compRequestBuilder.ifNoneMatch("*");
}
@@ -238,6 +246,8 @@ public class AWS2S3Producer extends DefaultProducer {
.abortMultipartUpload(AbortMultipartUploadRequest.builder().bucket(getConfiguration().getBucketName())
.key(keyName).uploadId(initResponse.uploadId()).build());
throw e;
+ } finally {
+ IOHelper.close(inputStream);
}
Message message = getMessageForResponse(exchange);
@@ -248,16 +258,12 @@ public class AWS2S3Producer extends DefaultProducer {
message.setHeader(AWS2S3Constants.VERSION_ID, uploadResult.versionId());
}
- if (getConfiguration().isDeleteAfterWrite()) {
+ if (filePayload != null && getConfiguration().isDeleteAfterWrite()) {
FileUtil.deleteFile(filePayload);
}
}
public void processSingleOp(final Exchange exchange) throws Exception {
- PutObjectRequest.Builder putObjectRequest = PutObjectRequest.builder();
-
- Map<String, String> objectMetadata = determineMetadata(exchange);
-
// the content-length may already be known
long contentLength = exchange.getIn().getHeader(AWS2S3Constants.CONTENT_LENGTH, -1, Long.class);
@@ -292,7 +298,8 @@ public class AWS2S3Producer extends DefaultProducer {
}
}
- doPutObject(exchange, putObjectRequest, objectMetadata, filePayload, inputStream, contentLength);
+ Map<String, String> objectMetadata = determineMetadata(exchange);
+ doPutObject(exchange, objectMetadata, filePayload, inputStream, contentLength);
} finally {
IOHelper.close(inputStream);
}
@@ -303,10 +310,12 @@ public class AWS2S3Producer extends DefaultProducer {
}
private void doPutObject(
- Exchange exchange, PutObjectRequest.Builder putObjectRequest, Map<String, String> objectMetadata,
+ Exchange exchange, Map<String, String> objectMetadata,
File file, InputStream inputStream, long contentLength) {
final String bucketName = AWS2S3Utils.determineBucketName(exchange, getConfiguration());
final String keyName = AWS2S3Utils.determineKey(exchange, getConfiguration());
+
+ PutObjectRequest.Builder putObjectRequest = PutObjectRequest.builder();
putObjectRequest.bucket(bucketName).key(keyName).metadata(objectMetadata);
String storageClass = AWS2S3Utils.determineStorageClass(exchange, getConfiguration());
diff --git a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/utils/AWS2S3Utils.java b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/utils/AWS2S3Utils.java
index 3ff68d36065..37c008ca8f9 100644
--- a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/utils/AWS2S3Utils.java
+++ b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/utils/AWS2S3Utils.java
@@ -18,6 +18,7 @@ package org.apache.camel.component.aws2.s3.utils;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
+import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
@@ -95,6 +96,8 @@ public final class AWS2S3Utils {
if (len > 0) {
return len;
}
+ } else if (is instanceof FileInputStream fis) {
+ return fis.getChannel().size();
}
if (!is.markSupported()) {
diff --git a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadMultipartIT.java b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadMultipartIT.java
index d4401155514..cec920bfeed 100644
--- a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadMultipartIT.java
+++ b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadMultipartIT.java
@@ -24,7 +24,6 @@ import java.util.List;
import org.apache.camel.EndpointInject;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
-import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.aws2.s3.AWS2S3Constants;
import org.apache.camel.component.aws2.s3.AWS2S3Operations;
@@ -33,21 +32,19 @@ import org.junit.jupiter.api.Test;
import software.amazon.awssdk.services.s3.model.S3Object;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
public class S3StreamUploadMultipartIT extends Aws2S3Base {
- @EndpointInject
- private ProducerTemplate template;
-
@EndpointInject("mock:result")
private MockEndpoint result;
@Test
- public void sendIn() throws Exception {
+ public void sendFile() throws Exception {
result.expectedMessageCount(10);
for (int i = 0; i < 10; i++) {
- template.send("direct:stream1", new Processor() {
+ Exchange out = template.send("direct:stream1", new Processor() {
@Override
public void process(Exchange exchange) {
@@ -55,6 +52,7 @@ public class S3StreamUploadMultipartIT extends Aws2S3Base {
exchange.getIn().setBody(new File("src/test/resources/empty.bin"));
}
});
+ assertFalse(out.isFailed());
}
MockEndpoint.assertIsSatisfied(context);
diff --git a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadMultipartIT.java b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3UploadFileMultipartIT.java
similarity index 73%
copy from components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadMultipartIT.java
copy to components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3UploadFileMultipartIT.java
index d4401155514..a7845fe8137 100644
--- a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadMultipartIT.java
+++ b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3UploadFileMultipartIT.java
@@ -24,7 +24,6 @@ import java.util.List;
import org.apache.camel.EndpointInject;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
-import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.aws2.s3.AWS2S3Constants;
import org.apache.camel.component.aws2.s3.AWS2S3Operations;
@@ -33,29 +32,26 @@ import org.junit.jupiter.api.Test;
import software.amazon.awssdk.services.s3.model.S3Object;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
-public class S3StreamUploadMultipartIT extends Aws2S3Base {
-
- @EndpointInject
- private ProducerTemplate template;
+public class S3UploadFileMultipartIT extends Aws2S3Base {
@EndpointInject("mock:result")
private MockEndpoint result;
@Test
- public void sendIn() throws Exception {
- result.expectedMessageCount(10);
+ public void sendFile() throws Exception {
+ result.expectedMessageCount(1);
- for (int i = 0; i < 10; i++) {
- template.send("direct:stream1", new Processor() {
+ Exchange out = template.send("direct:stream1", new Processor() {
- @Override
- public void process(Exchange exchange) {
- exchange.getIn().setHeader(AWS2S3Constants.KEY, "empty.bin");
- exchange.getIn().setBody(new File("src/test/resources/empty.bin"));
- }
- });
- }
+ @Override
+ public void process(Exchange exchange) {
+ exchange.getIn().setHeader(AWS2S3Constants.KEY, "empty-big.bin");
+ exchange.getIn().setBody(new File("src/test/resources/empty-big.bin"));
+ }
+ });
+ assertFalse(out.isFailed());
MockEndpoint.assertIsSatisfied(context);
@@ -67,11 +63,10 @@ public class S3StreamUploadMultipartIT extends Aws2S3Base {
}
});
- // file size: 5,242,880 bytes, 10 * (5 chunks of 1,000,000 + remainder of 242,880)
List<S3Object> resp = ex.getMessage().getBody(List.class);
- assertEquals(60, resp.size());
+ assertEquals(1, resp.size());
- assertEquals(10 * Files.size(Paths.get("src/test/resources/empty.bin")),
+ assertEquals(1 * Files.size(Paths.get("src/test/resources/empty-big.bin")),
resp.stream().mapToLong(S3Object::size).sum());
}
@@ -82,7 +77,7 @@ public class S3StreamUploadMultipartIT extends Aws2S3Base {
public void configure() {
String awsEndpoint1
= String.format(
- "aws2-s3://%s?autoCreateBucket=true&streamingUploadMode=true&keyName=fileTest.txt&batchMessageNumber=25&namingStrategy=random",
+ "aws2-s3://%s?autoCreateBucket=true&keyName=fileTest.txt&multiPartUpload=true&partSize=6000000",
name.get());
from("direct:stream1").to(awsEndpoint1).to("mock:result");
diff --git a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadMultipartIT.java b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3UploadInputStreamMultipartIT.java
similarity index 72%
copy from components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadMultipartIT.java
copy to components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3UploadInputStreamMultipartIT.java
index d4401155514..9f0127a6eb8 100644
--- a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadMultipartIT.java
+++ b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3UploadInputStreamMultipartIT.java
@@ -16,7 +16,7 @@
*/
package org.apache.camel.component.aws2.s3.integration;
-import java.io.File;
+import java.io.FileInputStream;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.List;
@@ -24,38 +24,34 @@ import java.util.List;
import org.apache.camel.EndpointInject;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
-import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.aws2.s3.AWS2S3Constants;
import org.apache.camel.component.aws2.s3.AWS2S3Operations;
import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import software.amazon.awssdk.services.s3.model.S3Object;
import static org.junit.jupiter.api.Assertions.assertEquals;
-public class S3StreamUploadMultipartIT extends Aws2S3Base {
-
- @EndpointInject
- private ProducerTemplate template;
+public class S3UploadInputStreamMultipartIT extends Aws2S3Base {
@EndpointInject("mock:result")
private MockEndpoint result;
@Test
- public void sendIn() throws Exception {
- result.expectedMessageCount(10);
+ public void sendInputStream() throws Exception {
+ result.expectedMessageCount(1);
- for (int i = 0; i < 10; i++) {
- template.send("direct:stream1", new Processor() {
+ Exchange out = template.send("direct:stream1", new Processor() {
- @Override
- public void process(Exchange exchange) {
- exchange.getIn().setHeader(AWS2S3Constants.KEY, "empty.bin");
- exchange.getIn().setBody(new File("src/test/resources/empty.bin"));
- }
- });
- }
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ exchange.getIn().setHeader(AWS2S3Constants.KEY, "empty-big.bin");
+ exchange.getIn().setBody(new FileInputStream("src/test/resources/empty-big.bin"));
+ }
+ });
+ Assertions.assertFalse(out.isFailed());
MockEndpoint.assertIsSatisfied(context);
@@ -67,11 +63,10 @@ public class S3StreamUploadMultipartIT extends Aws2S3Base {
}
});
- // file size: 5,242,880 bytes, 10 * (5 chunks of 1,000,000 + remainder of 242,880)
List<S3Object> resp = ex.getMessage().getBody(List.class);
- assertEquals(60, resp.size());
+ assertEquals(1, resp.size());
- assertEquals(10 * Files.size(Paths.get("src/test/resources/empty.bin")),
+ assertEquals(1 * Files.size(Paths.get("src/test/resources/empty-big.bin")),
resp.stream().mapToLong(S3Object::size).sum());
}
@@ -82,7 +77,7 @@ public class S3StreamUploadMultipartIT extends Aws2S3Base {
public void configure() {
String awsEndpoint1
= String.format(
- "aws2-s3://%s?autoCreateBucket=true&streamingUploadMode=true&keyName=fileTest.txt&batchMessageNumber=25&namingStrategy=random",
+ "aws2-s3://%s?autoCreateBucket=true&keyName=fileTest.txt&multiPartUpload=true&partSize=6000000",
name.get());
from("direct:stream1").to(awsEndpoint1).to("mock:result");
diff --git a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadMultipartIT.java b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3UploadInputStreamMultipartNoStreamCacheIT.java
similarity index 72%
copy from components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadMultipartIT.java
copy to components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3UploadInputStreamMultipartNoStreamCacheIT.java
index d4401155514..d4c17a7d6b2 100644
--- a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadMultipartIT.java
+++ b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3UploadInputStreamMultipartNoStreamCacheIT.java
@@ -16,7 +16,7 @@
*/
package org.apache.camel.component.aws2.s3.integration;
-import java.io.File;
+import java.io.FileInputStream;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.List;
@@ -24,38 +24,34 @@ import java.util.List;
import org.apache.camel.EndpointInject;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
-import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.aws2.s3.AWS2S3Constants;
import org.apache.camel.component.aws2.s3.AWS2S3Operations;
import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import software.amazon.awssdk.services.s3.model.S3Object;
import static org.junit.jupiter.api.Assertions.assertEquals;
-public class S3StreamUploadMultipartIT extends Aws2S3Base {
-
- @EndpointInject
- private ProducerTemplate template;
+public class S3UploadInputStreamMultipartNoStreamCacheIT extends Aws2S3Base {
@EndpointInject("mock:result")
private MockEndpoint result;
@Test
- public void sendIn() throws Exception {
- result.expectedMessageCount(10);
+ public void sendInputStream() throws Exception {
+ result.expectedMessageCount(1);
- for (int i = 0; i < 10; i++) {
- template.send("direct:stream1", new Processor() {
+ Exchange out = template.send("direct:stream1", new Processor() {
- @Override
- public void process(Exchange exchange) {
- exchange.getIn().setHeader(AWS2S3Constants.KEY, "empty.bin");
- exchange.getIn().setBody(new File("src/test/resources/empty.bin"));
- }
- });
- }
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ exchange.getIn().setHeader(AWS2S3Constants.KEY, "empty-big.bin");
+ exchange.getIn().setBody(new FileInputStream("src/test/resources/empty-big.bin"));
+ }
+ });
+ Assertions.assertFalse(out.isFailed());
MockEndpoint.assertIsSatisfied(context);
@@ -67,11 +63,10 @@ public class S3StreamUploadMultipartIT extends Aws2S3Base {
}
});
- // file size: 5,242,880 bytes, 10 * (5 chunks of 1,000,000 + remainder of 242,880)
List<S3Object> resp = ex.getMessage().getBody(List.class);
- assertEquals(60, resp.size());
+ assertEquals(1, resp.size());
- assertEquals(10 * Files.size(Paths.get("src/test/resources/empty.bin")),
+ assertEquals(1 * Files.size(Paths.get("src/test/resources/empty-big.bin")),
resp.stream().mapToLong(S3Object::size).sum());
}
@@ -80,9 +75,11 @@ public class S3StreamUploadMultipartIT extends Aws2S3Base {
return new RouteBuilder() {
@Override
public void configure() {
+ context.setStreamCaching(false);
+
String awsEndpoint1
= String.format(
- "aws2-s3://%s?autoCreateBucket=true&streamingUploadMode=true&keyName=fileTest.txt&batchMessageNumber=25&namingStrategy=random",
+ "aws2-s3://%s?autoCreateBucket=true&keyName=fileTest.txt&multiPartUpload=true&partSize=6000000",
name.get());
from("direct:stream1").to(awsEndpoint1).to("mock:result");
diff --git a/components/camel-aws/camel-aws2-s3/src/test/resources/empty-big.bin b/components/camel-aws/camel-aws2-s3/src/test/resources/empty-big.bin
new file mode 100644
index 00000000000..6c5d4031e03
Binary files /dev/null and b/components/camel-aws/camel-aws2-s3/src/test/resources/empty-big.bin differ
diff --git a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/AWS2S3EndpointBuilderFactory.java b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/AWS2S3EndpointBuilderFactory.java
index 8f1fc44bea5..0f2bcccc41e 100644
--- a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/AWS2S3EndpointBuilderFactory.java
+++ b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/AWS2S3EndpointBuilderFactory.java
@@ -2220,9 +2220,10 @@ public interface AWS2S3EndpointBuilderFactory {
}
/**
* Set up the partSize which is used in multipart upload, the default
- * size is 25M. Camel will only do multipart uploads for files that are
- * larger than the part-size thresholds. Files that are smaller will be
- * uploaded in a single operation.
+ * size is 25 MB. The minimum size in AWS is 5 MB. Camel will only do
+ * multipart uploads for files that are larger than the part-size
+ * thresholds. Files that are smaller will be uploaded in a single
+ * operation.
*
* The option is a: <code>long</code> type.
*
@@ -2238,9 +2239,10 @@ public interface AWS2S3EndpointBuilderFactory {
}
/**
* Set up the partSize which is used in multipart upload, the default
- * size is 25M. Camel will only do multipart uploads for files that are
- * larger than the part-size thresholds. Files that are smaller will be
- * uploaded in a single operation.
+ * size is 25 MB. The minimum size in AWS is 5 MB. Camel will only do
+ * multipart uploads for files that are larger than the part-size
+ * thresholds. Files that are smaller will be uploaded in a single
+ * operation.
*
* The option will be converted to a <code>long</code> type.
*