(fluss) branch main updated: [tiering] fix flink tiering duplicate write result (#3608)

[yuxia]

This is an automated email from the ASF dual-hosted git repository.

luoyuxia pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git


The following commit(s) were added to refs/heads/main by this push:
     new 4da3ebf98 [tiering] fix flink tiering duplicate write result (#3608)
4da3ebf98 is described below

commit 4da3ebf9849abcf398fb01b0db425dba32d7445f
Author: Junfan Zhang <zus...@apache.org>
AuthorDate: Tue Jul 7 19:16:51 2026 +0800

    [tiering] fix flink tiering duplicate write result (#3608)
---
 .../flink/tiering/source/enumerator/TieringSourceEnumerator.java      | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git 
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java
 
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java
index 5545e9a50..ab902e194 100644
--- 
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java
+++ 
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java
@@ -423,6 +423,7 @@ public class TieringSourceEnumerator
                                 TablePath.of(
                                         
tieringTable.getTablePath().getDatabaseName(),
                                         
tieringTable.getTablePath().getTableName()));
+                tieringTableEpochs.put(lakeTieringInfo.f0, lakeTieringInfo.f1);
                 LOG.info("Tiering table {} has been requested.", 
lakeTieringInfo);
             } else {
                 LOG.info("No available Tiering table found, will poll later.");
@@ -463,9 +464,9 @@ public class TieringSourceEnumerator
                 LOG.info(
                         "Generate Tiering splits for table {} is empty, no 
need to tier data.",
                         tieringTable.f2.getTableName());
+                tieringTableEpochs.remove(tieringTable.f0);
                 finishedTables.put(tieringTable.f0, 
TieringFinishInfo.from(tieringTable.f1));
             } else {
-                tieringTableEpochs.put(tieringTable.f0, tieringTable.f1);
                 pendingSplits.addAll(tieringSplits);
 
                 timerService.schedule(
@@ -484,6 +485,7 @@ public class TieringSourceEnumerator
         } catch (Exception e) {
             LOG.warn("Fail to generate Tiering splits for table {}.", 
tieringTable.f2, e);
             failedTableEpochs.put(tieringTable.f0, tieringTable.f1);
+            tieringTableEpochs.remove(tieringTable.f0);
         }
     }