[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

Author: wattt3

fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java

@@ -687,7 +687,9 @@ public void flush(long exclusiveUpToLogOffset, FatalErrorHandler fatalErrorHandl
                     } 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;

fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java

@@ -59,6 +59,7 @@
 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 @@ void testRowCountWithMixedOperations() throws Exception {
         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<>());