From c3e63df192c7ea16efeeda75a0603de2b617cf65 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Sun, 01 Nov 2026 17:54:04 +0000
Subject: [PATCH] [#1256] Remove incomplete changelog records completely during crash recovery (#1257)
---
opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/BlockLogReader.java | 12 +++++-
opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/LogFileTest.java | 74 +++++++++++++++++++++++++++++++++++++
2 files changed, 84 insertions(+), 2 deletions(-)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/BlockLogReader.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/BlockLogReader.java
index 8b33a59..ed706da 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/BlockLogReader.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/BlockLogReader.java
@@ -12,6 +12,7 @@
* information: "Portions Copyright [year] [name of copyright owner]".
*
* Copyright 2014-2016 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
*/
package org.opends.server.replication.server.changelog.file;
@@ -600,10 +601,17 @@
{
try
{
- final long lastBlockStart = getClosestBlockStartToEndOfFile();
+ long lastBlockStart = getClosestBlockStartToEndOfFile();
+ if (lastBlockStart > 0 && getFileLength() - lastBlockStart < SIZE_OF_BLOCK_OFFSET)
+ {
+ // A crash may interrupt the block offset itself. Read from the preceding complete block.
+ lastBlockStart -= blockSize;
+ }
positionToRecordFromBlockStart(lastBlockStart);
- long lastValidPosition = lastBlockStart;
+ // The first record may be incomplete and start before this block: remove it in its entirety.
+ // With a zero offset the reader skipped the offset field, which must be removed with the record.
+ long lastValidPosition = Math.min(lastBlockStart, reader.getFilePointer());
for (ByteString recordData = readNextRecord(); recordData != null; recordData = readNextRecord()) {
parser.decodeRecord(recordData);
lastValidPosition = reader.getFilePointer();
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/LogFileTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/LogFileTest.java
index 8dfc2d0..eeb9b7f 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/LogFileTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/LogFileTest.java
@@ -12,6 +12,7 @@
* information: "Portions Copyright [year] [name of copyright owner]".
*
* Copyright 2014-2016 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
*/
package org.opends.server.replication.server.changelog.file;
@@ -252,6 +253,79 @@
}
}
+ @DataProvider
+ Object[][] interruptedRecordData()
+ {
+ return new Object[][]
+ {
+ // First value length, complete prefix length, interrupted file length (256-byte blocks).
+ { 93, 104, 350 }, // incomplete body crossing a block
+ { 93, 104, 700 }, // incomplete body crossing several blocks
+ { 242, 253, 300 }, // length header split across a block, as in issue #1256
+ { 494, 509, 600 }, // split length header after a complete record spanning blocks
+ { 245, 256, 260 }, // zero block offset, with no record bytes after it
+ { 245, 256, 270 }, // incomplete record starting at a block boundary
+ { 93, 104, 256 }, // incomplete body ending exactly at a block boundary
+ { 93, 104, 260 }, // complete block offset, but no continuation bytes
+ { 93, 104, 257 }, // the block offset itself is incomplete
+ { 93, 104, 258 },
+ { 93, 104, 259 },
+ { 245, 256, 257 }, // incomplete offset following a complete record
+ { 0, 0, 300 }, // the first record is incomplete
+ };
+ }
+
+ @Test(dataProvider="interruptedRecordData")
+ public void recoveryRemovesTheWholeInterruptedRecordBeforeAppending(
+ int firstValueLength, long completePrefixLength, long interruptedLength) throws Exception
+ {
+ try (RandomAccessFile output = new RandomAccessFile(TEST_LOG_FILE, "rw"))
+ {
+ output.setLength(0);
+ }
+ final Record<String, String> first =
+ Record.from("key01", new String(new char[firstValueLength]).replace('\0', 'a'));
+ final Record<String, String> interrupted =
+ Record.from("key02", new String(new char[900]).replace('\0', 'b'));
+ final Record<String, String> appended = Record.from("key03", "after recovery");
+ try (LogFile<String, String> logFile = getLogFile(RECORD_PARSER))
+ {
+ if (completePrefixLength > 0)
+ {
+ assertThat(logFile.append(first)).isTrue();
+ }
+ assertThat(TEST_LOG_FILE.length()).isEqualTo(completePrefixLength);
+ assertThat(logFile.append(interrupted)).isTrue();
+ }
+ assertThat(TEST_LOG_FILE.length()).isGreaterThan(interruptedLength);
+ try (RandomAccessFile output = new RandomAccessFile(TEST_LOG_FILE, "rw"))
+ {
+ output.setLength(interruptedLength);
+ }
+
+ try (LogFile<String, String> logFile = getLogFile(RECORD_PARSER))
+ {
+ assertThat(TEST_LOG_FILE.length()).as("the entire incomplete record must be removed")
+ .isEqualTo(completePrefixLength);
+ assertThat(logFile.getNewestRecord()).isEqualTo(completePrefixLength > 0 ? first : null);
+ assertThat(logFile.append(appended)).isTrue();
+ }
+ // A second open must preserve both the complete prefix and the newly appended record.
+ try (LogFile<String, String> logFile = getLogFile(RECORD_PARSER);
+ DBCursor<Record<String, String>> cursor = logFile.getCursor())
+ {
+ assertThat(logFile.getNewestRecord()).isEqualTo(appended);
+ if (completePrefixLength > 0)
+ {
+ assertThat(cursor.next()).isTrue();
+ assertThat(cursor.getRecord()).isEqualTo(first);
+ }
+ assertThat(cursor.next()).isTrue();
+ assertThat(cursor.getRecord()).isEqualTo(appended);
+ assertThatCursorIsExhausted(cursor);
+ }
+ }
+
/** Append some raw data to the TEST_LOG_FILE. Intended to corrupt the log file. */
private void corruptTestLogFile(ByteStringBuilder corruptedRecordData) throws Exception
{
--
Gitblit v1.10.0