You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ja...@apache.org on 2026/04/16 16:04:31 UTC
(fluss) branch main updated: [lake/tiering] add tiering table pendingTime and freshness metrics (#3076)
This is an automated email from the ASF dual-hosted git repository.
jark 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 92ca8c6ad [lake/tiering] add tiering table pendingTime and freshness metrics (#3076)
92ca8c6ad is described below
commit 92ca8c6ad2b6425d3dc21f95624fb5ee756e8515
Author: Junbo Wang <be...@gmail.com>
AuthorDate: Fri Apr 17 00:04:27 2026 +0800
[lake/tiering] add tiering table pendingTime and freshness metrics (#3076)
---
.../java/org/apache/fluss/metrics/MetricNames.java | 2 ++
.../coordinator/LakeTableTieringManager.java | 26 ++++++++++++++++++
.../coordinator/LakeTableTieringManagerTest.java | 32 ++++++++++++++++++++++
.../maintenance/observability/monitor-metrics.md | 14 ++++++++--
4 files changed, 72 insertions(+), 2 deletions(-)
diff --git a/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java b/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java
index c1cc2b5b6..650a93abb 100644
--- a/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java
+++ b/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java
@@ -70,6 +70,8 @@ public class MetricNames {
public static final String LAKE_TIERING_TABLE_FAILURES_TOTAL = "failuresTotal";
public static final String LAKE_TIERING_TABLE_FILE_SIZE = "fileSize";
public static final String LAKE_TIERING_TABLE_RECORD_COUNT = "recordCount";
+ public static final String LAKE_TIERING_TABLE_PENDING_TIME = "pendingTime";
+ public static final String LAKE_TIERING_TABLE_FRESHNESS = "freshness";
// --------------------------------------------------------------------------------------------
// metrics for tablet server
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 a3448d148..60589a999 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
@@ -141,6 +141,9 @@ public class LakeTableTieringManager implements AutoCloseable {
// table_id -> start time (ms) of the currently in-progress tiering round
private final Map<Long, Long> currentTieringStartTime;
+ // table_id -> time (ms) when the table entered pending queue
+ private final Map<Long, Long> pendingEnterTime;
+
// the live tables that are tiering,
// from table_id -> last heartbeat time by the tiering service
private final Map<Long, Long> liveTieringTableIds;
@@ -184,6 +187,7 @@ public class LakeTableTieringManager implements AutoCloseable {
this.delayedTieringByTableId = new HashMap<>();
this.tableFailureCounters = new HashMap<>();
this.currentTieringStartTime = new HashMap<>();
+ this.pendingEnterTime = new HashMap<>();
this.tieringMetricGroup = lakeTieringMetricGroup;
registerMetrics();
}
@@ -279,6 +283,10 @@ public class LakeTableTieringManager implements AutoCloseable {
MetricNames.LAKE_TIERING_TABLE_TIER_DURATION,
() -> inReadLock(lock, () -> getLastResultField(tableId, r -> r.tierDuration)));
+ // pendingTime: how long the table has been waiting in the pending queue
+ tableMetricGroup.gauge(
+ MetricNames.LAKE_TIERING_TABLE_PENDING_TIME, () -> getTablePendingTime(tableId));
+
// failuresTotal: total failure count for this table
Counter failuresCounter =
tableMetricGroup.counter(MetricNames.LAKE_TIERING_TABLE_FAILURES_TOTAL);
@@ -293,6 +301,11 @@ public class LakeTableTieringManager implements AutoCloseable {
tableMetricGroup.gauge(
MetricNames.LAKE_TIERING_TABLE_RECORD_COUNT,
() -> inReadLock(lock, () -> getLastResultField(tableId, r -> r.recordCount)));
+
+ // freshness: the user-configured table data freshness interval in milliseconds
+ tableMetricGroup.gauge(
+ MetricNames.LAKE_TIERING_TABLE_FRESHNESS,
+ () -> inReadLock(lock, () -> tableLakeFreshness.getOrDefault(tableId, -1L)));
}
/**
@@ -316,6 +329,7 @@ public class LakeTableTieringManager implements AutoCloseable {
tableLakeFreshness.remove(tableId);
lastTieringResult.remove(tableId);
currentTieringStartTime.remove(tableId);
+ pendingEnterTime.remove(tableId);
tableFailureCounters.remove(tableId);
// close and remove the metric group to unregister metrics
tieringMetricGroup.removeTableLakeTieringMetricGroup(tableId);
@@ -579,10 +593,12 @@ public class LakeTableTieringManager implements AutoCloseable {
// increase tiering epoch and initialize the heartbeat of the tiering table
tableTierEpoch.computeIfPresent(tableId, (t, v) -> v + 1);
pendingTieringTables.add(tableId);
+ pendingEnterTime.put(tableId, clock.milliseconds());
break;
case Tiering:
liveTieringTableIds.put(tableId, clock.milliseconds());
currentTieringStartTime.put(tableId, clock.milliseconds());
+ pendingEnterTime.remove(tableId);
break;
case Tiered:
liveTieringTableIds.remove(tableId);
@@ -825,4 +841,14 @@ public class LakeTableTieringManager implements AutoCloseable {
long getLastTieringResultField(long tableId, ToLongFunction<LastTieringResult> fieldExtractor) {
return inReadLock(lock, () -> getLastResultField(tableId, fieldExtractor));
}
+
+ @VisibleForTesting
+ long getTablePendingTime(long tableId) {
+ return inReadLock(
+ lock,
+ () -> {
+ Long enterTime = pendingEnterTime.get(tableId);
+ return enterTime != null ? clock.milliseconds() - enterTime : 0L;
+ });
+ }
}
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 e2108d78a..4c12edbeb 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
@@ -304,6 +304,31 @@ class LakeTableTieringManagerTest {
// Advance time and request table
manualClock.advanceTime(Duration.ofSeconds(10));
+
+ // Wait for the delayed task to execute and move to pending
+ waitValue(
+ () ->
+ tableTieringManager.getTableState(tableId1)
+ == LakeTableTieringManager.TieringState.Pending
+ ? Optional.of(true)
+ : Optional.empty(),
+ Duration.ofSeconds(5),
+ "Table should be in pending state");
+
+ // pendingTime should be > 0 when in pending state
+ // The table entered pending after the 10s delay, so pendingTime should reflect time since
+ // then
+ assertThat(tableTieringManager.getTableState(tableId1))
+ .isEqualTo(LakeTableTieringManager.TieringState.Pending);
+
+ // Verify pendingTime is 0 initially (just entered pending state)
+ assertThat(tableTieringManager.getTablePendingTime(tableId1)).isEqualTo(0L);
+
+ // Advance time while in pending state - pendingTime should increase
+ manualClock.advanceTime(Duration.ofSeconds(5));
+ assertThat(tableTieringManager.getTablePendingTime(tableId1)).isEqualTo(5000L);
+
+ // Request table - should transition to Tiering and pendingTime should reset to 0
assertRequestTable(tableId1, tablePath1, 1);
// State should be Tiering (4)
@@ -372,6 +397,13 @@ class LakeTableTieringManagerTest {
// State should be Pending (3) after failure
assertThat(tableTieringManager.getTableState(tableId1))
.isEqualTo(LakeTableTieringManager.TieringState.Pending);
+
+ // After failure, table re-enters pending state, pendingTime should be 0 initially
+ assertThat(tableTieringManager.getTablePendingTime(tableId1)).isEqualTo(0L);
+
+ // Advance time - pendingTime should increase from failure time
+ manualClock.advanceTime(Duration.ofSeconds(3));
+ assertThat(tableTieringManager.getTablePendingTime(tableId1)).isEqualTo(3000L);
}
@Test
diff --git a/website/docs/maintenance/observability/monitor-metrics.md b/website/docs/maintenance/observability/monitor-metrics.md
index 330a326b2..961d0bc8f 100644
--- a/website/docs/maintenance/observability/monitor-metrics.md
+++ b/website/docs/maintenance/observability/monitor-metrics.md
@@ -294,7 +294,7 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM
</thead>
<tbody>
<tr>
- <th rowspan="25"><strong>coordinator</strong></th>
+ <th rowspan="27"><strong>coordinator</strong></th>
<td style={{textAlign: 'center', verticalAlign: 'middle' }} rowspan="10">-</td>
<td>activeCoordinatorCount</td>
<td>The number of active CoordinatorServer (only leader) in this cluster.</td>
@@ -401,7 +401,7 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM
<td>Gauge</td>
</tr>
<tr>
- <td rowspan="5">lakeTiering_table</td>
+ <td rowspan="7">lakeTiering_table</td>
<td>tierLag</td>
<td>Time in milliseconds since the last successful tiering operation for this table. For newly registered tables that have never completed a tiering round, the lag is measured from the time the table was registered.</td>
<td>Gauge</td>
@@ -426,6 +426,16 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM
<td>Cumulative total record count of the lake table after the last tiering round. Returns -1 if no tiering has completed yet.</td>
<td>Gauge</td>
</tr>
+ <tr>
+ <td>pendingTime</td>
+ <td>How long (in milliseconds) the table has been waiting in the pending queue for tiering. Returns 0 when the table is not currently pending.</td>
+ <td>Gauge</td>
+ </tr>
+ <tr>
+ <td>freshness</td>
+ <td>The user-configured data freshness interval (in milliseconds) for this table.</td>
+ <td>Gauge</td>
+ </tr>
</tbody>
</table>