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 93b7f0643186 CAMEL-25785: camel-aws2-s3 - a timestamp window uploaded
more than once is stored in more than one object (#28136)
93b7f0643186 is described below
commit 93b7f064318628d7290d33bc615b22dea57acbe6
Author: allthingssecurity <[email protected]>
AuthorDate: Wed Oct 21 00:30:49 2026 +0530
CAMEL-25785: camel-aws2-s3 - a timestamp window uploaded more than once is
stored in more than one object (#28136)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/catalog/docs/aws2-s3-streaming.adoc | 18 +-
.../src/main/docs/aws2-s3-streaming.adoc | 18 +-
.../aws2/s3/stream/AWS2S3StreamUploadProducer.java | 101 +++++++-
...S3StreamUploadTimestampGroupingEdgeCasesIT.java | 20 +-
.../S3StreamUploadTimestampGroupingIT.java | 40 +--
.../AWS2S3StreamUploadProducerTimestampTest.java | 22 +-
...S3StreamUploadTimestampWindowOverwriteTest.java | 282 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 15 ++
8 files changed, 484 insertions(+), 32 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 8bb1352c9381..1fa950c8cc9d 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
@@ -98,7 +98,23 @@ Messages are grouped based on their timestamps extracted
from the specified head
- Date: Java Date object
- String: String representation of Unix timestamp in milliseconds
-Files are automatically named using a timestamp-based pattern that includes
the time window information.
+Files are automatically named using a timestamp-based pattern that includes
the time window information, for example
+`grouped-messages_20240101_0800_0800-0805.txt` for the window from 08:00 to
08:05 (UTC). The upload of a window is
+completed when it reaches `batchMessageNumber` messages or `batchSize` bytes,
or when `streamingUploadTimeout` runs. When
+more messages of that window come after it, they are uploaded to a new object
of the window, named with the number of
+the uploads of the window before it:
`grouped-messages_20240101_0800_0800-0805-1.txt`, then `-2`, and so on. The
first
+time a producer uploads a window, it lists the objects of the window in the
bucket (one `ListObjectsV2` request with the
+window's key as prefix), so after a restart it continues after the highest
number already stored instead of replacing
+that object. This needs the `s3:ListBucket` permission: without it the
producer logs a WARN and numbers the uploads of
+the window from the first one, so an upload after a restart can replace an
object of the window again.
+
+These keys do not sort in upload order: `...0805-1.txt` sorts before
`...0805.txt` (`-` sorts before `.`), and `-10`
+sorts before `-2`. To read the uploads of a window in order, sort them by the
number after the time range (none for
+the first upload).
+
+Producers that run at the same time (for example two routes or two
applications) with the same `keyName` and
+timestamp grouping still choose the same names for the same window, so one can
replace the object of the other. Give
+each of them its own `keyName`.
==== Example Configuration
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 8bb1352c9381..1fa950c8cc9d 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
@@ -98,7 +98,23 @@ Messages are grouped based on their timestamps extracted
from the specified head
- Date: Java Date object
- String: String representation of Unix timestamp in milliseconds
-Files are automatically named using a timestamp-based pattern that includes
the time window information.
+Files are automatically named using a timestamp-based pattern that includes
the time window information, for example
+`grouped-messages_20240101_0800_0800-0805.txt` for the window from 08:00 to
08:05 (UTC). The upload of a window is
+completed when it reaches `batchMessageNumber` messages or `batchSize` bytes,
or when `streamingUploadTimeout` runs. When
+more messages of that window come after it, they are uploaded to a new object
of the window, named with the number of
+the uploads of the window before it:
`grouped-messages_20240101_0800_0800-0805-1.txt`, then `-2`, and so on. The
first
+time a producer uploads a window, it lists the objects of the window in the
bucket (one `ListObjectsV2` request with the
+window's key as prefix), so after a restart it continues after the highest
number already stored instead of replacing
+that object. This needs the `s3:ListBucket` permission: without it the
producer logs a WARN and numbers the uploads of
+the window from the first one, so an upload after a restart can replace an
object of the window again.
+
+These keys do not sort in upload order: `...0805-1.txt` sorts before
`...0805.txt` (`-` sorts before `.`), and `-10`
+sorts before `-2`. To read the uploads of a window in order, sort them by the
number after the time range (none for
+the first upload).
+
+Producers that run at the same time (for example two routes or two
applications) with the same `keyName` and
+timestamp grouping still choose the same names for the same window, so one can
replace the object of the other. Give
+each of them its own `keyName`.
==== Example Configuration
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 2e7cc2c75a16..660062fd6a9b 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
@@ -22,6 +22,7 @@ import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Date;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.TimeZone;
@@ -57,6 +58,7 @@ import software.amazon.awssdk.services.s3.model.CompletedPart;
import software.amazon.awssdk.services.s3.model.CreateMultipartUploadRequest;
import software.amazon.awssdk.services.s3.model.CreateMultipartUploadResponse;
import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
+import software.amazon.awssdk.services.s3.model.ListObjectsV2Response;
import software.amazon.awssdk.services.s3.model.ObjectCannedACL;
import software.amazon.awssdk.services.s3.model.S3Object;
import software.amazon.awssdk.services.s3.model.UploadPartRequest;
@@ -71,9 +73,22 @@ public class AWS2S3StreamUploadProducer extends
DefaultProducer {
private static final Logger LOG =
LoggerFactory.getLogger(AWS2S3StreamUploadProducer.class);
private static final String TIMEOUT_CHECKER_EXECUTOR_NAME =
"S3_Streaming_Upload_Timeout_Checker";
+ // the number of timestamp windows whose completed uploads are counted
+ private static final int MAX_COUNTED_WINDOWS = 10000;
private AtomicInteger part = new AtomicInteger();
private UploadState uploadAggregate = null;
private final Map<Long, UploadState> timestampBasedUploads = new
ConcurrentHashMap<>();
+ // The uploads completed for each timestamp window, guarded by the lock. A
window is uploaded again when it gets
+ // more messages after its upload was completed (batchMessageNumber,
batchSize or streamingUploadTimeout); the new
+ // object gets a name of its own, as an object with the name of the first
one would replace it. The count of a
+ // window starts from the objects of the window already in the bucket
(uploaded before a restart). Windows come in
+ // time order, so only the windows seen last are kept, and a long-running
producer does not keep every window.
+ private final Map<Long, Integer> completedUploadsByWindow = new
LinkedHashMap<>() {
+ @Override
+ protected boolean removeEldestEntry(Map.Entry<Long, Integer> eldest) {
+ return size() > MAX_COUNTED_WINDOWS;
+ }
+ };
private final Lock lock = new ReentrantLock();
private transient String s3ProducerToString;
private ScheduledExecutorService timeoutCheckerExecutorService;
@@ -109,10 +124,12 @@ public class AWS2S3StreamUploadProducer extends
DefaultProducer {
}
// Complete any pending timestamp-based uploads
- for (UploadState state : timestampBasedUploads.values()) {
+ for (Map.Entry<Long, UploadState> entry :
timestampBasedUploads.entrySet()) {
+ UploadState state = entry.getValue();
if (ObjectHelper.isNotEmpty(state) && state.buffer.size() > 0)
{
uploadPart(state);
completeUpload(state);
+ completedUploadsByWindow.merge(entry.getKey(), 1,
Integer::sum);
}
}
timestampBasedUploads.clear();
@@ -155,6 +172,7 @@ public class AWS2S3StreamUploadProducer extends
DefaultProducer {
try {
uploadPart(state);
completeUpload(state);
+ completedUploadsByWindow.merge(entry.getKey(),
1, Integer::sum);
} catch (Exception e) {
LOG.warn("Error during timeout-triggered
upload flush for {}", state.dynamicKeyName, e);
}
@@ -373,7 +391,8 @@ public class AWS2S3StreamUploadProducer extends
DefaultProducer {
final String keyName = getConfiguration().getKeyName();
final String fileName =
AWS2S3Utils.determineFileName(keyName);
final String extension =
AWS2S3Utils.determineFileExtension(keyName);
- newState.dynamicKeyName =
generateTimestampBasedFileName(fileName, extension, timestampWindow);
+ newState.dynamicKeyName =
generateTimestampBasedFileName(fileName, extension, timestampWindow,
+ completedUploads(timestampWindow, fileName,
extension));
// Initialize multipart upload
CreateMultipartUploadRequest.Builder
createMultipartUploadRequest
@@ -423,6 +442,7 @@ public class AWS2S3StreamUploadProducer extends
DefaultProducer {
uploadPart(state);
CompleteMultipartUploadResponse uploadResult =
completeUpload(state);
timestampBasedUploads.remove(timestampWindow);
+ completedUploadsByWindow.merge(timestampWindow, 1,
Integer::sum);
if (getConfiguration().isMultiPartUpload()) {
maxRead =
Math.toIntExact(getConfiguration().getPartSize());
}
@@ -544,7 +564,79 @@ public class AWS2S3StreamUploadProducer extends
DefaultProducer {
return (timestamp / windowSize) * windowSize;
}
- private String generateTimestampBasedFileName(String baseFileName, String
extension, long timestampWindow) {
+ /**
+ * The number of the completed uploads of a timestamp window. The first
time this producer sees the window, it is
+ * the number of the objects of the window in the bucket, so an upload
after a restart does not replace them.
+ */
+ private int completedUploads(long timestampWindow, String fileName, String
extension) {
+ Integer completed = completedUploadsByWindow.get(timestampWindow);
+ if (completed == null) {
+ completed =
uploadsInBucket(generateTimestampBasedFileName(fileName, null, timestampWindow,
0), extension);
+ completedUploadsByWindow.put(timestampWindow, completed);
+ }
+ return completed;
+ }
+
+ /**
+ * The number of the uploads of a timestamp window stored in the bucket:
an object named like the first upload of
+ * the window counts 1, an object of the later upload number N counts N +
1, and the highest wins. When the bucket
+ * cannot be listed (for example without the s3:ListBucket permission),
the count starts from 0.
+ */
+ private int uploadsInBucket(String windowKey, String extension) {
+ String ext = extension != null ? extension : "";
+ int uploads = 0;
+ try {
+ String continuationToken = null;
+ do {
+ ListObjectsV2Request.Builder request
+ =
ListObjectsV2Request.builder().bucket(getConfiguration().getBucketName()).prefix(windowKey);
+ if (continuationToken != null) {
+ request.continuationToken(continuationToken);
+ }
+ ListObjectsV2Response response =
getEndpoint().getS3Client().listObjectsV2(request.build());
+ for (S3Object object : response.contents()) {
+ uploads = Math.max(uploads, uploadNumber(object.key(),
windowKey, ext));
+ }
+ continuationToken =
Boolean.TRUE.equals(response.isTruncated()) ? response.nextContinuationToken()
: null;
+ } while (continuationToken != null);
+ } catch (Exception e) {
+ LOG.warn("Cannot list the objects of {} in bucket {}, so its
uploads are numbered from the first one and may"
+ + " replace objects uploaded before this producer
started: {}",
+ windowKey + ext, getConfiguration().getBucketName(),
e.getMessage());
+ LOG.debug("Listing the objects of {} failed", windowKey + ext, e);
+ return 0;
+ }
+ return uploads;
+ }
+
+ /**
+ * 1 for the object of the first upload of the window, N + 1 for the
object of the later upload number N, 0 for an
+ * object that is not an upload of the window.
+ */
+ private static int uploadNumber(String key, String windowKey, String
extension) {
+ if (key == null || key.length() < windowKey.length() +
extension.length() || !key.startsWith(windowKey)
+ || !key.endsWith(extension)) {
+ return 0;
+ }
+ String suffix = key.substring(windowKey.length(), key.length() -
extension.length());
+ if (suffix.isEmpty()) {
+ return 1;
+ }
+ // at most 9 digits, so the number fits an int
+ if (suffix.length() > 1 && suffix.length() <= 10 && suffix.charAt(0)
== '-'
+ && suffix.substring(1).chars().allMatch(c -> c >= '0' && c <=
'9')) {
+ return Integer.parseInt(suffix.substring(1)) + 1;
+ }
+ return 0;
+ }
+
+ /**
+ * The name of an upload of a timestamp window: the first upload of the
window has the name of the window, a later
+ * one (the window got more messages after its upload was completed) gets
the number of the uploads before it as a
+ * suffix, so it does not replace them.
+ */
+ private String generateTimestampBasedFileName(
+ String baseFileName, String extension, long timestampWindow, int
completedUploads) {
Date windowStart = new Date(timestampWindow);
Date windowEnd = new Date(timestampWindow +
getConfiguration().getTimestampWindowSizeMillis());
@@ -564,6 +656,9 @@ public class AWS2S3StreamUploadProducer extends
DefaultProducer {
SimpleDateFormat timeFormat = new SimpleDateFormat(timePattern);
timeFormat.setTimeZone(TimeZone.getTimeZone("UTC"));
String timeRange = timeFormat.format(windowStart) + "-" +
timeFormat.format(windowEnd);
+ if (completedUploads > 0) {
+ timeRange = timeRange + "-" + completedUploads;
+ }
if (extension != null && !extension.isEmpty()) {
return baseFileName + "_" + sdf.format(windowStart) + "_" +
timeRange + extension;
diff --git
a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadTimestampGroupingEdgeCasesIT.java
b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadTimestampGroupingEdgeCasesIT.java
index bf26f73164f4..d75cfc9cacfa 100644
---
a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadTimestampGroupingEdgeCasesIT.java
+++
b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadTimestampGroupingEdgeCasesIT.java
@@ -27,6 +27,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;
@@ -41,6 +42,21 @@ public class S3StreamUploadTimestampGroupingEdgeCasesIT
extends Aws2S3Base {
@EndpointInject("mock:result")
private MockEndpoint result;
+ @BeforeEach
+ void removeTheObjectsOfAnEarlierRun() {
+ // a surefire rerun uses the same bucket, and a new producer numbers
the uploads of a window after the objects of
+ // that window already in the bucket: start from an empty bucket, so
only the objects of this test are listed
+ 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 testTimestampGroupingWithInvalidStringTimestamp() throws
Exception {
result.reset();
@@ -128,7 +144,9 @@ public class S3StreamUploadTimestampGroupingEdgeCasesIT
extends Aws2S3Base {
});
List<S3Object> resp = ex.getMessage().getBody(List.class);
- assertEquals(1, resp.size(), "Should create only 1 file for large time
window");
+ // one window; batchMessageNumber=5 uploads it twice, and the second
upload gets its own object
+ assertEquals(List.of("largeWindowTest_20240101_0800_0800-0900-1.txt",
"largeWindowTest_20240101_0800_0800-0900.txt"),
+ resp.stream().map(S3Object::key).sorted().toList());
}
@Test
diff --git
a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadTimestampGroupingIT.java
b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadTimestampGroupingIT.java
index 548b00730545..a3320be01642 100644
---
a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadTimestampGroupingIT.java
+++
b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadTimestampGroupingIT.java
@@ -27,6 +27,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;
@@ -41,6 +42,21 @@ public class S3StreamUploadTimestampGroupingIT extends
Aws2S3Base {
@EndpointInject("mock:result")
private MockEndpoint result;
+ @BeforeEach
+ void removeTheObjectsOfAnEarlierRun() {
+ // a surefire rerun uses the same bucket, and a new producer numbers
the uploads of a window after the objects of
+ // that window already in the bucket: start from an empty bucket, so
only the objects of this test are listed
+ 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 testTimestampGroupingBasic() throws Exception {
result.reset();
@@ -75,23 +91,13 @@ public class S3StreamUploadTimestampGroupingIT extends
Aws2S3Base {
});
List<S3Object> resp = ex.getMessage().getBody(List.class);
- assertEquals(2, resp.size(), "Should create 2 files for 2 different
timestamp windows");
-
- // Verify files have timestamp-based naming
- boolean foundWindow1File = false;
- boolean foundWindow2File = false;
-
- for (S3Object s3Object : resp) {
- String key = s3Object.key();
- if (key.contains("20240101_0800_0800-0805")) {
- foundWindow1File = true;
- } else if (key.contains("20240101_0805_0805-0810")) {
- foundWindow2File = true;
- }
- }
-
- assertTrue(foundWindow1File, "Should find file for first timestamp
window");
- assertTrue(foundWindow2File, "Should find file for second timestamp
window");
+ // batchMessageNumber=5: each window is uploaded twice, and the second
upload of a window gets its own object
+ assertEquals(List.of(
+ "basicTest_20240101_0800_0800-0805-1.txt",
+ "basicTest_20240101_0800_0800-0805.txt",
+ "basicTest_20240101_0805_0805-0810-1.txt",
+ "basicTest_20240101_0805_0805-0810.txt"),
+ resp.stream().map(S3Object::key).sorted().toList());
}
@Test
diff --git
a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadProducerTimestampTest.java
b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadProducerTimestampTest.java
index 4089a9669d8a..a1ef35f116fa 100644
---
a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadProducerTimestampTest.java
+++
b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadProducerTimestampTest.java
@@ -85,28 +85,32 @@ public class AWS2S3StreamUploadProducerTimestampTest {
@Test
public void testGenerateTimestampBasedFileName() throws Exception {
Method method = AWS2S3StreamUploadProducer.class.getDeclaredMethod(
- "generateTimestampBasedFileName", String.class, String.class,
long.class);
+ "generateTimestampBasedFileName", String.class, String.class,
long.class, int.class);
method.setAccessible(true);
when(configuration.getTimestampWindowSizeMillis()).thenReturn(300000L); // 5
minutes
// Test with extension
long timestampWindow = 1704096000000L; // 2024-01-01 08:00:00 UTC
- String fileName = (String) method.invoke(producer, "testfile", ".txt",
timestampWindow);
+ String fileName = (String) method.invoke(producer, "testfile", ".txt",
timestampWindow, 0);
assertEquals("testfile_20240101_0800_0800-0805.txt", fileName);
// Test without extension
- fileName = (String) method.invoke(producer, "testfile", null,
timestampWindow);
+ fileName = (String) method.invoke(producer, "testfile", null,
timestampWindow, 0);
assertEquals("testfile_20240101_0800_0800-0805", fileName);
// Test with empty extension
- fileName = (String) method.invoke(producer, "testfile", "",
timestampWindow);
+ fileName = (String) method.invoke(producer, "testfile", "",
timestampWindow, 0);
assertEquals("testfile_20240101_0800_0800-0805", fileName);
// Test different time window
long timestampWindow2 = 1704096300000L; // 2024-01-01 08:05:00 UTC
- fileName = (String) method.invoke(producer, "myfile", ".log",
timestampWindow2);
+ fileName = (String) method.invoke(producer, "myfile", ".log",
timestampWindow2, 0);
assertEquals("myfile_20240101_0805_0805-0810.log", fileName);
+
+ // a later upload of a window that was already uploaded
+ fileName = (String) method.invoke(producer, "myfile", ".log",
timestampWindow2, 2);
+ assertEquals("myfile_20240101_0805_0805-0810-2.log", fileName);
}
@Test
@@ -259,24 +263,24 @@ public class AWS2S3StreamUploadProducerTimestampTest {
@Test
public void testGenerateTimestampBasedFileNameWithDifferentWindowSizes()
throws Exception {
Method method = AWS2S3StreamUploadProducer.class.getDeclaredMethod(
- "generateTimestampBasedFileName", String.class, String.class,
long.class);
+ "generateTimestampBasedFileName", String.class, String.class,
long.class, int.class);
method.setAccessible(true);
long timestampWindow = 1704096000000L; // 2024-01-01 08:00:00 UTC
// Test 1-minute window
when(configuration.getTimestampWindowSizeMillis()).thenReturn(60000L);
- String fileName1min = (String) method.invoke(producer, "test", ".txt",
timestampWindow);
+ String fileName1min = (String) method.invoke(producer, "test", ".txt",
timestampWindow, 0);
assertEquals("test_20240101_0800_0800-0801.txt", fileName1min);
// Test 1-hour window
when(configuration.getTimestampWindowSizeMillis()).thenReturn(3600000L);
- String fileName1hour = (String) method.invoke(producer, "test",
".txt", timestampWindow);
+ String fileName1hour = (String) method.invoke(producer, "test",
".txt", timestampWindow, 0);
assertEquals("test_20240101_0800_0800-0900.txt", fileName1hour);
// Test 30-minute window
when(configuration.getTimestampWindowSizeMillis()).thenReturn(1800000L);
- String fileName30min = (String) method.invoke(producer, "test",
".txt", timestampWindow);
+ String fileName30min = (String) method.invoke(producer, "test",
".txt", timestampWindow, 0);
assertEquals("test_20240101_0800_0800-0830.txt", fileName30min);
}
}
diff --git
a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadTimestampWindowOverwriteTest.java
b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadTimestampWindowOverwriteTest.java
new file mode 100644
index 000000000000..87f304d5b47c
--- /dev/null
+++
b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadTimestampWindowOverwriteTest.java
@@ -0,0 +1,282 @@
+/*
+ * 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.lang.reflect.Constructor;
+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.ListObjectsV2Request;
+import software.amazon.awssdk.services.s3.model.ListObjectsV2Response;
+import software.amazon.awssdk.services.s3.model.S3Exception;
+import software.amazon.awssdk.services.s3.model.S3Object;
+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 timestamp grouping, a window whose upload is completed (by
batchMessageNumber, batchSize or the streaming upload
+ * timeout) and then gets more messages is uploaded again. The new object must
not replace the first one: a bucket keeps
+ * one object per key. Uses a mocked S3 client that keeps the completed
objects by key, like a bucket.
+ */
+public class AWS2S3StreamUploadTimestampWindowOverwriteTest {
+
+ private static final long WINDOW_START = 1704096000000L; // 2024-01-01
08:00:00 UTC
+
+ @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 AWS2S3StreamUploadProducer producer;
+ 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("grouped-messages.txt");
+ when(configuration.isMultiPartUpload()).thenReturn(false);
+ when(configuration.getPartSize()).thenReturn(25L * 1024 * 1024);
+
when(configuration.getNamingStrategy()).thenReturn(AWSS3NamingStrategyEnum.progressive);
+
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);
+ when(configuration.isTimestampGroupingEnabled()).thenReturn(true);
+ when(configuration.getTimestampWindowSizeMillis()).thenReturn(300000L);
+
when(configuration.getTimestampHeaderName()).thenReturn(Exchange.MESSAGE_TIMESTAMP);
+
+ 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();
+ });
+
+ // like S3: the keys with the prefix in lexical order, at most two per
page
+
when(s3Client.listObjectsV2(any(ListObjectsV2Request.class))).thenAnswer(invocation
-> {
+ ListObjectsV2Request request = invocation.getArgument(0);
+ List<String> keys = bucket.keySet().stream().filter(key ->
key.startsWith(request.prefix())).sorted()
+ .toList();
+ int from = request.continuationToken() != null ?
Integer.parseInt(request.continuationToken()) : 0;
+ int to = Math.min(from + 2, keys.size());
+ ListObjectsV2Response.Builder response =
ListObjectsV2Response.builder()
+ .contents(keys.subList(from, to).stream().map(key ->
S3Object.builder().key(key).build()).toList())
+ .isTruncated(to < keys.size());
+ if (to < keys.size()) {
+ response.nextContinuationToken(String.valueOf(to));
+ }
+ return response.build();
+ });
+
+ producer = new AWS2S3StreamUploadProducer(endpoint);
+ }
+
+ @Test
+ public void
everyMessageOfAWindowIsStoredWhenTheWindowIsUploadedMoreThanOnce() throws
Exception {
+ // the example of the documentation: a message every 10 seconds, in 5
minute windows, so 30 messages per
+ // window; with batchMessageNumber=10 the window is uploaded after the
10th, the 20th and the 30th message
+ List<String> sent = new ArrayList<>();
+ for (int i = 0; i < 30; i++) {
+ String body = "message-" + i + "\n";
+ sent.add(body);
+ Exchange exchange = new DefaultExchange(camelContext);
+ exchange.getIn().setBody(body);
+ exchange.getIn().setHeader(Exchange.MESSAGE_TIMESTAMP,
WINDOW_START + i * 10000L);
+ producer.process(exchange);
+ }
+
+ assertEquals(3, uploadIds.get(), "the window is uploaded three times");
+ assertEquals(3, bucket.size(),
+ "each upload of the window must be its own object, but the
bucket has " + bucket.keySet());
+ StringBuilder stored = new StringBuilder();
+ bucket.keySet().stream().sorted().forEach(key ->
stored.append(bucket.get(key)));
+ List<String> storedMessages = new ArrayList<>();
+ for (String line : stored.toString().split("\n")) {
+ storedMessages.add(line + "\n");
+ }
+ assertEquals(sent.size(), storedMessages.size(), "every message sent
must be in the bucket");
+ assertEquals(sent.stream().sorted().toList(),
storedMessages.stream().sorted().toList());
+ }
+
+ @Test
+ public void
everyMessageOfAWindowIsStoredWhenTheTimeoutUploadsTheWindowBeforeItEnds()
throws Exception {
+ // streamingUploadTimeout: the timer uploads the window, and more
messages of the window come after it
+ Runnable timeoutTask = timeoutTask();
+ for (int i = 0; i < 10; i++) {
+ Exchange exchange = new DefaultExchange(camelContext);
+ exchange.getIn().setBody("message-" + i + "\n");
+ exchange.getIn().setHeader(Exchange.MESSAGE_TIMESTAMP,
WINDOW_START + i * 10000L);
+ producer.process(exchange);
+ if (i == 4) {
+ timeoutTask.run();
+ }
+ }
+ timeoutTask.run();
+
+ assertEquals(2, uploadIds.get(), "the window is uploaded twice");
+ assertEquals(2, bucket.size(),
+ "each upload of the window must be its own object, but the
bucket has " + bucket.keySet());
+ assertEquals("message-0\nmessage-1\nmessage-2\nmessage-3\nmessage-4\n",
+ bucket.get("grouped-messages_20240101_0800_0800-0805.txt"));
+ assertEquals("message-5\nmessage-6\nmessage-7\nmessage-8\nmessage-9\n",
+ bucket.get("grouped-messages_20240101_0800_0800-0805-1.txt"));
+ }
+
+ private Runnable timeoutTask() throws Exception {
+ Class<?> taskClass =
Class.forName(AWS2S3StreamUploadProducer.class.getName() +
"$StreamingUploadTimeoutTask");
+ Constructor<?> constructor =
taskClass.getDeclaredConstructor(AWS2S3StreamUploadProducer.class);
+ constructor.setAccessible(true);
+ return (Runnable) constructor.newInstance(producer);
+ }
+
+ @Test
+ public void theFirstUploadOfAWindowKeepsItsName() throws Exception {
+ for (int i = 0; i < 10; i++) {
+ Exchange exchange = new DefaultExchange(camelContext);
+ exchange.getIn().setBody("message-" + i + "\n");
+ exchange.getIn().setHeader(Exchange.MESSAGE_TIMESTAMP,
WINDOW_START + i * 10000L);
+ producer.process(exchange);
+ }
+
+ assertEquals(List.of("grouped-messages_20240101_0800_0800-0805.txt"),
new ArrayList<>(bucket.keySet()));
+ }
+
+ @Test
+ public void anUploadAfterARestartDoesNotReplaceTheObjectsOfTheWindow()
throws Exception {
+ // stopping the producer uploads the open window; after a restart (a
new producer) the next upload of the
+ // window must not replace that object
+ producer.start();
+ sendMessages(0, 5);
+ producer.stop();
+ assertEquals(List.of("grouped-messages_20240101_0800_0800-0805.txt"),
new ArrayList<>(bucket.keySet()));
+
+ producer = new AWS2S3StreamUploadProducer(endpoint);
+ producer.start();
+ sendMessages(5, 10);
+ producer.stop();
+
+ assertEquals("message-0\nmessage-1\nmessage-2\nmessage-3\nmessage-4\n",
+ bucket.get("grouped-messages_20240101_0800_0800-0805.txt"));
+ assertEquals("message-5\nmessage-6\nmessage-7\nmessage-8\nmessage-9\n",
+ bucket.get("grouped-messages_20240101_0800_0800-0805-1.txt"));
+ assertEquals(2, bucket.size(), "the bucket has " + bucket.keySet());
+ }
+
+ @Test
+ public void theNextUploadAfterARestartFollowsTheHighestUploadInTheBucket()
throws Exception {
+ // listed in lexical order "-1", "-10", "-2": the highest number wins,
not the last key listed
+ bucket.put("grouped-messages_20240101_0800_0800-0805.txt", "a");
+ bucket.put("grouped-messages_20240101_0800_0800-0805-1.txt", "b");
+ bucket.put("grouped-messages_20240101_0800_0800-0805-10.txt", "c");
+ bucket.put("grouped-messages_20240101_0800_0800-0805-2.txt", "d");
+ // not uploads of the window
+ bucket.put("grouped-messages_20240101_0800_0800-0805-99.log", "e");
+ bucket.put("grouped-messages_20240101_0800_0800-0805-x.txt", "f");
+ bucket.put("grouped-messages_20240101_0800_0800-0805-12-1.txt", "g");
+
+ sendMessages(0, 10);
+
+
assertEquals("message-0\nmessage-1\nmessage-2\nmessage-3\nmessage-4\nmessage-5\nmessage-6\nmessage-7\n"
+ + "message-8\nmessage-9\n",
+ bucket.get("grouped-messages_20240101_0800_0800-0805-11.txt"));
+ assertEquals(8, bucket.size(), "the bucket has " + bucket.keySet());
+ }
+
+ @Test
+ public void
theUploadsOfAWindowAreNumberedFromTheFirstWhenTheBucketCannotBeListed() throws
Exception {
+ // without s3:ListBucket the producer still uploads, numbering the
uploads of the window from the first one
+ when(s3Client.listObjectsV2(any(ListObjectsV2Request.class)))
+
.thenThrow(S3Exception.builder().statusCode(403).message("Access
Denied").build());
+
+ sendMessages(0, 20);
+
+ assertEquals(2, bucket.size(), "the bucket has " + bucket.keySet());
+
assertTrue(bucket.containsKey("grouped-messages_20240101_0800_0800-0805.txt"));
+
assertTrue(bucket.containsKey("grouped-messages_20240101_0800_0800-0805-1.txt"));
+ }
+
+ private void sendMessages(int from, int to) throws Exception {
+ for (int i = from; i < to; i++) {
+ Exchange exchange = new DefaultExchange(camelContext);
+ exchange.getIn().setBody("message-" + i + "\n");
+ exchange.getIn().setHeader(Exchange.MESSAGE_TIMESTAMP,
WINDOW_START + i * 10000L);
+ producer.process(exchange);
+ }
+ }
+}
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 21a1aef9398e..7e9cb0ad2b31 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
@@ -1445,6 +1445,21 @@ context. A check sends two `LookupEvents` requests, one
per event name, and one
always one request, for all Secrets Manager events); reads of secrets no
longer add requests. LookupEvents allows two
requests per second per account and Region.
+=== camel-aws2-s3 - a timestamp window uploaded more than once is stored in
more than one object
+
+With `streamingUploadMode=true` and `timestampGroupingEnabled=true`, the
upload of a time window is completed when it
+reaches `batchMessageNumber` messages (10 by default) or `batchSize` bytes, or
when `streamingUploadTimeout` runs. A
+later message of the same window started a new upload with the same key, and
S3 replaced the object of the first upload
+with it, so the messages of the earlier uploads were lost (unless the bucket
is versioned). A later upload of a window
+now gets the number of the uploads of that window before it after the time
range: for example
+`grouped-messages_20240101_0800_0800-0805.txt`, then
`grouped-messages_20240101_0800_0800-0805-1.txt`,
+`grouped-messages_20240101_0800_0800-0805-2.txt`, and so on. The first upload
of a window keeps the name it had. The
+first time a producer uploads a window, it lists the objects of the window in
the bucket (one `ListObjectsV2` request
+per window), so after a restart it continues after the highest number stored;
this needs the `s3:ListBucket`
+permission, and without it the producer logs a WARN and numbers the uploads of
the window from the first one. These
+keys do not sort in upload order (`-1` sorts before the first upload's name,
`-10` before `-2`). Producers running at
+the same time with the same `keyName` still need distinct key names.
+
=== camel-aws2-kinesis - the consumer continues after the last record when a
shard iterator expired
When a cached shard iterator has expired (an iterator is valid for 5 minutes,
so after a route was suspended for
