CAMEL-25785: camel-aws2-s3 - a timestamp window uploaded more than once is stored in more than one object by allthingssecurity · Pull Request #28136 · apache/camel

allthingssecurity · 2026-10-20T00:11:52Z

Description

With streamingUploadMode=true and timestampGroupingEnabled=true, the aws2-s3 stream upload producer names the object of a time window only after the window (grouped-messages_20240101_0800_0800-0805.txt). The upload of a window is completed when it reaches batchMessageNumber messages (10 by default) or batchSize bytes, or when streamingUploadTimeout runs, and the window is then forgotten. The next message of the same window started a new multipart upload with the same key, and S3 replaced the first object with it. The messages of the earlier uploads were lost (unless the bucket is versioned), and every exchange had completed normally.

The example in the component documentation (a message every 10 seconds, 5-minute windows, default batchMessageNumber) has 30 messages per window: the window is uploaded three times and the bucket keeps only the last 10 messages.

Change (in AWS2S3StreamUploadProducer): the producer counts the completed uploads of each window, under its existing lock. The count lives in an insertion-ordered map that holds the last 10000 windows (windows come in time order). The first time a producer uploads a window, it lists the window's objects in the bucket (ListObjectsV2 with the window key as prefix, following continuation tokens) and starts the count at the highest upload stored (the first upload's name counts 1, -N counts N+1), so an upload after a restart does not replace the object the stop flushed. If the listing fails (for example without s3:ListBucket), it logs a WARN and numbers the window from the first upload. The windows flushed by doStop are counted too. A later upload of a window gets that count after the time range: grouped-messages_20240101_0800_0800-0805-1.txt, then -2, and so on. The first upload of a window keeps the name it had, so a window that is uploaded once is unchanged.

