You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by "allthingssecurity (via GitHub)" <gi...@apache.org> on 2026/10/20 00:11:55 UTC
[PR] CAMEL-25785: camel-aws2-s3 - a timestamp window uploaded more than once is stored in more than one object [camel]
allthingssecurity opened a new pull request, #28136:
URL: https://github.com/apache/camel/pull/28136
# Description
[CAMEL-25785](https://issues.apache.org/jira/browse/CAMEL-25785)
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 access-ordered map that holds the last 10000 windows. 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).
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.
- With the change the camel-aws2-s3 tests pass: 107 run, 0 failures, 0 errors, 0 skipped.
What is not changed: after a restart the count starts again, so a message for a window that was already uploaded before the restart still uses the name of the first upload. A failed timeout flush still drops its window with a `WARN` (as decided in CAMEL-24195).
What was not run: the full root build; no test against S3.
# Target
- [x] I checked that the commit is targeting the correct branch (Camel 4 uses the `main` branch)
# Tracking
- [x] If this is a large change, bug fix, or code improvement, I checked there is a [JIRA issue](https://issues.apache.org/jira/browse/CAMEL) filed for the change (usually before you start working on it): [CAMEL-25785](https://issues.apache.org/jira/browse/CAMEL-25785).
# Apache Camel coding standards and style
- [x] 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
- [x] 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 commit carries a `Co-Authored-By` trailer.
_Claude Code on behalf of allthingssecurity_
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25785: camel-aws2-s3 - a timestamp window uploaded more than once is stored in more than one object [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #28136:
URL: https://github.com/apache/camel/pull/28136#issuecomment-6284334552
:star2: Thank you for your contribution to the Apache Camel project! :star2:
:robot: CI automation will test this PR automatically.
:camel: 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](https://camel.apache.org/community/team/#committers) have access to the summary.
:warning: Be careful when sharing logs. Review their contents before sharing them publicly.
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25785: camel-aws2-s3 - a timestamp window uploaded more than once is stored in more than one object [camel]
Posted by "allthingssecurity (via GitHub)" <gi...@apache.org>.
allthingssecurity commented on code in PR #28136:
URL: https://github.com/apache/camel/pull/28136#discussion_r4320247744
##########
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:
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_
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25785: camel-aws2-s3 - a timestamp window uploaded more than once is stored in more than one object [camel]
Posted by "allthingssecurity (via GitHub)" <gi...@apache.org>.
allthingssecurity commented on code in PR #28136:
URL: https://github.com/apache/camel/pull/28136#discussion_r4324515446
##########
components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadTimestampGroupingIT.java:
##########
@@ -75,23 +75,13 @@ public void process(Exchange exchange) {
});
List<S3Object> resp = ex.getMessage().getBody(List.class);
- assertEquals(2, resp.size(), "Should create 2 files for 2 different timestamp windows");
-
- // Verify files have timestamp-based naming
- boolean foundWindow1File = false;
- boolean foundWindow2File = false;
-
- for (S3Object s3Object : resp) {
- String key = s3Object.key();
- if (key.contains("20240101_0800_0800-0805")) {
- foundWindow1File = true;
- } else if (key.contains("20240101_0805_0805-0810")) {
- foundWindow2File = true;
- }
- }
-
- assertTrue(foundWindow1File, "Should find file for first timestamp window");
- assertTrue(foundWindow2File, "Should find file for second timestamp window");
+ // batchMessageNumber=5: each window is uploaded twice, and the second upload of a window gets its own object
+ assertEquals(List.of(
Review Comment:
Thanks, done in c3290bed822d26902bbfed1e20ea3461ec9b85e1: `S3StreamUploadTimestampGroupingIT` and `S3StreamUploadTimestampGroupingEdgeCasesIT` now delete the objects of the bucket in a `@BeforeEach`, with the same helper as the ITs in #28137. Besides the exact key lists, this also covers the exact counts of the large span, custom window, multipart and boundary tests, which a rerun broke the same way. Compiled with `test-compile`; not run here (no Docker locally).
_Claude Code on behalf of allthingssecurity_
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25785: camel-aws2-s3 - a timestamp window uploaded more than once is stored in more than one object [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #28136:
URL: https://github.com/apache/camel/pull/28136#issuecomment-6284527666
<!-- ci-tested-modules -->
:test_tube: **CI tested the following changed modules:**
- `catalog/camel-catalog`
- `components/camel-aws/camel-aws2-s3`
- `docs`
---
<details><summary>:microscope: Scalpel shadow comparison — Scalpel: 9 of 707 tested, 25 compile-only — current: 9 all tested</summary>
[Maveniverse Scalpel](https://github.com/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)
<details><summary>Modules Scalpel would test (9)</summary>
- `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
</details>
<details><summary>Modules with tests skipped (25)</summary>
- `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`
</details>
> :information_source: Shadow mode — Scalpel observes but does not affect test execution. [Learn more](https://github.com/maveniverse/scalpel)
</details>
<details><summary>All tested modules (36 modules, 0.0s total)</summary>
**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 | |
</details>
---
:gear: [View full build and test results](https://github.com/apache/camel/actions/runs/39346204430)
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25785: camel-aws2-s3 - a timestamp window uploaded more than once is stored in more than one object [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus commented on code in PR #28136:
URL: https://github.com/apache/camel/pull/28136#discussion_r4324150528
##########
components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadTimestampGroupingIT.java:
##########
@@ -75,23 +75,13 @@ public void process(Exchange exchange) {
});
List<S3Object> resp = ex.getMessage().getBody(List.class);
- assertEquals(2, resp.size(), "Should create 2 files for 2 different timestamp windows");
-
- // Verify files have timestamp-based naming
- boolean foundWindow1File = false;
- boolean foundWindow2File = false;
-
- for (S3Object s3Object : resp) {
- String key = s3Object.key();
- if (key.contains("20240101_0800_0800-0805")) {
- foundWindow1File = true;
- } else if (key.contains("20240101_0805_0805-0810")) {
- foundWindow2File = true;
- }
- }
-
- assertTrue(foundWindow1File, "Should find file for first timestamp window");
- assertTrue(foundWindow2File, "Should find file for second timestamp window");
+ // batchMessageNumber=5: each window is uploaded twice, and the second upload of a window gets its own object
+ assertEquals(List.of(
Review Comment:
Nit: on a surefire rerun the bucket still holds the four objects of the first run, and the new producer lists them and continues with `-2` and `-3`, so this exact list fails on the rerun. Emptying the bucket in a `@BeforeEach` (like the ITs in #28137) avoids that, here and in the EdgeCases IT.
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25785: camel-aws2-s3 - a timestamp window uploaded more than once is stored in more than one object [camel]
Posted by "allthingssecurity (via GitHub)" <gi...@apache.org>.
allthingssecurity commented on PR #28136:
URL: https://github.com/apache/camel/pull/28136#issuecomment-6285513385
Thanks for the review. I went with option (a), in 00f39093950e:
- 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_
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25785: camel-aws2-s3 - a timestamp window uploaded more than once is stored in more than one object [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus merged PR #28136:
URL: https://github.com/apache/camel/pull/28136
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25785: camel-aws2-s3 - a timestamp window uploaded more than once is stored in more than one object [camel]
Posted by "allthingssecurity (via GitHub)" <gi...@apache.org>.
allthingssecurity commented on PR #28136:
URL: https://github.com/apache/camel/pull/28136#issuecomment-6285747707
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_
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25785: camel-aws2-s3 - a timestamp window uploaded more than once is stored in more than one object [camel]
Posted by "gnodet-bot (via GitHub)" <gi...@apache.org>.
gnodet-bot commented on code in PR #28136:
URL: https://github.com/apache/camel/pull/28136#discussion_r4320237872
##########
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:
⚠️ **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`.
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25785: camel-aws2-s3 - a timestamp window uploaded more than once is stored in more than one object [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25785: camel-aws2-s3 - a timestamp window uploaded more than once is stored in more than one object [camel]
Posted by "allthingssecurity (via GitHub)" <gi...@apache.org>.
allthingssecurity commented on code in PR #28136:
URL: https://github.com/apache/camel/pull/28136#discussion_r4322378380
##########
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:
Done in 00f39093950e: the map is insertion-ordered now (`new LinkedHashMap<>()` with `removeEldestEntry`), so a lookup no longer reorders it.
_Claude Code on behalf of allthingssecurity_
##########
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:
Right, removed in 00f39093950e.
_Claude Code on behalf of allthingssecurity_
##########
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:
Added in 00f39093950e: 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_
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org