From 5e46e799e32f30c130578b999131ce1815d66940 Mon Sep 17 00:00:00 2001 From: fhan Date: Sun, 20 Sep 2026 19:21:54 +0800 Subject: [PATCH 1/2] [server] Handle duplicate tiering completion reports idempotently --- .../coordinator/LakeTableTieringManager.java | 14 +++ .../LakeTableTieringManagerTest.java | 96 +++++++++++++++++++ 2 files changed, 110 insertions(+) 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 9498c45ecdf..586bc4be046 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,20 @@ public void finishTableTiering( lock, () -> { validateTieringServiceRequest(tableId, tieredEpoch); + TieringState tieringState = tieringStates.get(tableId); + if (!isForceFinished + && tieredEpoch > 0 + && tieringState == TieringState.Scheduled) { + // The normal tiering round has already completed. Treat a repeated report + // for the same epoch as a successful no-op. + return; + } + if (tieringState != TieringState.Tiering) { + throw new IllegalStateException( + String.format( + "The table %d to finish tiering must in Tiering state, but in %s state.", + tableId, tieringState)); + } 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 4c12edbeb24..1ee1e2ba87e 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 @@ -175,6 +175,91 @@ void testFinishTableTieringReTriggerSchedule() { assertRequestTable(tableId1, tablePath1, 2); } + @Test + void testDuplicateNormalFinishIsIdempotent() { + long tableId = 1L; + TablePath tablePath = TablePath.of("db", "table"); + TableInfo tableInfo = createTableInfo(tableId, tablePath, Duration.ofSeconds(10)); + tableTieringManager.addNewLakeTable(tableInfo); + + manualClock.advanceTime(Duration.ofSeconds(10)); + assertRequestTable(tableId, tablePath, 1); + + manualClock.advanceTime(Duration.ofSeconds(1)); + tableTieringManager.finishTableTiering(tableId, 1, false, new TieringStats(1024L, 100L)); + + long firstCompletionTime = manualClock.milliseconds(); + manualClock.advanceTime(Duration.ofMillis(50)); + tableTieringManager.finishTableTiering(tableId, 1, false, new TieringStats(2048L, 200L)); + + assertThat(tableTieringManager.getTableState(tableId)) + .isEqualTo(LakeTableTieringManager.TieringState.Scheduled); + assertThat(tableTieringManager.getTableLastSuccessTime(tableId)) + .isEqualTo(firstCompletionTime); + assertThat(tableTieringManager.getLastTieringResultField(tableId, r -> r.tierDuration)) + .isEqualTo(1000L); + assertThat(tableTieringManager.getLastTieringResultField(tableId, r -> r.fileSize)) + .isEqualTo(1024L); + assertThat(tableTieringManager.getLastTieringResultField(tableId, r -> r.recordCount)) + .isEqualTo(100L); + + // The duplicate must not reschedule the next round from the duplicate report time. + manualClock.advanceTime(Duration.ofMillis(9950)); + assertRequestTable(tableId, tablePath, 2); + + // Once the next round starts, the old completion must still be fenced. + assertThatThrownBy( + () -> + tableTieringManager.finishTableTiering( + tableId, 1, false, TieringStats.UNKNOWN)) + .isInstanceOf(FencedTieringEpochException.class) + .hasMessage( + "The tiering epoch %d is not match current epoch %d in coordinator for table %d.", + 1, 2, tableId); + } + + @Test + void testFinishTableTieringRequiresTieringState() { + long tableId = 1L; + TablePath tablePath = TablePath.of("db", "table"); + TableInfo tableInfo = createTableInfo(tableId, tablePath, Duration.ofSeconds(10)); + tableTieringManager.addNewLakeTable(tableInfo); + + assertThatThrownBy( + () -> + tableTieringManager.finishTableTiering( + tableId, 0, false, new TieringStats(1024L, 100L))) + .isInstanceOf(IllegalStateException.class) + .hasMessage( + "The table %d to finish tiering must in Tiering state, but in %s state.", + tableId, LakeTableTieringManager.TieringState.Scheduled); + assertThat(tableTieringManager.getTableLastSuccessTime(tableId)).isEqualTo(0L); + assertThat(tableTieringManager.getLastTieringResultField(tableId, r -> r.tierDuration)) + .isEqualTo(-1L); + + manualClock.advanceTime(Duration.ofSeconds(10)); + waitValue( + () -> + tableTieringManager.getTableState(tableId) + == LakeTableTieringManager.TieringState.Pending + ? Optional.of(true) + : Optional.empty(), + Duration.ofSeconds(5), + "Table should be in pending state"); + + assertThatThrownBy( + () -> + tableTieringManager.finishTableTiering( + tableId, 1, false, new TieringStats(1024L, 100L))) + .isInstanceOf(IllegalStateException.class) + .hasMessage( + "The table %d to finish tiering must in Tiering state, but in %s state.", + tableId, LakeTableTieringManager.TieringState.Pending); + assertThat(tableTieringManager.getTableLastSuccessTime(tableId)).isEqualTo(0L); + assertThat(tableTieringManager.getLastTieringResultField(tableId, r -> r.tierDuration)) + .isEqualTo(-1L); + } + @Test void testTieringServiceTimeOutReTriggerPending() { long tableId1 = 1L; @@ -421,6 +506,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); From fe9a57189b1010e5d3d64af1b902fe389845f8e8 Mon Sep 17 00:00:00 2001 From: fhan Date: Mon, 21 Sep 2026 14:52:53 +0800 Subject: [PATCH 2/2] [lake/paimon] refine code impl according to review comments --- .../coordinator/LakeTableTieringManager.java | 21 ++-- .../LakeTableTieringManagerTest.java | 100 +++--------------- 2 files changed, 24 insertions(+), 97 deletions(-) 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 586bc4be046..019ec9858df 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 @@ -448,19 +448,18 @@ public void finishTableTiering( () -> { validateTieringServiceRequest(tableId, tieredEpoch); TieringState tieringState = tieringStates.get(tableId); - if (!isForceFinished - && tieredEpoch > 0 - && tieringState == TieringState.Scheduled) { - // The normal tiering round has already completed. Treat a repeated report - // for the same epoch as a successful no-op. + 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; } - if (tieringState != TieringState.Tiering) { - throw new IllegalStateException( - String.format( - "The table %d to finish tiering must in Tiering state, but in %s state.", - tableId, tieringState)); - } 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 1ee1e2ba87e..13b273c1cb8 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,104 +160,32 @@ 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(); - - // now, advance 1 second to trigger the table tiering - 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 - assertRequestTable(tableId1, tablePath1, 2); - } - - @Test - void testDuplicateNormalFinishIsIdempotent() { - long tableId = 1L; - TablePath tablePath = TablePath.of("db", "table"); - TableInfo tableInfo = createTableInfo(tableId, tablePath, Duration.ofSeconds(10)); - tableTieringManager.addNewLakeTable(tableInfo); - - manualClock.advanceTime(Duration.ofSeconds(10)); - assertRequestTable(tableId, tablePath, 1); - manualClock.advanceTime(Duration.ofSeconds(1)); - tableTieringManager.finishTableTiering(tableId, 1, false, new TieringStats(1024L, 100L)); + tableTieringManager.finishTableTiering( + tableId1, tieredEpoch, false, new TieringStats(1024L, 100L)); long firstCompletionTime = manualClock.milliseconds(); manualClock.advanceTime(Duration.ofMillis(50)); - tableTieringManager.finishTableTiering(tableId, 1, false, new TieringStats(2048L, 200L)); + tableTieringManager.finishTableTiering( + tableId1, tieredEpoch, false, new TieringStats(2048L, 200L)); - assertThat(tableTieringManager.getTableState(tableId)) + assertThat(tableTieringManager.getTableState(tableId1)) .isEqualTo(LakeTableTieringManager.TieringState.Scheduled); - assertThat(tableTieringManager.getTableLastSuccessTime(tableId)) + assertThat(tableTieringManager.getTableLastSuccessTime(tableId1)) .isEqualTo(firstCompletionTime); - assertThat(tableTieringManager.getLastTieringResultField(tableId, r -> r.tierDuration)) + assertThat(tableTieringManager.getLastTieringResultField(tableId1, r -> r.tierDuration)) .isEqualTo(1000L); - assertThat(tableTieringManager.getLastTieringResultField(tableId, r -> r.fileSize)) + assertThat(tableTieringManager.getLastTieringResultField(tableId1, r -> r.fileSize)) .isEqualTo(1024L); - assertThat(tableTieringManager.getLastTieringResultField(tableId, r -> r.recordCount)) + assertThat(tableTieringManager.getLastTieringResultField(tableId1, r -> r.recordCount)) .isEqualTo(100L); // The duplicate must not reschedule the next round from the duplicate report time. - manualClock.advanceTime(Duration.ofMillis(9950)); - assertRequestTable(tableId, tablePath, 2); - - // Once the next round starts, the old completion must still be fenced. - assertThatThrownBy( - () -> - tableTieringManager.finishTableTiering( - tableId, 1, false, TieringStats.UNKNOWN)) - .isInstanceOf(FencedTieringEpochException.class) - .hasMessage( - "The tiering epoch %d is not match current epoch %d in coordinator for table %d.", - 1, 2, tableId); - } - - @Test - void testFinishTableTieringRequiresTieringState() { - long tableId = 1L; - TablePath tablePath = TablePath.of("db", "table"); - TableInfo tableInfo = createTableInfo(tableId, tablePath, Duration.ofSeconds(10)); - tableTieringManager.addNewLakeTable(tableInfo); - - assertThatThrownBy( - () -> - tableTieringManager.finishTableTiering( - tableId, 0, false, new TieringStats(1024L, 100L))) - .isInstanceOf(IllegalStateException.class) - .hasMessage( - "The table %d to finish tiering must in Tiering state, but in %s state.", - tableId, LakeTableTieringManager.TieringState.Scheduled); - assertThat(tableTieringManager.getTableLastSuccessTime(tableId)).isEqualTo(0L); - assertThat(tableTieringManager.getLastTieringResultField(tableId, r -> r.tierDuration)) - .isEqualTo(-1L); - - manualClock.advanceTime(Duration.ofSeconds(10)); - waitValue( - () -> - tableTieringManager.getTableState(tableId) - == LakeTableTieringManager.TieringState.Pending - ? Optional.of(true) - : Optional.empty(), - Duration.ofSeconds(5), - "Table should be in pending state"); - - assertThatThrownBy( - () -> - tableTieringManager.finishTableTiering( - tableId, 1, false, new TieringStats(1024L, 100L))) - .isInstanceOf(IllegalStateException.class) - .hasMessage( - "The table %d to finish tiering must in Tiering state, but in %s state.", - tableId, LakeTableTieringManager.TieringState.Pending); - assertThat(tableTieringManager.getTableLastSuccessTime(tableId)).isEqualTo(0L); - assertThat(tableTieringManager.getLastTieringResultField(tableId, r -> r.tierDuration)) - .isEqualTo(-1L); + assertThat(tableTieringManager.requestTable()).isNull(); + manualClock.advanceTime(Duration.ofSeconds(4)); + assertThat(tableTieringManager.requestTable()).isNull(); + manualClock.advanceTime(Duration.ofMillis(5950)); + assertRequestTable(tableId1, tablePath1, 2); } @Test