[server] Cdc log visible only when the data already flush to rocksDb from preWriteBuffer

Author: swuferhong

fluss-server/src/main/java/com/alibaba/fluss/server/log/LogTablet.java

@@ -385,7 +385,7 @@ public void updateHighWatermark(long highWatermark) {
         updateHighWatermarkMetadata(newHighWatermarkMetadata);
     }
 
-    private void updateHighWatermarkMetadata(LogOffsetMetadata newHighWatermark) {
+    public void updateHighWatermarkMetadata(LogOffsetMetadata newHighWatermark) {
         if (newHighWatermark.getMessageOffset() < 0) {
             throw new IllegalArgumentException("High watermark offset should be non-negative");
         }
@@ -403,11 +403,14 @@ private void updateHighWatermarkMetadata(LogOffsetMetadata newHighWatermark) {
     }
 
     /**
-     * Update the highWatermark to a new value if and only if it is larger than the old value. It is
-     * an error to update to a value which is larger than the log end offset.
+     * Judge whether to update the highWatermark to a new value if and only if it is larger than the
+     * old value. It is an error to update to a value which is larger than the log end offset.
      *
      * <p>This method is intended to be used by the leader to update the highWatermark after
      * follower fetch offsets have been updated.
+     *
+     * @param newHighWatermark the suggested new value for the high watermark.
+     * @return the old high watermark if the new high watermark can be updated, otherwise empty.
      */
     public Optional<LogOffsetMetadata> maybeIncrementHighWatermark(
             LogOffsetMetadata newHighWatermark) throws IOException {
@@ -425,7 +428,6 @@ public Optional<LogOffsetMetadata> maybeIncrementHighWatermark(
             if (oldHighWatermark.getMessageOffset() < newHighWatermark.getMessageOffset()
                     || (oldHighWatermark.getMessageOffset() == newHighWatermark.getMessageOffset()
                             && oldHighWatermark.onOlderSegment(newHighWatermark))) {
-                updateHighWatermarkMetadata(newHighWatermark);
                 return Optional.of(oldHighWatermark);
             } else {
                 return Optional.empty();

fluss-server/src/main/java/com/alibaba/fluss/server/replica/Replica.java

@@ -936,8 +936,12 @@ private boolean maybeIncrementLeaderHW(LogTablet leaderLog, long currentTimeMs)
                 leaderLog.maybeIncrementHighWatermark(newHighWatermark);
         if (oldWatermark.isPresent()) {
             LOG.debug("High watermark update from {} to {}.", oldWatermark.get(), newHighWatermark);
-            // when watermark advanced, we may need to flush kv if it's kv replica
+            // when the watermark can be advanced, we may need to flush kv first if it's kv replica,
+            // and then update highWatermark.
+            // TODO The flushKV and updateHighWatermark need to be atomic operation. See
+            // https://github.com/alibaba/fluss/issues/513
             mayFlushKv(newHighWatermark.getMessageOffset());
+            logTablet.updateHighWatermarkMetadata(newHighWatermark);
             return true;
         } else {
             return false;

fluss-server/src/test/java/com/alibaba/fluss/server/log/LogTabletTest.java

@@ -161,7 +161,11 @@ void testHighWatermarkMaintenance() throws Exception {
         logTablet.appendAsLeader(mr1);
         assertHighWatermark(0L);
         // Update highWatermark as leader.
-        logTablet.maybeIncrementHighWatermark(new LogOffsetMetadata(1L));
+        LogOffsetMetadata logOffsetMetadata = new LogOffsetMetadata(1L);
+        Optional<LogOffsetMetadata> oldLogOffsetMetadata =
+                logTablet.maybeIncrementHighWatermark(logOffsetMetadata);
+        assertThat(oldLogOffsetMetadata).isPresent();
+        logTablet.updateHighWatermarkMetadata(logOffsetMetadata);
         assertHighWatermark(1L);
         // Cannot update past the log end offset.
         logTablet.updateHighWatermark(5L);

fluss-server/src/test/java/com/alibaba/fluss/server/replica/ReplicaManagerTest.java

@@ -64,6 +64,7 @@
 
 import javax.annotation.Nullable;
 
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -80,6 +81,7 @@
 import static com.alibaba.fluss.record.TestData.DATA1_KEY_TYPE;
 import static com.alibaba.fluss.record.TestData.DATA1_ROW_TYPE;
 import static com.alibaba.fluss.record.TestData.DATA1_SCHEMA;
+import static com.alibaba.fluss.record.TestData.DATA1_SCHEMA_PK;
 import static com.alibaba.fluss.record.TestData.DATA1_TABLE_ID;
 import static com.alibaba.fluss.record.TestData.DATA1_TABLE_ID_PK;
 import static com.alibaba.fluss.record.TestData.DATA1_TABLE_PATH;
@@ -102,6 +104,7 @@
 import static com.alibaba.fluss.testutils.DataTestUtils.genMemoryLogRecordsByObject;
 import static com.alibaba.fluss.testutils.DataTestUtils.getKeyValuePairs;
 import static com.alibaba.fluss.testutils.DataTestUtils.row;
+import static com.alibaba.fluss.testutils.common.CommonTestUtils.retry;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
@@ -1163,6 +1166,51 @@ void testStopReplica() throws Exception {
         replicaManager.getReplicaOrException(tb);
     }
 
+    @Test
+    void testKvDataVisibility() throws Exception {
+        // The CDC log is only visible after the KV has been flushed to RocksDB. In other words,
+        // when we can read the CDC log, the associated kv record must have been
+        // inserted/updated/deleted in RocksDB. The reason for ensuring this visibility is that we
+        // first buffer the data in memory before flushing it to RocksDB. Thus, we need to guarantee
+        // visibility.
+        TableBucket tb = new TableBucket(DATA1_TABLE_ID_PK, 1);
+        makeKvTableAsLeader(DATA1_TABLE_ID_PK, DATA1_TABLE_PATH_PK, tb.getBucket());
+        Replica replica = replicaManager.getReplicaOrException(tb);
+
+        // retry send kv records to kv store, if the highWatermark increased, the kv record must be
+        // visible in rocksdb.
+        int round = 1000;
+        CompactedKeyEncoder keyEncoder = new CompactedKeyEncoder(DATA1_ROW_TYPE, new int[] {0});
+        CompletableFuture<List<PutKvResultForBucket>> future;
+        for (int i = 0; i < round; i++) {
+            future = new CompletableFuture<>();
+            Object[] key = {i};
+            Object[] value = {i, "a"};
+            byte[] keyBytes = keyEncoder.encodeKey(row(key));
+            byte[] valueBytes =
+                    ValueEncoder.encodeValue(
+                            DEFAULT_SCHEMA_ID, compactedRow(DATA1_SCHEMA_PK.getRowType(), value));
+            List<byte[]> lookups = replica.lookups(Collections.singletonList(keyBytes));
+            assertThat(lookups.size()).isEqualTo(1);
+            assertThat(lookups.get(0)).isNull();
+            replicaManager.putRecordsToKv(
+                    20000,
+                    1,
+                    Collections.singletonMap(
+                            tb, genKvRecordBatch(Collections.singletonList(Tuple2.of(key, value)))),
+                    null,
+                    future::complete);
+            assertThat(future.get()).containsOnly(new PutKvResultForBucket(tb, i + 1));
+
+            int expectedHw = i + 1;
+            retry(
+                    Duration.ofMinutes(1),
+                    () -> assertThat(replica.getLogHighWatermark()).isEqualTo(expectedHw));
+            assertThat(replica.lookups(Collections.singletonList(keyBytes)))
+                    .containsExactlyInAnyOrder(valueBytes);
+        }
+    }
+
     @Test
     void testSnapshotKvReplicas() throws Exception {
         // create multiple kv replicas and all do the snapshot operation