diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java index 9498c45ecd..019ec9858d 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java +++ b/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); diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/LakeTableTieringManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/LakeTableTieringManagerTest.java index 4c12edbeb2..13b273c1cb 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/LakeTableTieringManagerTest.java +++ b/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);