shashank created CAMEL-26107:
--------------------------------
Summary: 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
{{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)
