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 b130c548223e CAMEL-26539: camel-aws2-s3 - streaming upload writes
concurrent exchanges one at a time (#29122)
b130c548223e is described below
commit b130c548223e0cb6e28486e411779361dbdd59cf
Author: allthingssecurity <[email protected]>
AuthorDate: Wed Nov 4 15:05:15 2026 +0530
CAMEL-26539: camel-aws2-s3 - streaming upload writes concurrent exchanges
one at a time (#29122)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../aws2/s3/stream/AWS2S3StreamUploadProducer.java | 91 ++++----
.../AWS2S3StreamUploadConcurrentExchangesTest.java | 242 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 8 +
3 files changed, 293 insertions(+), 48 deletions(-)
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 660062fd6a9b..cf466437f177 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
@@ -199,59 +199,64 @@ public class AWS2S3StreamUploadProducer extends
DefaultProducer {
private void processWithoutTimestampGrouping(final Exchange exchange)
throws Exception {
InputStream is = exchange.getIn().getMandatoryBody(InputStream.class);
+ // one exchange at a time: the exchanges share the open upload and the
part number of the object names
+ lock.lock();
+ try {
+ writeWithoutTimestampGrouping(exchange, is);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ private void writeWithoutTimestampGrouping(final Exchange exchange, final
InputStream is) throws Exception {
UploadState state = null;
int totalSize = 0;
byte[] b;
int maxRead = (getConfiguration().isMultiPartUpload()
? Math.toIntExact(getConfiguration().getPartSize()) :
getConfiguration().getBufferSize());
- lock.lock();
- try {
- if (uploadAggregate != null) {
- uploadAggregate.index++;
- maxRead -= uploadAggregate.buffer.size();
- maxRead = Math.max(1, maxRead);
- }
- } finally {
- lock.unlock();
+ if (uploadAggregate != null) {
+ uploadAggregate.index++;
+ maxRead -= uploadAggregate.buffer.size();
+ maxRead = Math.max(1, maxRead);
}
while ((b = AWS2S3Utils.toByteArray(is, maxRead)) != null && b.length
> 0) {
totalSize += b.length;
if (getConfiguration().isMultiPartUpload())
maxRead -= b.length;
- lock.lock();
- try {
- // aggregate with previously received exchanges
- if (ObjectHelper.isNotEmpty(uploadAggregate)) {
- uploadAggregate.buffer.write(b);
- if (getConfiguration().isMultiPartUpload() &&
- uploadAggregate.buffer.size() >=
getConfiguration().getPartSize()) {
+ // aggregate with previously received exchanges
+ if (ObjectHelper.isNotEmpty(uploadAggregate)) {
+ uploadAggregate.buffer.write(b);
+ if (getConfiguration().isMultiPartUpload() &&
+ uploadAggregate.buffer.size() >=
getConfiguration().getPartSize()) {
+ uploadPart(uploadAggregate);
+ maxRead =
Math.toIntExact(getConfiguration().getPartSize());
+ continue;
+ }
+ if (uploadAggregate.buffer.size() >=
getConfiguration().getBatchSize()
+ || (uploadAggregate.index >=
getConfiguration().getBatchMessageNumber()
+ && uploadAggregate.buffer.size() <
getConfiguration().getPartSize())) {
+
+ if (uploadAggregate.buffer.size() > 0)
uploadPart(uploadAggregate);
+ CompleteMultipartUploadResponse uploadResult;
+ try {
+ uploadResult = completeUpload(uploadAggregate);
+ } finally {
+ // completeUpload aborts the upload when it fails: do
not append to it any longer
+ this.uploadAggregate = null;
+ }
+ if (getConfiguration().isMultiPartUpload()) {
maxRead =
Math.toIntExact(getConfiguration().getPartSize());
- continue;
}
- if (uploadAggregate.buffer.size() >=
getConfiguration().getBatchSize()
- || (uploadAggregate.index >=
getConfiguration().getBatchMessageNumber()
- && uploadAggregate.buffer.size() <
getConfiguration().getPartSize())) {
-
- if (uploadAggregate.buffer.size() > 0)
- uploadPart(uploadAggregate);
- CompleteMultipartUploadResponse uploadResult =
completeUpload(uploadAggregate);
- this.uploadAggregate = null;
- if (getConfiguration().isMultiPartUpload()) {
- maxRead =
Math.toIntExact(getConfiguration().getPartSize());
- }
- Message message = getMessageForResponse(exchange);
- message.setHeader(AWS2S3Constants.E_TAG,
uploadResult.eTag());
- if (uploadResult.versionId() != null) {
- message.setHeader(AWS2S3Constants.VERSION_ID,
uploadResult.versionId());
- }
- populateHttpResponseCode(uploadResult, message);
+ Message message = getMessageForResponse(exchange);
+ message.setHeader(AWS2S3Constants.E_TAG,
uploadResult.eTag());
+ if (uploadResult.versionId() != null) {
+ message.setHeader(AWS2S3Constants.VERSION_ID,
uploadResult.versionId());
}
- continue;
+ populateHttpResponseCode(uploadResult, message);
}
- } finally {
- lock.unlock();
+ continue;
}
if (state == null) {
state = new UploadState();
@@ -337,17 +342,7 @@ public class AWS2S3StreamUploadProducer extends
DefaultProducer {
if (ObjectHelper.isNotEmpty(state)) {
// exchange wasn't large enough to send, batch it with subsequent
exchanges.
- lock.lock();
- try {
- if (ObjectHelper.isEmpty(this.uploadAggregate)) {
- this.uploadAggregate = state;
- } else {
- // handle potential race condition.
-
this.uploadAggregate.buffer.write(state.buffer.toByteArray());
- }
- } finally {
- lock.unlock();
- }
+ this.uploadAggregate = state;
}
}
diff --git
a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadConcurrentExchangesTest.java
b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadConcurrentExchangesTest.java
new file mode 100644
index 000000000000..83542bc3e9cb
--- /dev/null
+++
b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadConcurrentExchangesTest.java
@@ -0,0 +1,242 @@
+/*
+ * 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.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+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.AfterEach;
+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.AbortMultipartUploadRequest;
+import software.amazon.awssdk.services.s3.model.AbortMultipartUploadResponse;
+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.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
+
+/**
+ * Exchanges sent at the same time through one streaming upload producer must
not start two uploads. The mocked S3Client
+ * keeps the completed objects by key and holds the first
createMultipartUpload call until the test releases it.
+ */
+public class AWS2S3StreamUploadConcurrentExchangesTest {
+
+ @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<>();
+ // the uploads neither completed nor aborted
+ private final Map<String, String> openUploads = new ConcurrentHashMap<>();
+ private final AtomicInteger uploadIds = new AtomicInteger();
+ // a failure of an exchange sent from another thread
+ private final AtomicReference<Throwable> failure = new AtomicReference<>();
+
+ // the first createMultipartUpload call waits in the S3 client until
released
+ private final CountDownLatch firstCreateEntered = new CountDownLatch(1);
+ private final CountDownLatch releaseFirstCreate = new CountDownLatch(1);
+
+ private DefaultCamelContext camelContext;
+ private AWS2S3StreamUploadProducer producer;
+
+ @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.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);
+
+ SdkHttpResponse httpResponse =
SdkHttpResponse.builder().statusCode(200).build();
+
+
when(s3Client.createMultipartUpload(any(CreateMultipartUploadRequest.class))).thenAnswer(invocation
-> {
+ CreateMultipartUploadRequest request = invocation.getArgument(0);
+ int id = uploadIds.incrementAndGet();
+ if (id == 1) {
+ firstCreateEntered.countDown();
+ assertTrue(releaseFirstCreate.await(20, TimeUnit.SECONDS),
"the first upload was not released");
+ }
+ String uploadId = "upload-" + id;
+ keysByUploadId.put(uploadId, request.key());
+ partsByUploadId.put(uploadId, new ByteArrayOutputStream());
+ openUploads.put(uploadId, request.key());
+ 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);
+ ByteArrayOutputStream parts =
partsByUploadId.get(request.uploadId());
+ synchronized (parts) {
+
parts.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);
+ openUploads.remove(request.uploadId());
+ // 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();
+ });
+
+
when(s3Client.abortMultipartUpload(any(AbortMultipartUploadRequest.class))).thenAnswer(invocation
-> {
+ AbortMultipartUploadRequest request = invocation.getArgument(0);
+ openUploads.remove(request.uploadId());
+ return AbortMultipartUploadResponse.builder().build();
+ });
+
+ producer = new AWS2S3StreamUploadProducer(endpoint);
+ }
+
+ @AfterEach
+ public void tearDown() {
+ releaseFirstCreate.countDown();
+ camelContext.stop();
+ }
+
+ @Test
+ public void
anExchangeThatStartedAnUploadWhileAnotherRanDoesNotReplaceAnObject() throws
Exception {
+ producer.start();
+
+ // exchange 0 waits in createMultipartUpload while 10 other exchanges
arrive (batchMessageNumber=10)
+ sendWhileTheFirstUploadIsCreated("message-0\n", () -> send(1, 10));
+ send(11, 9);
+ producer.stop();
+
+ assertNull(failure.get(), () -> "an exchange failed: " +
failure.get());
+ List<String> stored = storedLines();
+ for (int i = 0; i < 20; i++) {
+ assertEquals(1, Collections.frequency(stored, "message-" + i),
+ "message-" + i + " must be stored once, but the bucket has
" + bucket);
+ }
+ assertEquals(20, stored.size(), "every message sent must be in the
bucket: " + bucket);
+ assertEquals(Collections.emptyMap(), openUploads, "every upload is
completed or aborted");
+ }
+
+ @Test
+ public void
theFirstPartOfALargeBodyIsNotLostWhenAnotherExchangeOpensTheUpload() throws
Exception {
+ // multipart upload; the body of exchange A is one part and a half
(the reads of the producer are 4096 bytes)
+ when(configuration.isMultiPartUpload()).thenReturn(true);
+ when(configuration.getPartSize()).thenReturn(8192L);
+ String large = "A".repeat(8191) + "\n" + "B".repeat(4095) + "\n";
+ producer.start();
+
+ // the large exchange reads its first part and waits in
createMultipartUpload while a small exchange arrives
+ sendWhileTheFirstUploadIsCreated(large, () -> send(0, 1));
+ producer.stop();
+
+ assertNull(failure.get(), () -> "an exchange failed: " +
failure.get());
+ String all = String.join("", bucket.values());
+ assertTrue(all.contains(large), "the large body must be stored whole,
but the bucket has "
+ + bucket.keySet() + " with " +
all.length() + " characters");
+ assertTrue(all.contains("message-0\n"), "the small body must be
stored");
+ assertEquals(large.length() + "message-0\n".length(), all.length(),
"nothing else is stored");
+ assertEquals(Collections.emptyMap(), openUploads, "every upload is
completed or aborted");
+ }
+
+ private void sendWhileTheFirstUploadIsCreated(String firstBody, Runnable
others) throws Exception {
+ Thread first = new Thread(() -> {
+ try {
+ Exchange exchange = new DefaultExchange(camelContext);
+ exchange.getIn().setBody(firstBody);
+ producer.process(exchange);
+ } catch (Throwable e) {
+ failure.set(e);
+ }
+ }, "first");
+ first.start();
+ assertTrue(firstCreateEntered.await(20, TimeUnit.SECONDS));
+
+ Thread other = new Thread(others, "others");
+ other.start();
+ // the other exchanges either ran (the first one holds no lock) or
wait for the first one
+ await().atMost(20, TimeUnit.SECONDS)
+ .until(() -> !other.isAlive() || other.getState() ==
Thread.State.WAITING);
+ releaseFirstCreate.countDown();
+ first.join(20000);
+ other.join(20000);
+ assertFalse(first.isAlive() || other.isAlive(), "the exchanges did not
complete");
+ }
+
+ private void send(int from, int count) {
+ for (int i = from; i < from + count; i++) {
+ Exchange exchange = new DefaultExchange(camelContext);
+ exchange.getIn().setBody("message-" + i + "\n");
+ try {
+ producer.process(exchange);
+ } catch (Throwable e) {
+ failure.set(e);
+ }
+ }
+ }
+
+ private List<String> storedLines() {
+ return bucket.values().stream().flatMap(content ->
Arrays.stream(content.split("\n"))).toList();
+ }
+}
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 3e329077add8..ed00683ba05a 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
@@ -3560,6 +3560,14 @@ named after `keyName` alone (for example `camel.txt`),
and only the later files
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-aws2-s3 - streaming upload writes concurrent exchanges one at a time
+
+With `streamingUploadMode=true` (without `timestampGroupingEnabled`), the
producer now writes one exchange at a time,
+so exchanges sent concurrently (concurrent consumers, parallel splits,
`threads()`) no longer lose objects or parts.
+The body of an exchange is read while the other exchanges of the same producer
wait, and bodies that complete an
+upload of their own (`batchSize` or more, or `batchMessageNumber=1`) are no
longer uploaded in parallel. To upload
+large bodies in parallel, use the producer without `streamingUploadMode`.
+
=== 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
