[flink] Expose the tiering reader waiting ratio

Author: beryllw

fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java

@@ -373,4 +373,7 @@ public class MetricNames {
     // for lake tiering metrics - operator level
     public static final String TIERING_SERVICE_READ_BYTES = "readBytes";
     public static final String TIERING_SERVICE_READ_BYTES_RATE = "readBytesPerSecond";
+
+    // for lake tiering metrics - enumerator level
+    public static final String TIERING_SERVICE_WAITING_READERS_RATIO = "waitingReadersRatio";
 }

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java

@@ -31,6 +31,7 @@
 import org.apache.fluss.flink.tiering.event.FailedTieringEvent;
 import org.apache.fluss.flink.tiering.event.FinishedTieringEvent;
 import org.apache.fluss.flink.tiering.event.TieringReachMaxDurationEvent;
+import org.apache.fluss.flink.tiering.source.metrics.TieringMetrics;
 import org.apache.fluss.flink.tiering.source.split.FlinkTieringSplit;
 import org.apache.fluss.flink.tiering.source.state.TieringSourceEnumeratorState;
 import org.apache.fluss.lake.committer.TieringStats;
@@ -189,6 +190,9 @@ public TieringSourceEnumerator(
         this.finishedTables = new ConcurrentHashMap<>();
         this.failedTableEpochs = new ConcurrentHashMap<>();
         this.tieringReachMaxDurationsTables = Collections.synchronizedSet(new TreeSet<>());
+        TieringMetrics.registerWaitingReadersRatio(
+                enumeratorMetricGroup,
+                () -> (double) readersAwaitingSplit.size() / context.currentParallelism());
     }
 
     @Override

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/metrics/TieringMetrics.java

@@ -21,15 +21,18 @@
 import org.apache.fluss.metrics.MetricNames;
 
 import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Gauge;
 import org.apache.flink.metrics.MeterView;
 import org.apache.flink.metrics.MetricGroup;
 import org.apache.flink.metrics.groups.SourceReaderMetricGroup;
+import org.apache.flink.metrics.groups.SplitEnumeratorMetricGroup;
 
 /**
- * A collection class for handling metrics in Tiering source reader.
+ * Metrics for the Flink-based tiering source.
  *
- * <p>All metrics are registered under group "fluss.tieringService", which is a child group of
- * {@link org.apache.flink.metrics.groups.OperatorMetricGroup}.
+ * <p>All metrics are registered under group "fluss.tieringService", as a child of {@link
+ * SourceReaderMetricGroup} for reader metrics or {@link SplitEnumeratorMetricGroup} for enumerator
+ * metrics.
  *
  * <p>The following metrics are available:
  *
@@ -38,6 +41,8 @@
  *       since the job started (actual Fluss records size).
  *   <li>{@code fluss.tieringService.readBytesPerSecond} - Meter: bytes-per-second rate derived from
  *       the counter using a 60-second sliding window.
+ *   <li>{@code fluss.tieringService.waitingReadersRatio} - Gauge: fraction of source readers
+ *       waiting for splits, reported by the enumerator.
  * </ul>
  */
 @Internal
@@ -64,4 +69,13 @@ public TieringMetrics(SourceReaderMetricGroup sourceReaderMetricGroup) {
     public void recordBytesRead(long bytes) {
         readBytesCounter.inc(bytes);
     }
+
+    /** Registers the proportion of source readers waiting for splits. */
+    public static void registerWaitingReadersRatio(
+            SplitEnumeratorMetricGroup metricGroup, Gauge<Double> waitingReadersRatio) {
+        metricGroup
+                .addGroup(FLUSS_METRIC_GROUP)
+                .addGroup(TIERING_SERVICE_GROUP)
+                .gauge(MetricNames.TIERING_SERVICE_WAITING_READERS_RATIO, waitingReadersRatio);
+    }
 }

fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorSchedulingTest.java

@@ -23,18 +23,23 @@
 import org.apache.fluss.client.tiering.TieringSplitGenerator;
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.flink.tiering.TestingLakeTieringFactory;
+import org.apache.fluss.flink.tiering.source.metrics.TieringMetrics;
 import org.apache.fluss.flink.tiering.source.split.FlinkTieringSplit;
 import org.apache.fluss.metadata.Schema;
 import org.apache.fluss.metadata.TableBucket;
 import org.apache.fluss.metadata.TableDescriptor;
 import org.apache.fluss.metadata.TableInfo;
 import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.metrics.MetricNames;
 import org.apache.fluss.rpc.gateway.CoordinatorGateway;
 import org.apache.fluss.rpc.messages.LakeTieringHeartbeatRequest;
 import org.apache.fluss.rpc.messages.LakeTieringHeartbeatResponse;
 import org.apache.fluss.testutils.common.ManuallyTriggeredScheduledExecutorService;
 import org.apache.fluss.types.DataTypes;
 
+import org.apache.flink.metrics.groups.SplitEnumeratorMetricGroup;
+import org.apache.flink.metrics.testutils.MetricListener;
+import org.apache.flink.runtime.metrics.groups.InternalSplitEnumeratorMetricGroup;
 import org.junit.jupiter.api.Test;
 
 import java.io.IOException;
