address jark's comments

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);
     }
 
-    public void updateHighWatermarkMetadata(LogOffsetMetadata newHighWatermark) {
+    private void updateHighWatermarkMetadata(LogOffsetMetadata newHighWatermark) {
         if (newHighWatermark.getMessageOffset() < 0) {
             throw new IllegalArgumentException("High watermark offset should be non-negative");
         }
@@ -403,14 +403,11 @@ public void updateHighWatermarkMetadata(LogOffsetMetadata newHighWatermark) {
     }
 
     /**
-     * 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.
+     * 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 {
@@ -428,6 +425,7 @@ 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

@@ -932,16 +932,15 @@ private boolean maybeIncrementLeaderHW(LogTablet leaderLog, long currentTimeMs)
             }
         }
 
+        // 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());
         Optional<LogOffsetMetadata> oldWatermark =
                 leaderLog.maybeIncrementHighWatermark(newHighWatermark);
         if (oldWatermark.isPresent()) {
             LOG.debug("High watermark update from {} to {}.", oldWatermark.get(), newHighWatermark);
-            // 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,11 +161,7 @@ void testHighWatermarkMaintenance() throws Exception {
         logTablet.appendAsLeader(mr1);
         assertHighWatermark(0L);
         // Update highWatermark as leader.
-        LogOffsetMetadata logOffsetMetadata = new LogOffsetMetadata(1L);
-        Optional<LogOffsetMetadata> oldLogOffsetMetadata =
-                logTablet.maybeIncrementHighWatermark(logOffsetMetadata);
-        assertThat(oldLogOffsetMetadata).isPresent();
-        logTablet.updateHighWatermarkMetadata(logOffsetMetadata);
+        logTablet.maybeIncrementHighWatermark(new LogOffsetMetadata(1L));
         assertHighWatermark(1L);
         // Cannot update past the log end offset.
         logTablet.updateHighWatermark(5L);