You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ja...@apache.org on 2025/08/12 11:48:03 UTC
(fluss) branch main updated: [log] Fix the recovery log failed situation because of OutOfOrderSequenceException cause by writerId expire (#1386)
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 72e4c4334 [log] Fix the recovery log failed situation because of OutOfOrderSequenceException cause by writerId expire (#1386)
72e4c4334 is described below
commit 72e4c43345ebca50db0e8b71aac4cfea2e029c0b
Author: yunhong <33...@qq.com>
AuthorDate: Tue Aug 12 19:47:50 2025 +0800
[log] Fix the recovery log failed situation because of OutOfOrderSequenceException cause by writerId expire (#1386)
---
.../com/alibaba/fluss/server/log/LogTablet.java | 4 +-
.../alibaba/fluss/server/log/WriterAppendInfo.java | 37 ++++++++++----
.../fluss/server/log/WriterStateManager.java | 4 ++
.../alibaba/fluss/server/log/LogTabletTest.java | 2 +-
.../fluss/server/log/WriterStateManagerTest.java | 56 ++++++++++++++++++----
5 files changed, 82 insertions(+), 21 deletions(-)
diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/log/LogTablet.java b/fluss-server/src/main/java/com/alibaba/fluss/server/log/LogTablet.java
index 490c02aa6..68231bc47 100644
--- a/fluss-server/src/main/java/com/alibaba/fluss/server/log/LogTablet.java
+++ b/fluss-server/src/main/java/com/alibaba/fluss/server/log/LogTablet.java
@@ -1115,7 +1115,9 @@ public final class LogTablet {
// update writers.
WriterAppendInfo appendInfo =
writers.computeIfAbsent(writerId, id -> writerStateManager.prepareUpdate(writerId));
- appendInfo.append(batch);
+ appendInfo.append(
+ batch,
+ writerStateManager.isWriterInBatchExpired(System.currentTimeMillis(), batch));
}
static void rebuildWriterState(
diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/log/WriterAppendInfo.java b/fluss-server/src/main/java/com/alibaba/fluss/server/log/WriterAppendInfo.java
index a27d5299a..ea755e6c8 100644
--- a/fluss-server/src/main/java/com/alibaba/fluss/server/log/WriterAppendInfo.java
+++ b/fluss-server/src/main/java/com/alibaba/fluss/server/log/WriterAppendInfo.java
@@ -21,6 +21,8 @@ import com.alibaba.fluss.exception.OutOfOrderSequenceException;
import com.alibaba.fluss.metadata.TableBucket;
import com.alibaba.fluss.record.LogRecordBatch;
+import static com.alibaba.fluss.record.LogRecordBatch.NO_BATCH_SEQUENCE;
+
/**
* This class is used to validate the records appended by a given writer before they are written to
* log. It's initialized with writer's state after the last successful append.
@@ -42,35 +44,38 @@ public class WriterAppendInfo {
return writerId;
}
- public void append(LogRecordBatch batch) {
+ public void append(LogRecordBatch batch, boolean isWriterInBatchExpired) {
LogOffsetMetadata firstOffsetMetadata = new LogOffsetMetadata(batch.baseLogOffset());
appendDataBatch(
batch.batchSequence(),
firstOffsetMetadata,
batch.lastLogOffset(),
- System.currentTimeMillis()); // TODO, add timestamp to record batch.
+ isWriterInBatchExpired,
+ batch.commitTimestamp());
}
public void appendDataBatch(
int batchSequence,
LogOffsetMetadata firstOffsetMetadata,
long lastOffset,
- long lastTimestamp) {
- maybeValidateDataBatch(batchSequence, lastOffset);
+ boolean isWriterInBatchExpired,
+ long batchTimestamp) {
+ maybeValidateDataBatch(batchSequence, isWriterInBatchExpired, lastOffset);
updatedEntry.addBath(
batchSequence,
lastOffset,
(int) (lastOffset - firstOffsetMetadata.getMessageOffset()),
- lastTimestamp);
+ batchTimestamp);
}
- private void maybeValidateDataBatch(int appendFirstSeq, long lastOffset) {
+ private void maybeValidateDataBatch(
+ int appendFirstSeq, boolean isWriterInBatchExpired, long lastOffset) {
int currentLastSeq =
!updatedEntry.isEmpty()
? updatedEntry.lastBatchSequence()
: currentEntry.lastBatchSequence();
// must be in sequence, even for the first batch should start from 0
- if (!inSequence(currentLastSeq, appendFirstSeq)) {
+ if (!inSequence(currentLastSeq, appendFirstSeq, isWriterInBatchExpired)) {
throw new OutOfOrderSequenceException(
String.format(
"Out of order batch sequence for writer %s at offset %s in "
@@ -83,8 +88,22 @@ public class WriterAppendInfo {
return updatedEntry;
}
- private boolean inSequence(int lastBatchSeq, int nextBatchSeq) {
- return nextBatchSeq == lastBatchSeq + 1L
+ /**
+ * Check if the next batch sequence is in sequence with the last batch sequence. The following
+ * three scenarios will be judged as in sequence:
+ *
+ * <ul>
+ * <li>If lastBatchSeq equals NO_BATCH_SEQUENCE, we need to check whether the committed
+ * timestamp of the next batch under the current writerId has expired. If it has expired,
+ * we consider this a special case caused by writerId expiration, for this case, to ensure
+ * the correctness of follower sync, we still treat it as in sequence.
+ * <li>nextBatchSeq == lastBatchSeq + 1L
+ * <li>lastBatchSeq reaches its maximum value
+ * </ul>
+ */
+ private boolean inSequence(int lastBatchSeq, int nextBatchSeq, boolean isWriterInBatchExpired) {
+ return (lastBatchSeq == NO_BATCH_SEQUENCE && isWriterInBatchExpired)
+ || nextBatchSeq == lastBatchSeq + 1L
|| (nextBatchSeq == 0 && lastBatchSeq == Integer.MAX_VALUE);
}
}
diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/log/WriterStateManager.java b/fluss-server/src/main/java/com/alibaba/fluss/server/log/WriterStateManager.java
index 2d0c2b921..6163d86eb 100644
--- a/fluss-server/src/main/java/com/alibaba/fluss/server/log/WriterStateManager.java
+++ b/fluss-server/src/main/java/com/alibaba/fluss/server/log/WriterStateManager.java
@@ -419,6 +419,10 @@ public class WriterStateManager {
return currentTimeMs - writerStateEntry.lastBatchTimestamp() > writerExpirationMs;
}
+ public boolean isWriterInBatchExpired(long currentTimeMs, LogRecordBatch recordBatch) {
+ return currentTimeMs - recordBatch.commitTimestamp() > writerExpirationMs;
+ }
+
private static List<WriterStateEntry> readSnapshot(File file) {
try {
byte[] json = Files.readAllBytes(file.toPath());
diff --git a/fluss-server/src/test/java/com/alibaba/fluss/server/log/LogTabletTest.java b/fluss-server/src/test/java/com/alibaba/fluss/server/log/LogTabletTest.java
index 4da484579..5b69d5065 100644
--- a/fluss-server/src/test/java/com/alibaba/fluss/server/log/LogTabletTest.java
+++ b/fluss-server/src/test/java/com/alibaba/fluss/server/log/LogTabletTest.java
@@ -411,7 +411,7 @@ final class LogTabletTest extends LogTestBase {
@Test
void testPeriodicWriterIdExpiration() throws Exception {
- conf.set(ConfigOptions.WRITER_ID_EXPIRATION_TIME, Duration.ofMillis(3000));
+ conf.set(ConfigOptions.WRITER_ID_EXPIRATION_TIME, Duration.ofMillis(1000));
conf.set(ConfigOptions.WRITER_ID_EXPIRATION_CHECK_INTERVAL, Duration.ofMillis(1000));
long writerId1 = 23L;
LogTablet log = createLogTablet(conf);
diff --git a/fluss-server/src/test/java/com/alibaba/fluss/server/log/WriterStateManagerTest.java b/fluss-server/src/test/java/com/alibaba/fluss/server/log/WriterStateManagerTest.java
index 6ebe62978..d02138fbe 100644
--- a/fluss-server/src/test/java/com/alibaba/fluss/server/log/WriterStateManagerTest.java
+++ b/fluss-server/src/test/java/com/alibaba/fluss/server/log/WriterStateManagerTest.java
@@ -21,6 +21,7 @@ import com.alibaba.fluss.config.ConfigOptions;
import com.alibaba.fluss.config.Configuration;
import com.alibaba.fluss.exception.OutOfOrderSequenceException;
import com.alibaba.fluss.metadata.TableBucket;
+import com.alibaba.fluss.utils.clock.ManualClock;
import com.alibaba.fluss.utils.types.Tuple2;
import org.junit.jupiter.api.BeforeEach;
@@ -124,14 +125,15 @@ public class WriterStateManagerTest {
@Test
void testPrepareUpdateDoesNotMutate() {
WriterAppendInfo appendInfo = stateManager.prepareUpdate(writerId);
- appendInfo.appendDataBatch(0, new LogOffsetMetadata(15L), 20L, System.currentTimeMillis());
+ appendInfo.appendDataBatch(
+ 0, new LogOffsetMetadata(15L), 20L, false, System.currentTimeMillis());
assertThat(stateManager.lastEntry(writerId)).isNotPresent();
stateManager.update(appendInfo);
assertThat(stateManager.lastEntry(writerId)).isPresent();
WriterAppendInfo nextAppendInfo = stateManager.prepareUpdate(writerId);
nextAppendInfo.appendDataBatch(
- 1, new LogOffsetMetadata(26L), 30L, System.currentTimeMillis());
+ 1, new LogOffsetMetadata(26L), 30L, false, System.currentTimeMillis());
assertThat(stateManager.lastEntry(writerId)).isPresent();
WriterStateEntry lastEntry = stateManager.lastEntry(writerId).get();
@@ -179,8 +181,8 @@ public class WriterStateManagerTest {
@Test
void testRemoveExpiredWritersOnReload() throws IOException {
- append(stateManager, writerId, 0, 0L, 0);
- append(stateManager, writerId, 1, 1L, 1);
+ append(stateManager, writerId, 0, 0L, false, 0);
+ append(stateManager, writerId, 1, 1L, false, 1);
stateManager.takeSnapshot();
WriterStateManager recoveredMapping =
@@ -194,14 +196,14 @@ public class WriterStateManagerTest {
// the writer mapping. If writing with the same writerId and non-zero batch sequence, the
// OutOfOrderSequenceException will throw. If you want to continue to write, you need to get
// a new writer id.
- assertThatThrownBy(() -> append(recoveredMapping, writerId, 2, 2L, 70001))
+ assertThatThrownBy(() -> append(recoveredMapping, writerId, 2, 2L, false, 3000L))
.isInstanceOf(OutOfOrderSequenceException.class)
.hasMessageContaining(
"Out of order batch sequence for writer 1 at offset 2 in "
+ "table-bucket TableBucket{tableId=1001, bucket=0}"
+ " : 2 (incoming batch seq.), -1 (current batch seq.)");
- append(recoveredMapping, 2L, 0, 2L, 70002);
+ append(recoveredMapping, 2L, 0, 2L, false, 70002);
assertThat(recoveredMapping.activeWriters().size()).isEqualTo(1);
assertThat(recoveredMapping.activeWriters().values().iterator().next().lastBatchSequence())
@@ -209,6 +211,34 @@ public class WriterStateManagerTest {
assertThat(recoveredMapping.mapEndOffset()).isEqualTo(3L);
}
+ @Test
+ void testAppendAnExpiredBatchWithEmptyWriterStatus() throws Exception {
+ ManualClock clock = new ManualClock(5000L);
+
+ // 2 seconds to expire the writer.
+ conf.set(ConfigOptions.WRITER_ID_EXPIRATION_TIME, Duration.ofSeconds(2));
+ WriterStateManager stateManager1 =
+ new WriterStateManager(
+ tableBucket,
+ logDir,
+ (int) conf.get(ConfigOptions.WRITER_ID_EXPIRATION_TIME).toMillis());
+
+ // If we try to append an expired batch with none zero batch sequence, the
+ // OutOfOrderSequenceException will not been throw.
+ append(stateManager1, 1L, 10, 10L, true, clock.milliseconds());
+ assertThat(stateManager1.activeWriters().size()).isEqualTo(1);
+ assertThat(stateManager1.activeWriters().values().iterator().next().lastBatchSequence())
+ .isEqualTo(10);
+
+ // If we try to append a none-expired batch with none zero batch sequence, the
+ // OutOfOrderSequenceException will throw.
+ assertThatThrownBy(() -> append(stateManager1, 2L, 10, 10L, false, 1000L))
+ .isInstanceOf(OutOfOrderSequenceException.class)
+ .hasMessageContaining(
+ "Out of order batch sequence for writer 2 at offset 10 in table-bucket "
+ + "TableBucket{tableId=1001, bucket=0} : 10 (incoming batch seq.), -1 (current batch seq.)");
+ }
+
@Test
void testDeleteSnapshotsBefore() throws IOException {
append(stateManager, writerId, 0, 0L);
@@ -322,7 +352,7 @@ public class WriterStateManagerTest {
@Test
void testSkipSnapshotIfOffsetUnchanged() throws IOException {
- append(stateManager, writerId, 0, 0L, 0L);
+ append(stateManager, writerId, 0, 0L, false, 0L);
stateManager.takeSnapshot();
assertThat(Objects.requireNonNull(logDir.listFiles()).length).isEqualTo(1);
@@ -475,7 +505,7 @@ public class WriterStateManagerTest {
private void append(
WriterStateManager stateManager, long writerId, int batchSequence, long offset) {
- append(stateManager, writerId, batchSequence, offset, System.currentTimeMillis());
+ append(stateManager, writerId, batchSequence, offset, false, System.currentTimeMillis());
}
private void append(
@@ -483,9 +513,15 @@ public class WriterStateManagerTest {
long writerId,
int batchSequence,
long offset,
- long timestamp) {
+ boolean isWriterInBatchExpired,
+ long lastTimestamp) {
WriterAppendInfo appendInfo = stateManager.prepareUpdate(writerId);
- appendInfo.appendDataBatch(batchSequence, new LogOffsetMetadata(offset), offset, timestamp);
+ appendInfo.appendDataBatch(
+ batchSequence,
+ new LogOffsetMetadata(offset),
+ offset,
+ isWriterInBatchExpired,
+ lastTimestamp);
stateManager.update(appendInfo);
stateManager.updateMapEndOffset(offset + 1);
}