allthingssecurity opened a new pull request, #28638: URL: https://github.com/apache/camel/pull/28638
# Description [CAMEL-26107](https://issues.apache.org/jira/browse/CAMEL-26107) `WriteAheadResumeStrategy` keeps the offsets that the wrapped resume strategy has not confirmed yet in a log file, and is meant to replay them (the entries in state `NEW` or `FAILED`) on the next start. After a restart nothing is replayed: - `start()` creates `new LogWriter(logFile, flushPolicy)`, and the `LogWriter` constructor opens the file with `CREATE, WRITE, TRUNCATE_EXISTING` and writes a new header. - `loadCache()` reads the file with a `LogReader` and replays the `NEW`/`FAILED` entries. - Camel always calls `start()` first. `ResumableProcessor.doStart` starts the strategy, and the processors of a route start before its consumer. The consumers call `loadCache()` in their own `doStart`: cassandraql, couchbase, couchdb and the mongodb change streams consumer through `ResumeStrategyHelper.resume`, file and aws2-kinesis directly. The `Splitter` (`doInit`) also starts the strategy before `loadCache()`. So `loadCache()` reads a file with only the 12-byte header, and the offsets that were in flight when the JVM stopped are lost without any log line. I checked it with a test: three `NEW` records, 135 bytes before `start()`, 12 bytes after, nothing replayed; with the 135 bytes written back after `start()`, all three are replayed. Once the log is read, `LogReader` has to cope with the file a crash leaves. It did not: a record cut short at the end of the file throws `BufferUnderflowException` when the record spans two reads of its 512 KiB buffer, a negative slot length throws `NegativeArraySizeException`, and a file shorter than the header throws `BufferUnderflowException` from the constructor. Changes: - `start()` reads the unconfirmed entries of the existing log before it creates the writer. It then appends them to the new log, so they are not lost if the JVM stops again before `loadCache()`. This keeps the writer as it is: a new log in `TransactionLog`'s layout, with nothing to know about the old file. Opening the writer without truncation would have meant teaching `LogWriter` and `TransactionLog` where the old records end and which of them are still live, and the next roll-over rewrites the file from memory anyway. - Only the last entry of each key in the log counts, as the wrapped strategy keeps the last offset of each key: it is carried and replayed if it is `NEW` or `FAILED`, and an earlier entry of the same key is not. Replaying every unconfirmed entry would let an old `FAILED` offset replace a later offset of the same key that the wrapped strategy confirmed (the consumer would then resume from the older offset), and with a wrapped strategy that never calls the callback (mongodb, the transient strategy) every offset of a run would be carried and replayed again at each start, until the log rolls over. - `loadCache()` replays these entries through the same path as a new offset (`tryUpdateDelegate` with the entry's `CachedEntryInfo`), so the callback of the wrapped strategy marks each one `PROCESSED` or `FAILED` at its position in the new log. A `FAILED` or still unconfirmed entry is replayed again on the next start. The replay no longer writes through `updateState(PersistedLogEntry, ...)`, and the three private methods for persisted entries are removed. - A replay that throws (the wrapped strategy fails, or the adapter cannot deserialize the key) is logged with a WARN and the other entries are still replayed; an entry whose key or value deserializes to `null` is not replayed, with a WARN. These entries stay in the log. Otherwise one such entry would fail the start of the route at every restart. - `loadCache()` no longer calls `logWriter.reset()` when nothing was replayed. On main this was a no-op, as the file had just been truncated (unless an offset was appended between `start()` and `loadCache()`, which it then dropped). Now it would drop entries that the adapter cannot deserialize (they are kept in the log, with a WARN) and leave `TransactionLog` pointing at records that are no longer in the file. - `LogReader`: a `fill(bytes)` helper reads until the buffer holds what is needed. An incomplete record or header now ends in `InvalidRecordException` (or `IllegalArgumentException` for the header, as for an empty read before), also when the record spans two buffer reads. A negative slot length also ends in `InvalidRecordException`. A record cut inside its first 4 bytes, which was silently taken as the end of the file, now ends in `InvalidRecordException` too. - `start()` stops reading at such a record and logs a WARN. The entries before it are replayed and the route starts. A file with another format name or version is replaced by a new log, as before, with a WARN. - A log that holds as many entries as it keeps before it rolls over (524,288, `LogWriter.DEFAULT_CAPACITY`) may have rolled over. After a roll-over the file holds the newer layer before the older one and has no layer information, so the last entry of a key in the file can be an older offset of the key, and replaying it would make the consumer resume from it (with `MongoDbResumeStrategy`, which never confirms, about 524,288 events back). Such a log is not replayed, as on main: a WARN names the number of unconfirmed entries, which are discarded with it. The strategy has a package-private `maxRecordCount` (the capacity it gives the `LogWriter`) so that the test can use a log of 3 entries. - Folded in from the review of CAMEL-26088 (#28596), as offered there: - `TransactionLog` checks the slot of the entry: an entry can be updated while its slot still holds it (the same `LayerInfo` object). This replaces the index arithmetic, and it also covers a case this PR makes visible: a route stop/start keeps the strategy and creates a new `LogWriter` and `TransactionLog`, and a callback of the previous start that reports late carried a `LayerInfo` that the arithmetic accepted for whatever entry is now at that index of the new log (with this PR, a carried entry or a new offset). Comparing the layer number would not catch that, as the new log starts at layer 0 again. - `LogWriter.updateState` does not write an entry without a position (`position < 0`). With the slot check this is defensive: the layer information of an entry whose write failed is never given to a caller. - A javadoc line on `EntryInfo.CachedEntryInfo.getPosition()`: the position where the entry was written when the information was created. - A test for a shorter record at the end of a layer, with nothing written after it. - `LogReader.doReadEntry` gives a `PersistedLogEntry` the position where its record starts, not the channel position after the buffer was filled, so `LogWriter.updateState(PersistedLogEntry, ...)` (public, no caller in Camel) writes over that record. - Upgrade guide 4.24: users now get the replay after a restart, which is the documented intent, and the roll-over case. - Backport: camel-4.23.x, 4.22.x, 4.18.x and 4.14.x have the same camel-wal Java sources and the same `start()` before `loadCache()` in `ResumableProcessor` and `ResumeStrategyHelper` (read with `git show`, not built). Callers checked: - `ResumableProcessor.doStart` (start), `ResumeStrategyHelper.resume` (loadCache, then the adapter's `resume()`), and `Splitter.doInit` (start, then loadCache): the replay happens in `loadCache()`, before the adapter resumes, as it was meant to. - camel-kafka's consumer does not call `loadCache()`: with this strategy around a Kafka strategy, the entries stay in the log and are not replayed (on main they were dropped). - The delegate gets the key and value that the adapter's `deserializeKey`/`deserializeValue` read back, wrapped in `OffsetKeys.of`/`Offsets.of`: the same values as in the original run for the types `Serializable` writes (`Long`, `String`, `File`). `SingleNodeKafkaResumeStrategy` adds the replayed offset to the adapter's cache in `updateLastOffset`, so `ResumeStrategyHelper` resumes from it; its callback runs later, on the producer thread. - `KinesisResumeStrategy` throws `UnsupportedOperationException` from `updateLastOffset(key, offset, callback)`, which is what this strategy calls, so this strategy cannot store offsets through it on main either; the replay of such an entry is logged and the consumer still starts. `MongoDbResumeStrategy` and `TransientResumeStrategy` never call the callback, so their entries stay `NEW` and the last offset of each key is replayed at every start. - Concurrency: the replay runs in `loadCache()` before the consumers that call it start polling, except aws2-kinesis, which calls it after its scheduler started (first poll after the initial delay). The callbacks of the replay update the log as the callbacks of new offsets do; `LogWriter` itself is thread safe since CAMEL-26088 (#28596). - No file, an empty file, and a header-only file: nothing to replay, and a new log. There has only ever been version 1 of the format (`camel-wa`, since 3.20), so no older format exists. A file with a different name or version is not read. - No other module has tests that use `WriteAheadResumeStrategy` (`git grep`), so I ran only the camel-wal tests. Tests: - New `WriteAheadResumeStrategyRestartTest`, 26 tests. Each calls `start()` and then `loadCache()` on a new strategy, as Camel does, with a recording wrapped strategy that confirms or fails through the callback: - `restartReplaysTheNewAndFailedEntriesButNotTheProcessedOnes`: `NEW`, `PROCESSED`, `FAILED` and `NEW` entries; the three unconfirmed ones are replayed and then `PROCESSED` in the log, and a second restart replays nothing; - `onlyTheLastOffsetOfEachKeyIsReplayed`: a `FAILED` offset followed by a `PROCESSED` one of the same key is not replayed, of two `NEW` offsets only the later, and a `FAILED` offset after a `PROCESSED` one is; - `entriesOfADelegateThatNeverConfirmsDoNotPileUpAcrossRestarts`: two restarts each replay the last offset of each key, and the log keeps one entry per key; - `entryThatCannotBeReplayedDoesNotStopTheOthersNorTheStart`: a key too short for its type, a key of an unknown type and a wrapped strategy that throws; the other entry is replayed, the route starts, and the failed one is replayed on the next start; - `entryWhoseReplayFailsIsReplayedOnTheNextStart`; - `unconfirmedEntriesStayInTheLogUntilTheyAreReplayed` (stopped after `start()`, before `loadCache()`); - `truncatedLastRecordIsSkippedAndThePreviousOnesAreReplayed`: the last record is cut in the state, the key metadata, the key, the value length and the value (5 cases); - `truncatedLastRecordAcrossTheReadBufferIsSkipped`: the last record starts in the first 512 KiB read and the file ends 2 bytes after it (11,897 records replayed); - `invalidLastRecordDoesNotFailTheStart`: key length -5, value length -1, key length `Integer.MAX_VALUE` (larger than the buffer), key length 1000 (past the end of the file), state 99; - `startsWithNothingToReplay`: no file, empty, header only, shorter than a header, another format name, another version; - `logThatMayHaveRolledOverIsNotReplayed` (a log of capacity 3 with 4 offsets of one key: the file is `[4, 2, 3]`, and without the check `key-0=3` is replayed) and `logThatHasNotRolledOverIsReplayed` (a control: it passes without the check too); - `callbackOfThePreviousStartDoesNotUpdateAnEntryOfTheNewLog`: the strategy is stopped and started again while a callback is pending; the late callback no longer marks the entry now at its index in the new log. - `TransactionLogTest` now tests through `add`/`update` (same layer, rolled over, two layers behind, previous layer still there, an entry of another log). New `LogReaderEntryPositionTest` (2 tests) and `LogWriterUpdateAfterRollOverPositionTest#shorterRecordAtTheEndOfALayerKeepsFileReadable`. - Negative run on main (origin/main 7b58ce0cdaf6 sources, the first 23 tests of the class, before the three tests added for the roll-over and the late callback, which use the package-private `maxRecordCount` and do not compile on main; their checks are in the mutation bullet below): 17 of 23 fail, all with nothing replayed (`expected: <[key-0=0, key-2=2, key-3=3]> but was: <[]>`). The 6 `startsWithNothingToReplay` cases pass on main and are the controls. - With the strategy change but main's `LogReader`: 4 fail. The negative lengths fail with `NegativeArraySizeException: -5` / `-1`, and the header too short and the record across the buffer fail with `BufferUnderflowException`, all from `start()`. - The three tests for the last offset of a key and for a failing replay fail without those two changes (all unconfirmed entries replayed: `but was: <[key-0=0, key-1=2, key-1=3, key-2=5]>`; the start fails with `BufferUnderflowException`). - The checks for the folded-in changes, each mutation run alone: the old `LogReader` position fails both `LogReaderEntryPositionTest` tests; a slot check that compares the layer number fails `TransactionLogTest#testCannotUpdateRecordOfAnotherLog` and `callbackOfThePreviousStartDoesNotUpdateAnEntryOfTheNewLog` (`key-2=FAILED`); without the roll-over check `logThatMayHaveRolledOverIsNotReplayed` fails (`key-0=3` replayed); without the truncate after a roll-over (from #28596) the shorter-record tests fail. - camel-wal with the fix, on main with CAMEL-26088 (#28596): 57 run, 0 failures. - The formatter and import-sort plugins ran. I built offline with `-Djackson2-version=2.22.3 -Djackson3-version=3.2.3` and `-Dasm-version=9.11` (the versions on main are not in my local repository). Not changed: - A log that rolled over and was then cut short, because the system stopped while the log rewrote itself after the roll-over (each append after a roll-over rewrites the older layer), holds fewer entries than the capacity, with the newer ones first. It is still replayed, and the last entry of a key in it can be an older offset, so the consumer can process again the messages after that offset (duplicates, nothing lost). The guide says so. Closing this needs the file to say it rolled over (a marker in the header, or a sequence number per record), or the writer to keep the file in write order; I'll file a follow-up JIRA for it, linked to this one. (After a roll-over every append rewrites up to 524,287 records, so a run that rolls over is very slow already; that is a second reason for the follow-up.) - The writer still truncates the file before the unconfirmed entries are appended again, so a crash in between (a few writes during the route start) loses them. Writing a new file and renaming it would close that window, at the cost of a bigger change to `LogWriter`. # 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-26107](https://issues.apache.org/jira/browse/CAMEL-26107). # 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 camel-wal and ran its tests, including the formatter and import-sort plugins. No generated file changes. 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
