From c1cf7c11938ad5dcbdb388359abd55e49fd7e8d4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=99=BD=E9=B5=BA?= Date: Wed, 9 Sep 2026 17:49:31 +0800 Subject: [PATCH 1/7] [server] Keep async KV flushes on WAL batch boundaries Track pending WAL batch ends through append, recovery, flush completion and truncation. Keep each independently completed native write on whole WAL batches so storage backpressure cannot publish an interior high watermark. Cover native write rejection, retry, byte and entry budgets, duplicate and empty batches, leader HW publication, follower truncation and WAL recovery with real storage tests. --- .../fluss/server/kv/KvRecoverHelper.java | 3 + .../org/apache/fluss/server/kv/KvTablet.java | 45 ++- .../server/kv/prewrite/KvPreWriteBuffer.java | 115 +++++--- .../server/kv/KvFlushBatchBoundaryTest.java | 264 ++++++++++++++++++ .../apache/fluss/server/kv/KvTabletTest.java | 80 +++++- .../kv/prewrite/KvPreWriteBufferTest.java | 82 +++++- 6 files changed, 529 insertions(+), 60 deletions(-) create mode 100644 fluss-server/src/test/java/org/apache/fluss/server/kv/KvFlushBatchBoundaryTest.java diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvRecoverHelper.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvRecoverHelper.java index a3afeec4bb..8852d64018 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvRecoverHelper.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvRecoverHelper.java @@ -253,6 +253,9 @@ private long readLogRecordsAndApply( rowCountUpdater, autoIncIdRangeUpdater, resumeRecordConsumer); + if (fetchIsolation == FetchIsolation.LOG_END) { + kvTablet.getKvPreWriteBuffer().registerBatchEnd(nextFetchOffset); + } } } return nextFetchOffset; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java index 76c603c8ec..12b8cdf748 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java @@ -123,9 +123,9 @@ public final class KvTablet { private static final long ROW_COUNT_DISABLED = -1; /** - * Max records per native write of the asynchronous flush; mirrors the batching capacity of + * KV entry budget per native write of the asynchronous flush; mirrors the batching capacity of * {@code RocksDBWriteBatchWrapper} (hundreds of keys per write batch is RocksDB best practice). - * Together with {@code writeBatchSize} this bounds one atomic native write. + * A single WAL batch may exceed this budget to keep its KV mutations atomic. */ private static final int MAX_RECORDS_PER_NATIVE_WRITE = 500; @@ -178,6 +178,8 @@ public final class KvTablet { private volatile long rowCount; + @Nullable private Runnable beforeNativeWrite; + @GuardedBy("kvLock") private volatile boolean isClosed = false; @@ -804,16 +806,27 @@ private LogAppendInfo putAsLeader( tableBucket)); } - return kvWriteProcessor.putAsLeader( - kvRecords, - targetColumns, - mergeMode, - kvStateAccessor, - originalPartitionName, - memoizedLakeLookup); + LogAppendInfo appendInfo = + kvWriteProcessor.putAsLeader( + kvRecords, + targetColumns, + mergeMode, + kvStateAccessor, + originalPartitionName, + memoizedLakeLookup); + if (!appendInfo.duplicated()) { + // KvWriteProcessor appends one WAL batch for each accepted KV batch. + kvPreWriteBuffer.registerBatchEnd(appendInfo.lastOffset() + 1); + } + return appendInfo; }); } + @VisibleForTesting + void setBeforeNativeWrite(@Nullable Runnable beforeNativeWrite) { + this.beforeNativeWrite = beforeNativeWrite; + } + @VisibleForTesting long localLogEndOffset() { return logTablet.localLogEndOffset(); @@ -958,11 +971,12 @@ private boolean tryAcquireScheduledFlush() { } /** - * Writes the prepared entries to RocksDB in segments of at most {@code - * MAX_RECORDS_PER_NATIVE_WRITE} records / {@code writeBatchSize} bytes. Each segment forms - * exactly one atomic native write (the writer has implicit flushes disabled) and is completed - * immediately after it lands, so {@code flushedLogOffset}/{@code rowCount} stay consistent with - * the RocksDB content even if a later segment is rejected by the no-slowdown gate. + * Writes the prepared entries to RocksDB in complete WAL batch groups targeting {@code + * MAX_RECORDS_PER_NATIVE_WRITE} records / {@code writeBatchSize} bytes. A single WAL batch may + * exceed these budgets. Each segment forms exactly one atomic native write (the writer has + * implicit flushes disabled) and is completed immediately after it lands, so {@code + * flushedLogOffset}/{@code rowCount} stay consistent with the RocksDB content even if a later + * segment is rejected by the no-slowdown gate. */ @GuardedBy("kvLock") private void writePreparedFlush(PreparedFlush preparedFlush) throws Exception { @@ -987,6 +1001,9 @@ private void writePreparedFlush(PreparedFlush preparedFlush) throws Exception { kvBatchWriter.put(entry.getKey().get(), value.get()); } } + if (beforeNativeWrite != null) { + beforeNativeWrite.run(); + } kvBatchWriter.flush(); completeFlushedSegment(segment); nextSegment++; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBuffer.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBuffer.java index aae74e19b7..0464a12cbe 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBuffer.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBuffer.java @@ -27,10 +27,12 @@ import javax.annotation.Nullable; import javax.annotation.concurrent.NotThreadSafe; +import java.util.ArrayDeque; import java.util.ArrayList; import java.util.Arrays; import java.util.Base64; import java.util.Collections; +import java.util.Deque; import java.util.HashMap; import java.util.Iterator; import java.util.LinkedList; @@ -95,6 +97,9 @@ public class KvPreWriteBuffer { // a linked list for all kv entries private final LinkedList allKvEntries = new LinkedList<>(); + // Exclusive ends of WAL batches whose KV updates have not yet completed flushing. + private final Deque batchEndOffsets = new ArrayDeque<>(); + // metrics related. private final Counter truncateAsDuplicatedCount; private final Counter truncateAsErrorCount; @@ -110,6 +115,15 @@ public KvPreWriteBuffer(TabletServerMetricGroup serverMetricGroup) { truncateAsErrorCount = serverMetricGroup.kvTruncateAsErrorCount(); } + /** Registers the exclusive end of a successfully appended or recovered WAL batch. */ + public void registerBatchEnd(long batchEndOffset) { + checkArgument( + batchEndOffsets.isEmpty() || batchEndOffset > batchEndOffsets.peekLast(), + "WAL batch ends must increase: %s", + batchEndOffset); + batchEndOffsets.addLast(batchEndOffset); + } + /** * Delete a key-value pair with the given key. * @@ -194,6 +208,9 @@ public void truncateTo(long targetLogSequenceNumber, TruncateReason truncateReas truncateAsErrorCount.inc(); } + while (!batchEndOffsets.isEmpty() && batchEndOffsets.peekLast() > targetLogSequenceNumber) { + batchEndOffsets.removeLast(); + } Iterator descIter = allKvEntries.descendingIterator(); while (descIter.hasNext()) { KvEntry entry = descIter.next(); @@ -255,7 +272,18 @@ public PreparedFlush prepareFlush(long exclusiveUpToLogSequenceNumber) { entries.add(entry); rowCountDiff += rowCountDelta(entry); } - return new PreparedFlush(exclusiveUpToLogSequenceNumber, entries, rowCountDiff); + List preparedBatchEnds = new ArrayList<>(); + for (long batchEnd : batchEndOffsets) { + if (batchEnd >= exclusiveUpToLogSequenceNumber) { + break; + } + preparedBatchEnds.add(batchEnd); + } + // The requested flush target is itself a WAL batch boundary. Keeping it also covers + // ranges with no KV mutations, including empty WAL batches. + preparedBatchEnds.add(exclusiveUpToLogSequenceNumber); + return new PreparedFlush( + exclusiveUpToLogSequenceNumber, entries, rowCountDiff, preparedBatchEnds); } /** Completes a prepared async flush and removes flushed entries from the buffer. */ @@ -281,6 +309,10 @@ public int completeFlush(PreparedFlush preparedFlush) { if (allKvEntries.isEmpty()) { maxLogSequenceNumber = -1; } + while (!batchEndOffsets.isEmpty() + && batchEndOffsets.peekFirst() <= preparedFlush.exclusiveUpToLogSequenceNumber) { + batchEndOffsets.removeFirst(); + } return preparedFlush.rowCountDiff; } @@ -471,12 +503,17 @@ public static final class PreparedFlush { private final long exclusiveUpToLogSequenceNumber; private final List entries; private final int rowCountDiff; + private final List batchEndOffsets; private PreparedFlush( - long exclusiveUpToLogSequenceNumber, List entries, int rowCountDiff) { + long exclusiveUpToLogSequenceNumber, + List entries, + int rowCountDiff, + List batchEndOffsets) { this.exclusiveUpToLogSequenceNumber = exclusiveUpToLogSequenceNumber; this.entries = entries; this.rowCountDiff = rowCountDiff; + this.batchEndOffsets = batchEndOffsets; } public long exclusiveUpToLogSequenceNumber() { @@ -496,64 +533,60 @@ public boolean isEmpty() { } /** - * Splits this prepared flush into consecutive segments so that each segment can be written - * to the kv storage as one native write and completed via {@link #completeFlush} - * independently. Segments preserve the entry order, so completing them in order keeps the - * list-prefix invariant checked by {@link #completeFlush}. + * Splits the prepared prefix only between complete WAL batches. Each segment is one atomic + * native write and can therefore publish its end independently after success. * - *

The upper log sequence number of every segment except the last one is the log sequence - * number of the first entry of the next segment, so advancing {@code flushedLogOffset} to a - * segment boundary never claims an entry that has not been written yet. The last segment - * keeps the original target so an empty tail still publishes the full flush range. - * - * @param maxBytesPerSegment max key/value payload bytes per segment, {@code <= 0} means - * unlimited; a single entry larger than the limit is kept as an oversized singleton - * @param maxRecordsPerSegment max record count per segment + * @param maxBytesPerSegment key/value payload budget, or non-positive for unlimited + * @param maxRecordsPerSegment KV entry budget; a single WAL batch may exceed either budget */ public List split(long maxBytesPerSegment, int maxRecordsPerSegment) { checkArgument(maxRecordsPerSegment > 0, "maxRecordsPerSegment must be positive."); - // Single pass: only boundary detection needs to visit every entry (byte sizes and - // row-count deltas). Segments are zero-copy subList views of the entry list. - List segments = null; + List segments = new ArrayList<>(); int segmentStart = 0; + int entryIndex = 0; + int segmentBatchStart = 0; long segmentBytes = 0; int segmentRowCountDiff = 0; - for (int i = 0; i < entries.size(); i++) { - KvEntry entry = entries.get(i); - long currentEntryBytes = entryBytes(entry.getKey(), entry.getValue()); - boolean hasEntries = i > segmentStart; - boolean recordLimitReached = hasEntries && i - segmentStart >= maxRecordsPerSegment; - boolean byteLimitExceeded = - hasEntries - && maxBytesPerSegment > 0 - && currentEntryBytes > maxBytesPerSegment - segmentBytes; - if (recordLimitReached || byteLimitExceeded) { - // Seal the current segment right before this entry: all entries below this - // entry's log sequence number are contained in the sealed segments. - if (segments == null) { - segments = new ArrayList<>(); - } + for (int batchIndex = 0; batchIndex < batchEndOffsets.size(); batchIndex++) { + long batchEnd = batchEndOffsets.get(batchIndex); + int batchStart = entryIndex; + long batchBytes = 0; + int batchRowCountDiff = 0; + while (entryIndex < entries.size() + && entries.get(entryIndex).getLogSequenceNumber() < batchEnd) { + KvEntry entry = entries.get(entryIndex++); + batchBytes += entryBytes(entry.getKey(), entry.getValue()); + batchRowCountDiff += rowCountDelta(entry); + } + boolean hasEntries = batchStart > segmentStart; + if (hasEntries + && entryIndex > batchStart + && (entryIndex - segmentStart > maxRecordsPerSegment + || (maxBytesPerSegment > 0 + && batchBytes > maxBytesPerSegment - segmentBytes))) { segments.add( new PreparedFlush( - entry.getLogSequenceNumber(), - entries.subList(segmentStart, i), - segmentRowCountDiff)); - segmentStart = i; + batchEndOffsets.get(batchIndex - 1), + entries.subList(segmentStart, batchStart), + segmentRowCountDiff, + batchEndOffsets.subList(segmentBatchStart, batchIndex))); + segmentStart = batchStart; + segmentBatchStart = batchIndex; segmentBytes = 0; segmentRowCountDiff = 0; } - segmentBytes += currentEntryBytes; - segmentRowCountDiff += rowCountDelta(entry); + segmentBytes += batchBytes; + segmentRowCountDiff += batchRowCountDiff; } - if (segments == null) { - // Everything fits into a single segment: reuse this prepared flush as-is. + if (segments.isEmpty()) { return Collections.singletonList(this); } segments.add( new PreparedFlush( exclusiveUpToLogSequenceNumber, entries.subList(segmentStart, entries.size()), - segmentRowCountDiff)); + segmentRowCountDiff, + batchEndOffsets.subList(segmentBatchStart, batchEndOffsets.size()))); return segments; } } diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvFlushBatchBoundaryTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvFlushBatchBoundaryTest.java new file mode 100644 index 0000000000..4c7c9ba927 --- /dev/null +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvFlushBatchBoundaryTest.java @@ -0,0 +1,264 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.server.kv; + +import org.apache.fluss.config.Configuration; +import org.apache.fluss.exception.StorageBackpressureException; +import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.record.KvRecordBatch; +import org.apache.fluss.record.KvRecordTestUtils; +import org.apache.fluss.rpc.protocol.MergeMode; +import org.apache.fluss.server.entity.NotifyLeaderAndIsrData; +import org.apache.fluss.server.log.FetchIsolation; +import org.apache.fluss.server.replica.Replica; +import org.apache.fluss.server.replica.ReplicaTestBase; +import org.apache.fluss.server.zk.data.LeaderAndIsr; +import org.apache.fluss.utils.types.Tuple2; + +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.apache.fluss.record.TestData.DATA1_PHYSICAL_TABLE_PATH_PK; +import static org.apache.fluss.record.TestData.DATA1_ROW_TYPE; +import static org.apache.fluss.record.TestData.DATA1_TABLE_ID_PK; +import static org.apache.fluss.record.TestData.DEFAULT_SCHEMA_ID; +import static org.apache.fluss.server.coordinator.CoordinatorContext.INITIAL_COORDINATOR_EPOCH; +import static org.apache.fluss.testutils.DataTestUtils.genKvRecordBatch; +import static org.apache.fluss.testutils.DataTestUtils.genKvRecords; +import static org.apache.fluss.testutils.DataTestUtils.getKeyValuePairs; +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests asynchronous KV flush boundaries through leader HW publication and WAL recovery. */ +class KvFlushBatchBoundaryTest extends ReplicaTestBase { + @Override + protected KvFlushScheduler createTestKvFlushScheduler(Configuration conf) { + return new KvFlushScheduler() { + @Override + public void enqueue(KvTablet tablet) {} + + @Override + public void retryLater(KvTablet tablet, long delayMs) {} + + @Override + public void close() {} + }; + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void testBackpressureAndRetryAfterWalRecovery(boolean recover) throws Exception { + Replica replica = createLeader(); + try { + appendBatches(replica); + long target = replica.getLocalLogEndOffset(); + assertThat(replica.getLogHighWatermark()).isZero(); + if (recover) { + replica.makeFollower(followerState()); + assertThat(replica.isLeader()).isFalse(); + replica.makeLeader(state(2)); + } + KvTablet tablet = replica.getKvTablet(); + rejectSecondWrite(tablet); + tablet.requestFlush( + target, + failure -> { + throw new AssertionError(failure); + }); + tablet.runScheduledFlush(); + assertCommittedPrefix(replica); + + tablet.setBeforeNativeWrite(null); + tablet.requestFlushRetry(); + tablet.runScheduledFlush(); + assertThat(tablet.getFlushedLogOffset()).isEqualTo(target); + assertThat(replica.getLogHighWatermark()).isEqualTo(target); + assertThat(replica.checkEnoughReplicasReachOffset(target).f0).isTrue(); + assertThat(tablet.getRowCount()).isEqualTo(601); + assertThat(tablet.getKvPreWriteBuffer().pendingFlushBytes()).isZero(); + assertValue(tablet, 0, "after"); + assertValue(tablet, 600, "tail"); + List ends = new ArrayList<>(); + replica.getLogTablet() + .read(0, Integer.MAX_VALUE, FetchIsolation.HIGH_WATERMARK, true) + .getRecords() + .batches() + .forEach(batch -> ends.add(batch.nextLogOffset())); + assertThat(ends).containsExactly(600L, 1800L, 1801L, 1802L); + } finally { + try { + replica.delete(); + } finally { + replicaManager + .getServerMetricGroup() + .removeTableBucketMetricGroup( + DATA1_PHYSICAL_TABLE_PATH_PK.getTablePath(), + new TableBucket(DATA1_TABLE_ID_PK, 1)); + } + } + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void testFollowerTruncationPreservesCommittedBatch(boolean recover) throws Exception { + Replica replica = createLeader(); + try { + appendBatches(replica); + KvTablet tablet = replica.getKvTablet(); + rejectSecondWrite(tablet); + tablet.runScheduledFlush(); + assertCommittedPrefix(replica); + long hw = replica.getLogHighWatermark(); + replica.makeFollower(followerState()); + assertThat(replica.isLeader()).isFalse(); + replica.truncateTo(hw); + assertThat(replica.getLocalLogEndOffset()).isEqualTo(hw); + assertThat( + replica.getLogTablet() + .read(0, Integer.MAX_VALUE, FetchIsolation.LOG_END, true) + .getRecords() + .batches()) + .singleElement() + .satisfies(batch -> assertThat(batch.nextLogOffset()).isEqualTo(hw)); + if (recover) { + replica.makeLeader(state(2)); + assertThat(replica.getKvTablet().getRowCount()).isEqualTo(600); + assertValue(replica.getKvTablet(), 0, "before"); + replica.putRecordsToLeader( + genKvRecordBatch(new Object[] {600, "tail"}), null, MergeMode.DEFAULT, 0); + replica.getKvTablet().runScheduledFlush(); + assertThat(replica.getLogHighWatermark()).isEqualTo(601); + assertValue(replica.getKvTablet(), 600, "tail"); + } + } finally { + try { + replica.delete(); + } finally { + replicaManager + .getServerMetricGroup() + .removeTableBucketMetricGroup( + DATA1_PHYSICAL_TABLE_PATH_PK.getTablePath(), + new TableBucket(DATA1_TABLE_ID_PK, 1)); + } + } + } + + private Replica createLeader() throws Exception { + Replica replica = + makeKvReplica(DATA1_PHYSICAL_TABLE_PATH_PK, new TableBucket(DATA1_TABLE_ID_PK, 1)); + replica.makeLeader(state(0)); + return replica; + } + + private NotifyLeaderAndIsrData followerState() { + int leader = TABLET_SERVER_ID + 1; + return new NotifyLeaderAndIsrData( + DATA1_PHYSICAL_TABLE_PATH_PK, + new TableBucket(DATA1_TABLE_ID_PK, 1), + Arrays.asList(TABLET_SERVER_ID, leader), + new LeaderAndIsr( + leader, + 1, + Collections.singletonList(leader), + Collections.emptyList(), + INITIAL_COORDINATOR_EPOCH, + 1)); + } + + private NotifyLeaderAndIsrData state(int epoch) { + return new NotifyLeaderAndIsrData( + DATA1_PHYSICAL_TABLE_PATH_PK, + new TableBucket(DATA1_TABLE_ID_PK, 1), + Collections.singletonList(TABLET_SERVER_ID), + new LeaderAndIsr( + TABLET_SERVER_ID, + epoch, + Collections.singletonList(TABLET_SERVER_ID), + Collections.emptyList(), + INITIAL_COORDINATOR_EPOCH, + epoch)); + } + + private void appendBatches(Replica replica) throws Exception { + Object[][] before = new Object[600][]; + Object[][] after = new Object[600][]; + for (int i = 0; i < 600; i++) { + before[i] = new Object[] {i, "before"}; + after[i] = new Object[] {i, "after"}; + } + KvRecordBatch first = + KvRecordTestUtils.KvRecordBatchFactory.of(DEFAULT_SCHEMA_ID) + .ofRecords(genKvRecords(before), 42L, 0); + replica.putRecordsToLeader(first, null, MergeMode.DEFAULT, 0); + // A duplicate must neither retain speculative KV updates nor register a second boundary. + replica.putRecordsToLeader(first, null, MergeMode.DEFAULT, 0); + replica.putRecordsToLeader(genKvRecordBatch(after), null, MergeMode.DEFAULT, 0); + byte[] missingKey = getKeyValuePairs(genKvRecords(new Object[] {999, "missing"})).get(0).f0; + replica.putRecordsToLeader( + KvRecordTestUtils.KvRecordBatchFactory.of(DEFAULT_SCHEMA_ID) + .ofRecords( + KvRecordTestUtils.KvRecordFactory.of(DATA1_ROW_TYPE) + .ofRecord(missingKey, null)), + null, + MergeMode.DEFAULT, + 0); + replica.putRecordsToLeader( + genKvRecordBatch(new Object[] {600, "tail"}), null, MergeMode.DEFAULT, 0); + assertThat(replica.getLocalLogEndOffset()).isEqualTo(1802); + } + + private void rejectSecondWrite(KvTablet tablet) { + AtomicInteger attempts = new AtomicInteger(); + tablet.setBeforeNativeWrite( + () -> { + if (attempts.incrementAndGet() == 2) { + throw new StorageBackpressureException("Reject second native write"); + } + }); + } + + private void assertCommittedPrefix(Replica replica) throws Exception { + KvTablet tablet = replica.getKvTablet(); + assertThat(tablet.getFlushState()).isEqualTo(KvTablet.FlushState.STORAGE_BLOCKED); + assertThat(tablet.getFlushedLogOffset()).isEqualTo(600); + assertThat(replica.getLogHighWatermark()).isEqualTo(600); + assertThat(replica.checkEnoughReplicasReachOffset(600).f0).isTrue(); + assertThat(replica.checkEnoughReplicasReachOffset(1800).f0).isFalse(); + assertThat(tablet.getRowCount()).isEqualTo(600); + assertValue(tablet, 0, "before"); + assertThat( + replica.getLogTablet() + .read(0, Integer.MAX_VALUE, FetchIsolation.HIGH_WATERMARK, true) + .getRecords() + .batches()) + .singleElement() + .satisfies(batch -> assertThat(batch.nextLogOffset()).isEqualTo(600)); + } + + private void assertValue(KvTablet tablet, int key, String value) throws Exception { + Tuple2 expected = + getKeyValuePairs(genKvRecords(new Object[] {key, value})).get(0); + assertThat(tablet.multiGet(Collections.singletonList(expected.f0)).get(0).toByteArray()) + .containsExactly(expected.f1); + } +} diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java index 4bd4e58e5b..8ee1c4e401 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java @@ -113,6 +113,7 @@ import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; import java.util.stream.IntStream; @@ -1989,6 +1990,74 @@ void testOrphanedPreparedFlushEntriesAreRetried() throws Exception { assertThat(kvTablet.getKvPreWriteBuffer().pendingFlushBytes()).isGreaterThan(0); } + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void testBackpressurePreservesCompleteWalBatch(boolean byteLimit) throws Exception { + if (byteLimit) { + conf.set(ConfigOptions.KV_WRITE_BATCH_SIZE, new MemorySize(64)); + } + ManualKvFlushScheduler scheduler = new ManualKvFlushScheduler(conf); + PhysicalTablePath path = PhysicalTablePath.of(TablePath.of("testDb", "t1")); + schemaGetter = new TestingSchemaGetter(new SchemaInfo(DATA1_SCHEMA_PK, schemaId)); + logTablet = createLogTablet(tempLogDir, 0L, path); + kvTablet = + createKvTablet( + path, + logTablet.getTableBucket(), + logTablet, + tmpKvDir, + schemaGetter, + new HashMap<>(), + scheduler); + List records = new ArrayList<>(); + for (int i = 0; i < 600; i++) { + records.add(kvRecordFactory.ofRecord("key" + i, new Object[] {i, "value"})); + } + kvTablet.putAsLeader(kvRecordBatchFactory.ofRecords(records), null); + long firstBatchEnd = logTablet.localLogEndOffset(); + kvTablet.putAsLeader( + kvRecordBatchFactory.ofRecords( + kvRecordFactory.ofRecord("later", new Object[] {601, "later"})), + null); + long target = logTablet.localLogEndOffset(); + AtomicInteger writes = new AtomicInteger(); + kvTablet.setBeforeNativeWrite( + () -> { + if (writes.incrementAndGet() == 2) { + throw new StorageBackpressureException("Reject the second native write"); + } + }); + kvTablet.requestFlush( + target, + failure -> { + throw new AssertionError(failure); + }); + kvTablet.runScheduledFlush(); + assertThat(kvTablet.getFlushState()).isEqualTo(KvTablet.FlushState.STORAGE_BLOCKED); + assertThat(kvTablet.getFlushedLogOffset()).isEqualTo(firstBatchEnd); + assertThat(kvTablet.getRowCount()).isEqualTo(600); + assertThat(kvTablet.multiGet(Arrays.asList("key0".getBytes(), "key599".getBytes()))) + .noneMatch(Objects::isNull); + assertThat(kvTablet.multiGet(Collections.singletonList("later".getBytes())).get(0)) + .isNull(); + logTablet.updateHighWatermark(kvTablet.getFlushedLogOffset()); + assertThat( + logTablet + .read(0, Integer.MAX_VALUE, FetchIsolation.HIGH_WATERMARK, true) + .getRecords() + .batches()) + .singleElement() + .satisfies(batch -> assertThat(batch.nextLogOffset()).isEqualTo(firstBatchEnd)); + kvTablet.setBeforeNativeWrite(null); + kvTablet.requestFlushRetry(); + kvTablet.runScheduledFlush(); + assertThat(kvTablet.getFlushedLogOffset()).isEqualTo(target); + assertThat(kvTablet.getRowCount()).isEqualTo(601); + assertThat(kvTablet.getKvPreWriteBuffer().pendingFlushBytes()).isZero(); + assertThat(kvTablet.multiGet(Collections.singletonList("later".getBytes())).get(0)) + .isNotNull(); + } + @Test void testScheduledFlushWritesLargePreparedRangeInSegments() throws Exception { ManualKvFlushScheduler manualScheduler = new ManualKvFlushScheduler(conf); @@ -2005,19 +2074,24 @@ void testScheduledFlushWritesLargePreparedRangeInSegments() throws Exception { new HashMap<>(), manualScheduler); - // 1200 records force the prepared range to be written as multiple native segments - // (500 records each), exercising the per-segment completion of the scheduled flush. + // Three complete 400-entry WAL batches require three native writes with a 500-entry budget. int recordCount = 1200; List records = new ArrayList<>(recordCount); for (int i = 0; i < recordCount; i++) { records.add(kvRecordFactory.ofRecord("key" + i, new Object[] {i, "v" + i})); } - kvTablet.putAsLeader(kvRecordBatchFactory.ofRecords(records), null); + for (int start = 0; start < recordCount; start += 400) { + kvTablet.putAsLeader( + kvRecordBatchFactory.ofRecords(records.subList(start, start + 400)), null); + } + AtomicInteger nativeWrites = new AtomicInteger(); + kvTablet.setBeforeNativeWrite(nativeWrites::incrementAndGet); long flushOffset = logTablet.localLogEndOffset(); kvTablet.requestFlush(flushOffset, NOPErrorHandler.INSTANCE); kvTablet.runScheduledFlush(); + assertThat(nativeWrites.get()).isEqualTo(3); assertThat(kvTablet.getFlushedLogOffset()).isEqualTo(flushOffset); assertThat(kvTablet.getRowCount()).isEqualTo(recordCount); assertThat(kvTablet.getKvPreWriteBuffer().pendingFlushBytes()).isEqualTo(0); diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferTest.java index 26d49807bb..4cca584985 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferTest.java @@ -235,12 +235,14 @@ void testSplitPreparedFlushByRecordCount() { } bufferDelete(buffer, "key2", 4); + for (KvPreWriteBuffer.KvEntry entry : buffer.getAllKvEntries()) { + buffer.registerBatchEnd(entry.getLogSequenceNumber() + 1); + } PreparedFlush preparedFlush = buffer.prepareFlush(10); List segments = preparedFlush.split(0, 2); assertThat(segments).hasSize(3); - // Every segment boundary is the lsn of the first entry of the next segment; the last - // segment keeps the original target so the full flush range gets published. + // Each entry here is a complete WAL batch. The last segment includes the empty tail. assertThat(segments.get(0).entries()).hasSize(2); assertThat(segments.get(0).exclusiveUpToLogSequenceNumber()).isEqualTo(2); assertThat(segments.get(1).entries()).hasSize(2); @@ -268,6 +270,9 @@ void testSplitPreparedFlushByByteSize() { bufferInsert(buffer, "b", "1234", 1); bufferInsert(buffer, "c", "1234", 2); + for (KvPreWriteBuffer.KvEntry entry : buffer.getAllKvEntries()) { + buffer.registerBatchEnd(entry.getLogSequenceNumber() + 1); + } PreparedFlush preparedFlush = buffer.prepareFlush(3); List segments = preparedFlush.split(10, Integer.MAX_VALUE); @@ -288,6 +293,9 @@ void testSplitPreparedFlushWithOversizedEntry() { bufferInsert(buffer, "a", "1234567890", 0); bufferInsert(buffer, "b", "123", 1); + for (KvPreWriteBuffer.KvEntry entry : buffer.getAllKvEntries()) { + buffer.registerBatchEnd(entry.getLogSequenceNumber() + 1); + } PreparedFlush preparedFlush = buffer.prepareFlush(2); List segments = preparedFlush.split(10, Integer.MAX_VALUE); @@ -307,6 +315,9 @@ void testSplitPreparedFlushUsesFirstReachedLimit() { bufferInsert(buffer, "c", "12345", 2); bufferInsert(buffer, "d", "1234", 3); + for (KvPreWriteBuffer.KvEntry entry : buffer.getAllKvEntries()) { + buffer.registerBatchEnd(entry.getLogSequenceNumber() + 1); + } PreparedFlush preparedFlush = buffer.prepareFlush(4); List segments = preparedFlush.split(10, 2); @@ -329,6 +340,9 @@ void testCompletePrefixSegmentsAndAbortRest() { bufferInsert(buffer, "c", "1234", 2); bufferInsert(buffer, "d", "12345", 3); + for (KvPreWriteBuffer.KvEntry entry : buffer.getAllKvEntries()) { + buffer.registerBatchEnd(entry.getLogSequenceNumber() + 1); + } PreparedFlush preparedFlush = buffer.prepareFlush(4); List segments = preparedFlush.split(10, Integer.MAX_VALUE); assertThat(segments).hasSize(3); @@ -349,6 +363,70 @@ void testCompletePrefixSegmentsAndAbortRest() { assertThat(retry.rowCountDiff()).isEqualTo(3); } + @Test + void testSplitKeepsOversizedBatchAndEmptyTailTogether() { + KvPreWriteBuffer buffer = new KvPreWriteBuffer(TestingMetricGroups.TABLET_SERVER_METRICS); + buffer.registerBatchEnd(1); + bufferInsert(buffer, "a", "large-value", 1); + bufferInsert(buffer, "b", "large-value", 2); + buffer.registerBatchEnd(3); + buffer.registerBatchEnd(4); + bufferUpdate(buffer, "a", "new-value", 5); + buffer.registerBatchEnd(6); + buffer.registerBatchEnd(7); + List segments = buffer.prepareFlush(7).split(2, 1); + assertThat(segments).hasSize(2); + assertThat(segments.get(0).exclusiveUpToLogSequenceNumber()).isEqualTo(4); + assertThat(segments.get(0).entries()).hasSize(2); + assertThat(buffer.completeFlush(segments.get(0))).isEqualTo(2); + buffer.abortFlush(segments.get(1)); + List retry = buffer.prepareFlush(7).split(2, 1); + assertThat(retry) + .singleElement() + .satisfies( + segment -> { + assertThat(segment.exclusiveUpToLogSequenceNumber()).isEqualTo(7); + assertThat(segment.entries()).hasSize(1); + assertThat(buffer.completeFlush(segment)).isZero(); + }); + assertThat(buffer.pendingFlushBytes()).isZero(); + } + + @Test + void testTruncateAndReappendBatchBoundaries() { + KvPreWriteBuffer buffer = new KvPreWriteBuffer(TestingMetricGroups.TABLET_SERVER_METRICS); + bufferInsert(buffer, "a", "first", 0); + buffer.registerBatchEnd(1); + bufferUpdate(buffer, "a", "discarded", 2); + buffer.registerBatchEnd(3); + buffer.registerBatchEnd(4); + buffer.truncateTo(1, TruncateReason.ERROR); + assertThat(getValue(buffer, "a")).isEqualTo("first"); + bufferUpdate(buffer, "a", "replacement", 2); + buffer.registerBatchEnd(3); + bufferInsert(buffer, "b", "second", 3); + buffer.registerBatchEnd(4); + List segments = buffer.prepareFlush(4).split(0, 1); + assertThat(segments) + .extracting(PreparedFlush::exclusiveUpToLogSequenceNumber) + .containsExactly(1L, 3L, 4L); + assertThat(buffer.completeFlush(segments.get(0))).isEqualTo(1); + assertThat(getValue(buffer, "a")).isEqualTo("replacement"); + assertThat(buffer.completeFlush(segments.get(1))).isZero(); + assertThat(buffer.completeFlush(segments.get(2))).isEqualTo(1); + assertThat(buffer.getAllKvEntries()).isEmpty(); + assertThat(buffer.pendingFlushBytes()).isZero(); + // The next range must not contain completed boundaries, even when it starts empty. + buffer.registerBatchEnd(5); + bufferInsert(buffer, "c", "next", 5); + buffer.registerBatchEnd(6); + assertThat(buffer.prepareFlush(6).split(0, 1)) + .singleElement() + .satisfies( + segment -> + assertThat(segment.exclusiveUpToLogSequenceNumber()).isEqualTo(6)); + } + private static void bufferInsert( KvPreWriteBuffer kvPreWriteBuffer, String key, String value, int elementCount) { kvPreWriteBuffer.insert(toKey(key), value.getBytes(), elementCount); From 810a6b3182fb1717e91a5b22629dd28835dfb87f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=99=BD=E9=B5=BA?= Date: Wed, 9 Sep 2026 19:00:57 +0800 Subject: [PATCH 2/7] [server] Reduce flush boundary allocations and duplicate tests Reuse boxed batch ends and bypass segment scanning for a single WAL batch. Allocate the segment list only when splitting is required. Consolidate byte and entry budget coverage in the real Replica recovery test and retain the complete truncation and recovery scenario. --- .../server/kv/prewrite/KvPreWriteBuffer.java | 15 ++-- .../server/kv/KvFlushBatchBoundaryTest.java | 68 +++++++++---------- .../apache/fluss/server/kv/KvTabletTest.java | 68 ------------------- 3 files changed, 44 insertions(+), 107 deletions(-) diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBuffer.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBuffer.java index 0464a12cbe..fd7f54fdde 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBuffer.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBuffer.java @@ -119,8 +119,7 @@ public KvPreWriteBuffer(TabletServerMetricGroup serverMetricGroup) { public void registerBatchEnd(long batchEndOffset) { checkArgument( batchEndOffsets.isEmpty() || batchEndOffset > batchEndOffsets.peekLast(), - "WAL batch ends must increase: %s", - batchEndOffset); + "WAL batch ends must increase."); batchEndOffsets.addLast(batchEndOffset); } @@ -273,7 +272,7 @@ public PreparedFlush prepareFlush(long exclusiveUpToLogSequenceNumber) { rowCountDiff += rowCountDelta(entry); } List preparedBatchEnds = new ArrayList<>(); - for (long batchEnd : batchEndOffsets) { + for (Long batchEnd : batchEndOffsets) { if (batchEnd >= exclusiveUpToLogSequenceNumber) { break; } @@ -541,7 +540,10 @@ public boolean isEmpty() { */ public List split(long maxBytesPerSegment, int maxRecordsPerSegment) { checkArgument(maxRecordsPerSegment > 0, "maxRecordsPerSegment must be positive."); - List segments = new ArrayList<>(); + if (batchEndOffsets.size() == 1) { + return Collections.singletonList(this); + } + List segments = null; int segmentStart = 0; int entryIndex = 0; int segmentBatchStart = 0; @@ -564,6 +566,9 @@ public List split(long maxBytesPerSegment, int maxRecordsPerSegme && (entryIndex - segmentStart > maxRecordsPerSegment || (maxBytesPerSegment > 0 && batchBytes > maxBytesPerSegment - segmentBytes))) { + if (segments == null) { + segments = new ArrayList<>(); + } segments.add( new PreparedFlush( batchEndOffsets.get(batchIndex - 1), @@ -578,7 +583,7 @@ public List split(long maxBytesPerSegment, int maxRecordsPerSegme segmentBytes += batchBytes; segmentRowCountDiff += batchRowCountDiff; } - if (segments.isEmpty()) { + if (segments == null) { return Collections.singletonList(this); } segments.add( diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvFlushBatchBoundaryTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvFlushBatchBoundaryTest.java index 4c7c9ba927..43b16bceb8 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvFlushBatchBoundaryTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvFlushBatchBoundaryTest.java @@ -17,7 +17,9 @@ package org.apache.fluss.server.kv; +import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; +import org.apache.fluss.config.MemorySize; import org.apache.fluss.exception.StorageBackpressureException; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.record.KvRecordBatch; @@ -30,8 +32,9 @@ import org.apache.fluss.server.zk.data.LeaderAndIsr; import org.apache.fluss.utils.types.Tuple2; +import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.ValueSource; +import org.junit.jupiter.params.provider.CsvSource; import java.util.ArrayList; import java.util.Arrays; @@ -66,8 +69,12 @@ public void close() {} } @ParameterizedTest - @ValueSource(booleans = {false, true}) - void testBackpressureAndRetryAfterWalRecovery(boolean recover) throws Exception { + @CsvSource({"false, false", "false, true", "true, false", "true, true"}) + void testBackpressureAndRetryAfterWalRecovery(boolean recover, boolean byteLimit) + throws Exception { + if (byteLimit) { + conf.set(ConfigOptions.KV_WRITE_BATCH_SIZE, new MemorySize(64)); + } Replica replica = createLeader(); try { appendBatches(replica); @@ -106,21 +113,12 @@ void testBackpressureAndRetryAfterWalRecovery(boolean recover) throws Exception .forEach(batch -> ends.add(batch.nextLogOffset())); assertThat(ends).containsExactly(600L, 1800L, 1801L, 1802L); } finally { - try { - replica.delete(); - } finally { - replicaManager - .getServerMetricGroup() - .removeTableBucketMetricGroup( - DATA1_PHYSICAL_TABLE_PATH_PK.getTablePath(), - new TableBucket(DATA1_TABLE_ID_PK, 1)); - } + deleteReplica(replica); } } - @ParameterizedTest - @ValueSource(booleans = {false, true}) - void testFollowerTruncationPreservesCommittedBatch(boolean recover) throws Exception { + @Test + void testFollowerTruncationPreservesCommittedBatch() throws Exception { Replica replica = createLeader(); try { appendBatches(replica); @@ -140,26 +138,28 @@ void testFollowerTruncationPreservesCommittedBatch(boolean recover) throws Excep .batches()) .singleElement() .satisfies(batch -> assertThat(batch.nextLogOffset()).isEqualTo(hw)); - if (recover) { - replica.makeLeader(state(2)); - assertThat(replica.getKvTablet().getRowCount()).isEqualTo(600); - assertValue(replica.getKvTablet(), 0, "before"); - replica.putRecordsToLeader( - genKvRecordBatch(new Object[] {600, "tail"}), null, MergeMode.DEFAULT, 0); - replica.getKvTablet().runScheduledFlush(); - assertThat(replica.getLogHighWatermark()).isEqualTo(601); - assertValue(replica.getKvTablet(), 600, "tail"); - } + replica.makeLeader(state(2)); + assertThat(replica.getKvTablet().getRowCount()).isEqualTo(600); + assertValue(replica.getKvTablet(), 0, "before"); + replica.putRecordsToLeader( + genKvRecordBatch(new Object[] {600, "tail"}), null, MergeMode.DEFAULT, 0); + replica.getKvTablet().runScheduledFlush(); + assertThat(replica.getLogHighWatermark()).isEqualTo(601); + assertValue(replica.getKvTablet(), 600, "tail"); } finally { - try { - replica.delete(); - } finally { - replicaManager - .getServerMetricGroup() - .removeTableBucketMetricGroup( - DATA1_PHYSICAL_TABLE_PATH_PK.getTablePath(), - new TableBucket(DATA1_TABLE_ID_PK, 1)); - } + deleteReplica(replica); + } + } + + private void deleteReplica(Replica replica) throws Exception { + try { + replica.delete(); + } finally { + replicaManager + .getServerMetricGroup() + .removeTableBucketMetricGroup( + DATA1_PHYSICAL_TABLE_PATH_PK.getTablePath(), + new TableBucket(DATA1_TABLE_ID_PK, 1)); } } diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java index 8ee1c4e401..8fd66e7443 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java @@ -1990,74 +1990,6 @@ void testOrphanedPreparedFlushEntriesAreRetried() throws Exception { assertThat(kvTablet.getKvPreWriteBuffer().pendingFlushBytes()).isGreaterThan(0); } - @ParameterizedTest - @ValueSource(booleans = {false, true}) - void testBackpressurePreservesCompleteWalBatch(boolean byteLimit) throws Exception { - if (byteLimit) { - conf.set(ConfigOptions.KV_WRITE_BATCH_SIZE, new MemorySize(64)); - } - ManualKvFlushScheduler scheduler = new ManualKvFlushScheduler(conf); - PhysicalTablePath path = PhysicalTablePath.of(TablePath.of("testDb", "t1")); - schemaGetter = new TestingSchemaGetter(new SchemaInfo(DATA1_SCHEMA_PK, schemaId)); - logTablet = createLogTablet(tempLogDir, 0L, path); - kvTablet = - createKvTablet( - path, - logTablet.getTableBucket(), - logTablet, - tmpKvDir, - schemaGetter, - new HashMap<>(), - scheduler); - List records = new ArrayList<>(); - for (int i = 0; i < 600; i++) { - records.add(kvRecordFactory.ofRecord("key" + i, new Object[] {i, "value"})); - } - kvTablet.putAsLeader(kvRecordBatchFactory.ofRecords(records), null); - long firstBatchEnd = logTablet.localLogEndOffset(); - kvTablet.putAsLeader( - kvRecordBatchFactory.ofRecords( - kvRecordFactory.ofRecord("later", new Object[] {601, "later"})), - null); - long target = logTablet.localLogEndOffset(); - AtomicInteger writes = new AtomicInteger(); - kvTablet.setBeforeNativeWrite( - () -> { - if (writes.incrementAndGet() == 2) { - throw new StorageBackpressureException("Reject the second native write"); - } - }); - kvTablet.requestFlush( - target, - failure -> { - throw new AssertionError(failure); - }); - kvTablet.runScheduledFlush(); - assertThat(kvTablet.getFlushState()).isEqualTo(KvTablet.FlushState.STORAGE_BLOCKED); - assertThat(kvTablet.getFlushedLogOffset()).isEqualTo(firstBatchEnd); - assertThat(kvTablet.getRowCount()).isEqualTo(600); - assertThat(kvTablet.multiGet(Arrays.asList("key0".getBytes(), "key599".getBytes()))) - .noneMatch(Objects::isNull); - assertThat(kvTablet.multiGet(Collections.singletonList("later".getBytes())).get(0)) - .isNull(); - logTablet.updateHighWatermark(kvTablet.getFlushedLogOffset()); - assertThat( - logTablet - .read(0, Integer.MAX_VALUE, FetchIsolation.HIGH_WATERMARK, true) - .getRecords() - .batches()) - .singleElement() - .satisfies(batch -> assertThat(batch.nextLogOffset()).isEqualTo(firstBatchEnd)); - kvTablet.setBeforeNativeWrite(null); - kvTablet.requestFlushRetry(); - kvTablet.runScheduledFlush(); - assertThat(kvTablet.getFlushedLogOffset()).isEqualTo(target); - assertThat(kvTablet.getRowCount()).isEqualTo(601); - assertThat(kvTablet.getKvPreWriteBuffer().pendingFlushBytes()).isZero(); - assertThat(kvTablet.multiGet(Collections.singletonList("later".getBytes())).get(0)) - .isNotNull(); - } - @Test void testScheduledFlushWritesLargePreparedRangeInSegments() throws Exception { ManualKvFlushScheduler manualScheduler = new ManualKvFlushScheduler(conf); From 444b7502b0e9646e0aefb70255f05ca3482f9560 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=99=BD=E9=B5=BA?= Date: Wed, 9 Sep 2026 19:25:56 +0800 Subject: [PATCH 3/7] [server] Check flush budgets after complete WAL batches Accumulate complete WAL batches before checking the soft entry and byte budgets. Remove lookahead batch counters and keep trailing empty batches in the final segment. Verify grouped native writes, partial completion and recovery through real storage tests. --- .../org/apache/fluss/server/kv/KvTablet.java | 8 ++--- .../server/kv/prewrite/KvPreWriteBuffer.java | 32 +++++++---------- .../apache/fluss/server/kv/KvTabletTest.java | 4 +-- .../kv/prewrite/KvPreWriteBufferTest.java | 35 +++++++++---------- 4 files changed, 36 insertions(+), 43 deletions(-) diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java index 12b8cdf748..7a3fa2a885 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java @@ -125,7 +125,7 @@ public final class KvTablet { /** * KV entry budget per native write of the asynchronous flush; mirrors the batching capacity of * {@code RocksDBWriteBatchWrapper} (hundreds of keys per write batch is RocksDB best practice). - * A single WAL batch may exceed this budget to keep its KV mutations atomic. + * The budget is checked after each complete WAL batch to keep its KV mutations atomic. */ private static final int MAX_RECORDS_PER_NATIVE_WRITE = 500; @@ -972,9 +972,9 @@ private boolean tryAcquireScheduledFlush() { /** * Writes the prepared entries to RocksDB in complete WAL batch groups targeting {@code - * MAX_RECORDS_PER_NATIVE_WRITE} records / {@code writeBatchSize} bytes. A single WAL batch may - * exceed these budgets. Each segment forms exactly one atomic native write (the writer has - * implicit flushes disabled) and is completed immediately after it lands, so {@code + * MAX_RECORDS_PER_NATIVE_WRITE} records / {@code writeBatchSize} bytes. Budgets are checked + * after each complete WAL batch. Each segment forms exactly one atomic native write (the writer + * has implicit flushes disabled) and is completed immediately after it lands, so {@code * flushedLogOffset}/{@code rowCount} stay consistent with the RocksDB content even if a later * segment is rejected by the no-slowdown gate. */ diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBuffer.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBuffer.java index fd7f54fdde..d956073dce 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBuffer.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBuffer.java @@ -533,10 +533,11 @@ public boolean isEmpty() { /** * Splits the prepared prefix only between complete WAL batches. Each segment is one atomic - * native write and can therefore publish its end independently after success. + * native write and can therefore publish its end independently after success. Budgets are + * checked after adding each complete WAL batch. * * @param maxBytesPerSegment key/value payload budget, or non-positive for unlimited - * @param maxRecordsPerSegment KV entry budget; a single WAL batch may exceed either budget + * @param maxRecordsPerSegment KV entry budget; the final batch may exceed either budget */ public List split(long maxBytesPerSegment, int maxRecordsPerSegment) { checkArgument(maxRecordsPerSegment > 0, "maxRecordsPerSegment must be positive."); @@ -551,37 +552,30 @@ public List split(long maxBytesPerSegment, int maxRecordsPerSegme int segmentRowCountDiff = 0; for (int batchIndex = 0; batchIndex < batchEndOffsets.size(); batchIndex++) { long batchEnd = batchEndOffsets.get(batchIndex); - int batchStart = entryIndex; - long batchBytes = 0; - int batchRowCountDiff = 0; while (entryIndex < entries.size() && entries.get(entryIndex).getLogSequenceNumber() < batchEnd) { KvEntry entry = entries.get(entryIndex++); - batchBytes += entryBytes(entry.getKey(), entry.getValue()); - batchRowCountDiff += rowCountDelta(entry); + segmentBytes += entryBytes(entry.getKey(), entry.getValue()); + segmentRowCountDiff += rowCountDelta(entry); } - boolean hasEntries = batchStart > segmentStart; - if (hasEntries - && entryIndex > batchStart - && (entryIndex - segmentStart > maxRecordsPerSegment + if (entryIndex < entries.size() + && (entryIndex - segmentStart >= maxRecordsPerSegment || (maxBytesPerSegment > 0 - && batchBytes > maxBytesPerSegment - segmentBytes))) { + && segmentBytes >= maxBytesPerSegment))) { if (segments == null) { segments = new ArrayList<>(); } segments.add( new PreparedFlush( - batchEndOffsets.get(batchIndex - 1), - entries.subList(segmentStart, batchStart), + batchEnd, + entries.subList(segmentStart, entryIndex), segmentRowCountDiff, - batchEndOffsets.subList(segmentBatchStart, batchIndex))); - segmentStart = batchStart; - segmentBatchStart = batchIndex; + batchEndOffsets.subList(segmentBatchStart, batchIndex + 1))); + segmentStart = entryIndex; + segmentBatchStart = batchIndex + 1; segmentBytes = 0; segmentRowCountDiff = 0; } - segmentBytes += batchBytes; - segmentRowCountDiff += batchRowCountDiff; } if (segments == null) { return Collections.singletonList(this); diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java index 8fd66e7443..568d9f9fa5 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java @@ -2006,7 +2006,7 @@ void testScheduledFlushWritesLargePreparedRangeInSegments() throws Exception { new HashMap<>(), manualScheduler); - // Three complete 400-entry WAL batches require three native writes with a 500-entry budget. + // The 500-entry budget groups complete 400-entry batches into writes of 800 and 400. int recordCount = 1200; List records = new ArrayList<>(recordCount); for (int i = 0; i < recordCount; i++) { @@ -2023,7 +2023,7 @@ void testScheduledFlushWritesLargePreparedRangeInSegments() throws Exception { kvTablet.requestFlush(flushOffset, NOPErrorHandler.INSTANCE); kvTablet.runScheduledFlush(); - assertThat(nativeWrites.get()).isEqualTo(3); + assertThat(nativeWrites.get()).isEqualTo(2); assertThat(kvTablet.getFlushedLogOffset()).isEqualTo(flushOffset); assertThat(kvTablet.getRowCount()).isEqualTo(recordCount); assertThat(kvTablet.getKvPreWriteBuffer().pendingFlushBytes()).isEqualTo(0); diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferTest.java index 4cca584985..cabf1a680f 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferTest.java @@ -276,14 +276,14 @@ void testSplitPreparedFlushByByteSize() { PreparedFlush preparedFlush = buffer.prepareFlush(3); List segments = preparedFlush.split(10, Integer.MAX_VALUE); - // Adding the second entry would exceed the limit, while the last two entries exactly fit. + // The first two complete batches cross the byte budget and close the first segment. assertThat(segments).hasSize(2); - assertThat(segments.get(0).entries()).hasSize(1); - assertThat(segments.get(0).exclusiveUpToLogSequenceNumber()).isEqualTo(1); - assertThat(segments.get(0).rowCountDiff()).isEqualTo(1); - assertThat(segments.get(1).entries()).hasSize(2); + assertThat(segments.get(0).entries()).hasSize(2); + assertThat(segments.get(0).exclusiveUpToLogSequenceNumber()).isEqualTo(2); + assertThat(segments.get(0).rowCountDiff()).isEqualTo(2); + assertThat(segments.get(1).entries()).hasSize(1); assertThat(segments.get(1).exclusiveUpToLogSequenceNumber()).isEqualTo(3); - assertThat(segments.get(1).rowCountDiff()).isEqualTo(2); + assertThat(segments.get(1).rowCountDiff()).isEqualTo(1); } @Test @@ -309,10 +309,10 @@ void testSplitPreparedFlushWithOversizedEntry() { @Test void testSplitPreparedFlushUsesFirstReachedLimit() { KvPreWriteBuffer buffer = new KvPreWriteBuffer(TestingMetricGroups.TABLET_SERVER_METRICS); - // Entry payload sizes are 2, 2, 6, and 5 bytes. + // Entry payload sizes are 2, 2, 11, and 5 bytes. bufferInsert(buffer, "a", "x", 0); bufferInsert(buffer, "b", "y", 1); - bufferInsert(buffer, "c", "12345", 2); + bufferInsert(buffer, "c", "1234567890", 2); bufferInsert(buffer, "d", "1234", 3); for (KvPreWriteBuffer.KvEntry entry : buffer.getAllKvEntries()) { @@ -345,26 +345,25 @@ void testCompletePrefixSegmentsAndAbortRest() { } PreparedFlush preparedFlush = buffer.prepareFlush(4); List segments = preparedFlush.split(10, Integer.MAX_VALUE); - assertThat(segments).hasSize(3); + assertThat(segments).hasSize(2); // Model a storage rejection after the first segment landed: complete the written prefix, // abort the rest. - assertThat(buffer.completeFlush(segments.get(0))).isEqualTo(1); + assertThat(buffer.completeFlush(segments.get(0))).isEqualTo(2); buffer.abortFlush(segments.get(1)); - buffer.abortFlush(segments.get(2)); // The completed entries are gone; the aborted ones are ACTIVE again and can be prepared // by the retry, which must cover exactly the remaining range. - assertThat(buffer.pendingFlushBytes()).isEqualTo(16); - assertThat(buffer.getAllKvEntries()).hasSize(3); + assertThat(buffer.pendingFlushBytes()).isEqualTo(11); + assertThat(buffer.getAllKvEntries()).hasSize(2); PreparedFlush retry = buffer.prepareFlush(4); - assertThat(retry.entries()).hasSize(3); - assertThat(retry.entries().get(0).getLogSequenceNumber()).isEqualTo(1); - assertThat(retry.rowCountDiff()).isEqualTo(3); + assertThat(retry.entries()).hasSize(2); + assertThat(retry.entries().get(0).getLogSequenceNumber()).isEqualTo(2); + assertThat(retry.rowCountDiff()).isEqualTo(2); } @Test - void testSplitKeepsOversizedBatchAndEmptyTailTogether() { + void testSplitKeepsOversizedBatchesAndSkipsEmptySegments() { KvPreWriteBuffer buffer = new KvPreWriteBuffer(TestingMetricGroups.TABLET_SERVER_METRICS); buffer.registerBatchEnd(1); bufferInsert(buffer, "a", "large-value", 1); @@ -376,7 +375,7 @@ void testSplitKeepsOversizedBatchAndEmptyTailTogether() { buffer.registerBatchEnd(7); List segments = buffer.prepareFlush(7).split(2, 1); assertThat(segments).hasSize(2); - assertThat(segments.get(0).exclusiveUpToLogSequenceNumber()).isEqualTo(4); + assertThat(segments.get(0).exclusiveUpToLogSequenceNumber()).isEqualTo(3); assertThat(segments.get(0).entries()).hasSize(2); assertThat(buffer.completeFlush(segments.get(0))).isEqualTo(2); buffer.abortFlush(segments.get(1)); From afa360756cfdae8f2cfc873dcd0ed786eb7b8976 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=99=BD=E9=B5=BA?= Date: Wed, 9 Sep 2026 19:35:46 +0800 Subject: [PATCH 4/7] [test] Focus WAL boundary coverage on two replica regressions Keep real append/backpressure/retry and WAL recovery/truncation scenarios. Remove duplicate parameter combinations and buffer regressions while retaining existing grouping tests. --- .../server/kv/KvFlushBatchBoundaryTest.java | 110 +++++------------- .../kv/prewrite/KvPreWriteBufferTest.java | 64 ---------- 2 files changed, 32 insertions(+), 142 deletions(-) diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvFlushBatchBoundaryTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvFlushBatchBoundaryTest.java index 43b16bceb8..a2c53dec42 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvFlushBatchBoundaryTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvFlushBatchBoundaryTest.java @@ -22,8 +22,6 @@ import org.apache.fluss.config.MemorySize; import org.apache.fluss.exception.StorageBackpressureException; import org.apache.fluss.metadata.TableBucket; -import org.apache.fluss.record.KvRecordBatch; -import org.apache.fluss.record.KvRecordTestUtils; import org.apache.fluss.rpc.protocol.MergeMode; import org.apache.fluss.server.entity.NotifyLeaderAndIsrData; import org.apache.fluss.server.log.FetchIsolation; @@ -33,19 +31,13 @@ import org.apache.fluss.utils.types.Tuple2; import org.junit.jupiter.api.Test; -import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.CsvSource; -import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; -import java.util.List; import java.util.concurrent.atomic.AtomicInteger; import static org.apache.fluss.record.TestData.DATA1_PHYSICAL_TABLE_PATH_PK; -import static org.apache.fluss.record.TestData.DATA1_ROW_TYPE; import static org.apache.fluss.record.TestData.DATA1_TABLE_ID_PK; -import static org.apache.fluss.record.TestData.DEFAULT_SCHEMA_ID; import static org.apache.fluss.server.coordinator.CoordinatorContext.INITIAL_COORDINATOR_EPOCH; import static org.apache.fluss.testutils.DataTestUtils.genKvRecordBatch; import static org.apache.fluss.testutils.DataTestUtils.genKvRecords; @@ -68,30 +60,15 @@ public void close() {} }; } - @ParameterizedTest - @CsvSource({"false, false", "false, true", "true, false", "true, true"}) - void testBackpressureAndRetryAfterWalRecovery(boolean recover, boolean byteLimit) - throws Exception { - if (byteLimit) { - conf.set(ConfigOptions.KV_WRITE_BATCH_SIZE, new MemorySize(64)); - } + @Test + void testBackpressurePublishesWholeBatchAndRetryCompletes() throws Exception { Replica replica = createLeader(); try { appendBatches(replica); long target = replica.getLocalLogEndOffset(); assertThat(replica.getLogHighWatermark()).isZero(); - if (recover) { - replica.makeFollower(followerState()); - assertThat(replica.isLeader()).isFalse(); - replica.makeLeader(state(2)); - } KvTablet tablet = replica.getKvTablet(); rejectSecondWrite(tablet); - tablet.requestFlush( - target, - failure -> { - throw new AssertionError(failure); - }); tablet.runScheduledFlush(); assertCommittedPrefix(replica); @@ -103,31 +80,39 @@ void testBackpressureAndRetryAfterWalRecovery(boolean recover, boolean byteLimit assertThat(replica.checkEnoughReplicasReachOffset(target).f0).isTrue(); assertThat(tablet.getRowCount()).isEqualTo(601); assertThat(tablet.getKvPreWriteBuffer().pendingFlushBytes()).isZero(); - assertValue(tablet, 0, "after"); + assertValue(tablet, 0, "before"); assertValue(tablet, 600, "tail"); - List ends = new ArrayList<>(); - replica.getLogTablet() - .read(0, Integer.MAX_VALUE, FetchIsolation.HIGH_WATERMARK, true) - .getRecords() - .batches() - .forEach(batch -> ends.add(batch.nextLogOffset())); - assertThat(ends).containsExactly(600L, 1800L, 1801L, 1802L); + assertThat( + replica.getLogTablet() + .read(0, Integer.MAX_VALUE, FetchIsolation.HIGH_WATERMARK, true) + .getRecords() + .batches()) + .extracting(batch -> batch.nextLogOffset()) + .containsExactly(600L, 601L); } finally { deleteReplica(replica); } } @Test - void testFollowerTruncationPreservesCommittedBatch() throws Exception { + void testRecoveredBatchBoundariesSurviveBackpressureAndTruncation() throws Exception { + conf.set(ConfigOptions.KV_WRITE_BATCH_SIZE, new MemorySize(64)); Replica replica = createLeader(); try { appendBatches(replica); + replica.makeFollower(state(TABLET_SERVER_ID + 1, 1)); + replica.makeLeader(state(TABLET_SERVER_ID, 2)); KvTablet tablet = replica.getKvTablet(); + tablet.requestFlush( + replica.getLocalLogEndOffset(), + failure -> { + throw new AssertionError(failure); + }); rejectSecondWrite(tablet); tablet.runScheduledFlush(); assertCommittedPrefix(replica); long hw = replica.getLogHighWatermark(); - replica.makeFollower(followerState()); + replica.makeFollower(state(TABLET_SERVER_ID + 1, 3)); assertThat(replica.isLeader()).isFalse(); replica.truncateTo(hw); assertThat(replica.getLocalLogEndOffset()).isEqualTo(hw); @@ -138,7 +123,7 @@ void testFollowerTruncationPreservesCommittedBatch() throws Exception { .batches()) .singleElement() .satisfies(batch -> assertThat(batch.nextLogOffset()).isEqualTo(hw)); - replica.makeLeader(state(2)); + replica.makeLeader(state(TABLET_SERVER_ID, 4)); assertThat(replica.getKvTablet().getRowCount()).isEqualTo(600); assertValue(replica.getKvTablet(), 0, "before"); replica.putRecordsToLeader( @@ -166,65 +151,33 @@ private void deleteReplica(Replica replica) throws Exception { private Replica createLeader() throws Exception { Replica replica = makeKvReplica(DATA1_PHYSICAL_TABLE_PATH_PK, new TableBucket(DATA1_TABLE_ID_PK, 1)); - replica.makeLeader(state(0)); + replica.makeLeader(state(TABLET_SERVER_ID, 0)); return replica; } - private NotifyLeaderAndIsrData followerState() { - int leader = TABLET_SERVER_ID + 1; + private NotifyLeaderAndIsrData state(int leader, int epoch) { return new NotifyLeaderAndIsrData( DATA1_PHYSICAL_TABLE_PATH_PK, new TableBucket(DATA1_TABLE_ID_PK, 1), - Arrays.asList(TABLET_SERVER_ID, leader), + Arrays.asList(TABLET_SERVER_ID, TABLET_SERVER_ID + 1), new LeaderAndIsr( leader, - 1, - Collections.singletonList(leader), - Collections.emptyList(), - INITIAL_COORDINATOR_EPOCH, - 1)); - } - - private NotifyLeaderAndIsrData state(int epoch) { - return new NotifyLeaderAndIsrData( - DATA1_PHYSICAL_TABLE_PATH_PK, - new TableBucket(DATA1_TABLE_ID_PK, 1), - Collections.singletonList(TABLET_SERVER_ID), - new LeaderAndIsr( - TABLET_SERVER_ID, epoch, - Collections.singletonList(TABLET_SERVER_ID), + Collections.singletonList(leader), Collections.emptyList(), INITIAL_COORDINATOR_EPOCH, epoch)); } private void appendBatches(Replica replica) throws Exception { - Object[][] before = new Object[600][]; - Object[][] after = new Object[600][]; - for (int i = 0; i < 600; i++) { - before[i] = new Object[] {i, "before"}; - after[i] = new Object[] {i, "after"}; + Object[][] rows = new Object[600][]; + for (int i = 0; i < rows.length; i++) { + rows[i] = new Object[] {i, "before"}; } - KvRecordBatch first = - KvRecordTestUtils.KvRecordBatchFactory.of(DEFAULT_SCHEMA_ID) - .ofRecords(genKvRecords(before), 42L, 0); - replica.putRecordsToLeader(first, null, MergeMode.DEFAULT, 0); - // A duplicate must neither retain speculative KV updates nor register a second boundary. - replica.putRecordsToLeader(first, null, MergeMode.DEFAULT, 0); - replica.putRecordsToLeader(genKvRecordBatch(after), null, MergeMode.DEFAULT, 0); - byte[] missingKey = getKeyValuePairs(genKvRecords(new Object[] {999, "missing"})).get(0).f0; - replica.putRecordsToLeader( - KvRecordTestUtils.KvRecordBatchFactory.of(DEFAULT_SCHEMA_ID) - .ofRecords( - KvRecordTestUtils.KvRecordFactory.of(DATA1_ROW_TYPE) - .ofRecord(missingKey, null)), - null, - MergeMode.DEFAULT, - 0); + replica.putRecordsToLeader(genKvRecordBatch(rows), null, MergeMode.DEFAULT, 0); replica.putRecordsToLeader( genKvRecordBatch(new Object[] {600, "tail"}), null, MergeMode.DEFAULT, 0); - assertThat(replica.getLocalLogEndOffset()).isEqualTo(1802); + assertThat(replica.getLocalLogEndOffset()).isEqualTo(601); } private void rejectSecondWrite(KvTablet tablet) { @@ -243,9 +196,10 @@ private void assertCommittedPrefix(Replica replica) throws Exception { assertThat(tablet.getFlushedLogOffset()).isEqualTo(600); assertThat(replica.getLogHighWatermark()).isEqualTo(600); assertThat(replica.checkEnoughReplicasReachOffset(600).f0).isTrue(); - assertThat(replica.checkEnoughReplicasReachOffset(1800).f0).isFalse(); + assertThat(replica.checkEnoughReplicasReachOffset(601).f0).isFalse(); assertThat(tablet.getRowCount()).isEqualTo(600); assertValue(tablet, 0, "before"); + assertValue(tablet, 599, "before"); assertThat( replica.getLogTablet() .read(0, Integer.MAX_VALUE, FetchIsolation.HIGH_WATERMARK, true) diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferTest.java index cabf1a680f..f414888eb1 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferTest.java @@ -362,70 +362,6 @@ void testCompletePrefixSegmentsAndAbortRest() { assertThat(retry.rowCountDiff()).isEqualTo(2); } - @Test - void testSplitKeepsOversizedBatchesAndSkipsEmptySegments() { - KvPreWriteBuffer buffer = new KvPreWriteBuffer(TestingMetricGroups.TABLET_SERVER_METRICS); - buffer.registerBatchEnd(1); - bufferInsert(buffer, "a", "large-value", 1); - bufferInsert(buffer, "b", "large-value", 2); - buffer.registerBatchEnd(3); - buffer.registerBatchEnd(4); - bufferUpdate(buffer, "a", "new-value", 5); - buffer.registerBatchEnd(6); - buffer.registerBatchEnd(7); - List segments = buffer.prepareFlush(7).split(2, 1); - assertThat(segments).hasSize(2); - assertThat(segments.get(0).exclusiveUpToLogSequenceNumber()).isEqualTo(3); - assertThat(segments.get(0).entries()).hasSize(2); - assertThat(buffer.completeFlush(segments.get(0))).isEqualTo(2); - buffer.abortFlush(segments.get(1)); - List retry = buffer.prepareFlush(7).split(2, 1); - assertThat(retry) - .singleElement() - .satisfies( - segment -> { - assertThat(segment.exclusiveUpToLogSequenceNumber()).isEqualTo(7); - assertThat(segment.entries()).hasSize(1); - assertThat(buffer.completeFlush(segment)).isZero(); - }); - assertThat(buffer.pendingFlushBytes()).isZero(); - } - - @Test - void testTruncateAndReappendBatchBoundaries() { - KvPreWriteBuffer buffer = new KvPreWriteBuffer(TestingMetricGroups.TABLET_SERVER_METRICS); - bufferInsert(buffer, "a", "first", 0); - buffer.registerBatchEnd(1); - bufferUpdate(buffer, "a", "discarded", 2); - buffer.registerBatchEnd(3); - buffer.registerBatchEnd(4); - buffer.truncateTo(1, TruncateReason.ERROR); - assertThat(getValue(buffer, "a")).isEqualTo("first"); - bufferUpdate(buffer, "a", "replacement", 2); - buffer.registerBatchEnd(3); - bufferInsert(buffer, "b", "second", 3); - buffer.registerBatchEnd(4); - List segments = buffer.prepareFlush(4).split(0, 1); - assertThat(segments) - .extracting(PreparedFlush::exclusiveUpToLogSequenceNumber) - .containsExactly(1L, 3L, 4L); - assertThat(buffer.completeFlush(segments.get(0))).isEqualTo(1); - assertThat(getValue(buffer, "a")).isEqualTo("replacement"); - assertThat(buffer.completeFlush(segments.get(1))).isZero(); - assertThat(buffer.completeFlush(segments.get(2))).isEqualTo(1); - assertThat(buffer.getAllKvEntries()).isEmpty(); - assertThat(buffer.pendingFlushBytes()).isZero(); - // The next range must not contain completed boundaries, even when it starts empty. - buffer.registerBatchEnd(5); - bufferInsert(buffer, "c", "next", 5); - buffer.registerBatchEnd(6); - assertThat(buffer.prepareFlush(6).split(0, 1)) - .singleElement() - .satisfies( - segment -> - assertThat(segment.exclusiveUpToLogSequenceNumber()).isEqualTo(6)); - } - private static void bufferInsert( KvPreWriteBuffer kvPreWriteBuffer, String key, String value, int elementCount) { kvPreWriteBuffer.insert(toKey(key), value.getBytes(), elementCount); From c595bc00ef27b6afeb7ea3c8b3f1b728e4870b06 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=99=BD=E9=B5=BA?= Date: Wed, 9 Sep 2026 19:56:38 +0800 Subject: [PATCH 5/7] [server] Mark WAL boundaries on existing KV entries Replace the boxed boundary queue and flush snapshots with a flag on the final KV mutation of each confirmed WAL batch. Split directly on marked entries and retain the requested flush target for empty tails. Boundary cleanup follows the existing entry completion and truncation lifecycle. --- .../server/kv/prewrite/KvPreWriteBuffer.java | 80 +++++-------------- .../kv/prewrite/KvPreWriteBufferTest.java | 31 +++---- 2 files changed, 38 insertions(+), 73 deletions(-) diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBuffer.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBuffer.java index d956073dce..19b10dae2e 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBuffer.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBuffer.java @@ -27,12 +27,10 @@ import javax.annotation.Nullable; import javax.annotation.concurrent.NotThreadSafe; -import java.util.ArrayDeque; import java.util.ArrayList; import java.util.Arrays; import java.util.Base64; import java.util.Collections; -import java.util.Deque; import java.util.HashMap; import java.util.Iterator; import java.util.LinkedList; @@ -97,9 +95,6 @@ public class KvPreWriteBuffer { // a linked list for all kv entries private final LinkedList allKvEntries = new LinkedList<>(); - // Exclusive ends of WAL batches whose KV updates have not yet completed flushing. - private final Deque batchEndOffsets = new ArrayDeque<>(); - // metrics related. private final Counter truncateAsDuplicatedCount; private final Counter truncateAsErrorCount; @@ -115,12 +110,14 @@ public KvPreWriteBuffer(TabletServerMetricGroup serverMetricGroup) { truncateAsErrorCount = serverMetricGroup.kvTruncateAsErrorCount(); } - /** Registers the exclusive end of a successfully appended or recovered WAL batch. */ + /** Marks the last KV mutation when it ends a successfully appended or recovered WAL batch. */ public void registerBatchEnd(long batchEndOffset) { - checkArgument( - batchEndOffsets.isEmpty() || batchEndOffset > batchEndOffsets.peekLast(), - "WAL batch ends must increase."); - batchEndOffsets.addLast(batchEndOffset); + KvEntry last = allKvEntries.peekLast(); + // Empty batches have no mutation to mark. Their offsets are covered by the flush target + // or a later batch; never infer a boundary from a gap between KV mutations. + if (last != null && last.logSequenceNumber + 1 == batchEndOffset) { + last.batchEnd = true; + } } /** @@ -207,9 +204,6 @@ public void truncateTo(long targetLogSequenceNumber, TruncateReason truncateReas truncateAsErrorCount.inc(); } - while (!batchEndOffsets.isEmpty() && batchEndOffsets.peekLast() > targetLogSequenceNumber) { - batchEndOffsets.removeLast(); - } Iterator descIter = allKvEntries.descendingIterator(); while (descIter.hasNext()) { KvEntry entry = descIter.next(); @@ -271,18 +265,7 @@ public PreparedFlush prepareFlush(long exclusiveUpToLogSequenceNumber) { entries.add(entry); rowCountDiff += rowCountDelta(entry); } - List preparedBatchEnds = new ArrayList<>(); - for (Long batchEnd : batchEndOffsets) { - if (batchEnd >= exclusiveUpToLogSequenceNumber) { - break; - } - preparedBatchEnds.add(batchEnd); - } - // The requested flush target is itself a WAL batch boundary. Keeping it also covers - // ranges with no KV mutations, including empty WAL batches. - preparedBatchEnds.add(exclusiveUpToLogSequenceNumber); - return new PreparedFlush( - exclusiveUpToLogSequenceNumber, entries, rowCountDiff, preparedBatchEnds); + return new PreparedFlush(exclusiveUpToLogSequenceNumber, entries, rowCountDiff); } /** Completes a prepared async flush and removes flushed entries from the buffer. */ @@ -308,10 +291,6 @@ public int completeFlush(PreparedFlush preparedFlush) { if (allKvEntries.isEmpty()) { maxLogSequenceNumber = -1; } - while (!batchEndOffsets.isEmpty() - && batchEndOffsets.peekFirst() <= preparedFlush.exclusiveUpToLogSequenceNumber) { - batchEndOffsets.removeFirst(); - } return preparedFlush.rowCountDiff; } @@ -394,6 +373,7 @@ public static class KvEntry { private final Key key; private final Value value; private final long logSequenceNumber; + private boolean batchEnd; // the previous mapped value in the buffer before this key-value put; null once the // referenced entry has been flushed (see completeFlush) @@ -502,17 +482,12 @@ public static final class PreparedFlush { private final long exclusiveUpToLogSequenceNumber; private final List entries; private final int rowCountDiff; - private final List batchEndOffsets; private PreparedFlush( - long exclusiveUpToLogSequenceNumber, - List entries, - int rowCountDiff, - List batchEndOffsets) { + long exclusiveUpToLogSequenceNumber, List entries, int rowCountDiff) { this.exclusiveUpToLogSequenceNumber = exclusiveUpToLogSequenceNumber; this.entries = entries; this.rowCountDiff = rowCountDiff; - this.batchEndOffsets = batchEndOffsets; } public long exclusiveUpToLogSequenceNumber() { @@ -541,25 +516,17 @@ public boolean isEmpty() { */ public List split(long maxBytesPerSegment, int maxRecordsPerSegment) { checkArgument(maxRecordsPerSegment > 0, "maxRecordsPerSegment must be positive."); - if (batchEndOffsets.size() == 1) { - return Collections.singletonList(this); - } List segments = null; int segmentStart = 0; - int entryIndex = 0; - int segmentBatchStart = 0; long segmentBytes = 0; int segmentRowCountDiff = 0; - for (int batchIndex = 0; batchIndex < batchEndOffsets.size(); batchIndex++) { - long batchEnd = batchEndOffsets.get(batchIndex); - while (entryIndex < entries.size() - && entries.get(entryIndex).getLogSequenceNumber() < batchEnd) { - KvEntry entry = entries.get(entryIndex++); - segmentBytes += entryBytes(entry.getKey(), entry.getValue()); - segmentRowCountDiff += rowCountDelta(entry); - } - if (entryIndex < entries.size() - && (entryIndex - segmentStart >= maxRecordsPerSegment + for (int i = 0; i < entries.size(); i++) { + KvEntry entry = entries.get(i); + segmentBytes += entryBytes(entry.getKey(), entry.getValue()); + segmentRowCountDiff += rowCountDelta(entry); + if (entry.batchEnd + && i + 1 < entries.size() + && (i + 1 - segmentStart >= maxRecordsPerSegment || (maxBytesPerSegment > 0 && segmentBytes >= maxBytesPerSegment))) { if (segments == null) { @@ -567,12 +534,10 @@ public List split(long maxBytesPerSegment, int maxRecordsPerSegme } segments.add( new PreparedFlush( - batchEnd, - entries.subList(segmentStart, entryIndex), - segmentRowCountDiff, - batchEndOffsets.subList(segmentBatchStart, batchIndex + 1))); - segmentStart = entryIndex; - segmentBatchStart = batchIndex + 1; + entry.logSequenceNumber + 1, + entries.subList(segmentStart, i + 1), + segmentRowCountDiff)); + segmentStart = i + 1; segmentBytes = 0; segmentRowCountDiff = 0; } @@ -584,8 +549,7 @@ public List split(long maxBytesPerSegment, int maxRecordsPerSegme new PreparedFlush( exclusiveUpToLogSequenceNumber, entries.subList(segmentStart, entries.size()), - segmentRowCountDiff, - batchEndOffsets.subList(segmentBatchStart, batchEndOffsets.size()))); + segmentRowCountDiff)); return segments; } } diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferTest.java index f414888eb1..443bd73efa 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferTest.java @@ -232,12 +232,12 @@ void testSplitPreparedFlushByRecordCount() { // +key0(lsn 0), +key1(lsn 1), +key2(lsn 2), +key3(lsn 3), -key2(lsn 4) for (int i = 0; i < 4; i++) { bufferInsert(buffer, "key" + i, "value" + i, i); + buffer.registerBatchEnd(i + 1); } bufferDelete(buffer, "key2", 4); + buffer.registerBatchEnd(5); + buffer.registerBatchEnd(10); - for (KvPreWriteBuffer.KvEntry entry : buffer.getAllKvEntries()) { - buffer.registerBatchEnd(entry.getLogSequenceNumber() + 1); - } PreparedFlush preparedFlush = buffer.prepareFlush(10); List segments = preparedFlush.split(0, 2); @@ -267,12 +267,12 @@ void testSplitPreparedFlushByByteSize() { KvPreWriteBuffer buffer = new KvPreWriteBuffer(TestingMetricGroups.TABLET_SERVER_METRICS); // Entry payload sizes are 6, 5, and 5 bytes. bufferInsert(buffer, "a", "12345", 0); + buffer.registerBatchEnd(1); bufferInsert(buffer, "b", "1234", 1); + buffer.registerBatchEnd(2); bufferInsert(buffer, "c", "1234", 2); + buffer.registerBatchEnd(3); - for (KvPreWriteBuffer.KvEntry entry : buffer.getAllKvEntries()) { - buffer.registerBatchEnd(entry.getLogSequenceNumber() + 1); - } PreparedFlush preparedFlush = buffer.prepareFlush(3); List segments = preparedFlush.split(10, Integer.MAX_VALUE); @@ -291,11 +291,10 @@ void testSplitPreparedFlushWithOversizedEntry() { KvPreWriteBuffer buffer = new KvPreWriteBuffer(TestingMetricGroups.TABLET_SERVER_METRICS); // The first entry is larger than the byte limit and must remain a non-empty singleton. bufferInsert(buffer, "a", "1234567890", 0); + buffer.registerBatchEnd(1); bufferInsert(buffer, "b", "123", 1); + buffer.registerBatchEnd(2); - for (KvPreWriteBuffer.KvEntry entry : buffer.getAllKvEntries()) { - buffer.registerBatchEnd(entry.getLogSequenceNumber() + 1); - } PreparedFlush preparedFlush = buffer.prepareFlush(2); List segments = preparedFlush.split(10, Integer.MAX_VALUE); @@ -311,13 +310,14 @@ void testSplitPreparedFlushUsesFirstReachedLimit() { KvPreWriteBuffer buffer = new KvPreWriteBuffer(TestingMetricGroups.TABLET_SERVER_METRICS); // Entry payload sizes are 2, 2, 11, and 5 bytes. bufferInsert(buffer, "a", "x", 0); + buffer.registerBatchEnd(1); bufferInsert(buffer, "b", "y", 1); + buffer.registerBatchEnd(2); bufferInsert(buffer, "c", "1234567890", 2); + buffer.registerBatchEnd(3); bufferInsert(buffer, "d", "1234", 3); + buffer.registerBatchEnd(4); - for (KvPreWriteBuffer.KvEntry entry : buffer.getAllKvEntries()) { - buffer.registerBatchEnd(entry.getLogSequenceNumber() + 1); - } PreparedFlush preparedFlush = buffer.prepareFlush(4); List segments = preparedFlush.split(10, 2); @@ -336,13 +336,14 @@ void testCompletePrefixSegmentsAndAbortRest() { KvPreWriteBuffer buffer = new KvPreWriteBuffer(TestingMetricGroups.TABLET_SERVER_METRICS); // Entry payload sizes are 6, 5, 5, and 6 bytes. bufferInsert(buffer, "a", "12345", 0); + buffer.registerBatchEnd(1); bufferInsert(buffer, "b", "1234", 1); + buffer.registerBatchEnd(2); bufferInsert(buffer, "c", "1234", 2); + buffer.registerBatchEnd(3); bufferInsert(buffer, "d", "12345", 3); + buffer.registerBatchEnd(4); - for (KvPreWriteBuffer.KvEntry entry : buffer.getAllKvEntries()) { - buffer.registerBatchEnd(entry.getLogSequenceNumber() + 1); - } PreparedFlush preparedFlush = buffer.prepareFlush(4); List segments = preparedFlush.split(10, Integer.MAX_VALUE); assertThat(segments).hasSize(2); From 1e5f62d208bef46c132c2c3f60f474a9f3b6ef15 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=99=BD=E9=B5=BA?= Date: Wed, 9 Sep 2026 20:22:06 +0800 Subject: [PATCH 6/7] [server] Clarify WAL boundary and flush target names Name the tail mutation marker explicitly for WAL batches and describe its per-batch call ordering. Name native write budgets as entry and byte targets rather than hard maxima. --- .../fluss/server/kv/KvRecoverHelper.java | 2 +- .../org/apache/fluss/server/kv/KvTablet.java | 8 ++--- .../server/kv/prewrite/KvPreWriteBuffer.java | 31 +++++++++++------- .../kv/prewrite/KvPreWriteBufferTest.java | 32 +++++++++---------- 4 files changed, 40 insertions(+), 33 deletions(-) diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvRecoverHelper.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvRecoverHelper.java index 8852d64018..109ccb21f4 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvRecoverHelper.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvRecoverHelper.java @@ -254,7 +254,7 @@ private long readLogRecordsAndApply( autoIncIdRangeUpdater, resumeRecordConsumer); if (fetchIsolation == FetchIsolation.LOG_END) { - kvTablet.getKvPreWriteBuffer().registerBatchEnd(nextFetchOffset); + kvTablet.getKvPreWriteBuffer().markWalBatchEnd(nextFetchOffset); } } } diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java index 7a3fa2a885..2312b6fd4e 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java @@ -127,7 +127,7 @@ public final class KvTablet { * {@code RocksDBWriteBatchWrapper} (hundreds of keys per write batch is RocksDB best practice). * The budget is checked after each complete WAL batch to keep its KV mutations atomic. */ - private static final int MAX_RECORDS_PER_NATIVE_WRITE = 500; + private static final int TARGET_ENTRIES_PER_NATIVE_WRITE = 500; private final PhysicalTablePath physicalPath; private final TableBucket tableBucket; @@ -816,7 +816,7 @@ private LogAppendInfo putAsLeader( memoizedLakeLookup); if (!appendInfo.duplicated()) { // KvWriteProcessor appends one WAL batch for each accepted KV batch. - kvPreWriteBuffer.registerBatchEnd(appendInfo.lastOffset() + 1); + kvPreWriteBuffer.markWalBatchEnd(appendInfo.lastOffset() + 1); } return appendInfo; }); @@ -972,7 +972,7 @@ private boolean tryAcquireScheduledFlush() { /** * Writes the prepared entries to RocksDB in complete WAL batch groups targeting {@code - * MAX_RECORDS_PER_NATIVE_WRITE} records / {@code writeBatchSize} bytes. Budgets are checked + * TARGET_ENTRIES_PER_NATIVE_WRITE} entries / {@code writeBatchSize} bytes. Budgets are checked * after each complete WAL batch. Each segment forms exactly one atomic native write (the writer * has implicit flushes disabled) and is completed immediately after it lands, so {@code * flushedLogOffset}/{@code rowCount} stay consistent with the RocksDB content even if a later @@ -981,7 +981,7 @@ private boolean tryAcquireScheduledFlush() { @GuardedBy("kvLock") private void writePreparedFlush(PreparedFlush preparedFlush) throws Exception { List segments = - preparedFlush.split(writeBatchSize, MAX_RECORDS_PER_NATIVE_WRITE); + preparedFlush.split(writeBatchSize, TARGET_ENTRIES_PER_NATIVE_WRITE); int nextSegment = 0; try (ResourceGuard.Lease lease = rocksDBKv.getResourceGuard().acquireResource(); KvBatchWriter kvBatchWriter = createNoSlowdownKvBatchWriter()) { diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBuffer.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBuffer.java index 19b10dae2e..18e922cf3f 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBuffer.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBuffer.java @@ -110,13 +110,20 @@ public KvPreWriteBuffer(TabletServerMetricGroup serverMetricGroup) { truncateAsErrorCount = serverMetricGroup.kvTruncateAsErrorCount(); } - /** Marks the last KV mutation when it ends a successfully appended or recovered WAL batch. */ - public void registerBatchEnd(long batchEndOffset) { + /** + * Marks the last KV mutation when it ends a successfully appended or recovered WAL batch. + * + *

Call after applying each batch and before staging any mutation from the next batch. Empty + * batches do not mark an entry; their offsets are covered by the flush target or a later batch. + * + * @param batchEndOffset the exclusive end offset of the completed WAL batch + */ + public void markWalBatchEnd(long batchEndOffset) { KvEntry last = allKvEntries.peekLast(); // Empty batches have no mutation to mark. Their offsets are covered by the flush target // or a later batch; never infer a boundary from a gap between KV mutations. if (last != null && last.logSequenceNumber + 1 == batchEndOffset) { - last.batchEnd = true; + last.endOfWalBatch = true; } } @@ -373,7 +380,7 @@ public static class KvEntry { private final Key key; private final Value value; private final long logSequenceNumber; - private boolean batchEnd; + private boolean endOfWalBatch; // the previous mapped value in the buffer before this key-value put; null once the // referenced entry has been flushed (see completeFlush) @@ -511,11 +518,11 @@ public boolean isEmpty() { * native write and can therefore publish its end independently after success. Budgets are * checked after adding each complete WAL batch. * - * @param maxBytesPerSegment key/value payload budget, or non-positive for unlimited - * @param maxRecordsPerSegment KV entry budget; the final batch may exceed either budget + * @param targetBytesPerSegment key/value payload budget, or non-positive for unlimited + * @param targetEntriesPerSegment KV entry budget; the final batch may exceed either budget */ - public List split(long maxBytesPerSegment, int maxRecordsPerSegment) { - checkArgument(maxRecordsPerSegment > 0, "maxRecordsPerSegment must be positive."); + public List split(long targetBytesPerSegment, int targetEntriesPerSegment) { + checkArgument(targetEntriesPerSegment > 0, "targetEntriesPerSegment must be positive."); List segments = null; int segmentStart = 0; long segmentBytes = 0; @@ -524,11 +531,11 @@ public List split(long maxBytesPerSegment, int maxRecordsPerSegme KvEntry entry = entries.get(i); segmentBytes += entryBytes(entry.getKey(), entry.getValue()); segmentRowCountDiff += rowCountDelta(entry); - if (entry.batchEnd + if (entry.endOfWalBatch && i + 1 < entries.size() - && (i + 1 - segmentStart >= maxRecordsPerSegment - || (maxBytesPerSegment > 0 - && segmentBytes >= maxBytesPerSegment))) { + && (i + 1 - segmentStart >= targetEntriesPerSegment + || (targetBytesPerSegment > 0 + && segmentBytes >= targetBytesPerSegment))) { if (segments == null) { segments = new ArrayList<>(); } diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferTest.java index 443bd73efa..45d983133a 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferTest.java @@ -232,11 +232,11 @@ void testSplitPreparedFlushByRecordCount() { // +key0(lsn 0), +key1(lsn 1), +key2(lsn 2), +key3(lsn 3), -key2(lsn 4) for (int i = 0; i < 4; i++) { bufferInsert(buffer, "key" + i, "value" + i, i); - buffer.registerBatchEnd(i + 1); + buffer.markWalBatchEnd(i + 1); } bufferDelete(buffer, "key2", 4); - buffer.registerBatchEnd(5); - buffer.registerBatchEnd(10); + buffer.markWalBatchEnd(5); + buffer.markWalBatchEnd(10); PreparedFlush preparedFlush = buffer.prepareFlush(10); List segments = preparedFlush.split(0, 2); @@ -267,11 +267,11 @@ void testSplitPreparedFlushByByteSize() { KvPreWriteBuffer buffer = new KvPreWriteBuffer(TestingMetricGroups.TABLET_SERVER_METRICS); // Entry payload sizes are 6, 5, and 5 bytes. bufferInsert(buffer, "a", "12345", 0); - buffer.registerBatchEnd(1); + buffer.markWalBatchEnd(1); bufferInsert(buffer, "b", "1234", 1); - buffer.registerBatchEnd(2); + buffer.markWalBatchEnd(2); bufferInsert(buffer, "c", "1234", 2); - buffer.registerBatchEnd(3); + buffer.markWalBatchEnd(3); PreparedFlush preparedFlush = buffer.prepareFlush(3); List segments = preparedFlush.split(10, Integer.MAX_VALUE); @@ -291,9 +291,9 @@ void testSplitPreparedFlushWithOversizedEntry() { KvPreWriteBuffer buffer = new KvPreWriteBuffer(TestingMetricGroups.TABLET_SERVER_METRICS); // The first entry is larger than the byte limit and must remain a non-empty singleton. bufferInsert(buffer, "a", "1234567890", 0); - buffer.registerBatchEnd(1); + buffer.markWalBatchEnd(1); bufferInsert(buffer, "b", "123", 1); - buffer.registerBatchEnd(2); + buffer.markWalBatchEnd(2); PreparedFlush preparedFlush = buffer.prepareFlush(2); List segments = preparedFlush.split(10, Integer.MAX_VALUE); @@ -310,13 +310,13 @@ void testSplitPreparedFlushUsesFirstReachedLimit() { KvPreWriteBuffer buffer = new KvPreWriteBuffer(TestingMetricGroups.TABLET_SERVER_METRICS); // Entry payload sizes are 2, 2, 11, and 5 bytes. bufferInsert(buffer, "a", "x", 0); - buffer.registerBatchEnd(1); + buffer.markWalBatchEnd(1); bufferInsert(buffer, "b", "y", 1); - buffer.registerBatchEnd(2); + buffer.markWalBatchEnd(2); bufferInsert(buffer, "c", "1234567890", 2); - buffer.registerBatchEnd(3); + buffer.markWalBatchEnd(3); bufferInsert(buffer, "d", "1234", 3); - buffer.registerBatchEnd(4); + buffer.markWalBatchEnd(4); PreparedFlush preparedFlush = buffer.prepareFlush(4); List segments = preparedFlush.split(10, 2); @@ -336,13 +336,13 @@ void testCompletePrefixSegmentsAndAbortRest() { KvPreWriteBuffer buffer = new KvPreWriteBuffer(TestingMetricGroups.TABLET_SERVER_METRICS); // Entry payload sizes are 6, 5, 5, and 6 bytes. bufferInsert(buffer, "a", "12345", 0); - buffer.registerBatchEnd(1); + buffer.markWalBatchEnd(1); bufferInsert(buffer, "b", "1234", 1); - buffer.registerBatchEnd(2); + buffer.markWalBatchEnd(2); bufferInsert(buffer, "c", "1234", 2); - buffer.registerBatchEnd(3); + buffer.markWalBatchEnd(3); bufferInsert(buffer, "d", "12345", 3); - buffer.registerBatchEnd(4); + buffer.markWalBatchEnd(4); PreparedFlush preparedFlush = buffer.prepareFlush(4); List segments = preparedFlush.split(10, Integer.MAX_VALUE); From 0ea1e7e1b3bba7a525bd586aa1c8bd8d4cd4ac7a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=99=BD=E9=B5=BA?= Date: Thu, 10 Sep 2026 06:04:24 +0800 Subject: [PATCH 7/7] [test] Trigger memtable rotation in KV replica restore test Continue writing an existing key after deleting the KV directory so RocksDB surfaces the storage error even when the large WAL batch is applied in one native write. Keep the existing failover and full restored-data assertions. --- .../fluss/server/replica/KvReplicaRestoreITCase.java | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/KvReplicaRestoreITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/KvReplicaRestoreITCase.java index c38d39a834..ed938ea5a4 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/replica/KvReplicaRestoreITCase.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/KvReplicaRestoreITCase.java @@ -135,9 +135,14 @@ void testRestore() throws Exception { // we delete the kv dir to make flush fail FileUtils.deleteDirectory(replica.getKvTablet().getKvTabletDir()); - // should fail + // Append the batch that must be recovered from WAL. putRecordBatch(tableBucket, leaderServer, toKvRecordBatch(records)); + // A complete WAL batch is one native write and may remain in the active memtable. + // Further writes rotate the memtable and surface the deleted directory as an I/O error. + KvRecordBatch triggerBatch = + genKvRecordBatch(new Object[] {recordsNum - 1, "k" + (recordsNum - 1)}); + // wait for the replica to restore in another server AtomicInteger newLeaderServer = new AtomicInteger(-1); waitUntil( @@ -147,6 +152,7 @@ void testRestore() throws Exception { newLeaderServer.set(restoreServer); return true; } + putRecordBatch(tableBucket, leaderServer, triggerBatch); return false; }, Duration.ofMinutes(2),