davsclaus commented on code in PR #28136:
URL: https://github.com/apache/camel/pull/28136#discussion_r4321905193
##########
components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadProducer.java:
##########
@@ -71,9 +72,21 @@ 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. Only the windows
+ // used last are counted, so a long-running producer does not keep every
window it has seen.
+ private final Map<Long, Integer> completedUploadsByWindow = new
LinkedHashMap<>(16, 0.75f, true) {
Review Comment:
Windows arrive in time order, so insertion order is enough for evicting old
windows. Dropping `accessOrder` also makes `getOrDefault` a plain read, which
settles the iteration concern in the thread above.
```suggestion
private final Map<Long, Integer> completedUploadsByWindow = new
LinkedHashMap<>() {
```
##########
components/camel-aws/camel-aws2-s3/src/main/docs/aws2-s3-streaming.adoc:
##########
@@ -96,7 +96,11 @@ 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.
Review Comment:
Worth one sentence here: these keys do not sort in upload order.
`...0805-1.txt` sorts before `...0805.txt` (`-` < `.`), and `-10` sorts before
`-2`, so anyone listing a window's objects in lexical order gets them out of
order.
##########
components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadTimestampWindowOverwriteTest.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * 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.UploadPartRequest;
+import software.amazon.awssdk.services.s3.model.UploadPartResponse;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+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();
+ });
+
+ 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 {
+ when(configuration.getStreamingUploadTimeout()).thenReturn(60000L);
Review Comment:
Nit: `StreamingUploadTimeoutTask.run()` does not read
`getStreamingUploadTimeout()`, so this stub is not needed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
