[
https://issues.apache.org/jira/browse/CAMEL-26107?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Work on CAMEL-26107 started by shashank.
----------------------------------------
> camel-wal - WriteAheadResumeStrategy never replays the unconfirmed offsets
> after a restart: start() truncates the log before loadCache() reads it
> -------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-26107
> URL: https://issues.apache.org/jira/browse/CAMEL-26107
> Project: Camel
> Issue Type: Bug
> Reporter: shashank
> Assignee: shashank
> Priority: Major
>
> {{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 (main 7b58ce0cdaf6):
> * {{start()}} creates {{new LogWriter(logFile, flushPolicy)}}; 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 calls {{start()}} first: {{ResumableProcessor.doStart}} starts the
> strategy, and the consumers (cassandraql, couchbase, couchdb, mongodb change
> streams, through {{ResumeStrategyHelper.resume}}; file and aws2-kinesis call
> {{loadCache()}} directly) call {{loadCache()}} in their own {{doStart}}. The
> {{Splitter}} also starts the strategy before {{loadCache()}}.
> So {{loadCache()}} reads a file with only the 12-byte header. Checked with a
> test: three {{NEW}} records, 135 bytes before {{start()}}, 12 bytes after,
> nothing replayed; with the bytes written back after {{start()}}, all three
> are replayed. The offsets that were in flight when the JVM stopped are lost
> silently, which is what the component is for.
> Once the log is read, two {{LogReader}} cases become reachable: a record cut
> short at the end of the file (a crash while writing it) throws
> {{BufferUnderflowException}} when the record spans two reads of the reader's
> 512 KiB buffer, and a negative slot length throws
> {{NegativeArraySizeException}}; a file shorter than the header throws
> {{BufferUnderflowException}} from the constructor.
> *Fix*: {{start()}} reads the unconfirmed entries of the existing log before
> it creates the writer, and appends them to the new log, so they survive
> another crash before the replay. As the wrapped strategy keeps the last
> offset of each key, only the last entry of each key in the log counts: it is
> replayed if it is {{NEW}} or {{FAILED}}, and an earlier entry of the same key
> is not (an old {{FAILED}} offset must not replace a later offset of the key
> that was confirmed, and the entries of a wrapped strategy that never calls
> back, such as the mongodb one, must not pile up across restarts).
> {{loadCache()}} replays them through the same path as a new offset, so the
> callback of the wrapped strategy marks each one {{PROCESSED}} or {{FAILED}}
> at its position in the new log (the replay no longer writes through
> {{updateState(PersistedLogEntry, ...)}}). A replay that fails, or an entry
> the adapter cannot deserialize, is logged with a WARN and does not stop the
> other entries nor the start of the route; the entry stays in the log.
> {{LogReader}} throws {{InvalidRecordException}} for an incomplete record or
> header (also across two buffer reads) and for a negative slot length;
> {{start()}} logs a WARN and replays the entries before such a record, and a
> file of another format is replaced by a new log with a WARN, as before. The
> route start does not fail on a damaged log.
> *Tests*: new {{WriteAheadResumeStrategyRestartTest}}, 23 tests through
> {{start()}} + {{loadCache()}}: NEW and FAILED replayed, PROCESSED not; only
> the last offset of each key; no pile-up with a wrapped strategy that never
> confirms; an entry that cannot be replayed does not stop the others nor the
> start; a failed replay is replayed on the next start; entries stay in the log
> until replayed; a last record truncated at 5 places, and across the reader
> buffer; a negative, huge or too long slot length and an invalid state; no
> file, empty, header only, shorter than a header, other format, other version.
> 17 fail on main ({{expected: <[key-0=0, key-2=2, key-3=3]> but was: <[]>}}),
> the 6 'nothing to replay' cases pass on main. With the strategy change but
> the old {{LogReader}}, 4 fail with {{NegativeArraySizeException}} /
> {{BufferUnderflowException}}. camel-wal: 46 tests, 0 failures; merged with
> the CAMEL-26088 branch: 50, 0 failures.
> *Upgrade guide*: 4.24 entry, as users now get the replay after a restart (the
> documented intent).
> *Not changed*: after a roll-over the log holds the newer entries before the
> older ones, so the last entry of a key and the replay order are those of the
> file, not the order of the offsets (no layer information in the file).
> {{LogReader}} still gives a {{PersistedLogEntry}} the channel position after
> the buffer fill; no caller writes at it any more.
> *Affected*: main, camel-4.23.x, camel-4.22.x, camel-4.18.x and camel-4.14.x
> (same camel-wal Java sources, and {{start()}} before {{loadCache()}} in
> {{ResumableProcessor}}/{{ResumeStrategyHelper}}). Read, not run on the
> branches.
> _Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
