This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new e7c3e7ed2eeb CAMEL-25789: camel-aws2-s3 - streaming upload with the
random or timestamp naming strategy names the first file too (#28137)
e7c3e7ed2eeb is described below
commit e7c3e7ed2eeb7928ffc043e74c800fb8089c5ab5
Author: allthingssecurity <[email protected]>
AuthorDate: Tue Oct 20 18:37:52 2026 +0530
CAMEL-25789: camel-aws2-s3 - streaming upload with the random or timestamp
naming strategy names the first file too (#28137)
The first upload of a stream upload producer was named after keyName alone
also with the random and
timestamp naming strategies; only the later uploads got the UUID or
timestamp. A producer started again
(route or application restart), or a second producer with the same keyName,
wrote its first upload to
the same key and replaced the first file written before. Every upload now
gets the UUID or timestamp.
Follow-ups:
- camel-aws2-s3 - stream upload ITs expect the first file to be named by
the strategy too
- camel-aws2-s3 - stream upload ITs start from an empty bucket, fix the
Javadoc of the restart test
Closes #28137
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/catalog/docs/aws2-s3-streaming.adoc | 4 +-
.../src/main/docs/aws2-s3-streaming.adoc | 4 +-
.../aws2/s3/stream/AWS2S3StreamUploadProducer.java | 31 ++--
.../s3/integration/S3StreamUploadTimeoutIT.java | 21 ++-
.../S3StreamUploadTimestampNamingStrategyIT.java | 52 ++++---
.../S3StreamUploadTimestampTimeoutIT.java | 44 ++++--
.../AWS2S3StreamUploadRandomNamingRestartTest.java | 165 +++++++++++++++++++++
.../aws2/s3/stream/AWSS3NamingStrategyTest.java | 6 +-
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 8 +
9 files changed, 269 insertions(+), 66 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws2-s3-streaming.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws2-s3-streaming.adoc
index 1c9258dc10cb..8bb1352c9381 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws2-s3-streaming.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws2-s3-streaming.adoc
@@ -7,8 +7,10 @@ xref:ROOT:aws2-s3-component.adoc[Back to AWS S3 Component]
With the stream mode enabled, users will be able to upload data to S3 without
knowing ahead of time the dimension of the data, by leveraging multipart upload.
The upload will be completed when the batchSize has been completed or the
batchMessageNumber has been reached.
-There are two possible naming strategies: progressive and random.
+There are three possible naming strategies: progressive, random and timestamp.
With the progressive strategy, each file will have the name composed by
keyName option and a progressive counter, and eventually the file extension (if
any), while with the random strategy a UUID will be added after keyName and
eventually the file extension will be appended.
+With the timestamp strategy, the time the upload started (in milliseconds) is
added after keyName instead of the UUID.
+With the random and timestamp strategies every file gets this part, the first
file of the producer too, so a producer that is started again does not replace
the files written before.
Additionally, streaming upload mode supports timestamp-based file grouping,
which allows messages to be automatically grouped into time windows based on
their timestamps.
diff --git
a/components/camel-aws/camel-aws2-s3/src/main/docs/aws2-s3-streaming.adoc
b/components/camel-aws/camel-aws2-s3/src/main/docs/aws2-s3-streaming.adoc
index 1c9258dc10cb..8bb1352c9381 100644
--- a/components/camel-aws/camel-aws2-s3/src/main/docs/aws2-s3-streaming.adoc
+++ b/components/camel-aws/camel-aws2-s3/src/main/docs/aws2-s3-streaming.adoc
@@ -7,8 +7,10 @@ xref:ROOT:aws2-s3-component.adoc[Back to AWS S3 Component]
With the stream mode enabled, users will be able to upload data to S3 without
knowing ahead of time the dimension of the data, by leveraging multipart upload.
The upload will be completed when the batchSize has been completed or the
batchMessageNumber has been reached.
-There are two possible naming strategies: progressive and random.
+There are three possible naming strategies: progressive, random and timestamp.
With the progressive strategy, each file will have the name composed by
keyName option and a progressive counter, and eventually the file extension (if
any), while with the random strategy a UUID will be added after keyName and
eventually the file extension will be appended.
+With the timestamp strategy, the time the upload started (in milliseconds) is
added after keyName instead of the UUID.
+With the random and timestamp strategies every file gets this part, the first
file of the producer too, so a producer that is started again does not replace
the files written before.
Additionally, streaming upload mode supports timestamp-based file grouping,
which allows messages to be automatically grouped into time windows based on
their timestamps.
diff --git
a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadProducer.java
b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadProducer.java
index 0ed57fd2e99c..2e7cc2c75a16 100644
---
a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadProducer.java
+++
b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadProducer.java
@@ -517,33 +517,20 @@ public class AWS2S3StreamUploadProducer extends
DefaultProducer {
}
break;
case random:
- if (part > 0) {
- if (ObjectHelper.isNotEmpty(ext)) {
- dynamicKeyName = fileName + "-" + id.toString() + ext;
- } else {
- dynamicKeyName = fileName + "-" + id.toString();
- }
+ // every upload, the first one of the producer too: a producer
that is started again, or another
+ // producer with the same keyName, would otherwise replace the
first object with its own
+ if (ObjectHelper.isNotEmpty(ext)) {
+ dynamicKeyName = fileName + "-" + id.toString() + ext;
} else {
- if (ObjectHelper.isNotEmpty(ext)) {
- dynamicKeyName = fileName + ext;
- } else {
- dynamicKeyName = fileName;
- }
+ dynamicKeyName = fileName + "-" + id.toString();
}
break;
case timestamp:
- if (part > 0) {
- if (ObjectHelper.isNotEmpty(ext)) {
- dynamicKeyName = fileName + "-" + timestamp + ext;
- } else {
- dynamicKeyName = fileName + "-" + timestamp;
- }
+ // every upload, as for random
+ if (ObjectHelper.isNotEmpty(ext)) {
+ dynamicKeyName = fileName + "-" + timestamp + ext;
} else {
- if (ObjectHelper.isNotEmpty(ext)) {
- dynamicKeyName = fileName + ext;
- } else {
- dynamicKeyName = fileName;
- }
+ dynamicKeyName = fileName + "-" + timestamp;
}
break;
default:
diff --git
a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadTimeoutIT.java
b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadTimeoutIT.java
index e04f20ba7353..48887aef5f49 100644
---
a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadTimeoutIT.java
+++
b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadTimeoutIT.java
@@ -27,6 +27,7 @@ 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.awaitility.Awaitility;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import software.amazon.awssdk.services.s3.model.S3Object;
@@ -40,11 +41,26 @@ public class S3StreamUploadTimeoutIT extends Aws2S3Base {
@EndpointInject("mock:result")
private MockEndpoint result;
+ @BeforeEach
+ void removeTheObjectsOfAnEarlierRun() {
+ // a surefire rerun uses the same bucket: start from an empty bucket,
so only the objects of this run are counted
+ Exchange ex = template.request("direct:listObjects",
+ exchange ->
exchange.getIn().setHeader(AWS2S3Constants.S3_OPERATION,
AWS2S3Operations.listObjects));
+ List<S3Object> objects = ex.getMessage().getBody(List.class);
+ for (S3Object object : objects) {
+ template.request("direct:listObjects", exchange -> {
+ exchange.getIn().setHeader(AWS2S3Constants.S3_OPERATION,
AWS2S3Operations.deleteObject);
+ exchange.getIn().setHeader(AWS2S3Constants.KEY, object.key());
+ });
+ }
+ }
+
@Test
public void sendIn() throws Exception {
for (int i = 1; i <= 2; i++) {
int count = i * 23;
+ int uploads = i;
result.expectedMessageCount(count);
@@ -54,12 +70,15 @@ public class S3StreamUploadTimeoutIT extends Aws2S3Base {
MockEndpoint.assertIsSatisfied(context, 11, TimeUnit.SECONDS);
+ // each iteration is uploaded by the timeout to its own object
(random naming names the first one too)
Awaitility.await().atMost(11, TimeUnit.SECONDS)
.untilAsserted(() -> {
Exchange ex = template.request("direct:listObjects",
this::process);
List<S3Object> resp =
ex.getMessage().getBody(List.class);
- assertEquals(1, resp.size());
+ assertEquals(uploads, resp.size());
+ // 23 messages of "Andrea\n" per iteration
+ assertEquals(7L * 23 * uploads,
resp.stream().mapToLong(S3Object::size).sum());
});
}
}
diff --git
a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadTimestampNamingStrategyIT.java
b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadTimestampNamingStrategyIT.java
index 914fb20d0581..190fba87fc68 100644
---
a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadTimestampNamingStrategyIT.java
+++
b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadTimestampNamingStrategyIT.java
@@ -26,6 +26,7 @@ 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.BeforeEach;
import org.junit.jupiter.api.Test;
import software.amazon.awssdk.services.s3.model.S3Object;
@@ -40,6 +41,20 @@ public class S3StreamUploadTimestampNamingStrategyIT extends
Aws2S3Base {
@EndpointInject("mock:result")
private MockEndpoint result;
+ @BeforeEach
+ void removeTheObjectsOfAnEarlierRun() {
+ // a surefire rerun uses the same bucket: start from an empty bucket,
so only the objects of this run are counted
+ Exchange ex = template.request("direct:listObjects",
+ exchange ->
exchange.getIn().setHeader(AWS2S3Constants.S3_OPERATION,
AWS2S3Operations.listObjects));
+ List<S3Object> objects = ex.getMessage().getBody(List.class);
+ for (S3Object object : objects) {
+ template.request("direct:listObjects", exchange -> {
+ exchange.getIn().setHeader(AWS2S3Constants.S3_OPERATION,
AWS2S3Operations.deleteObject);
+ exchange.getIn().setHeader(AWS2S3Constants.KEY, object.key());
+ });
+ }
+ }
+
@Test
public void sendInWithTimestampNamingStrategy() throws Exception {
result.expectedMessageCount(1000);
@@ -65,34 +80,23 @@ public class S3StreamUploadTimestampNamingStrategyIT
extends Aws2S3Base {
List<S3Object> resp = ex.getMessage().getBody(List.class);
assertEquals(40, resp.size());
- // Verify that uploaded files use timestamp naming strategy
- // Files should have names like: fileTest.txt, fileTest-<timestamp>.txt
- boolean foundBaseFile = false;
- boolean foundTimestampFile = false;
-
+ // Every upload is named with its timestamp, the first one too:
fileTest-<timestamp>.txt, never fileTest.txt
for (S3Object s3Object : resp) {
String key = s3Object.key();
-
- if ("fileTest.txt".equals(key)) {
- foundBaseFile = true;
- } else if (key.startsWith("fileTest-") && key.endsWith(".txt")) {
- foundTimestampFile = true;
-
- // Extract timestamp from filename and verify it's within
expected range
- String timestampStr = key.substring("fileTest-".length(),
key.length() - ".txt".length());
- try {
- long timestamp = Long.parseLong(timestampStr);
- assertTrue(timestamp >= beforeUpload && timestamp <=
afterUpload,
- "Timestamp " + timestamp + " should be between " +
beforeUpload + " and " + afterUpload);
- } catch (NumberFormatException e) {
- // This shouldn't happen with timestamp naming strategy
- throw new AssertionError("Expected numeric timestamp in
filename: " + key, e);
- }
+ assertTrue(key.startsWith("fileTest-") && key.endsWith(".txt"),
+ "Expected a timestamp-named file
(fileTest-<timestamp>.txt): " + key);
+
+ // Extract timestamp from filename and verify it's within expected
range
+ String timestampStr = key.substring("fileTest-".length(),
key.length() - ".txt".length());
+ try {
+ long timestamp = Long.parseLong(timestampStr);
+ assertTrue(timestamp >= beforeUpload && timestamp <=
afterUpload,
+ "Timestamp " + timestamp + " should be between " +
beforeUpload + " and " + afterUpload);
+ } catch (NumberFormatException e) {
+ // This shouldn't happen with timestamp naming strategy
+ throw new AssertionError("Expected numeric timestamp in
filename: " + key, e);
}
}
-
- assertTrue(foundBaseFile, "Should find base file (fileTest.txt)");
- assertTrue(foundTimestampFile, "Should find timestamp-named files
(fileTest-<timestamp>.txt)");
}
@Override
diff --git
a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadTimestampTimeoutIT.java
b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadTimestampTimeoutIT.java
index 65a62a7e2638..d076fe4d9f4f 100644
---
a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadTimestampTimeoutIT.java
+++
b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadTimestampTimeoutIT.java
@@ -27,6 +27,7 @@ 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.awaitility.Awaitility;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import software.amazon.awssdk.services.s3.model.S3Object;
@@ -41,16 +42,31 @@ public class S3StreamUploadTimestampTimeoutIT extends
Aws2S3Base {
@EndpointInject("mock:result")
private MockEndpoint result;
+ @BeforeEach
+ void removeTheObjectsOfAnEarlierRun() {
+ // a surefire rerun uses the same bucket: start from an empty bucket,
so only the objects of this run are counted
+ Exchange ex = template.request("direct:listObjects",
+ exchange ->
exchange.getIn().setHeader(AWS2S3Constants.S3_OPERATION,
AWS2S3Operations.listObjects));
+ List<S3Object> objects = ex.getMessage().getBody(List.class);
+ for (S3Object object : objects) {
+ template.request("direct:listObjects", exchange -> {
+ exchange.getIn().setHeader(AWS2S3Constants.S3_OPERATION,
AWS2S3Operations.deleteObject);
+ exchange.getIn().setHeader(AWS2S3Constants.KEY, object.key());
+ });
+ }
+ }
+
@Test
public void sendInWithTimestampAndTimeout() throws Exception {
+ long firstUpload = System.currentTimeMillis();
+
for (int i = 1; i <= 2; i++) {
int count = i * 23;
+ int uploads = i;
result.expectedMessageCount(count);
- long beforeUpload = System.currentTimeMillis();
-
for (int j = 0; j < 23; j++) {
template.sendBody("direct:stream1", "TestData\n");
}
@@ -59,33 +75,31 @@ public class S3StreamUploadTimestampTimeoutIT extends
Aws2S3Base {
MockEndpoint.assertIsSatisfied(context, 11, TimeUnit.SECONDS);
+ // each iteration is uploaded by the timeout to its own object,
named with its timestamp (the first one too)
Awaitility.await().atMost(11, TimeUnit.SECONDS)
.untilAsserted(() -> {
Exchange ex = template.request("direct:listObjects",
this::process);
List<S3Object> resp =
ex.getMessage().getBody(List.class);
- assertEquals(1, resp.size());
-
- // Verify the uploaded file uses timestamp naming
strategy
- S3Object s3Object = resp.get(0);
- String key = s3Object.key();
+ assertEquals(uploads, resp.size());
- // The file should either be the base name or have a
timestamp suffix
- if ("fileTest.txt".equals(key)) {
- // This is fine - it's the base file
- } else if (key.startsWith("fileTest-") &&
key.endsWith(".txt")) {
- // Extract and validate timestamp
+ long size = 0;
+ for (S3Object s3Object : resp) {
+ String key = s3Object.key();
+ assertTrue(key.startsWith("fileTest-") &&
key.endsWith(".txt"),
+ "Expected a timestamp-named file
(fileTest-<timestamp>.txt): " + key);
String timestampStr =
key.substring("fileTest-".length(), key.length() - ".txt".length());
try {
long timestamp = Long.parseLong(timestampStr);
- assertTrue(timestamp >= beforeUpload &&
timestamp <= afterUpload + 11000, // Allow extra time for timeout
+ assertTrue(timestamp >= firstUpload &&
timestamp <= afterUpload + 11000, // Allow extra time for timeout
"Timestamp " + timestamp + " should be
within expected range");
} catch (NumberFormatException e) {
throw new AssertionError("Expected numeric
timestamp in filename: " + key, e);
}
- } else {
- throw new AssertionError("Unexpected filename
format: " + key);
+ size += s3Object.size();
}
+ // 23 messages of "TestData\n" per iteration
+ assertEquals(9L * 23 * uploads, size);
});
}
}
diff --git
a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadRandomNamingRestartTest.java
b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadRandomNamingRestartTest.java
new file mode 100644
index 000000000000..2486053b5ca4
--- /dev/null
+++
b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadRandomNamingRestartTest.java
@@ -0,0 +1,165 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.s3.stream;
+
+import java.io.ByteArrayOutputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.component.aws2.s3.AWS2S3Configuration;
+import org.apache.camel.component.aws2.s3.AWS2S3Endpoint;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.support.DefaultExchange;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import software.amazon.awssdk.core.sync.RequestBody;
+import software.amazon.awssdk.http.SdkHttpResponse;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest;
+import
software.amazon.awssdk.services.s3.model.CompleteMultipartUploadResponse;
+import software.amazon.awssdk.services.s3.model.CreateMultipartUploadRequest;
+import software.amazon.awssdk.services.s3.model.CreateMultipartUploadResponse;
+import software.amazon.awssdk.services.s3.model.UploadPartRequest;
+import software.amazon.awssdk.services.s3.model.UploadPartResponse;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
+
+/**
+ * With namingStrategy=random every upload must get the UUID, the first one of
a producer too: a producer that is
+ * started again (or a second producer with the same keyName) must not replace
the objects written before. Uses a mocked
+ * S3 client that keeps the completed objects by key, like a bucket.
+ */
+public class AWS2S3StreamUploadRandomNamingRestartTest {
+
+ @Mock
+ private AWS2S3Endpoint endpoint;
+
+ @Mock
+ private AWS2S3Configuration configuration;
+
+ @Mock
+ private S3Client s3Client;
+
+ // the bucket: completed objects by key
+ private final Map<String, String> bucket = new ConcurrentHashMap<>();
+ private final Map<String, String> keysByUploadId = new
ConcurrentHashMap<>();
+ private final Map<String, ByteArrayOutputStream> partsByUploadId = new
ConcurrentHashMap<>();
+ private final AtomicInteger uploadIds = new AtomicInteger();
+
+ private DefaultCamelContext camelContext;
+
+ @BeforeEach
+ public void setUp() {
+ MockitoAnnotations.openMocks(this);
+ camelContext = new DefaultCamelContext();
+
+ when(endpoint.getConfiguration()).thenReturn(configuration);
+ when(endpoint.getCamelContext()).thenReturn(camelContext);
+ when(endpoint.getS3Client()).thenReturn(s3Client);
+
+ when(configuration.getBucketName()).thenReturn("test-bucket");
+ when(configuration.getKeyName()).thenReturn("camel.txt");
+ when(configuration.isMultiPartUpload()).thenReturn(false);
+ when(configuration.getPartSize()).thenReturn(25L * 1024 * 1024);
+
when(configuration.getNamingStrategy()).thenReturn(AWSS3NamingStrategyEnum.random);
+
when(configuration.getRestartingPolicy()).thenReturn(AWSS3RestartingPolicyEnum.override);
+ // the defaults of the endpoint
+ when(configuration.getBatchMessageNumber()).thenReturn(10);
+ when(configuration.getBatchSize()).thenReturn(1000000);
+ when(configuration.getBufferSize()).thenReturn(1000000);
+
+ SdkHttpResponse httpResponse =
SdkHttpResponse.builder().statusCode(200).build();
+
+
when(s3Client.createMultipartUpload(any(CreateMultipartUploadRequest.class))).thenAnswer(invocation
-> {
+ CreateMultipartUploadRequest request = invocation.getArgument(0);
+ String uploadId = "upload-" + uploadIds.incrementAndGet();
+ keysByUploadId.put(uploadId, request.key());
+ partsByUploadId.put(uploadId, new ByteArrayOutputStream());
+ return
CreateMultipartUploadResponse.builder().uploadId(uploadId).sdkHttpResponse(httpResponse).build();
+ });
+
+ when(s3Client.uploadPart(any(UploadPartRequest.class),
any(RequestBody.class))).thenAnswer(invocation -> {
+ UploadPartRequest request = invocation.getArgument(0);
+ RequestBody body = invocation.getArgument(1);
+
partsByUploadId.get(request.uploadId()).write(body.contentStreamProvider().newStream().readAllBytes());
+ return
UploadPartResponse.builder().eTag("etag").sdkHttpResponse(httpResponse).build();
+ });
+
+
when(s3Client.completeMultipartUpload(any(CompleteMultipartUploadRequest.class))).thenAnswer(invocation
-> {
+ CompleteMultipartUploadRequest request = invocation.getArgument(0);
+ // like S3: the completed object replaces any object of the same
key
+ bucket.put(keysByUploadId.get(request.uploadId()),
+
partsByUploadId.get(request.uploadId()).toString(StandardCharsets.UTF_8));
+ return
CompleteMultipartUploadResponse.builder().eTag("final-etag").sdkHttpResponse(httpResponse).build();
+ });
+ }
+
+ @Test
+ public void aProducerStartedAgainDoesNotReplaceTheObjectsOfTheFirstOne()
throws Exception {
+ // the first run of the route: 15 messages, one object of 10 messages
and 5 messages flushed at stop
+ AWS2S3StreamUploadProducer first = new
AWS2S3StreamUploadProducer(endpoint);
+ first.start();
+ send(first, 0, 15);
+ first.stop();
+ assertEquals(2, bucket.size(), "the first run wrote two objects: " +
bucket.keySet());
+
+ // the route is started again: a new producer, same endpoint options
+ AWS2S3StreamUploadProducer second = new
AWS2S3StreamUploadProducer(endpoint);
+ second.start();
+ send(second, 15, 15);
+ second.stop();
+
+ assertEquals(4, uploadIds.get(), "four uploads were completed");
+ assertEquals(4, bucket.size(), "every upload must be its own object,
but the bucket has " + bucket.keySet());
+ List<String> stored = new ArrayList<>();
+ for (String content : bucket.values()) {
+ for (String line : content.split("\n")) {
+ stored.add(line);
+ }
+ }
+ assertEquals(30, stored.size(), "every message sent must be in the
bucket: " + bucket);
+ }
+
+ @Test
+ public void everyObjectHasTheKeyNameAndARandomPart() throws Exception {
+ AWS2S3StreamUploadProducer producer = new
AWS2S3StreamUploadProducer(endpoint);
+ send(producer, 0, 20);
+
+ assertEquals(2, bucket.size(), bucket.keySet().toString());
+ for (String key : bucket.keySet()) {
+ assertTrue(key.matches("camel-[0-9a-f-]{36}\\.txt"), key);
+ }
+ }
+
+ private void send(AWS2S3StreamUploadProducer producer, int from, int
count) throws Exception {
+ for (int i = from; i < from + count; i++) {
+ Exchange exchange = new DefaultExchange(camelContext);
+ exchange.getIn().setBody("message-" + i + "\n");
+ producer.process(exchange);
+ }
+ }
+}
diff --git
a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/stream/AWSS3NamingStrategyTest.java
b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/stream/AWSS3NamingStrategyTest.java
index b28589411cec..1b602b170a61 100644
---
a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/stream/AWSS3NamingStrategyTest.java
+++
b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/stream/AWSS3NamingStrategyTest.java
@@ -61,8 +61,9 @@ public class AWSS3NamingStrategyTest extends CamelTestSupport
{
UUID testUuid =
UUID.fromString("123e4567-e89b-12d3-a456-426614174000");
+ // the first upload of a producer gets the random part too
String result = (String) method.invoke(producer, "testFile",
AWSS3NamingStrategyEnum.random, ".txt", 0, testUuid, 0L);
- assertEquals("testFile.txt", result);
+ assertEquals("testFile-123e4567-e89b-12d3-a456-426614174000.txt",
result);
result = (String) method.invoke(producer, "testFile",
AWSS3NamingStrategyEnum.random, ".txt", 1, testUuid, 0L);
assertEquals("testFile-123e4567-e89b-12d3-a456-426614174000.txt",
result);
@@ -82,9 +83,10 @@ public class AWSS3NamingStrategyTest extends
CamelTestSupport {
long testTimestamp = 1632468273000L; // September 24, 2021 1:04:33 PM
GMT
+ // the first upload of a producer gets the timestamp too
String result = (String) method.invoke(producer, "testFile",
AWSS3NamingStrategyEnum.timestamp, ".txt", 0, null,
testTimestamp);
- assertEquals("testFile.txt", result);
+ assertEquals("testFile-1632468273000.txt", result);
result = (String) method.invoke(producer, "testFile",
AWSS3NamingStrategyEnum.timestamp, ".txt", 1, null,
testTimestamp);
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 6872b0344856..951f78c9f1e8 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -1323,6 +1323,14 @@ handshake; an IP address must be listed as a subject
alternative name of the cer
host name (such as `mllp://8888`) cannot pass the check. Use a certificate
that matches the host of the endpoint, or set
the new `hostnameVerification` option to `false` to keep the previous
behaviour.
+=== camel-aws2-s3 - streaming upload with the random or timestamp naming
strategy names the first file too
+
+With `streamingUploadMode=true` and `namingStrategy=random` (or `timestamp`),
the first file uploaded by a producer was
+named after `keyName` alone (for example `camel.txt`), and only the later
files got a UUID (or a timestamp) after
+`keyName`. A producer that was started again (a restart of the route or of the
application), or another producer with
+the same `keyName`, wrote its first file to the same key and replaced the
first file written before. Every file now
+gets the UUID (or the timestamp), the first one too: `camel-<uuid>.txt`. The
progressive naming strategy is unchanged.
+
=== camel-atom - sortEntries sorts by the updated date and routes the oldest
entry first
With `sortEntries=true` the `atom` consumer sorted the entries of a feed by
their published date and routed them from