@@ -59,6 +64,7 @@ class TieringSourceEnumeratorSchedulingTest {
     @Test
     void testPendingSplitsAssignedInOrder() throws Exception {
         try (Fixture fixture = new Fixture()) {
+            assertThat(fixture.waitingReadersRatio()).isEqualTo(1.0);
             List<FlinkTieringSplit> splits = new ArrayList<>();
             for (int bucket = 0; bucket < 4; bucket++) {
                 splits.add(
@@ -83,11 +89,13 @@ void testPendingSplitsAssignedInOrder() throws Exception {
                             .collect(Collectors.toList());
             assertThat(assignedSplits).containsExactlyElementsOf(splits);
             assertThat(fixture.claims()).isEmpty();
+            assertThat(fixture.waitingReadersRatio()).isZero();
 
             // Once drained, another request polls for a new table without reassigning old splits.
             fixture.requestTable();
             assertThat(fixture.claims()).hasSize(1);
             assertThat(fixture.context.getSplitsAssignmentSequence()).hasSize(splits.size());
+            assertThat(fixture.waitingReadersRatio()).isEqualTo(1.0);
         }
     }
 
@@ -176,6 +184,7 @@ void testGenerationFailureDoesNotScheduleContinuation() throws Throwable {
                     .thenThrow(new IOException("metadata unavailable"));
 
             fixture.requestTable();
+            assertThat(fixture.waitingReadersRatio()).isEqualTo(1.0);
 
             // Failures must not drive immediate retries of a table re-queued by the coordinator.
             assertThat(fixture.timer.getActiveNonPeriodicScheduledTask()).isEmpty();
@@ -193,12 +202,15 @@ void testGenerationFailureDoesNotScheduleContinuation() throws Throwable {
     @Test
     void testFailOverPreventsContinuationClaim() throws Throwable {
         try (Fixture fixture = new Fixture(2)) {
+            assertThat(fixture.waitingReadersRatio()).isEqualTo(0.5);
             fixture.registerReader(1, 0);
+            assertThat(fixture.waitingReadersRatio()).isEqualTo(1.0);
             fixture.addTable(1, true);
             fixture.requestTable();
 
             // Reader 0 restarts while reader 1 is still on attempt 0, so failover is in progress.
             fixture.registerReader(0, 1);
+            assertThat(fixture.waitingReadersRatio()).isEqualTo(1.0);
 
             // The failover guard must leave the next table to the periodic poll.
             fixture.fireContinuation();
@@ -273,6 +285,17 @@ private void registerReader(int subtaskId, int attemptNumber) {
             enumerator.addReader(subtaskId);
         }
 
+        private double waitingReadersRatio() {
+            return context.metricListener
+                    .<Double>getGauge(
+                            "enumerator",
+                            TieringMetrics.FLUSS_METRIC_GROUP,
+                            TieringMetrics.TIERING_SERVICE_GROUP,
+                            MetricNames.TIERING_SERVICE_WAITING_READERS_RATIO)
+                    .get()
+                    .getValue();
+        }
+
         private CompletableFuture<LakeTieringHeartbeatResponse> respondToHeartbeat(
                 LakeTieringHeartbeatRequest request) {
             requests.add(request);
@@ -344,11 +367,17 @@ public void close() throws Exception {
 
     private static class TestingContext extends FlussMockSplitEnumeratorContext<FlinkTieringSplit> {
         private final Queue<Runnable> coordinatorCalls = new ArrayDeque<>();
+        private final MetricListener metricListener = new MetricListener();
 
         private TestingContext(int parallelism) {
             super(parallelism);
         }
 
+        @Override
+        public SplitEnumeratorMetricGroup metricGroup() {
+            return new InternalSplitEnumeratorMetricGroup(metricListener.getMetricGroup());
+        }
+
         @Override
         public void runInCoordinatorThread(Runnable runnable) {
             coordinatorCalls.add(runnable);

website/docs/maintenance/observability/monitor-metrics.md

@@ -1329,8 +1329,8 @@ How to Use Flink Metrics, you can see [Flink Metrics](https://nightlies.apache.o
 
 ### Tiering Service Metrics
 
-These metrics are exposed by the Flink-based tiering source reader when running the Lake Tiering Service.
-All metrics are registered under the `fluss.tieringService` metric group, which is a child of the Flink `SourceReaderMetricGroup`.
+These metrics are exposed by the Flink-based tiering source when running the Lake Tiering Service.
+All metrics are registered under the `fluss.tieringService` metric group, as a child of the Flink `SourceReaderMetricGroup` for reader metrics or `SplitEnumeratorMetricGroup` for enumerator metrics.
 
 <table class="table table-bordered">
     <thead>
@@ -1354,5 +1354,11 @@ All metrics are registered under the `fluss.tieringService` metric group, which
             <td>The read throughput rate in bytes per second, derived from the readBytes counter using a 60-second sliding window.</td>
             <td>Meter</td>
         </tr>
+        <tr>
+            <td>waitingReadersRatio</td>
+            <td>Flink Source Enumerator</td>
+            <td>The number of readers tracked by the enumerator as waiting for a split divided by the current source parallelism.</td>
+            <td>Gauge</td>
+        </tr>
     </tbody>
 </table>