This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new f4e22922dec4 CAMEL-26107: camel-wal - replay the unconfirmed entries
of the previous run after a restart (#28638)
f4e22922dec4 is described below
commit f4e22922dec4cae27577c41639ea56b5ddd0dc45
Author: allthingssecurity <[email protected]>
AuthorDate: Thu Oct 29 13:32:43 2026 +0530
CAMEL-26107: camel-wal - replay the unconfirmed entries of the previous run
after a restart (#28638)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../org/apache/camel/component/wal/EntryInfo.java | 11 +
.../org/apache/camel/component/wal/LogReader.java | 55 +-
.../org/apache/camel/component/wal/LogWriter.java | 7 +
.../apache/camel/component/wal/TransactionLog.java | 54 +-
.../component/wal/WriteAheadResumeStrategy.java | 243 ++++++---
.../component/wal/LogReaderEntryPositionTest.java | 105 ++++
.../LogWriterUpdateAfterRollOverPositionTest.java | 18 +
.../camel/component/wal/TransactionLogTest.java | 72 ++-
.../wal/WriteAheadResumeStrategyRestartTest.java | 587 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 21 +
10 files changed, 1022 insertions(+), 151 deletions(-)
diff --git
a/components/camel-wal/src/main/java/org/apache/camel/component/wal/EntryInfo.java
b/components/camel-wal/src/main/java/org/apache/camel/component/wal/EntryInfo.java
index c4fd076b0393..591ce2d63bfa 100644
---
a/components/camel-wal/src/main/java/org/apache/camel/component/wal/EntryInfo.java
+++
b/components/camel-wal/src/main/java/org/apache/camel/component/wal/EntryInfo.java
@@ -46,6 +46,17 @@ public class EntryInfo {
return layerInfo;
}
+ /**
+ * The position where the entry was written when this information was
created. A roll-over can write the entry
+ * again at another position later; an update of its state writes
where the entry is then, not at this position.
+ *
+ * @return the position where the entry was written when this
information was created
+ */
+ @Override
+ public long getPosition() {
+ return super.getPosition();
+ }
+
}
/**
diff --git
a/components/camel-wal/src/main/java/org/apache/camel/component/wal/LogReader.java
b/components/camel-wal/src/main/java/org/apache/camel/component/wal/LogReader.java
index 24a43c4b9881..ebdc6275eb69 100644
---
a/components/camel-wal/src/main/java/org/apache/camel/component/wal/LogReader.java
+++
b/components/camel-wal/src/main/java/org/apache/camel/component/wal/LogReader.java
@@ -92,6 +92,10 @@ public class LogReader implements AutoCloseable {
LOG.trace("Read {} bytes from the file channel", bytesRead);
ioBuffer.flip();
+ if (!fill(Header.BYTES)) {
+ throw new IllegalArgumentException("The file does not contain a
valid header");
+ }
+
byte[] name = new byte[Header.FORMAT_NAME_SIZE];
ioBuffer.get(name, 0, Header.FORMAT_NAME_SIZE);
LOG.trace("File format name: '{}'", new String(name));
@@ -105,8 +109,10 @@ public class LogReader implements AutoCloseable {
/**
* Read an entry from the file.
*
- * @return A log entry from the file or null when reaching the
end-of-file or if the file is empty
- * @throws IOException if unable to read the entry
+ * @return A log entry from the file or null when
reaching the end-of-file or if the file is
+ * empty
+ * @throws IOException if unable to read the entry
+ * @throws InvalidRecordException if the entry is incomplete (i.e.: the
file ends within it) or malformed
*/
public PersistedLogEntry readEntry() throws IOException {
if (header == null) {
@@ -137,10 +143,12 @@ public class LogReader implements AutoCloseable {
* @throws IOException if unable to read the entry
*/
private PersistedLogEntry doReadEntry() throws IOException {
- if (ioBuffer.remaining() < Integer.BYTES) {
- if (!reload()) {
- return null;
- }
+ // the position of the record in the file: the channel is ahead of the
buffer by the bytes not read from it yet
+ final long recordPosition = fileChannel.position() -
ioBuffer.remaining();
+
+ // there is at least one byte of the record in the buffer
+ if (!fill(Integer.BYTES)) {
+ throw new InvalidRecordException("A record is incomplete or
malformed");
}
int state = ioBuffer.getInt();
@@ -148,7 +156,7 @@ public class LogReader implements AutoCloseable {
Slot keySlot = readSlot();
Slot valueSlot = readSlot();
- EntryInfo entryInfo =
EntryInfo.createForPersisted(fileChannel.position());
+ EntryInfo entryInfo = EntryInfo.createForPersisted(recordPosition);
return new PersistedLogEntry(
entryInfo, LogEntry.EntryState.fromInt(state),
keySlot.metadata, keySlot.data,
@@ -165,22 +173,22 @@ public class LogReader implements AutoCloseable {
Slot slot = new Slot();
// The buffer needs to have enough space for the metadata and length.
- if (ioBuffer.remaining() < (Integer.BYTES * 2)) {
- if (!reload()) {
- throw new InvalidRecordException("A data slot within a record
is incomplete or malformed");
- }
+ if (!fill(Integer.BYTES * 2)) {
+ throw new InvalidRecordException("A data slot within a record is
incomplete or malformed");
}
slot.metadata = ioBuffer.getInt();
slot.length = ioBuffer.getInt();
+ if (slot.length < 0) {
+ throw new InvalidRecordException("A data slot within a record has
an invalid length: " + slot.length);
+ }
+
if (ioBuffer.capacity() < slot.length) {
throw new BufferTooSmallException(ioBuffer.capacity(),
slot.length);
}
- if (ioBuffer.remaining() < slot.length) {
- if (!reload()) {
- throw new InvalidRecordException("A data slot within a record
is incomplete or malformed");
- }
+ if (!fill(slot.length)) {
+ throw new InvalidRecordException("A data slot within a record is
incomplete or malformed");
}
slot.data = new byte[slot.length];
@@ -189,6 +197,23 @@ public class LogReader implements AutoCloseable {
return slot;
}
+ /**
+ * Makes sure that the intermediate buffer holds at least the given number
of bytes, reading more from the file if
+ * needed
+ *
+ * @param bytes the number of bytes needed
+ * @return true if the buffer holds them or false if the file
ends before
+ * @throws IOException in case of lower-level I/O errors
+ */
+ private boolean fill(int bytes) throws IOException {
+ while (ioBuffer.remaining() < bytes) {
+ if (!reload()) {
+ return false;
+ }
+ }
+ return true;
+ }
+
/**
* Reloads data into the intermediate buffer, compacting it on the process
*
diff --git
a/components/camel-wal/src/main/java/org/apache/camel/component/wal/LogWriter.java
b/components/camel-wal/src/main/java/org/apache/camel/component/wal/LogWriter.java
index bcbc8e887b11..ccb927759205 100644
---
a/components/camel-wal/src/main/java/org/apache/camel/component/wal/LogWriter.java
+++
b/components/camel-wal/src/main/java/org/apache/camel/component/wal/LogWriter.java
@@ -267,6 +267,13 @@ public final class LogWriter implements AutoCloseable {
final TransactionLog.EntryContainer container =
transactionLog.update(layerInfo, state);
if (container != null) {
+ if (container.position < 0) {
+ // the entry was never written, so there is nothing to
update in the file. Defensive: the layer
+ // information of an entry whose append failed is not
given to a caller
+ LOG.warn("Not updating the state of a record that was not
written to the transaction log: {}",
+ layerInfo);
+ return;
+ }
// not the position of the entry info: a roll-over may have
written the entry again elsewhere
persist(layerInfo, container.logEntry, container.position);
}
diff --git
a/components/camel-wal/src/main/java/org/apache/camel/component/wal/TransactionLog.java
b/components/camel-wal/src/main/java/org/apache/camel/component/wal/TransactionLog.java
index b647c3af878d..d885cd844476 100644
---
a/components/camel-wal/src/main/java/org/apache/camel/component/wal/TransactionLog.java
+++
b/components/camel-wal/src/main/java/org/apache/camel/component/wal/TransactionLog.java
@@ -74,8 +74,8 @@ class TransactionLog {
* it's entry.
*/
static class EntryContainer {
- LayerInfo layerInfo;
- LogEntry logEntry;
+ final LayerInfo layerInfo;
+ final LogEntry logEntry;
/**
* The current position of the entry in the file: a roll-over writes
the entries of the previous layer again,
* after the newer ones
@@ -128,42 +128,20 @@ class TransactionLog {
}
/**
- * Given the information about an entry, it determines whether it can be
updated or not
+ * Determines whether the entry with the given layer information can be
updated: its slot in the buffer still holds
+ * it, so no entry of a newer layer has replaced it. The layer information
of an entry of another transaction log,
+ * such as the log of a writer that was closed when the route stopped, is
never in a slot of this one.
*
- * @param transactionLogLayer the layer in the transaction log
- * @param transactionLogIndex the index in the transaction log
- * @param entryInfoLayer the layer for the entry
- * @param entryInfoIndex the index for the entry
- * @return true if it can be updated or false otherwise
- */
- static boolean canUpdate(int transactionLogLayer, int transactionLogIndex,
int entryInfoLayer, int entryInfoIndex) {
- if (transactionLogLayer == entryInfoLayer) {
- // Must make sure to not update beyond the current index
- if (transactionLogIndex >= entryInfoIndex) {
- return true;
- }
- }
-
- // The current layer has replaced the entries before the current index
(the index of the next entry to add).
- // The entries from there on are still those of the previous layer,
and those of older layers are all gone.
- if (transactionLogLayer == entryInfoLayer + 1) {
- if (transactionLogIndex <= entryInfoIndex) {
- return true;
- }
- }
-
- return false;
- }
-
- /**
- * Given the layer information for an entry, it determines if it can be
updated via
- * {@link TransactionLog#canUpdate(int, int, int, int)}
- *
- * @param layerInfo the layer information
+ * @param layerInfo the layer information of the entry, as {@link
#add(LogEntry)} created it
* @return true if it can be updated or false otherwise
*/
private boolean canUpdate(LayerInfo layerInfo) {
- return canUpdate(currentLayer, currentIndex, layerInfo.getLayer(),
layerInfo.getIndex());
+ final int index = layerInfo.getIndex();
+ if (index < 0 || index >= logEntries.length) {
+ return false;
+ }
+ final EntryContainer container = logEntries[index];
+ return container != null && container.layerInfo == layerInfo;
}
/**
@@ -171,7 +149,8 @@ class TransactionLog {
*
* @param layerInfo the layer information for the entry
* @param state the state to update the entry to
- * @return the container of the updated entry or null if the
record was rolled-over and discarded
+ * @return the container of the updated entry or null if the
record was rolled-over and discarded (or
+ * belongs to another transaction log, such as the log
of a writer closed when the route stopped)
*/
public EntryContainer update(LayerInfo layerInfo, LogEntry.EntryState
state) {
if (layerInfo == null) {
@@ -192,12 +171,13 @@ class TransactionLog {
return container;
} else {
if (LOG.isDebugEnabled()) {
- LOG.debug("Discarded a record because it has been rolled-over.
Record layer: {}", layerInfo);
+ LOG.debug("Discarded a record because it has been rolled-over,
or it is not in this log. Record layer: {}",
+ layerInfo);
}
if (state == LogEntry.EntryState.FAILED) {
LOG.warn(
- "An update of failed record was discarded because it
has been rolled over (it may have taken too long to report back)");
+ "An update of failed record was discarded because it
has been rolled over (it may have taken too long to report back), or because it
belongs to the log of a previous start of the route");
}
}
diff --git
a/components/camel-wal/src/main/java/org/apache/camel/component/wal/WriteAheadResumeStrategy.java
b/components/camel-wal/src/main/java/org/apache/camel/component/wal/WriteAheadResumeStrategy.java
index f184a0b64934..d08d133d1a85 100644
---
a/components/camel-wal/src/main/java/org/apache/camel/component/wal/WriteAheadResumeStrategy.java
+++
b/components/camel-wal/src/main/java/org/apache/camel/component/wal/WriteAheadResumeStrategy.java
@@ -20,11 +20,18 @@ package org.apache.camel.component.wal;
import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
import org.apache.camel.CamelContext;
import org.apache.camel.CamelContextAware;
import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.component.wal.exceptions.BufferTooSmallException;
+import org.apache.camel.component.wal.exceptions.InvalidRecordException;
import org.apache.camel.resume.Deserializable;
import org.apache.camel.resume.Offset;
import org.apache.camel.resume.OffsetKey;
@@ -70,18 +77,34 @@ public class WriteAheadResumeStrategy implements
ResumeStrategy, CamelContextAwa
}
}
+ /**
+ * An entry of the previous run that was not confirmed by the delegate
resume strategy, and where it is in the new
+ * transaction log
+ */
+ private record UnconfirmedEntry(LogEntry logEntry,
EntryInfo.CachedEntryInfo entryInfo) {
+ }
+
+ /**
+ * The key of a log entry: its serialized key and its metadata (the type
of the key)
+ */
+ private record EntryKey(int keyMetadata, ByteBuffer key) {
+ }
+
private static final Logger LOG =
LoggerFactory.getLogger(WriteAheadResumeStrategy.class);
private File logFile;
private LogWriter logWriter;
+ private List<UnconfirmedEntry> unconfirmedEntries = List.of();
private ResumeStrategy resumeStrategy;
private WriteAheadResumeStrategyConfiguration resumeStrategyConfiguration;
private CamelContext camelContext;
+ // the number of entries the log keeps before it rolls over
+ private final int maxRecordCount;
/**
* Creates a new write-ahead resume strategy
*/
public WriteAheadResumeStrategy() {
-
+ this.maxRecordCount = LogWriter.DEFAULT_CAPACITY;
}
/**
@@ -90,7 +113,16 @@ public class WriteAheadResumeStrategy implements
ResumeStrategy, CamelContextAwa
* @param resumeStrategyConfiguration the configuration to use for this
strategy instance
*/
public WriteAheadResumeStrategy(WriteAheadResumeStrategyConfiguration
resumeStrategyConfiguration) {
+ this(resumeStrategyConfiguration, LogWriter.DEFAULT_CAPACITY);
+ }
+
+ /**
+ * Creates a new write-ahead resume strategy whose log keeps the given
number of entries before it rolls over
+ * (visible for testing)
+ */
+ WriteAheadResumeStrategy(WriteAheadResumeStrategyConfiguration
resumeStrategyConfiguration, int maxRecordCount) {
this.resumeStrategyConfiguration = resumeStrategyConfiguration;
+ this.maxRecordCount = maxRecordCount;
}
@Override
@@ -143,29 +175,6 @@ public class WriteAheadResumeStrategy implements
ResumeStrategy, CamelContextAwa
}
}
- /**
- * Handles the result of an offset update for persisted entries (i.e.:
those already saved to permanent storage)
- *
- * @param entry the information about the entry that was updated
- * @param t a instance of any throwable class that was thrown by the
delegate resume strategy during update, if
- * none, then can be null
- */
- private void handleResult(PersistedLogEntry entry, Throwable t) {
- try {
- if (t == null) {
- logWriter.updateState(entry, LogEntry.EntryState.PROCESSED);
- } else {
- logWriter.updateState(entry, LogEntry.EntryState.FAILED);
- }
- } catch (IOException e) {
- if (t == null) {
- LOG.error("Unable to update state: {}", e.getMessage(), e);
- } else {
- LOG.error("Unable to mark the record as failed: {}",
e.getMessage(), e);
- }
- }
- }
-
@Override
public void updateLastOffset(OffsetKey<?> offsetKey, Offset<?>
offsetValue) throws Exception {
updateLastOffset(offsetKey, offsetValue, null);
@@ -224,30 +233,6 @@ public class WriteAheadResumeStrategy implements
ResumeStrategy, CamelContextAwa
}
}
- /**
- * Tries to update the offset in the delegate strategy, ensuring the entry
on log reflects the success or failure of
- * the update request
- *
- * @param offsetKey the offset key to update
- * @param offsetValue the offset value to update
- * @param entry the entry being updated
- * @param updateCallBack a callback to be executed after the updated has
occurred (null if not available)
- * @throws Exception
- */
- private void tryUpdateDelegate(
- OffsetKey<?> offsetKey, Offset<?> offsetValue, PersistedLogEntry
entry, UpdateCallBack updateCallBack)
- throws Exception {
- try {
- UpdateCallBack delegateCallback = resolveUpdateCallBack(entry,
updateCallBack);
-
- resumeStrategy.updateLastOffset(offsetKey, offsetValue,
delegateCallback);
- } catch (Throwable throwable) {
- logWriter.updateState(entry, LogEntry.EntryState.FAILED);
-
- throw throwable;
- }
- }
-
private UpdateCallBack resolveUpdateCallBack(EntryInfo.CachedEntryInfo
entryInfo, UpdateCallBack updateCallBack) {
if (updateCallBack == null) {
return t -> handleResult(entryInfo, t);
@@ -256,51 +241,133 @@ public class WriteAheadResumeStrategy implements
ResumeStrategy, CamelContextAwa
}
}
- private UpdateCallBack resolveUpdateCallBack(PersistedLogEntry entry,
UpdateCallBack updateCallBack) {
- if (updateCallBack == null) {
- return t -> handleResult(entry, t);
- } else {
- return new DelegateCallback(updateCallBack, t ->
handleResult(entry, t));
- }
- }
-
@Override
public void loadCache() throws Exception {
LOG.debug("Loading cache for the delegate strategy");
resumeStrategy.loadCache();
LOG.debug("Done loading cache for the delegate strategy");
+ final List<UnconfirmedEntry> entries = unconfirmedEntries;
+ unconfirmedEntries = List.of();
+ if (entries.isEmpty()) {
+ return;
+ }
+
+ final ResumeAdapter adapter = resumeStrategy.getAdapter();
+ if (!(adapter instanceof Deserializable deserializable)) {
+ LOG.warn("Not replaying the {} unconfirmed entries of the
transaction log {}: the resume adapter {} cannot "
+ + "deserialize them. They are kept in the log",
+ entries.size(), logFile, adapter);
+ return;
+ }
+
+ LOG.info("Replaying {} unconfirmed entries of the transaction log {}",
entries.size(), logFile);
+ for (UnconfirmedEntry entry : entries) {
+ // as for a new offset, a failure is logged and does not stop the
others (nor the start of the route)
+ try {
+ replay(deserializable, entry);
+ } catch (Exception e) {
+ LOG.warn("Unable to replay an unconfirmed entry of the
transaction log {}: it is kept in the log and "
+ + "replayed again on the next start. Reason: {}",
+ logFile, e.toString(), e);
+ }
+ }
+ LOG.trace("Finished replaying the unconfirmed entries");
+ }
+
+ private void replay(Deserializable deserializable, UnconfirmedEntry entry)
throws Exception {
+ Object oKey =
deserializable.deserializeKey(ByteBuffer.wrap(entry.logEntry().getKey()));
+ Object value =
deserializable.deserializeValue(ByteBuffer.wrap(entry.logEntry().getValue()));
+ if (oKey == null || value == null) {
+ LOG.warn("Not replaying an unconfirmed entry of the transaction
log {}: the resume adapter cannot "
+ + "deserialize its key or value. It is kept in the log",
+ logFile);
+ return;
+ }
+
+ tryUpdateDelegate(OffsetKeys.of(oKey), Offsets.of(value),
entry.entryInfo(), null);
+ }
+
+ /**
+ * Reads the entries that were not confirmed by the delegate resume
strategy from the transaction log of the
+ * previous run. As the delegate keeps the last offset of each key, only
the last entry of each key in the log
+ * counts: it is unconfirmed if it is in state NEW or FAILED. An earlier
entry of the same key is superseded, so a
+ * FAILED offset does not replace a later one that the delegate confirmed.
The log ends where a record is incomplete
+ * or invalid (i.e.: the last one, when the system crashed while writing
it): the entries before it are kept.
+ * <p/>
+ * A log that holds as many entries as it keeps before it rolls over may
have rolled over. Then the entries of the
+ * newer layer come first in the file, and the file has no layer
information, so the last entry of a key in the file
+ * may be an older offset of the key. Such a log is not replayed:
replaying an older offset would make the consumer
+ * resume from it. This does not cover a log that rolled over and was cut
short while an append rewrote it (the
+ * system crashed then): it holds fewer entries, the newer ones first.
+ *
+ * @param logFile the transaction log file, which may not exist
+ * @param maxRecordCount the number of entries the log keeps before it
rolls over
+ * @return the unconfirmed entries, in the order of the log
+ */
+ private static List<LogEntry> readUnconfirmedEntries(File logFile, int
maxRecordCount) {
+ final Map<EntryKey, LogEntry> lastEntries = new LinkedHashMap<>();
+ if (!logFile.isFile()) {
+ return List.of();
+ }
+
+ int count = 0;
+
try (LogReader reader = new LogReader(logFile)) {
+ final Header header = reader.getHeader();
+ if (header == null) {
+ return List.of();
+ }
+
+ if (!Header.FORMAT_NAME.equals(header.getFormatName().trim())
+ || header.getFileVersion() != Header.CURRENT_FILE_VERSION)
{
+ LOG.warn("Not replaying the transaction log {}: it has an
unknown format '{}' version {}, and a new log "
+ + "replaces it",
+ logFile, header.getFormatName().trim(),
header.getFileVersion());
+ return List.of();
+ }
- int updatedCount = 0;
- LOG.trace("Starting to read log entries");
PersistedLogEntry logEntry;
- do {
- logEntry = reader.readEntry();
- if (logEntry != null) {
- final LogEntry.EntryState entryState =
logEntry.getEntryState();
- if (entryState == LogEntry.EntryState.NEW || entryState ==
LogEntry.EntryState.FAILED) {
- final ResumeAdapter adapter =
resumeStrategy.getAdapter();
-
- if (adapter instanceof Deserializable) {
- Deserializable deserializable = (Deserializable)
adapter;
-
- Object oKey =
deserializable.deserializeKey(ByteBuffer.wrap(logEntry.getKey()));
- Object value =
deserializable.deserializeValue(ByteBuffer.wrap(logEntry.getValue()));
-
- tryUpdateDelegate(OffsetKeys.of(oKey),
Offsets.of(value), logEntry, null);
- updatedCount++;
- }
- }
- }
-
- } while (logEntry != null);
- LOG.trace("Finished reading log entries");
-
- if (updatedCount == 0) {
- logWriter.reset();
+ while ((logEntry = reader.readEntry()) != null) {
+ count++;
+ final EntryKey key = new EntryKey(logEntry.getKeyMetadata(),
ByteBuffer.wrap(logEntry.getKey()));
+ // keep the keys in the order of their last entry
+ lastEntries.remove(key);
+ lastEntries.put(key, logEntry);
+ }
+ } catch (InvalidRecordException | IllegalArgumentException |
BufferTooSmallException e) {
+ LOG.warn("Stopped reading the transaction log {} at an incomplete
or invalid record, the rest of the log is "
+ + "skipped: the {} entries before it are read. Reason:
{}",
+ logFile, count, e.toString());
+ } catch (IOException e) {
+ // such as no permission to read the file: not a record of the
log, but the log cannot be read further
+ LOG.warn("Cannot read the transaction log {}, the rest of the log
is skipped: the {} entries before the "
+ + "error are read. Reason: {}",
+ logFile, count, e.toString());
+ LOG.debug("Cannot read the transaction log {}", logFile, e);
+ }
+
+ final List<LogEntry> unconfirmed = unconfirmed(lastEntries.values());
+ if (count >= maxRecordCount && !unconfirmed.isEmpty()) {
+ LOG.warn("Not replaying the {} unconfirmed entries of the
transaction log {}: it holds {} entries, so it may "
+ + "have rolled over, and then the order of its entries is
not the order of the offsets. A new log "
+ + "replaces it, so these entries are discarded",
+ unconfirmed.size(), logFile, count);
+ return List.of();
+ }
+ return unconfirmed;
+ }
+
+ private static List<LogEntry> unconfirmed(Collection<LogEntry> entries) {
+ final List<LogEntry> unconfirmed = new ArrayList<>();
+ for (LogEntry entry : entries) {
+ final LogEntry.EntryState entryState = entry.getEntryState();
+ if (entryState == LogEntry.EntryState.NEW || entryState ==
LogEntry.EntryState.FAILED) {
+ unconfirmed.add(new LogEntry(
+ entryState, entry.getKeyMetadata(), entry.getKey(),
entry.getValueMetadata(), entry.getValue()));
}
}
+ return unconfirmed;
}
@Override
@@ -309,13 +376,23 @@ public class WriteAheadResumeStrategy implements
ResumeStrategy, CamelContextAwa
this.logFile = resumeStrategyConfiguration.getLogFile();
this.resumeStrategy =
resumeStrategyConfiguration.getDelegateResumeStrategy();
+ // the writer starts a new log, so read the entries of the
previous run first
+ final List<LogEntry> entries = readUnconfirmedEntries(logFile,
maxRecordCount);
+
final ScheduledExecutorService executorService =
camelContext.getExecutorServiceManager()
.newScheduledThreadPool(this,
"SingleNodeKafkaResumeStrategy", 1);
DefaultLogSupervisor flushPolicy = new DefaultLogSupervisor(
resumeStrategyConfiguration.getSupervisorInterval(),
executorService);
- logWriter = new LogWriter(logFile, flushPolicy);
+ logWriter = new LogWriter(logFile, flushPolicy, maxRecordCount);
+
+ // keep them in the new log until loadCache() replays them, in
case the system crashes again before
+ final List<UnconfirmedEntry> carried = new
ArrayList<>(entries.size());
+ for (LogEntry entry : entries) {
+ carried.add(new UnconfirmedEntry(entry,
logWriter.append(entry)));
+ }
+ unconfirmedEntries = carried;
} catch (Exception e) {
throw new RuntimeCamelException(e);
}
diff --git
a/components/camel-wal/src/test/java/org/apache/camel/component/wal/LogReaderEntryPositionTest.java
b/components/camel-wal/src/test/java/org/apache/camel/component/wal/LogReaderEntryPositionTest.java
new file mode 100644
index 000000000000..651f651c6652
--- /dev/null
+++
b/components/camel-wal/src/test/java/org/apache/camel/component/wal/LogReaderEntryPositionTest.java
@@ -0,0 +1,105 @@
+/*
+ * 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.wal;
+
+import java.io.File;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * The reader gives each entry the position where its record starts in the
file, so that
+ * {@link LogWriter#updateState(PersistedLogEntry, LogEntry.EntryState)}
writes the new state over that record. The
+ * records have different sizes, and the reader's buffer is small, so that it
reads the file in several parts.
+ */
+class LogReaderEntryPositionTest {
+
+ @TempDir
+ File testDir;
+
+ private static LogEntry entry(String key) {
+ return new LogEntry(
+ LogEntry.EntryState.NEW, 0,
key.getBytes(StandardCharsets.UTF_8), 0,
+ ("offset-of-" + key).getBytes(StandardCharsets.UTF_8));
+ }
+
+ private static List<PersistedLogEntry> readAll(File file, int capacity)
throws Exception {
+ List<PersistedLogEntry> entries = new ArrayList<>();
+ try (LogReader reader = new LogReader(file, capacity)) {
+ PersistedLogEntry entry;
+ while ((entry = reader.readEntry()) != null) {
+ entries.add(entry);
+ }
+ }
+ return entries;
+ }
+
+ private static String describe(List<PersistedLogEntry> entries) {
+ List<String> answer = new ArrayList<>();
+ for (PersistedLogEntry entry : entries) {
+ String key = new String(entry.getKey(), StandardCharsets.UTF_8);
+ assertEquals("offset-of-" + key, new String(entry.getValue(),
StandardCharsets.UTF_8),
+ "value of record " + key + " is corrupted");
+ answer.add(key + "=" + entry.getEntryState());
+ }
+ return answer.toString();
+ }
+
+ @Test
+ void entriesHaveThePositionOfTheirRecord() throws Exception {
+ File file = new File(testDir, "wal.data");
+ List<Long> positions = new ArrayList<>();
+ try (LogWriter writer = new LogWriter(file, new
DefaultLogSupervisor(100_000))) {
+ for (String key : List.of("a", "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
"cc", "dddddddddddddddd", "e")) {
+ positions.add(writer.append(entry(key)).getPosition());
+ }
+ }
+
+ // 64 bytes: the buffer holds one or two of the shorter records at a
time, and the 94 bytes record of "bbb..."
+ // spans two reads of the buffer
+ List<PersistedLogEntry> entries = readAll(file, 64);
+ List<Long> read = new ArrayList<>();
+ for (PersistedLogEntry entry : entries) {
+ read.add(entry.getEntryInfo().getPosition());
+ }
+ assertEquals(positions, read);
+ }
+
+ @Test
+ void updateOfAReadEntryChangesThatRecordOnly() throws Exception {
+ File file = new File(testDir, "wal.data");
+ try (LogWriter writer = new LogWriter(file, new
DefaultLogSupervisor(100_000))) {
+ for (String key : List.of("a", "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
"cc", "dddddddddddddddd")) {
+ writer.append(entry(key));
+ }
+ writer.flush();
+
+ List<PersistedLogEntry> entries = readAll(file, 64);
+ writer.updateState(entries.get(1), LogEntry.EntryState.PROCESSED);
+ writer.updateState(entries.get(2), LogEntry.EntryState.FAILED);
+ writer.flush();
+ }
+
+ assertEquals("[a=NEW, bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb=PROCESSED,
cc=FAILED, dddddddddddddddd=NEW]",
+ describe(readAll(file, LogReader.DEFAULT_CAPACITY)));
+ }
+}
diff --git
a/components/camel-wal/src/test/java/org/apache/camel/component/wal/LogWriterUpdateAfterRollOverPositionTest.java
b/components/camel-wal/src/test/java/org/apache/camel/component/wal/LogWriterUpdateAfterRollOverPositionTest.java
index beef37891f39..7635bbead06f 100644
---
a/components/camel-wal/src/test/java/org/apache/camel/component/wal/LogWriterUpdateAfterRollOverPositionTest.java
+++
b/components/camel-wal/src/test/java/org/apache/camel/component/wal/LogWriterUpdateAfterRollOverPositionTest.java
@@ -115,6 +115,24 @@ class LogWriterUpdateAfterRollOverPositionTest {
assertEquals("{d=NEW, e=NEW, cc=NEW}", readAll(file).toString());
}
+ @Test
+ void shorterRecordAtTheEndOfALayerKeepsFileReadable() throws Exception {
+ File file = new File(testDir, "wal.data");
+ try (LogWriter writer = new LogWriter(file, new
DefaultLogSupervisor(100_000), 3)) {
+ writer.append(entry("aa"));
+ writer.append(entry("bb"));
+ writer.append(entry("cccccccccccccccc"));
+ // roll-over: "d" and "e" replace "aa" and "bb"
+ writer.append(entry("d"));
+ writer.append(entry("e"));
+ // "f" replaces the longer "cccccccccccccccc", the last record of
the file: nothing is written after it
+ writer.append(entry("f"));
+ writer.flush();
+ }
+
+ assertEquals("{d=NEW, e=NEW, f=NEW}", readAll(file).toString());
+ }
+
@Test
void updateOfEntryTwoLayersBehindDoesNotTouchNewerEntry() throws Exception
{
File file = new File(testDir, "wal.data");
diff --git
a/components/camel-wal/src/test/java/org/apache/camel/component/wal/TransactionLogTest.java
b/components/camel-wal/src/test/java/org/apache/camel/component/wal/TransactionLogTest.java
index 111dbbb32246..4939ea803446 100644
---
a/components/camel-wal/src/test/java/org/apache/camel/component/wal/TransactionLogTest.java
+++
b/components/camel-wal/src/test/java/org/apache/camel/component/wal/TransactionLogTest.java
@@ -14,40 +14,80 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
package org.apache.camel.component.wal;
+import java.nio.charset.StandardCharsets;
+
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+class TransactionLogTest {
+
+ private static LogEntry entry(String key) {
+ return new LogEntry(LogEntry.EntryState.NEW, 0,
key.getBytes(StandardCharsets.UTF_8), 0, new byte[] { 1 });
+ }
+
+ private static TransactionLog.EntryContainer[] add(TransactionLog log, int
count) {
+ TransactionLog.EntryContainer[] containers = new
TransactionLog.EntryContainer[count];
+ for (int i = 0; i < count; i++) {
+ containers[i] = log.add(entry("record-" + i));
+ }
+ return containers;
+ }
-public class TransactionLogTest {
@DisplayName("Tests that can update records on the same layer")
@Test
void testCanUpdateSameLayer() {
- assertTrue(TransactionLog.canUpdate(0, 10, 0, 1));
- assertTrue(TransactionLog.canUpdate(0, 10, 0, 10));
- assertFalse(TransactionLog.canUpdate(0, 10, 0, 11));
+ TransactionLog log = new TransactionLog(10);
+ TransactionLog.EntryContainer[] containers = add(log, 3);
+
+ assertSame(containers[0], log.update(containers[0].layerInfo,
LogEntry.EntryState.PROCESSED));
+ assertSame(containers[2], log.update(containers[2].layerInfo,
LogEntry.EntryState.FAILED));
}
@DisplayName("Tests that prevent updating a record that has been
rolled-over")
@Test
void testCannotUpdateRolledOverRecord() {
- assertFalse(TransactionLog.canUpdate(1, 3, 0, 1));
- assertFalse(TransactionLog.canUpdate(1, 3, 0, 2));
- // two layers behind: the previous layer has replaced it already
- assertFalse(TransactionLog.canUpdate(2, 3, 0, 5));
+ TransactionLog log = new TransactionLog(3);
+ TransactionLog.EntryContainer[] containers = add(log, 5);
+
+ // records 3 and 4 (layer 1) replaced records 0 and 1 (layer 0)
+ assertNull(log.update(containers[0].layerInfo,
LogEntry.EntryState.PROCESSED));
+ assertNull(log.update(containers[1].layerInfo,
LogEntry.EntryState.PROCESSED));
+
+ // two layers behind: record 7 (layer 2) replaced record 4 (layer 1),
which had replaced record 1 (layer 0)
+ TransactionLog.EntryContainer[] next = add(log, 3);
+ assertNull(log.update(containers[1].layerInfo,
LogEntry.EntryState.PROCESSED));
+ assertNull(log.update(containers[4].layerInfo,
LogEntry.EntryState.PROCESSED));
+ assertSame(next[2], log.update(next[2].layerInfo,
LogEntry.EntryState.PROCESSED));
}
@DisplayName("Tests that can update records after a roll-over has started")
@Test
void testCanUpdateDifferentLayerWithRollOver() {
- assertTrue(TransactionLog.canUpdate(1, 10, 0, 11));
- assertTrue(TransactionLog.canUpdate(1, 10, 1, 10));
- // the current index is the index of the next entry to add, so the
entry there is not replaced yet
- assertTrue(TransactionLog.canUpdate(1, 10, 0, 10));
- assertTrue(TransactionLog.canUpdate(1, 3, 0, 3));
+ TransactionLog log = new TransactionLog(3);
+ TransactionLog.EntryContainer[] containers = add(log, 4);
+
+ // record 3 (layer 1) replaced record 0; record 1, the next to
replace, and record 2 are still there
+ assertSame(containers[1], log.update(containers[1].layerInfo,
LogEntry.EntryState.PROCESSED));
+ assertSame(containers[2], log.update(containers[2].layerInfo,
LogEntry.EntryState.PROCESSED));
+ assertSame(containers[3], log.update(containers[3].layerInfo,
LogEntry.EntryState.PROCESSED));
+ }
+
+ @DisplayName("Tests that a record of another transaction log is not
updated in this one")
+ @Test
+ void testCannotUpdateRecordOfAnotherLog() {
+ // as after a restart of the route: the writer of the previous start
had its own transaction log
+ TransactionLog previous = new TransactionLog(10);
+ TransactionLog.EntryContainer[] stale = add(previous, 2);
+ TransactionLog log = new TransactionLog(10);
+ TransactionLog.EntryContainer[] containers = add(log, 2);
+
+ assertNull(log.update(stale[1].layerInfo,
LogEntry.EntryState.PROCESSED));
+ assertEquals(LogEntry.EntryState.NEW,
containers[1].logEntry.getEntryState());
}
}
diff --git
a/components/camel-wal/src/test/java/org/apache/camel/component/wal/WriteAheadResumeStrategyRestartTest.java
b/components/camel-wal/src/test/java/org/apache/camel/component/wal/WriteAheadResumeStrategyRestartTest.java
new file mode 100644
index 000000000000..dfe092d22940
--- /dev/null
+++
b/components/camel-wal/src/test/java/org/apache/camel/component/wal/WriteAheadResumeStrategyRestartTest.java
@@ -0,0 +1,587 @@
+/*
+ * 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.wal;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.lang.reflect.Proxy;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Executors;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.resume.Deserializable;
+import org.apache.camel.resume.Offset;
+import org.apache.camel.resume.OffsetKey;
+import org.apache.camel.resume.Resumable;
+import org.apache.camel.resume.ResumeAdapter;
+import org.apache.camel.resume.ResumeStrategy;
+import org.apache.camel.resume.ResumeStrategy.UpdateCallBack;
+import org.apache.camel.resume.ResumeStrategyConfiguration;
+import org.apache.camel.spi.ExecutorServiceManager;
+import org.apache.camel.support.resume.OffsetKeys;
+import org.apache.camel.support.resume.Offsets;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The entries that the delegate strategy did not confirm before the system
stopped are replayed on the next start, in
+ * the order Camel calls the strategy: start() (from
ResumableProcessor.doStart) and then loadCache() (from the
+ * consumer).
+ */
+class WriteAheadResumeStrategyRestartTest {
+
+ /**
+ * The size of a record with a key "key-N" (one digit) and a long value
+ */
+ private static final int RECORD_SIZE = 41;
+
+ @TempDir
+ File testDir;
+
+ private final List<WriteAheadResumeStrategy> running = new ArrayList<>();
+
+ static class Adapter implements ResumeAdapter, Deserializable {
+ @Override
+ public void resume() {
+ }
+
+ @Override
+ public boolean deserialize(ByteBuffer keyBuffer, ByteBuffer
valueBuffer) {
+ return true;
+ }
+ }
+
+ /**
+ * A delegate strategy that records the replayed offsets and confirms
them, except for the keys it fails
+ */
+ static class Delegate implements ResumeStrategy {
+ final List<String> replayed = new CopyOnWriteArrayList<>();
+ final Set<String> failingKeys;
+ final Adapter adapter = new Adapter();
+
+ Delegate(String... failingKeys) {
+ this.failingKeys = Set.of(failingKeys);
+ }
+
+ @Override
+ public void setAdapter(ResumeAdapter adapter) {
+ }
+
+ @Override
+ public ResumeAdapter getAdapter() {
+ return adapter;
+ }
+
+ @Override
+ public <T extends Resumable> void updateLastOffset(T offset) {
+ }
+
+ @Override
+ public <T extends Resumable> void updateLastOffset(T offset,
UpdateCallBack updateCallBack) {
+ }
+
+ @Override
+ public void updateLastOffset(OffsetKey<?> offsetKey, Offset<?>
offsetValue) {
+ updateLastOffset(offsetKey, offsetValue, null);
+ }
+
+ @Override
+ public void updateLastOffset(OffsetKey<?> offsetKey, Offset<?> offset,
UpdateCallBack updateCallBack) {
+ replayed.add(offsetKey.getValue() + "=" + offset.getValue());
+ if (updateCallBack != null) {
+
updateCallBack.onUpdate(failingKeys.contains(offsetKey.getValue())
+ ? new IllegalStateException("the delegate failed") :
null);
+ }
+ }
+
+ @Override
+ public void setResumeStrategyConfiguration(ResumeStrategyConfiguration
resumeStrategyConfiguration) {
+ }
+
+ @Override
+ public ResumeStrategyConfiguration getResumeStrategyConfiguration() {
+ return null;
+ }
+
+ @Override
+ public void start() {
+ }
+
+ @Override
+ public void stop() {
+ }
+ }
+
+ /**
+ * A delegate strategy that never calls back, so the entries stay
unconfirmed (as with the MongoDB strategy)
+ */
+ static class NeverConfirmingDelegate extends Delegate {
+ @Override
+ public void updateLastOffset(OffsetKey<?> offsetKey, Offset<?> offset,
UpdateCallBack updateCallBack) {
+ replayed.add(offsetKey.getValue() + "=" + offset.getValue());
+ }
+ }
+
+ /**
+ * A delegate strategy that keeps the callbacks of some keys without
calling them (as a delegate that confirms
+ * later, on another thread), and confirms the others
+ */
+ static class LateConfirmingDelegate extends Delegate {
+ final Set<String> lateKeys;
+ final List<UpdateCallBack> callbacks = new CopyOnWriteArrayList<>();
+
+ LateConfirmingDelegate(String... lateKeys) {
+ this.lateKeys = Set.of(lateKeys);
+ }
+
+ @Override
+ public void updateLastOffset(OffsetKey<?> offsetKey, Offset<?> offset,
UpdateCallBack updateCallBack) {
+ if (lateKeys.contains(offsetKey.getValue())) {
+ replayed.add(offsetKey.getValue() + "=" + offset.getValue());
+ callbacks.add(updateCallBack);
+ } else {
+ super.updateLastOffset(offsetKey, offset, updateCallBack);
+ }
+ }
+ }
+
+ /**
+ * A delegate strategy that throws for one key (as when its storage is not
available)
+ */
+ static class ThrowingDelegate extends Delegate {
+ private final String throwingKey;
+
+ ThrowingDelegate(String throwingKey) {
+ this.throwingKey = throwingKey;
+ }
+
+ @Override
+ public void updateLastOffset(OffsetKey<?> offsetKey, Offset<?> offset,
UpdateCallBack updateCallBack) {
+ if (throwingKey.equals(offsetKey.getValue())) {
+ replayed.add(offsetKey.getValue() + "=" + offset.getValue());
+ throw new IllegalStateException("the delegate is not
available");
+ }
+ super.updateLastOffset(offsetKey, offset, updateCallBack);
+ }
+ }
+
+ @AfterEach
+ void stopRunningStrategies() {
+ for (WriteAheadResumeStrategy strategy : List.copyOf(running)) {
+ stop(strategy);
+ }
+ }
+
+ private static LogEntry entry(LogEntry.EntryState state, String key, long
value) {
+ return new LogEntry(
+ state, 0, OffsetKeys.of(key).serialize().array(), 0,
Offsets.of(value).serialize().array());
+ }
+
+ /**
+ * Writes a log as a previous run would leave it
+ */
+ private File writeLog(LogEntry... entries) throws IOException {
+ return writeLog(LogWriter.DEFAULT_CAPACITY, entries);
+ }
+
+ /**
+ * Writes a log as a previous run would leave it, with a log that keeps
the given number of entries before it rolls
+ * over
+ */
+ private File writeLog(int maxRecordCount, LogEntry... entries) throws
IOException {
+ File logFile = new File(testDir, "wal.data");
+ try (LogWriter writer = new LogWriter(logFile, new
DefaultLogSupervisor(100_000), maxRecordCount)) {
+ for (LogEntry entry : entries) {
+ writer.append(entry);
+ }
+ }
+ return logFile;
+ }
+
+ private File writeNewEntries(int count) throws IOException {
+ LogEntry[] entries = new LogEntry[count];
+ for (int i = 0; i < count; i++) {
+ entries[i] = entry(LogEntry.EntryState.NEW, "key-" + i, i);
+ }
+ return writeLog(entries);
+ }
+
+ private WriteAheadResumeStrategy start(File logFile, ResumeStrategy
delegate) {
+ return start(logFile, delegate, LogWriter.DEFAULT_CAPACITY);
+ }
+
+ private WriteAheadResumeStrategy start(File logFile, ResumeStrategy
delegate, int maxRecordCount) {
+ WriteAheadResumeStrategy strategy = new WriteAheadResumeStrategy(
+ WriteAheadResumeStrategyConfigurationBuilder.newBuilder()
+
.withLogFile(logFile).withDelegateResumeStrategy(delegate).build(),
+ maxRecordCount);
+
+ ExecutorServiceManager executorServiceManager =
(ExecutorServiceManager) Proxy.newProxyInstance(
+ getClass().getClassLoader(), new Class<?>[] {
ExecutorServiceManager.class },
+ (proxy, method, args) ->
"newScheduledThreadPool".equals(method.getName())
+ ? Executors.newSingleThreadScheduledExecutor() : null);
+ CamelContext camelContext = (CamelContext) Proxy.newProxyInstance(
+ getClass().getClassLoader(), new Class<?>[] {
CamelContext.class },
+ (proxy, method, args) ->
"getExecutorServiceManager".equals(method.getName())
+ ? executorServiceManager : null);
+ strategy.setCamelContext(camelContext);
+
+ strategy.start();
+ running.add(strategy);
+ return strategy;
+ }
+
+ private WriteAheadResumeStrategy restart(File logFile, ResumeStrategy
delegate) throws Exception {
+ WriteAheadResumeStrategy strategy = start(logFile, delegate);
+ strategy.loadCache();
+ return strategy;
+ }
+
+ private void stop(WriteAheadResumeStrategy strategy) {
+ running.remove(strategy);
+ strategy.stop();
+ }
+
+ /**
+ * Reads the log, as "key=STATE" for each entry
+ */
+ private static List<String> readLog(File logFile) throws IOException {
+ Adapter adapter = new Adapter();
+ List<String> entries = new ArrayList<>();
+ try (LogReader reader = new LogReader(logFile)) {
+ assertNotNull(reader.getHeader(), "the log has a header");
+ assertEquals(Header.FORMAT_NAME,
reader.getHeader().getFormatName().trim());
+ assertEquals(Header.CURRENT_FILE_VERSION,
reader.getHeader().getFileVersion());
+
+ PersistedLogEntry entry;
+ while ((entry = reader.readEntry()) != null) {
+
entries.add(adapter.deserializeKey(ByteBuffer.wrap(entry.getKey())) + "=" +
entry.getEntryState());
+ }
+ }
+ return entries;
+ }
+
+ private static List<LogEntry.EntryState> readStates(File logFile) throws
IOException {
+ List<LogEntry.EntryState> states = new ArrayList<>();
+ try (LogReader reader = new LogReader(logFile)) {
+ PersistedLogEntry entry;
+ while ((entry = reader.readEntry()) != null) {
+ states.add(entry.getEntryState());
+ }
+ }
+ return states;
+ }
+
+ private static void writeIntAt(File file, long position, int value) throws
IOException {
+ try (RandomAccessFile raf = new RandomAccessFile(file, "rw")) {
+ raf.seek(position);
+ raf.writeInt(value);
+ }
+ }
+
+ private static void truncate(File file, long length) throws IOException {
+ try (RandomAccessFile raf = new RandomAccessFile(file, "rw")) {
+ raf.setLength(length);
+ }
+ }
+
+ @Test
+ void entriesOfTheSameSerializedKeyWithAnotherKeyMetadataAreDifferentKeys()
throws Exception {
+ byte[] key = OffsetKeys.of("key-0").serialize().array();
+ File logFile = writeLog(
+ new LogEntry(LogEntry.EntryState.NEW, 0, key, 0,
Offsets.of(0L).serialize().array()),
+ new LogEntry(LogEntry.EntryState.NEW, 1, key, 0,
Offsets.of(1L).serialize().array()),
+ new LogEntry(LogEntry.EntryState.NEW, 0, key, 0,
Offsets.of(2L).serialize().array()));
+
+ Delegate delegate = new Delegate();
+ stop(restart(logFile, delegate));
+
+ // the last entry of each (key metadata, key), in the order of their
last entry
+ assertEquals(List.of("key-0=1", "key-0=2"), delegate.replayed);
+ }
+
+ @Test
+ void restartReplaysTheNewAndFailedEntriesButNotTheProcessedOnes() throws
Exception {
+ File logFile = writeLog(
+ entry(LogEntry.EntryState.NEW, "key-0", 0),
+ entry(LogEntry.EntryState.PROCESSED, "key-1", 1),
+ entry(LogEntry.EntryState.FAILED, "key-2", 2),
+ entry(LogEntry.EntryState.NEW, "key-3", 3));
+
+ Delegate delegate = new Delegate();
+ stop(restart(logFile, delegate));
+
+ assertEquals(List.of("key-0=0", "key-2=2", "key-3=3"),
delegate.replayed);
+ assertEquals(List.of("key-0=PROCESSED", "key-2=PROCESSED",
"key-3=PROCESSED"), readLog(logFile));
+
+ Delegate next = new Delegate();
+ stop(restart(logFile, next));
+ assertEquals(List.of(), next.replayed, "the confirmed entries are not
replayed again");
+ }
+
+ @Test
+ void onlyTheLastOffsetOfEachKeyIsReplayed() throws Exception {
+ File logFile = writeLog(
+ entry(LogEntry.EntryState.FAILED, "key-0", 0),
+ entry(LogEntry.EntryState.PROCESSED, "key-0", 1),
+ entry(LogEntry.EntryState.NEW, "key-1", 2),
+ entry(LogEntry.EntryState.NEW, "key-1", 3),
+ entry(LogEntry.EntryState.PROCESSED, "key-2", 4),
+ entry(LogEntry.EntryState.FAILED, "key-2", 5));
+
+ Delegate delegate = new Delegate();
+ stop(restart(logFile, delegate));
+
+ // a failed offset does not replace a later offset of its key that the
delegate confirmed
+ assertEquals(List.of("key-1=3", "key-2=5"), delegate.replayed);
+ assertEquals(List.of("key-1=PROCESSED", "key-2=PROCESSED"),
readLog(logFile));
+ }
+
+ @Test
+ void entriesOfADelegateThatNeverConfirmsDoNotPileUpAcrossRestarts() throws
Exception {
+ File logFile = new File(testDir, "wal.data");
+ WriteAheadResumeStrategy strategy = restart(logFile, new
NeverConfirmingDelegate());
+ strategy.updateLastOffset(OffsetKeys.of("key-0"), Offsets.of(0L));
+ strategy.updateLastOffset(OffsetKeys.of("key-0"), Offsets.of(1L));
+ strategy.updateLastOffset(OffsetKeys.of("key-1"), Offsets.of(2L));
+ stop(strategy);
+
+ for (int i = 0; i < 2; i++) {
+ Delegate delegate = new NeverConfirmingDelegate();
+ stop(restart(logFile, delegate));
+
+ assertEquals(List.of("key-0=1", "key-1=2"), delegate.replayed);
+ assertEquals(List.of("key-0=NEW", "key-1=NEW"), readLog(logFile));
+ }
+ }
+
+ @Test
+ void entryThatCannotBeReplayedDoesNotStopTheOthersNorTheStart() throws
Exception {
+ byte[] value = Offsets.of(7L).serialize().array();
+ File logFile = writeLog(
+ // a key too short for its type, and a key of an unknown type
+ new LogEntry(LogEntry.EntryState.NEW, 0, new byte[] { 1, 2 },
0, value),
+ new LogEntry(LogEntry.EntryState.NEW, 0,
ByteBuffer.allocate(8).putInt(99).array(), 0, value),
+ entry(LogEntry.EntryState.NEW, "key-0", 0),
+ entry(LogEntry.EntryState.NEW, "key-1", 1));
+
+ Delegate delegate = new ThrowingDelegate("key-0");
+ stop(restart(logFile, delegate));
+
+ assertEquals(List.of("key-0=0", "key-1=1"), delegate.replayed);
+ assertEquals(List.of(LogEntry.EntryState.NEW, LogEntry.EntryState.NEW,
LogEntry.EntryState.FAILED,
+ LogEntry.EntryState.PROCESSED), readStates(logFile));
+
+ Delegate next = new Delegate();
+ stop(restart(logFile, next));
+ assertEquals(List.of("key-0=0"), next.replayed);
+ assertEquals(List.of(LogEntry.EntryState.NEW, LogEntry.EntryState.NEW,
LogEntry.EntryState.PROCESSED),
+ readStates(logFile));
+ }
+
+ @Test
+ void entryWhoseReplayFailsIsReplayedOnTheNextStart() throws Exception {
+ File logFile = writeNewEntries(2);
+
+ Delegate delegate = new Delegate("key-0");
+ stop(restart(logFile, delegate));
+
+ assertEquals(List.of("key-0=0", "key-1=1"), delegate.replayed);
+ assertEquals(List.of("key-0=FAILED", "key-1=PROCESSED"),
readLog(logFile));
+
+ Delegate next = new Delegate();
+ stop(restart(logFile, next));
+ assertEquals(List.of("key-0=0"), next.replayed);
+ }
+
+ @Test
+ void unconfirmedEntriesStayInTheLogUntilTheyAreReplayed() throws Exception
{
+ File logFile = writeNewEntries(2);
+
+ // stopped (or crashed) after start() but before loadCache() replayed
the entries
+ Delegate notReplayed = new Delegate();
+ stop(start(logFile, notReplayed));
+ assertEquals(List.of(), notReplayed.replayed);
+ assertEquals(List.of("key-0=NEW", "key-1=NEW"), readLog(logFile));
+
+ Delegate delegate = new Delegate();
+ stop(restart(logFile, delegate));
+ assertEquals(List.of("key-0=0", "key-1=1"), delegate.replayed);
+ }
+
+ @Test
+ void callbackOfThePreviousStartDoesNotUpdateAnEntryOfTheNewLog() throws
Exception {
+ File logFile = new File(testDir, "wal.data");
+ LateConfirmingDelegate delegate = new LateConfirmingDelegate("key-1");
+ WriteAheadResumeStrategy strategy = restart(logFile, delegate);
+ strategy.updateLastOffset(OffsetKeys.of("key-0"), Offsets.of(0L));
+ strategy.updateLastOffset(OffsetKeys.of("key-1"), Offsets.of(1L));
+ UpdateCallBack late = delegate.callbacks.get(0);
+
+ // the route is stopped and started again (the same strategy) while
the callback of key-1 is pending: key-1 is
+ // carried into the new log, and a new offset of key-2 is confirmed
there, where key-1 was in the old log
+ strategy.stop();
+ strategy.start();
+ strategy.loadCache();
+ strategy.updateLastOffset(OffsetKeys.of("key-2"), Offsets.of(2L));
+
+ // the callback of the previous start reports late
+ late.onUpdate(new IllegalStateException("the delegate failed late"));
+ stop(strategy);
+
+ assertEquals(List.of("key-1=NEW", "key-2=PROCESSED"),
readLog(logFile));
+ }
+
+ @Test
+ void logThatMayHaveRolledOverIsNotReplayed() throws Exception {
+ // the log keeps 3 entries: the fourth offset rolled it over, so the
file holds 4, then the older 2 and 3
+ File logFile = writeLog(3,
+ entry(LogEntry.EntryState.NEW, "key-0", 1),
+ entry(LogEntry.EntryState.NEW, "key-0", 2),
+ entry(LogEntry.EntryState.NEW, "key-0", 3),
+ entry(LogEntry.EntryState.NEW, "key-0", 4));
+
+ Delegate delegate = new NeverConfirmingDelegate();
+ WriteAheadResumeStrategy strategy = start(logFile, delegate, 3);
+ strategy.loadCache();
+ stop(strategy);
+
+ // the last entry of the key in the file is the older offset 3:
replaying it would resume before 4
+ assertEquals(List.of(), delegate.replayed);
+ assertEquals(List.of(), readLog(logFile), "a new log replaces the
previous one");
+ }
+
+ @Test
+ void logThatHasNotRolledOverIsReplayed() throws Exception {
+ File logFile = writeLog(3,
+ entry(LogEntry.EntryState.NEW, "key-0", 1),
+ entry(LogEntry.EntryState.NEW, "key-0", 2));
+
+ Delegate delegate = new NeverConfirmingDelegate();
+ WriteAheadResumeStrategy strategy = start(logFile, delegate, 3);
+ strategy.loadCache();
+ stop(strategy);
+
+ assertEquals(List.of("key-0=2"), delegate.replayed);
+ assertEquals(List.of("key-0=NEW"), readLog(logFile));
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = { 3, 6, 16, 27, 40 })
+ void truncatedLastRecordIsSkippedAndThePreviousOnesAreReplayed(int
bytesLeftOfTheLastRecord) throws Exception {
+ File logFile = writeNewEntries(3);
+ truncate(logFile, Header.BYTES + 2 * RECORD_SIZE +
bytesLeftOfTheLastRecord);
+
+ Delegate delegate = new Delegate();
+ stop(restart(logFile, delegate));
+
+ assertEquals(List.of("key-0=0", "key-1=1"), delegate.replayed);
+ assertEquals(List.of("key-0=PROCESSED", "key-1=PROCESSED"),
readLog(logFile), "the new log is complete");
+ }
+
+ @Test
+ void truncatedLastRecordAcrossTheReadBufferIsSkipped() throws Exception {
+ // the last record starts in the first read of the reader's buffer,
and the file ends 2 bytes after it
+ File logFile = new File(testDir, "wal.data");
+ int count = 0;
+ try (LogWriter writer = new LogWriter(logFile, new
DefaultLogSupervisor(100_000))) {
+ long end;
+ do {
+ LogEntry entry = entry(LogEntry.EntryState.NEW, "key-" +
count, count);
+ end = writer.append(entry).getPosition() + entry.size();
+ count++;
+ } while (end <= LogReader.DEFAULT_CAPACITY + 2);
+ }
+ truncate(logFile, LogReader.DEFAULT_CAPACITY + 2);
+
+ Delegate delegate = new Delegate();
+ stop(restart(logFile, delegate));
+
+ assertEquals(count - 1, delegate.replayed.size());
+ assertEquals("key-" + (count - 2) + "=" + (count - 2),
delegate.replayed.get(count - 2));
+ }
+
+ @ParameterizedTest
+ @CsvSource({
+ // the position in the last record, the value written there
+ "8, -5", // key length
+ "25, -1", // value length
+ "8, 2147483647", // key length, larger than the reader's buffer
+ "8, 1000", // key length, beyond the end of the file
+ "0, 99" // state
+ })
+ void invalidLastRecordDoesNotFailTheStart(int positionInRecord, int value)
throws Exception {
+ File logFile = writeNewEntries(3);
+ writeIntAt(logFile, Header.BYTES + 2 * RECORD_SIZE + positionInRecord,
value);
+
+ Delegate delegate = new Delegate();
+ stop(restart(logFile, delegate));
+
+ assertEquals(List.of("key-0=0", "key-1=1"), delegate.replayed);
+ assertEquals(List.of("key-0=PROCESSED", "key-1=PROCESSED"),
readLog(logFile));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = { "no file", "empty", "header only", "shorter than
a header", "other format", "other version" })
+ void startsWithNothingToReplay(String content) throws Exception {
+ File logFile = new File(testDir, "wal.data");
+ switch (content) {
+ case "no file":
+ break;
+ case "empty":
+ assertTrue(logFile.createNewFile());
+ break;
+ case "header only":
+ writeLog();
+ break;
+ case "shorter than a header":
+ Files.write(logFile.toPath(), new byte[] { 'c', 'a', 'm', 'e',
'l' });
+ break;
+ case "other format":
+ writeNewEntries(1);
+ Files.write(logFile.toPath(), "other-fm".getBytes(),
StandardOpenOption.WRITE);
+ break;
+ default:
+ writeNewEntries(1);
+ writeIntAt(logFile, Header.FORMAT_NAME_SIZE,
Header.CURRENT_FILE_VERSION + 1);
+ break;
+ }
+
+ Delegate delegate = new Delegate();
+ WriteAheadResumeStrategy strategy = restart(logFile, delegate);
+ strategy.updateLastOffset(OffsetKeys.of("key-9"), Offsets.of(9L));
+ stop(strategy);
+
+ assertEquals(List.of("key-9=9"), delegate.replayed, "only the offset
of this run is updated");
+ assertEquals(List.of("key-9=PROCESSED"), readLog(logFile), "a new log
replaces the previous one");
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 0d6dd068dcbb..70a0b2850701 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -4075,3 +4075,24 @@ were delivered on the wrong topic. The producer now
publishes a new `MqttPropert
message expiry interval, content type, response topic, correlation data and
user properties of the header object, which
a publish no longer modifies. A topic alias or subscription identifiers set in
the header object are not sent: the
client assigns topic aliases for its connection, and subscription identifiers
are sent by the server only.
+
+=== camel-wal - the unconfirmed offsets of the previous run are replayed after
a restart
+
+`WriteAheadResumeStrategy` now does what it is documented to do after a
restart: when the consumer starts and loads the
+resume cache (`loadCache()`), the strategy replays the offsets of the previous
run that the wrapped resume strategy did
+not confirm, by calling `updateLastOffset` on the wrapped strategy. Before,
`start()` truncated the log file before
+`loadCache()` read it, so nothing was replayed. Only the last offset of each
key in the log counts: it is replayed when
+its entry is in state `NEW` or `FAILED`, and an earlier offset of the same key
is not replayed. These entries stay in
+the log until the wrapped strategy confirms them through its callback; a
wrapped strategy that never calls back gets
+the last offset of each such key again at every start. The Kafka consumer does
not call `loadCache()`, so with it the
+entries are kept in the log but not replayed. A replay that fails, or an entry
that the resume adapter cannot
+deserialize, is logged with a WARN and does not stop the start of the route.
If the last record of the log is
+incomplete (the JVM stopped while writing it) or invalid, the entries before
it are replayed and a WARN is logged; the
+route still starts. A file that is not a log of this format is replaced by a
new log, as before, with a WARN.
+
+The log keeps 524,288 entries; after that it rolls over and holds the newest
entries before the older ones, and the
+file does not say which are newer. So a log that holds that many entries is
not replayed, as before: a WARN names the
+number of unconfirmed entries, which are discarded with the log. Replaying the
wrong offset of a key would make the
+consumer resume from an older offset. A log that rolled over and was cut short
because the system stopped while the
+log rewrote itself after the roll-over holds fewer entries and is replayed:
the last entry of a key in it can then be
+an older offset, so the consumer can process again the messages after that
offset.
