[
https://issues.apache.org/jira/browse/CAMEL-26539?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Work on CAMEL-26539 started by shashank.
----------------------------------------
> 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
> Priority: Major
>
> 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)
