(fluss) branch main updated: [server] Keep flushedLogOffset monotonic to fix row count after failover (#3427)

[yunhong]

This is an automated email from the ASF dual-hosted git repository.

swuferhong pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git


The following commit(s) were added to refs/heads/main by this push:
     new 6d3ba388e [server] Keep flushedLogOffset monotonic to fix row count 
after failover (#3427)
6d3ba388e is described below

commit 6d3ba388e6c0f4a9d893a9f6d6ebdf45bfea5e30
Author: Daeuk Choi <37104691+wat...@users.noreply.github.com>
AuthorDate: Mon Jun 8 11:08:30 2026 +0900

    [server] Keep flushedLogOffset monotonic to fix row count after failover 
(#3427)
    
    * [server] Keep flushedLogOffset monotonic to fix row count after failover
    
    * add a ut test
---
 .../java/org/apache/fluss/server/kv/KvTablet.java  |  4 ++-
 .../org/apache/fluss/server/kv/KvTabletTest.java   | 35 ++++++++++++++++++++++
 2 files changed, 38 insertions(+), 1 deletion(-)

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 f38019399..0683f5e05 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
@@ -687,7 +687,9 @@ public final class KvTablet {
                     } else {
                         try {
                             int rowCountDiff = 
kvPreWriteBuffer.flush(exclusiveUpToLogOffset);
-                            flushedLogOffset = exclusiveUpToLogOffset;
+                            if (exclusiveUpToLogOffset > flushedLogOffset) {
+                                flushedLogOffset = exclusiveUpToLogOffset;
+                            }
                             if (rowCount != ROW_COUNT_DISABLED) {
                                 // row count is enabled, we update the row 
count after flush.
                                 long currentRowCount = rowCount;
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 d7ef34701..3425fa75c 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
@@ -59,6 +59,7 @@ import org.apache.fluss.server.kv.rocksdb.RocksDBStatistics;
 import org.apache.fluss.server.kv.rowmerger.RowMerger;
 import org.apache.fluss.server.kv.scan.OpenScanResult;
 import org.apache.fluss.server.kv.scan.ScannerContext;
+import org.apache.fluss.server.kv.snapshot.TabletState;
 import org.apache.fluss.server.log.FetchIsolation;
 import org.apache.fluss.server.log.LogAppendInfo;
 import org.apache.fluss.server.log.LogTablet;
@@ -1861,6 +1862,40 @@ class KvTabletTest {
         kvTablet.close();
     }
 
+    @Test
+    void testFlushDoesNotRegressFlushedLogOffset() throws Exception {
+        initLogTabletAndKvTablet(DATA1_SCHEMA_PK, new HashMap<>());
+
+        List<KvRecord> inserts = new ArrayList<>();
+        for (int i = 1; i <= 5; i++) {
+            inserts.add(kvRecordFactory.ofRecord("key" + i, new Object[] {i, 
"val" + i}));
+        }
+        kvTablet.putAsLeader(kvRecordBatchFactory.ofRecords(inserts), null);
+
+        List<KvRecord> deletes = new ArrayList<>();
+        for (int i = 1; i <= 2; i++) {
+            deletes.add(kvRecordFactory.ofRecord("key" + i, null));
+        }
+        kvTablet.putAsLeader(kvRecordBatchFactory.ofRecords(deletes), null);
+
+        long highOffset = logTablet.localLogEndOffset();
+        kvTablet.flush(highOffset, NOPErrorHandler.INSTANCE);
+        TabletState stateAfterHighFlush = kvTablet.getTabletState();
+        
assertThat(stateAfterHighFlush.getFlushedLogOffset()).isEqualTo(highOffset);
+        assertThat(stateAfterHighFlush.getRowCount()).isEqualTo(3L);
+        assertThat(kvTablet.getRowCount()).isEqualTo(3);
+
+        long lowOffset = highOffset - 2;
+        kvTablet.flush(lowOffset, NOPErrorHandler.INSTANCE);
+        TabletState stateAfterLowFlush = kvTablet.getTabletState();
+
+        
assertThat(stateAfterLowFlush.getFlushedLogOffset()).isEqualTo(highOffset);
+        assertThat(stateAfterLowFlush.getRowCount()).isEqualTo(3L);
+        assertThat(kvTablet.getRowCount()).isEqualTo(3);
+
+        kvTablet.close();
+    }
+
     @Test
     void testOpenScan_emptyBucket_returnsNullContext() throws Exception {
         initLogTabletAndKvTablet(DATA1_SCHEMA_PK, new HashMap<>());