shashank created CAMEL-26539:
--------------------------------
Summary: camel-aws2-s3 - streaming upload loses objects and parts
when exchanges are sent concurrently
Key: CAMEL-26539
URL: https://issues.apache.org/jira/browse/CAMEL-26539
Project: Camel
Issue Type: Bug
Components: camel-aws2
Reporter: shashank
Assignee: shashank
Found with a SPIN model of {{AWS2S3StreamUploadProducer}} (exchanges sent at
the same time through the one producer of a route, the lifecycle thread, and a
fake bucket in which a completed upload replaces the object of the same key),
then reproduced on the real producer with a mocked {{S3Client}}.
With {{streamingUploadMode=true}} (without {{timestampGroupingEnabled}}), the
producer collects the exchanges into one open multipart upload
({{uploadAggregate}}). {{processWithoutTimestampGrouping}} checks for the open
upload under the producer's lock (since CAMEL-24195), but when there is none it
starts an upload of its own *outside* the lock: it creates the upload state
(which takes the object name from the producer's part number, {{progressive}}
naming by default), calls {{createMultipartUpload}}, and only at the end of the
exchange takes the lock again to make its state the open upload, or, if another
exchange did that meanwhile, to copy its buffered bytes into that one ("handle
potential race condition"). Exchanges run at the same time whenever the route
has more than one thread: concurrent consumers (seda, jms, kafka), parallel
splits, {{threads()}}, HTTP consumers.
Two exchanges that find no open upload at the same time (at the start, and
again after every completed object, so after every {{batchMessageNumber}}
messages) both start an upload, and:
* *an object replaces another one*: both states take the same part number, so
both uploads get the same key. If the other exchange's upload is completed
first (for example once the next messages reach {{batchMessageNumber}}), the
slower exchange then makes its state the open upload, and when that one is
completed it replaces the object. With the defaults ({{batchMessageNumber=10}},
progressive naming) and two threads, the 10 messages of the first object are
lost, although their exchanges completed without an error (the one that
completed the object even got its ETag). When every exchange completes its own
upload ({{batchMessageNumber=1}}, or bodies of {{batchSize}} or more), two
concurrent exchanges can write the same key.
* *the first part of a large body is lost*: with {{multiPartUpload=true}}, an
exchange whose body is larger than {{partSize}} uploads its first part to its
own upload; if another exchange opened the shared upload meanwhile, the rest of
the body goes to the shared upload and the own upload, with the first part, is
neither completed nor aborted (an incomplete multipart upload that is billed
until a lifecycle rule removes it).
* the upload of the exchange whose bytes are copied is neither completed nor
aborted either.
Short trace of the model (3 exchanges, {{batchMessageNumber=2}}; line numbers
on main at dff263d86f08):
{noformat}
1 E2 : no open upload -> new state, part 0 (key camel.txt),
createMultipartUpload (AWS2S3StreamUploadProducer:256-304, no lock)
2 E1 : no open upload -> new state, part 0 (key camel.txt),
createMultipartUpload (no lock)
3 E2 : open upload := E2's state
(:338-351)
4 E0 : writes into the open upload, batchMessageNumber reached -> complete
camel.txt, part = 1, open upload := null (:222-255)
5 E1 : open upload is null -> open upload := E1's state (key camel.txt)
(:338-351)
6 stop (or the next messages): complete E1's upload -> camel.txt replaced, E0
and E2 lost
{noformat}
The real-class test holds the first {{createMultipartUpload}} call in the
mocked client: exchange 0 waits there while 10 more exchanges are sent, then 9
more. On main the bucket ends with one object of 10 messages ({{camel.txt}}:
message 0 and 11-19); messages 1-10 are gone. A second test sends a body of one
part and a half while a small exchange arrives: the first part is lost and its
upload stays open.
*Proposal:* write one exchange at a time: take the lock for the whole exchange
in {{processWithoutTimestampGrouping}} (the shared upload and the part number
of the names belong to the producer, and an exchange appended to the shared
upload already does its S3 calls under the lock). An exchange that finds no
open upload then starts it and makes it the open upload before the next
exchange looks, and the copy branch is no longer needed. The timestamp grouping
path is not changed.
Affected: main, 4.23.x, 4.22.x (the same code), 4.18.x, 4.14.x (the same race
in older code).
_Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