What users can notice: the second and later objects of a window have a new key. Upgrade guide (4.24): a note under its own camel-aws2-s3 heading. The naming is also described in aws2-s3-streaming.adoc (and its catalog copy). One ListObjectsV2 request per window per producer; it needs s3:ListBucket. The keys of a window do not sort in upload order (-1 before the first upload's name, -10 before -2); the docs say so.

Tests:

New AWS2S3StreamUploadTimestampWindowOverwriteTest (mocked S3Client that keeps completed objects by key, like a bucket). everyMessageOfAWindowIsStoredWhenTheWindowIsUploadedMoreThanOnce (the documentation example) and everyMessageOfAWindowIsStoredWhenTheTimeoutUploadsTheWindowBeforeItEnds fail without the change (expected: <3> but was: <1>, expected: <2> but was: <1>: the bucket holds one object for the window). theFirstUploadOfAWindowKeepsItsName passes with and without the change.
AWS2S3StreamUploadProducerTimestampTest calls the name method through reflection; it now uses the new signature and checks the suffix.
Review follow-up: anUploadAfterARestartDoesNotReplaceTheObjectsOfTheWindow and theNextUploadAfterARestartFollowsTheHighestUploadInTheBucket (objects -1, -10, -2 in the bucket, next is -11; listing paged two keys at a time) fail without the listing; theUploadsOfAWindowAreNumberedFromTheFirstWhenTheBucketCannotBeListed covers a 403 on the listing. S3StreamUploadTimestampGroupingIT.testTimestampGroupingBasic and S3StreamUploadTimestampGroupingEdgeCasesIT.testTimestampGroupingWithVeryLargeWindow now expect the exact keys, including the -1 object of each window (they asserted the old overwrite).
With the change the camel-aws2-s3 tests pass: 110 run, 0 failures, 0 errors, 0 skipped.

What is not changed: producers that run at the same time with the same keyName still choose the same names for a window, so they still need distinct key names (documented). A failed timeout flush still drops its window with a WARN (as decided in CAMEL-24195). Kept on main only (no backport), as agreed in the review.

What was not run: the full root build; no test against S3; the S3 integration tests (no Docker locally).

Target

I checked that the commit is targeting the correct branch (Camel 4 uses the main branch)

Tracking

If this is a large change, bug fix, or code improvement, I checked there is a JIRA issue filed for the change (usually before you start working on it): CAMEL-25785.

Apache Camel coding standards and style

I checked that each commit in the pull request has a meaningful subject line and body.
I have run mvn clean install -DskipTests locally from root folder and I have committed all auto-generated changes.
(I built and tested camel-aws2-s3, including the formatter and import-sort plugins. No option changed; the catalog copy of the edited doc page is updated by hand. I did not run the full root build.)

AI-assisted contributions

If this PR includes AI-generated code, commits have proper co-authorship attribution (e.g., Co-authored-by trailers) and the PR description identifies the AI tool used.
This PR was prepared with Claude Code (Claude Opus 5.5). The commits carry a Co-Authored-By trailer.

Claude Code on behalf of allthingssecurity

🤖 Generated with Claude Code

…ce is stored in more than one object With streamingUploadMode and timestampGroupingEnabled, the upload of a time window is completed at batchMessageNumber messages (10 by default), batchSize bytes or by streamingUploadTimeout. A later message of the same window started a new multipart upload with the same key, so S3 replaced the object of the first upload and its messages were lost. A later upload of a window now gets the number of the uploads of the window before it after the time range (-1, -2, ...); the first keeps its name. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

gnodet-bot

The fix correctly addresses the file naming collision that caused data loss when a timestamp window was uploaded multiple times. Good approach with the LinkedHashMap LRU eviction and the test coverage is solid.

Two minor observations:

1. Pre-existing (out of scope): In StreamingUploadTimeoutTask.run(), keysToRemove.add(entry.getKey()) runs outside the try/catch block. If uploadPart or completeUpload throws, the entry is removed from timestampBasedUploads without incrementing completedUploadsByWindow. On the next message for that window, generateTimestampBasedFileName will reuse the name of the first upload (count=0), potentially creating a filename collision. Pre-existing behavior, unrelated to this PR.

2. accessOrder=true iteration hazard (maintenance note): See inline comment on completedUploadsByWindow.

This review was generated by an AI agent, Hermès on behalf of @gnodet.

gnodet-bot · 2026-10-20T00:25:13Z

+    // 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) {


⚠️ Maintenance note: accessOrder=true (the third constructor argument) makes every get() a structural modification of the map. The existing comment notes it’s “guarded by the lock”, which keeps the current merge() and getOrDefault() calls safe. Consider adding a warning comment here so future maintainers don’t inadvertently call get() inside an iteration loop—it would throw ConcurrentModificationException.

Thanks. Every access to the map is under the producer's lock and nothing iterates over it (only merge and getOrDefault), so the access order is safe as it is; the field comment already says the lock guards it. I can spell out "no iteration" there if a maintainer prefers.

On the pre-existing point in the review: a timeout flush that fails is not counted, but a failed flush stores no object either (the multipart upload is not completed), so the next upload of that window takes the name the failed one would have had, and nothing is replaced.

Claude Code on behalf of allthingssecurity

github-actions · 2026-10-20T05:47:01Z

🌟 Thank you for your contribution to the Apache Camel project! 🌟
🤖 CI automation will test this PR automatically.

🐫 Apache Camel Committers, please review the following items:

First-time contributors require MANUAL approval for the GitHub Actions to run
You can use the command /component-test (camel-)component-name1 (camel-)component-name2.. to request a test from the test bot although they are normally detected and executed by CI.
You can label PRs using skip-tests and test-dependents to fine-tune the checks executed by this PR.
Build and test logs are available in the summary page. Only Apache Camel committers have access to the summary.

⚠️ Be careful when sharing logs. Review their contents before sharing them publicly.

davsclaus

Thanks for this. Within one producer instance the fix is correct: every map access is under lock, the timer counts an upload only after completeUpload succeeds, and the new tests fail without the fix.

Requested change: the restart and second-instance case. After a restart the producer is a new instance (SendProcessor.doStart creates a new one each time), so the count starts again at 0. A restart in the middle of a window still loses data: doStop flushes the open window to ..._0800-0805.txt, and the first upload of that window after the restart writes to the same key. Two producers with the same keyName collide the same way, because the window name is the same on both. #28137 fixes exactly this restart and second-instance case for the random strategy, so leaving it open here would be inconsistent. Please fix it in this PR. Some options:

(a) The first time a producer sees a window, list the objects with that window's key prefix (ListObjectsV2, as setStartingPart already does for lastPart) and start the count from there. That costs one request per window per producer and fixes restarts, but not concurrent producers.
(b) Fail closed with a conditional write: CompleteMultipartUploadRequest.ifNoneMatch("*") (S3-compatible stores may not support it).

(a) is probably enough if the docs say that concurrent producers on the same keyName still need distinct key names.

Backport: JIRA lists 4.18.x and 4.22.x as affected, but this changes object names users see, so I'd keep it on main.

Claude Code on behalf of davsclaus. This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.

davsclaus · 2026-10-20T05:57:30Z

+    // 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) {


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.

Suggested change

private final Map<Long, Integer> completedUploadsByWindow = new LinkedHashMap<>(16, 0.75f, true) {

private final Map<Long, Integer> completedUploadsByWindow = new LinkedHashMap<>() {

Done in 00f3909: the map is insertion-ordered now (new LinkedHashMap<>() with removeEldestEntry), so a lookup no longer reorders it.

Claude Code on behalf of allthingssecurity

davsclaus · 2026-10-20T05:57:31Z

+    }
+
+    private Runnable timeoutTask() throws Exception {
+        when(configuration.getStreamingUploadTimeout()).thenReturn(60000L);


Nit: StreamingUploadTimeoutTask.run() does not read getStreamingUploadTimeout(), so this stub is not needed.

Right, removed in 00f3909.

Claude Code on behalf of allthingssecurity

davsclaus · 2026-10-20T05:57:31Z

+`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.


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.

Added in 00f3909: the keys of a window do not sort in upload order (-1 before the first upload's name, -10 before -2), so to read a window in order, sort by the number after the time range. The upgrade guide says the same.

Claude Code on behalf of allthingssecurity

github-actions · 2026-10-20T06:02:47Z

🧪 CI tested the following changed modules:

catalog/camel-catalog
components/camel-aws/camel-aws2-s3
docs

🔬 Scalpel shadow comparison — Scalpel: 9 of 707 tested, 25 compile-only — current: 9 all tested

Maveniverse Scalpel detected 9 affected modules (current approach: 9).

Skip-tests mode would test 9 modules (3 direct + 8 downstream), skip tests for 25 (generated code, meta-modules)

Modules Scalpel would test (9)

camel-aws2-eventbridge ← downstream of org.apache.camel:camel-aws2-s3
camel-aws2-s3 ← components/camel-aws/camel-aws2-s3/src/main/docs/aws2-s3-streaming.adoc, components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadProducer.java, components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadProducerTimestampTest.java, components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadTimestampWindowOverwriteTest.java
camel-jbang-mcp ← downstream of org.apache.camel:camel-catalog
camel-jbang-plugin-mcp ← downstream of org.apache.camel:camel-jbang-core
camel-jbang-plugin-tui ← downstream of org.apache.camel:camel-catalog
camel-jbang-plugin-validate ← downstream of org.apache.camel:camel-yaml-dsl-validator
camel-launcher-container ← downstream of org.apache.camel:camel-launcher
camel-yaml-dsl-validator ← downstream of org.apache.camel:camel-catalog
camel-yaml-dsl-validator-maven-plugin ← downstream of org.apache.camel:camel-yaml-dsl-validator

Modules with tests skipped (25)

apache-camel
camel-allcomponents
camel-catalog-console
camel-catalog-maven
camel-catalog-suggest
camel-componentdsl
camel-endpointdsl
camel-endpointdsl-support
camel-itest
camel-jbang-core
camel-jbang-it
camel-jbang-main
camel-jbang-plugin-edit
camel-jbang-plugin-generate
camel-jbang-plugin-kubernetes
camel-jbang-plugin-test
camel-kamelet-main
camel-launcher
camel-report-maven-plugin
camel-route-parser
camel-yaml-dsl
camel-yaml-dsl-deserializers
camel-yaml-dsl-maven-plugin
coverage
dummy-component

ℹ️ Shadow mode — Scalpel observes but does not affect test execution. Learn more

All tested modules (36 modules, 0.0s total)

Total reactor time: 0.0s

Module	Duration	Status
Camel :: AWS2 Eventbridge	n/a
Camel :: AWS2 S3	n/a
Camel :: All Components Sync point	n/a
Camel :: Assembly	n/a
Camel :: Catalog :: Camel Catalog	n/a
Camel :: Catalog :: Camel Report Maven Plugin	n/a
Camel :: Catalog :: Camel Route Parser	n/a
Camel :: Catalog :: Console	n/a
Camel :: Catalog :: Dummy Component	n/a
Camel :: Catalog :: Maven	n/a
Camel :: Catalog :: Suggest (deprecated)	n/a
Camel :: Component DSL	n/a
Camel :: Coverage	n/a
Camel :: Docs	n/a
Camel :: Endpoint DSL	n/a
Camel :: Endpoint DSL :: Support	n/a
Camel :: Integration Tests	n/a
Camel :: JBang :: Core	n/a
Camel :: JBang :: Integration tests	n/a
Camel :: JBang :: MCP	n/a
Camel :: JBang :: Main	n/a
Camel :: JBang :: Plugin :: Edit	n/a
Camel :: JBang :: Plugin :: Generate	n/a
Camel :: JBang :: Plugin :: Kubernetes	n/a
Camel :: JBang :: Plugin :: MCP	n/a
Camel :: JBang :: Plugin :: TUI	n/a
Camel :: JBang :: Plugin :: Testing	n/a
Camel :: JBang :: Plugin :: Validate	n/a
Camel :: Kamelet Main	n/a
Camel :: Launcher	n/a
Camel :: Launcher :: Container	n/a
Camel :: YAML DSL	n/a
Camel :: YAML DSL :: Deserializers	n/a
Camel :: YAML DSL :: Maven Plugins	n/a
Camel :: YAML DSL :: Validator	n/a
Camel :: YAML DSL :: Validator Maven Plugin	n/a

⚙️ View full build and test results

…ow after the objects already in the bucket The first time a producer uploads a timestamp window, it lists the objects of the window (ListObjectsV2 with the window's key as prefix) and continues after the highest upload number stored, so an upload after a restart no longer replaces the object that the stop flushed. Without s3:ListBucket it logs a WARN and numbers from the first upload. The uploads flushed by doStop are counted too. The window count map keeps insertion order (windows come in time order). The docs say that window keys do not sort in upload order and that concurrent producers on the same keyName still need distinct key names. S3StreamUploadTimestampGroupingIT and S3StreamUploadTimestampGroupingEdgeCasesIT now expect the second upload of a window as its own object. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

allthingssecurity · 2026-10-20T07:13:45Z

Thanks for the review. I went with option (a), in 00f3909:

The first time a producer uploads a window, it lists the window's objects (ListObjectsV2 with the window key as prefix, following continuation tokens) and starts the count at the highest upload found: the first upload's name counts 1, -N counts N+1. Other keys under that prefix (another extension, -x) are ignored. After that the count is kept in memory as before. That's one listing per window per producer.
If the listing fails (for example without s3:ListBucket), the producer logs a WARN and numbers the window from the first upload, as before this PR. The streaming docs and the upgrade guide say so.
doStop now counts the windows it flushes, so a stopped and restarted producer instance does not reuse the name either.
New tests in AWS2S3StreamUploadTimestampWindowOverwriteTest: a restart in the middle of a window (stop flushes ..._0800-0805.txt, a new producer writes ..._0800-0805-1.txt), objects -1, -10, -2 already in the bucket (next is -11, with the listing paged two keys at a time), and a listing that fails with 403. With the listing removed, the two restart tests fail. The camel-aws2-s3 unit tests pass: 110 run, 0 failures.
Docs: producers running at the same time with the same keyName still need distinct key names. Option (a) does not prevent that collision.

Backport: agreed, main only.

CI: S3StreamUploadTimestampGroupingIT.testTimestampGroupingBasic and S3StreamUploadTimestampGroupingEdgeCasesIT.testTimestampGroupingWithVeryLargeWindow send 10 messages per window with batchMessageNumber=5. They expected one object per window, which was the overwrite: on main, half of the messages are lost. They now check the exact keys (basicTest_20240101_0800_0800-0805.txt and ...-0805-1.txt, the same for 0805-0810, and largeWindowTest_20240101_0800_0800-0900.txt and -1.txt). I could not run the ITs locally (no Docker here), so CI will be their first run.

Claude Code on behalf of allthingssecurity

gnodet-bot

Following up on my earlier review — all prior findings addressed in commit 00f3909.

gnodet-bot observations (from first review):

✅ accessOrder=true hazard — fixed: map now uses plain insertion-ordered new LinkedHashMap<>(), eliminating the structural-modification-on-get concern.
Pre-existing timeout flush issue (not addressed, correctly out of scope): acknowledged, no action needed.

davsclaus's CHANGES_REQUESTED:

✅ Restart-safety — implemented via completedUploads() + uploadsInBucket(): on first encounter of a window, the producer lists existing bucket objects and continues numbering from the highest stored upload. The result is cached (including on failure, where 0 is stored after the exception), so the ListObjectsV2 call occurs at most once per window per producer lifetime.
✅ Insertion-ordered map — confirmed (no accessOrder argument).
✅ getStreamingUploadTimeout() stub removed from test.
✅ Sort-order note added to docs.

Pre-fix test validation: ⚠️ Dynamic run skipped (no worktree checkout available). Static trace confirms: without completedUploads()/uploadsInBucket(), a second producer instance would start with empty completedUploadsByWindow, call generateTimestampBasedFileName(..., 0), produce the same key as the first upload, and overwrite it. The fix correctly queries the bucket on first window encounter; anUploadAfterARestartDoesNotReplaceTheObjectsOfTheWindow exercises exactly this path.

One low-severity nit in uploadsInBucket(): the early return 0 in the catch block bypasses the normal return uploads path. This is harmless (the caller completedUploads() caches whatever is returned, including 0), but a future reader might worry about repeated calls on persistent ListObjectsV2 failures. A brief comment noting the caller caches the result would preempt that concern:

            LOG.debug("Listing the objects of {} failed", windowKey + ext, e);
            return 0; // completedUploads() caches this; no retry storm on persistent failures

Otherwise the second commit is clean. The bounded LinkedHashMap with removeEldestEntry handles long-running producers correctly. uploadNumber() correctly handles adversarial keys (-12-1, wrong extension, non-digit suffixes), confirmed by theNextUploadAfterARestartFollowsTheHighestUploadInTheBucket. Lock discipline in doStop() is consistent — lock.lock() is acquired before the flush loop and covers the new completedUploadsByWindow.merge() call. Docs and upgrade guide are comprehensive.

This review was generated by an AI agent, Hermès on behalf of @gnodet.

allthingssecurity · 2026-10-20T07:28:20Z

Thanks for the re-check. On the return 0 nit: completedUploads() stores whatever uploadsInBucket() returns, including the 0 after a failed listing, so a window is listed at most once per producer even when the listing keeps failing. The Javadoc of completedUploads() says the bucket is listed the first time the producer sees the window, so I've left the code unchanged rather than push another commit for a comment.

Claude Code on behalf of allthingssecurity

gnodet-bot reviewed Oct 20, 2026

View reviewed changes

gnodet added the lts label Oct 20, 2026

github-actions Bot added components catalog docs components-aws and removed lts labels Oct 20, 2026

davsclaus requested changes Oct 20, 2026

View reviewed changes

allthingssecurity mentioned this pull request Oct 20, 2026

CAMEL-25789: camel-aws2-s3 - streaming upload with the random or timestamp naming strategy names the first file too #28137

Open

4 of 5 tasks

gnodet-bot reviewed Oct 20, 2026

View reviewed changes

	private final Map<Long, Integer> completedUploadsByWindow = new LinkedHashMap<>(16, 0.75f, true) {
	private final Map<Long, Integer> completedUploadsByWindow = new LinkedHashMap<>() {

Conversation

allthingssecurity commented Oct 20, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Target

Tracking

Apache Camel coding standards and style

AI-assisted contributions

Uh oh!

gnodet-bot left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

github-actions Bot commented Oct 20, 2026

Uh oh!

davsclaus left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

github-actions Bot commented Oct 20, 2026

Uh oh!

allthingssecurity commented Oct 20, 2026

Uh oh!

gnodet-bot left a comment

Choose a reason for hiding this comment

Uh oh!

allthingssecurity commented Oct 20, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

allthingssecurity commented Oct 20, 2026 •

edited

Loading