Skip to content

Commit b8cbb62

Browse files
swuferhongswuferhong
authored
[server] Fix orphan segments after full truncation and tolerate sequence gaps during writer recovery (#4013)
Delete the previous active segment after opening its replacement at a different offset. Strengthen deletion assertions for all segment files and cover truncation below the first segment across restart so a higher-offset orphan cannot advance the recovered LEO. Rebuild writer state from persisted batches without applying online sequence validation. Warn about discontinuities for observability and verify that the recovered state survives another restart.
1 parent 1c2c447 commit b8cbb62

6 files changed

Lines changed: 223 additions & 20 deletions

File tree

‎fluss-server/src/main/java/org/apache/fluss/server/log/LocalLog.java‎

Lines changed: 18 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -324,19 +324,30 @@ void removeAndDeleteSegments(List<LogSegment> segmentsToDelete, SegmentDeletionR
324324
LogSegment createAndDeleteSegment(
325325
long newOffset, LogSegment segmentToDelete, SegmentDeletionReason reason)
326326
throws IOException {
327-
// delete the old segment.
328-
if (newOffset == segmentToDelete.getBaseOffset()) {
329-
deleteSegmentFiles(Collections.singletonList(segmentToDelete), reason);
327+
boolean replaceAtSameOffset = newOffset == segmentToDelete.getBaseOffset();
328+
if (replaceAtSameOffset) {
329+
segmentToDelete.changeFileSuffixes("", FlussPaths.DELETED_FILE_SUFFIX);
330330
}
331-
reason.logReason(Collections.singletonList(segmentToDelete));
332331

333-
// open a new segment.
334-
LogSegment newSegment = LogSegment.open(logTabletDir, newOffset, config, logFormat);
332+
LogSegment newSegment;
333+
try {
334+
newSegment = LogSegment.open(logTabletDir, newOffset, config, logFormat);
335+
} catch (IOException e) {
336+
if (replaceAtSameOffset) {
337+
try {
338+
segmentToDelete.changeFileSuffixes(FlussPaths.DELETED_FILE_SUFFIX, "");
339+
} catch (IOException rollbackException) {
340+
e.addSuppressed(rollbackException);
341+
}
342+
}
343+
throw e;
344+
}
335345
segments.add(newSegment);
336346

337-
if (newOffset != segmentToDelete.getBaseOffset()) {
347+
if (!replaceAtSameOffset) {
338348
segments.remove(segmentToDelete.getBaseOffset());
339349
}
350+
deleteSegmentFiles(Collections.singletonList(segmentToDelete), reason);
340351
return newSegment;
341352
}
342353

‎fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java‎

Lines changed: 18 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1301,7 +1301,12 @@ private LogAppendInfo analyzeAndValidateRecords(MemoryLogRecords records) {
13011301
}
13021302

13031303
// update write append info.
1304-
updateWriterAppendInfo(writerStateManager, batch, updatedWriters, isAppendAsLeader);
1304+
updateWriterAppendInfo(
1305+
writerStateManager,
1306+
batch,
1307+
updatedWriters,
1308+
isAppendAsLeader,
1309+
WriterAppendInfo.SequenceValidation.ENFORCE);
13051310
}
13061311
}
13071312

@@ -1439,15 +1444,17 @@ private static void updateWriterAppendInfo(
14391444
WriterStateManager writerStateManager,
14401445
LogRecordBatch batch,
14411446
Map<Long, WriterAppendInfo> writers,
1442-
boolean isAppendAsLeader) {
1447+
boolean isAppendAsLeader,
1448+
WriterAppendInfo.SequenceValidation sequenceValidation) {
14431449
long writerId = batch.writerId();
14441450
// update writers.
14451451
WriterAppendInfo appendInfo =
14461452
writers.computeIfAbsent(writerId, id -> writerStateManager.prepareUpdate(writerId));
14471453
appendInfo.append(
14481454
batch,
14491455
writerStateManager.isWriterInBatchExpired(System.currentTimeMillis(), batch),
1450-
isAppendAsLeader);
1456+
isAppendAsLeader,
1457+
sequenceValidation);
14511458
}
14521459

14531460
static void rebuildWriterState(
@@ -1565,7 +1572,14 @@ private static void loadWritersFromRecords(
15651572
Map<Long, WriterAppendInfo> loadedWriters = new HashMap<>();
15661573
for (LogRecordBatch batch : records.batches()) {
15671574
if (batch.hasWriterId()) {
1568-
updateWriterAppendInfo(writerStateManager, batch, loadedWriters, false);
1575+
// The records have already been accepted and persisted. Recovery rebuilds writer
1576+
// state without applying online client sequence validation.
1577+
updateWriterAppendInfo(
1578+
writerStateManager,
1579+
batch,
1580+
loadedWriters,
1581+
false,
1582+
WriterAppendInfo.SequenceValidation.WARN_AND_ACCEPT);
15691583
}
15701584
}
15711585
loadedWriters.values().forEach(writerStateManager::update);

‎fluss-server/src/main/java/org/apache/fluss/server/log/WriterAppendInfo.java‎

Lines changed: 68 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -21,13 +21,23 @@
2121
import org.apache.fluss.metadata.TableBucket;
2222
import org.apache.fluss.record.LogRecordBatch;
2323

24+
import org.slf4j.Logger;
25+
import org.slf4j.LoggerFactory;
26+
2427
import static org.apache.fluss.record.LogRecordBatchFormat.NO_BATCH_SEQUENCE;
2528

2629
/**
2730
* This class is used to validate the records appended by a given writer before they are written to
2831
* log. It's initialized with writer's state after the last successful append.
2932
*/
3033
public class WriterAppendInfo {
34+
private static final Logger LOG = LoggerFactory.getLogger(WriterAppendInfo.class);
35+
36+
enum SequenceValidation {
37+
ENFORCE,
38+
WARN_AND_ACCEPT
39+
}
40+
3141
private final long writerId;
3242
private final TableBucket tableBucket;
3343
private final WriterStateEntry currentEntry;
@@ -46,14 +56,23 @@ public long writerId() {
4656

4757
public void append(
4858
LogRecordBatch batch, boolean isWriterInBatchExpired, boolean isAppendAsLeader) {
59+
append(batch, isWriterInBatchExpired, isAppendAsLeader, SequenceValidation.ENFORCE);
60+
}
61+
62+
void append(
63+
LogRecordBatch batch,
64+
boolean isWriterInBatchExpired,
65+
boolean isAppendAsLeader,
66+
SequenceValidation sequenceValidation) {
4967
LogOffsetMetadata firstOffsetMetadata = new LogOffsetMetadata(batch.baseLogOffset());
5068
appendDataBatch(
5169
batch.batchSequence(),
5270
firstOffsetMetadata,
5371
batch.lastLogOffset(),
5472
isWriterInBatchExpired,
5573
isAppendAsLeader,
56-
batch.commitTimestamp());
74+
batch.commitTimestamp(),
75+
sequenceValidation);
5776
}
5877

5978
public void appendDataBatch(
@@ -63,7 +82,38 @@ public void appendDataBatch(
6382
boolean isWriterInBatchExpired,
6483
boolean isAppendAsLeader,
6584
long batchTimestamp) {
66-
maybeValidateDataBatch(batchSequence, isWriterInBatchExpired, lastOffset, isAppendAsLeader);
85+
appendDataBatch(
86+
batchSequence,
87+
firstOffsetMetadata,
88+
lastOffset,
89+
isWriterInBatchExpired,
90+
isAppendAsLeader,
91+
batchTimestamp,
92+
SequenceValidation.ENFORCE);
93+
}
94+
95+
private void appendDataBatch(
96+
int batchSequence,
97+
LogOffsetMetadata firstOffsetMetadata,
98+
long lastOffset,
99+
boolean isWriterInBatchExpired,
100+
boolean isAppendAsLeader,
101+
long batchTimestamp,
102+
SequenceValidation sequenceValidation) {
103+
maybeValidateDataBatch(
104+
batchSequence,
105+
isWriterInBatchExpired,
106+
lastOffset,
107+
isAppendAsLeader,
108+
sequenceValidation);
109+
appendDataBatch(batchSequence, firstOffsetMetadata, lastOffset, batchTimestamp);
110+
}
111+
112+
private void appendDataBatch(
113+
int batchSequence,
114+
LogOffsetMetadata firstOffsetMetadata,
115+
long lastOffset,
116+
long batchTimestamp) {
67117
updatedEntry.addBath(
68118
batchSequence,
69119
lastOffset,
@@ -75,21 +125,30 @@ private void maybeValidateDataBatch(
75125
int appendFirstSeq,
76126
boolean isWriterInBatchExpired,
77127
long lastOffset,
78-
boolean isAppendAsLeader) {
79-
int currentLastSeq =
80-
!updatedEntry.isEmpty()
81-
? updatedEntry.lastBatchSequence()
82-
: currentEntry.lastBatchSequence();
128+
boolean isAppendAsLeader,
129+
SequenceValidation sequenceValidation) {
130+
int currentLastSeq = currentLastBatchSequence();
83131
// must be in sequence, even for the first batch should start from 0
84132
if (!inSequence(currentLastSeq, appendFirstSeq, isWriterInBatchExpired, isAppendAsLeader)) {
85-
throw new OutOfOrderSequenceException(
133+
String message =
86134
String.format(
87135
"Out of order batch sequence for writer %s at offset %s in "
88136
+ "table-bucket %s : %s (incoming batch seq.), %s (current batch seq.)",
89-
writerId, lastOffset, tableBucket, appendFirstSeq, currentLastSeq));
137+
writerId, lastOffset, tableBucket, appendFirstSeq, currentLastSeq);
138+
if (sequenceValidation == SequenceValidation.WARN_AND_ACCEPT) {
139+
LOG.warn("{}. Accepting the persisted batch.", message);
140+
return;
141+
}
142+
throw new OutOfOrderSequenceException(message);
90143
}
91144
}
92145

146+
private int currentLastBatchSequence() {
147+
return !updatedEntry.isEmpty()
148+
? updatedEntry.lastBatchSequence()
149+
: currentEntry.lastBatchSequence();
150+
}
151+
93152
public WriterStateEntry toEntry() {
94153
return updatedEntry;
95154
}

‎fluss-server/src/test/java/org/apache/fluss/server/log/LocalLogTest.java‎

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@
3232
import org.apache.fluss.server.log.LocalLog.SegmentDeletionReason;
3333
import org.apache.fluss.server.metrics.group.TestingMetricGroups;
3434
import org.apache.fluss.utils.CloseableIterator;
35+
import org.apache.fluss.utils.FlussPaths;
3536

3637
import org.junit.jupiter.api.AfterEach;
3738
import org.junit.jupiter.api.BeforeEach;
@@ -291,6 +292,7 @@ void testCreateAndDeleteSegment() throws Exception {
291292
assertThat(localLog.getSegments().activeSegment()).isEqualTo(newActiveSegment);
292293
assertThat(localLog.getSegments().activeSegment()).isNotEqualTo(oldActiveSegment);
293294
assertThat(localLog.getSegments().activeSegment().getBaseOffset()).isEqualTo(newOffset);
295+
assertThat(oldActiveSegment.deleted()).isTrue();
294296
assertThat(localLog.getRecoveryPoint()).isEqualTo(0L);
295297
assertThat(localLog.getLocalLogEndOffset()).isEqualTo(newOffset);
296298
FetchDataInfo read =
@@ -301,6 +303,39 @@ void testCreateAndDeleteSegment() throws Exception {
301303
assertThat(read.getRecords().sizeInBytes()).isEqualTo(0);
302304
}
303305

306+
@Test
307+
void testCreateAndDeleteSegmentWithSameOffset() throws Exception {
308+
LogSegment oldActiveSegment = localLog.getSegments().activeSegment();
309+
oldActiveSegment.offsetIndex();
310+
oldActiveSegment.timeIndex();
311+
long baseOffset = oldActiveSegment.getBaseOffset();
312+
File oldLogFile = oldActiveSegment.getFileLogRecords().file();
313+
File oldOffsetIndexFile = oldActiveSegment.getLazyOffsetIndex().file();
314+
File oldTimeIndexFile = oldActiveSegment.timeIndexFile();
315+
assertThat(oldLogFile).exists();
316+
assertThat(oldOffsetIndexFile).exists();
317+
assertThat(oldTimeIndexFile).exists();
318+
319+
LogSegment newActiveSegment =
320+
localLog.createAndDeleteSegment(
321+
baseOffset, oldActiveSegment, SegmentDeletionReason.LOG_ROLL);
322+
323+
assertThat(localLog.getSegments().activeSegment()).isEqualTo(newActiveSegment);
324+
assertThat(newActiveSegment.getFileLogRecords().file()).isEqualTo(oldLogFile);
325+
assertThat(newActiveSegment.getLazyOffsetIndex().file()).isEqualTo(oldOffsetIndexFile);
326+
assertThat(newActiveSegment.timeIndexFile()).isEqualTo(oldTimeIndexFile);
327+
assertThat(oldActiveSegment.getFileLogRecords().file().getName())
328+
.endsWith(FlussPaths.DELETED_FILE_SUFFIX);
329+
assertThat(oldActiveSegment.getLazyOffsetIndex().file().getName())
330+
.endsWith(FlussPaths.DELETED_FILE_SUFFIX);
331+
assertThat(oldActiveSegment.timeIndexFile().getName())
332+
.endsWith(FlussPaths.DELETED_FILE_SUFFIX);
333+
assertThat(oldActiveSegment.deleted()).isTrue();
334+
assertThat(newActiveSegment.getFileLogRecords().file()).exists();
335+
assertThat(newActiveSegment.offsetIndex().file()).exists();
336+
assertThat(newActiveSegment.timeIndex().file()).exists();
337+
}
338+
304339
@Test
305340
void testTruncateFullyAndStartAt() throws Exception {
306341
for (int i = 0; i <= 7; i++) {
@@ -337,6 +372,19 @@ void testTruncateFullyAndStartAt() throws Exception {
337372
assertThat(read.getRecords().sizeInBytes()).isEqualTo(0);
338373
}
339374

375+
@Test
376+
void testTruncateFullyAndStartAtDeletesOldActiveSegmentFile() throws Exception {
377+
LogSegment oldActiveSegment = localLog.getSegments().activeSegment();
378+
File oldLogFile = oldActiveSegment.getFileLogRecords().file();
379+
assertThat(oldLogFile).exists();
380+
381+
localLog.truncateFullyAndStartAt(10L);
382+
383+
assertThat(localLog.getSegments().baseOffsets()).containsExactly(10L);
384+
assertThat(oldActiveSegment.deleted()).isTrue();
385+
assertThat(oldLogFile).doesNotExist();
386+
}
387+
340388
@Test
341389
void testTruncateTo() throws Exception {
342390
for (int i = 0; i <= 11; i++) {

‎fluss-server/src/test/java/org/apache/fluss/server/log/LogLoaderTest.java‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -319,6 +319,37 @@ void testWriterSnapshotRecoveryFromDiscontinuousBatchSequence() throws Exception
319319
.isEqualTo(13);
320320
}
321321

322+
@Test
323+
void testWriterStateRecoveryAcceptsBatchSequenceGap() throws Exception {
324+
LogTablet log = createLogTablet(true);
325+
long writerId = 1L;
326+
327+
log.appendAsFollower(
328+
genMemoryLogRecordsWithWriterId(
329+
Collections.singletonList(new Object[] {1, "a"}), writerId, 10, 0L));
330+
log.appendAsFollower(
331+
genMemoryLogRecordsWithWriterId(
332+
Collections.singletonList(new Object[] {2, "b"}), writerId, 11, 1L));
333+
log.roll(Optional.empty());
334+
335+
MemoryLogRecords recordsWithSequenceGap =
336+
genMemoryLogRecordsWithWriterId(
337+
Collections.singletonList(new Object[] {3, "c"}), writerId, 100, 2L);
338+
log.activeLogSegment().append(2L, clock.milliseconds(), 2L, recordsWithSequenceGap);
339+
log.close();
340+
341+
log = createLogTablet(false);
342+
assertThat(log.localLogEndOffset()).isEqualTo(3L);
343+
assertThat(log.writerStateManager().activeWriters().get(writerId).lastBatchSequence())
344+
.isEqualTo(100);
345+
346+
// The recovered state should be persisted in the new snapshot and survive another restart.
347+
log.close();
348+
log = createLogTablet(false);
349+
assertThat(log.writerStateManager().activeWriters().get(writerId).lastBatchSequence())
350+
.isEqualTo(100);
351+
}
352+
322353
@Test
323354
void testWriterSnapshotsRecoveryAfterCleanShutdown() throws Exception {
324355
LogTablet log = createLogTablet(true);

‎fluss-server/src/test/java/org/apache/fluss/server/log/LogTabletTest.java‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -366,6 +366,46 @@ void testWriterStateTruncateFullyAndStartAt() throws Exception {
366366
assertThat(latestWriterSnapshotOffset(log).get()).isEqualTo(29);
367367
}
368368

369+
@Test
370+
void testTruncateToBeforeFirstSegmentDeletesHigherOffsetSegment() throws Exception {
371+
logTablet.truncateFullyAndStartAt(10L);
372+
logTablet.appendAsLeader(
373+
genMemoryLogRecordsByObject(Collections.singletonList(new Object[] {1, "a"})));
374+
LogSegment oldActiveSegment = logTablet.activeLogSegment();
375+
assertThat(oldActiveSegment.getBaseOffset()).isEqualTo(10L);
376+
377+
logTablet.truncateTo(5L);
378+
379+
assertThat(oldActiveSegment.deleted()).isTrue();
380+
assertThat(logTablet.logSegments())
381+
.extracting(LogSegment::getBaseOffset)
382+
.containsExactly(5L);
383+
assertThat(logTablet.localLogEndOffset()).isEqualTo(5L);
384+
385+
logTablet.close();
386+
logTablet =
387+
LogTablet.create(
388+
tempDir,
389+
PhysicalTablePath.of(DATA1_TABLE_PATH),
390+
logDir,
391+
conf,
392+
new AtomicBoolean(
393+
conf.get(ConfigOptions.LOG_RETENTION_ROLL_ACTIVE_SEGMENT_ENABLED)),
394+
TestingMetricGroups.TABLET_SERVER_METRICS,
395+
0,
396+
scheduler,
397+
LogFormat.ARROW,
398+
1,
399+
false,
400+
SystemClock.getInstance(),
401+
false);
402+
403+
assertThat(logTablet.logSegments())
404+
.extracting(LogSegment::getBaseOffset)
405+
.containsExactly(5L);
406+
assertThat(logTablet.localLogEndOffset()).isEqualTo(5L);
407+
}
408+
369409
@Test
370410
void testWriterIdExpirationOnSegmentDeletion() throws Exception {
371411
long writerId1 = 1L;

0 commit comments

Comments
 (0)