[#1256] Remove incomplete changelog records completely during crash recover...

Valery Kharseko

19 hours ago c3e63df192c7ea16efeeda75a0603de2b617cf65

 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();

 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
  {