From 53c6d33b0c2e612aff1352a8f237ee6e83cdfe42 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Wed, 21 Oct 2026 07:00:50 +0000
Subject: [PATCH] [#1227] Keep the record a changelog cursor read in advance when the head log file it is positioned in is rotated (#1228)
---
opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/LogFile.java | 27 +++++++++++--
opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/Log.java | 22 +++++++++--
opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/LogTest.java | 64 ++++++++++++++++++++++++++++++++
3 files changed, 104 insertions(+), 9 deletions(-)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/Log.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/Log.java
index 61d8233..b6c6dce 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/Log.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/Log.java
@@ -1048,7 +1048,8 @@
final K previousKey = logFiles.lowerKey(recordParser.getMaxKey());
final LogFile<K, V> logFile = findLogFileFor(previousKey, KeyMatchingStrategy.EQUAL_TO_KEY);
final AbortableLogCursor<K, V> cursor = pair.getFirst();
- cursor.reinitializeTo(new CursorState<K, V>(logFile, cursorState.filePosition, cursorState.record));
+ cursor.reinitializeTo(new CursorState<K, V>(
+ logFile, cursorState.filePosition, cursorState.record, cursorState.initialRecord));
}
}
}
@@ -1301,7 +1302,8 @@
log.sharedLock.lock();
try
{
- return new CursorState<>(currentLogFile, currentCursor.getFilePosition(), currentCursor.getRecord());
+ return new CursorState<>(currentLogFile, currentCursor.getFilePosition(), currentCursor.getRecord(),
+ currentCursor.getInitialRecord());
}
finally
{
@@ -1320,7 +1322,8 @@
void reinitializeTo(final CursorState<K, V> cursorState) throws ChangelogException
{
currentLogFile = cursorState.logFile;
- currentCursor = currentLogFile.getCursorInitialisedTo(cursorState.record, cursorState.filePosition);
+ currentCursor = currentLogFile.getCursorInitialisedTo(
+ cursorState.record, cursorState.initialRecord, cursorState.filePosition);
}
@Override
@@ -1605,6 +1608,14 @@
/** The record the cursor is pointing to. */
private final Record<K,V> record;
+ /**
+ * The record the cursor read in advance when it was positioned, and returns before it reads at
+ * {@link #filePosition}, which is past it. Without it, a cursor positioned on the head log file
+ * and moved to the rotated file before it read anything would skip the record it is positioned
+ * on - see issue #1227.
+ */
+ private final Record<K, V> initialRecord;
+
private final boolean isValid;
/** Creates a non-valid state. */
@@ -1612,15 +1623,18 @@
logFile = null;
filePosition = 0;
record = null;
+ initialRecord = null;
isValid = false;
}
/** Creates a valid state. */
- private CursorState(final LogFile<K, V> logFile, final long filePosition, final Record<K, V> record)
+ private CursorState(final LogFile<K, V> logFile, final long filePosition, final Record<K, V> record,
+ final Record<K, V> initialRecord)
{
this.logFile = logFile;
this.filePosition = filePosition;
this.record = record;
+ this.initialRecord = initialRecord;
isValid = true;
}
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/LogFile.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/LogFile.java
index 299ca53..8704c2b 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/LogFile.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/LogFile.java
@@ -368,15 +368,19 @@
*
* @param record
* The initial record this cursor points on
+ * @param initialRecord
+ * The record the cursor read in advance, which it returns before it reads at the
+ * provided position, or {@code null} if there is none
* @param position
* The file position this cursor points on
* @return the cursor
* @throws ChangelogException
* If a problem occurs while creating the cursor.
*/
- LogFileCursor<K, V> getCursorInitialisedTo(Record<K,V> record, long position) throws ChangelogException
+ LogFileCursor<K, V> getCursorInitialisedTo(Record<K,V> record, Record<K, V> initialRecord, long position)
+ throws ChangelogException
{
- return new LogFileCursor<>(this, record, position);
+ return new LogFileCursor<>(this, record, initialRecord, position);
}
/**
@@ -583,17 +587,19 @@
}
/**
- * Creates a cursor on the provided log, initialised to the provided record and
+ * Creates a cursor on the provided log, initialised to the provided records and
* pointing to the provided file position.
* <p>
- * Note: there is no check to ensure that provided record and file position are
+ * Note: there is no check to ensure that provided records and file position are
* consistent. It is the responsibility of the caller of this method.
*/
- private LogFileCursor(LogFile<K, V> logFile, Record<K, V> record, long filePosition) throws ChangelogException
+ private LogFileCursor(LogFile<K, V> logFile, Record<K, V> record, Record<K, V> initialRecord, long filePosition)
+ throws ChangelogException
{
this.logFile = logFile;
this.reader = logFile.getReader();
this.currentRecord = record;
+ this.initialRecord = initialRecord;
logFile.sharedLock.lock();
try
{
@@ -635,6 +641,17 @@
return currentRecord;
}
+ /**
+ * Returns the record this cursor read in advance when it was positioned, which the next call
+ * to {@link #next()} returns before anything is read from the file.
+ *
+ * @return the record read in advance, or {@code null} if there is none
+ */
+ Record<K, V> getInitialRecord()
+ {
+ return initialRecord;
+ }
+
/** {@inheritDoc} */
@Override
public boolean positionTo(final K key, final KeyMatchingStrategy match, final PositionStrategy pos)
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/LogTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/LogTest.java
index aee0ad1..1c7a1d1 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/LogTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/LogTest.java
@@ -32,6 +32,7 @@
import org.opends.server.replication.server.changelog.api.DBCursor.KeyMatchingStrategy;
import org.opends.server.replication.server.changelog.api.DBCursor.PositionStrategy;
import org.opends.server.replication.server.changelog.file.Log.LogRotationParameters;
+import org.opends.server.replication.server.changelog.file.Log.RepositionableCursor;
import org.opends.server.replication.server.changelog.file.LogFileTest.FailingStringRecordParser;
import org.opends.server.replication.server.changelog.file.Record.Mapper;
import org.opends.server.util.StaticUtils;
@@ -360,6 +361,69 @@
}
}
+ @DataProvider
+ Object[][] positionedOnTheHeadLogFile()
+ {
+ return new Object[][] {
+ // key to position to, key matching strategy, position strategy, index of the first record read
+ { "key009", EQUAL_TO_KEY, ON_MATCHING_KEY, 9 },
+ { "key009", EQUAL_TO_KEY, AFTER_MATCHING_KEY, 10 },
+ { "key009", GREATER_THAN_OR_EQUAL_TO_KEY, ON_MATCHING_KEY, 9 },
+ { "key009", GREATER_THAN_OR_EQUAL_TO_KEY, AFTER_MATCHING_KEY, 10 },
+ // key is between key009 and key010: the record before it is read in advance
+ { "key0095", LESS_THAN_OR_EQUAL_TO_KEY, ON_MATCHING_KEY, 9 },
+ // key is after the newest key: the newest record is read in advance
+ { "key0105", LESS_THAN_OR_EQUAL_TO_KEY, ON_MATCHING_KEY, 10 },
+ };
+ }
+
+ /**
+ * A cursor positioned on a key has read the record it is positioned on in advance, and is past it
+ * in the file: when the head log file it is positioned in is rotated before it reads anything, it
+ * must still read that record first (issue #1227).
+ */
+ @Test(dataProvider = "positionedOnTheHeadLogFile")
+ public void aCursorReadsTheRecordItIsPositionedOnWhenTheHeadLogFileIsRotatedBeforeItsFirstRead(String key,
+ KeyMatchingStrategy matchingStrategy, PositionStrategy positionStrategy, int firstIndex) throws Exception
+ {
+ try (Log<String, String> log = openLog(LogFileTest.RECORD_PARSER);
+ DBCursor<Record<String, String>> cursor = log.getCursor(key, matchingStrategy, positionStrategy))
+ {
+ // the head log file holds 2 records: this append rotates it
+ log.append(Record.from("key011", "value11"));
+
+ assertThatCursorCanBeFullyReadFromStart(cursor, firstIndex, 11);
+ }
+ }
+
+ /**
+ * A cursor which read the newest record is positioned again to read the records written since - the
+ * way FileReplicaDBCursor does - and reads the record it is positioned on in advance as well: when
+ * the head log file is rotated before its next read, it must still read that record first (issue
+ * #1227).
+ */
+ @Test
+ public void aCursorPositionedAgainReadsTheRecordItIsPositionedOnWhenTheHeadLogFileIsRotatedBeforeItsNextRead()
+ throws Exception
+ {
+ try (Log<String, String> log = openLog(LogFileTest.RECORD_PARSER))
+ {
+ // the head log file holds 2 records: this append rotates it, and the new one holds key011
+ log.append(Record.from("key011", "value11"));
+ try (RepositionableCursor<String, String> cursor = log.getCursor())
+ {
+ assertThatCursorCanBeFullyReadFromStart(cursor, 1, 11);
+ log.append(Record.from("key012", "value12"));
+
+ assertThat(cursor.positionTo("key011", GREATER_THAN_OR_EQUAL_TO_KEY, AFTER_MATCHING_KEY)).isTrue();
+ // the head log file holds 2 records: this append rotates it
+ log.append(Record.from("key013", "value13"));
+
+ assertThatCursorCanBeFullyRead(cursor, 12, 13);
+ }
+ }
+ }
+
@Test
public void testClear() throws Exception
{
--
Gitblit v1.10.0