[server] Handle duplicate tiering completion reports idempotently

fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java

@@ -447,6 +447,19 @@ public void finishTableTiering(
                 lock,
                 () -> {
                     validateTieringServiceRequest(tableId, tieredEpoch);
+                    TieringState tieringState = tieringStates.get(tableId);
+                    if (tieringState == TieringState.Scheduled) {
+                        // Completion reports use at-least-once delivery. Heartbeats can copy one
+                        // finished entry twice, and a lost response can trigger a retry. The first
+                        // report moves the table to Scheduled.
+                        // Ignore repeats to preserve statistics and state.
+                        LOG.debug(
+                                "Ignore the duplicate tiering completion report for table {} at epoch {} "
+                                        + "because the table is already in Scheduled state.",
+                                tableId,
+                                tieredEpoch);
+                        return;
+                    }
                     updateTableTieringResult(tableId, stats);
                     // to tiered state firstly
                     doHandleStateChange(tableId, TieringState.Tiered);

fluss-server/src/test/java/org/apache/fluss/server/coordinator/LakeTableTieringManagerTest.java

@@ -160,18 +160,31 @@ void testFinishTableTieringReTriggerSchedule() {
         assertThat(tableTieringManager.requestTable()).isNull();
 
         // mock lake tiering finish one-round tiering
-        tableTieringManager.finishTableTiering(tableId1, tieredEpoch, false, TieringStats.UNKNOWN);
-        // not advance time, request table should return null
-        assertThat(tableTieringManager.requestTable()).isNull();
+        manualClock.advanceTime(Duration.ofSeconds(1));
+        tableTieringManager.finishTableTiering(
+                tableId1, tieredEpoch, false, new TieringStats(1024L, 100L));
+
+        long firstCompletionTime = manualClock.milliseconds();
+        manualClock.advanceTime(Duration.ofMillis(50));
+        tableTieringManager.finishTableTiering(
+                tableId1, tieredEpoch, false, new TieringStats(2048L, 200L));
+
+        assertThat(tableTieringManager.getTableState(tableId1))
+                .isEqualTo(LakeTableTieringManager.TieringState.Scheduled);
+        assertThat(tableTieringManager.getTableLastSuccessTime(tableId1))
+                .isEqualTo(firstCompletionTime);
+        assertThat(tableTieringManager.getLastTieringResultField(tableId1, r -> r.tierDuration))
+                .isEqualTo(1000L);
+        assertThat(tableTieringManager.getLastTieringResultField(tableId1, r -> r.fileSize))
+                .isEqualTo(1024L);
+        assertThat(tableTieringManager.getLastTieringResultField(tableId1, r -> r.recordCount))
+                .isEqualTo(100L);
 
-        // now, advance 1 second to trigger the table tiering
+        // The duplicate must not reschedule the next round from the duplicate report time.
+        assertThat(tableTieringManager.requestTable()).isNull();
         manualClock.advanceTime(Duration.ofSeconds(4));
-        // not reach data freshness, shouldn't request table
         assertThat(tableTieringManager.requestTable()).isNull();
-
-        // advance 6 seconds again, should get table now
-        manualClock.advanceTime(Duration.ofSeconds(6));
-        // the tiered epoch should be 2 now
+        manualClock.advanceTime(Duration.ofMillis(5950));
         assertRequestTable(tableId1, tablePath1, 2);
     }
 
@@ -421,6 +434,17 @@ void testForceFinishTableTieringImmediatelyRePending() {
 
         // mock lake tiering force finish (e.g., due to exceeding tiering duration)
         tableTieringManager.finishTableTiering(tableId1, 1, true, TieringStats.UNKNOWN);
+
+        // a repeated forced completion uses the old epoch and must still be fenced
+        assertThatThrownBy(
+                        () ->
+                                tableTieringManager.finishTableTiering(
+                                        tableId1, 1, true, TieringStats.UNKNOWN))
+                .isInstanceOf(FencedTieringEpochException.class)
+                .hasMessage(
+                        "The tiering epoch %d is not match current epoch %d in coordinator for table %d.",
+                        1, 2, tableId1);
+
         // should immediately be re-pending and can be requested again without waiting
         assertRequestTable(tableId1, tablePath1, 2);