[flink] Add heartbeat and request-table metrics to TieringSourceEnumerator

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

@@ -303,4 +303,9 @@ 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 enumerator metrics
+    public static final String TIERING_HEARTBEAT_FAILURE = "tiering.heartbeat.failure";
+    public static final String TIERING_REQUEST_TABLE_EMPTY = "tiering.request-table.empty";
+    public static final String TIERING_REQUEST_TABLE_FAILURE = "tiering.request-table.failure";
 }

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

@@ -27,6 +27,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.TieringEnumeratorMetrics;
 import org.apache.fluss.flink.tiering.source.split.TieringSplit;
 import org.apache.fluss.flink.tiering.source.split.TieringSplitGenerator;
 import org.apache.fluss.flink.tiering.source.state.TieringSourceEnumeratorState;
@@ -78,7 +79,6 @@
 import static org.apache.fluss.flink.tiering.source.enumerator.TieringSourceEnumerator.HeartBeatHelper.failedTableHeartBeat;
 import static org.apache.fluss.flink.tiering.source.enumerator.TieringSourceEnumerator.HeartBeatHelper.heartBeatWithRequestNewTieringTable;
 import static org.apache.fluss.flink.tiering.source.enumerator.TieringSourceEnumerator.HeartBeatHelper.tieringTableHeartBeat;
-import static org.apache.fluss.flink.tiering.source.enumerator.TieringSourceEnumerator.HeartBeatHelper.waitHeartbeatResponse;
 
 /**
  * An implementation of {@link SplitEnumerator} used to request {@link TieringSplit} from Fluss
@@ -121,6 +121,8 @@ public class TieringSourceEnumerator
     private TieringSplitGenerator splitGenerator;
     private int flussCoordinatorEpoch;
 
+    private TieringEnumeratorMetrics tieringMetrics;
+
     private volatile boolean isFailOvering = false;
 
     private volatile boolean closed = false;
@@ -144,6 +146,17 @@ public TieringSourceEnumerator(
         this.finishedTables = new ConcurrentHashMap<>();
         this.failedTableEpochs = new ConcurrentHashMap<>();
         this.tieringReachMaxDurationsTables = Collections.synchronizedSet(new TreeSet<>());
+        this.tieringMetrics = new TieringEnumeratorMetrics(enumeratorMetricGroup);
+    }
+
+    @VisibleForTesting
+    TieringEnumeratorMetrics getTieringMetrics() {
+        return tieringMetrics;
+    }
+
+    @VisibleForTesting
+    void setCoordinatorGateway(CoordinatorGateway gateway) {
+        this.coordinatorGateway = gateway;
     }
 
     @Override
@@ -171,6 +184,7 @@ public void start() {
                     flussCoordinatorEpoch);
 
         } catch (Exception e) {
+            tieringMetrics.incHeartbeatFailure();
             LOG.error("Register Tiering Service failed due to ", e);
             throw new FlinkRuntimeException("Register Tiering Service failed due to ", e);
         }
@@ -362,6 +376,7 @@ protected void handleTableTieringReachMaxDuration(
     void generateAndAssignSplits(
             @Nullable Tuple3<Long, Long, TablePath> tieringTable, Throwable throwable) {
         if (throwable != null) {
+            tieringMetrics.incRequestTableFailure();
             ExceptionUtils.rethrow(throwable);
         }
         if (tieringTable != null) {
@@ -392,7 +407,9 @@ private void assignSplits() {
         }
     }
 
-    private @Nullable Tuple3<Long, Long, TablePath> requestTieringTableSplitsViaHeartBeat() {
+    @VisibleForTesting
+    @Nullable
+    Tuple3<Long, Long, TablePath> requestTieringTableSplitsViaHeartBeat() {
         if (closed) {
             return null;
         }
@@ -431,6 +448,7 @@ private void assignSplits() {
                 tieringTableEpochs.put(lakeTieringInfo.f0, lakeTieringInfo.f1);
                 LOG.info("Tiering table {} has been requested.", lakeTieringInfo);
             } else {
+                tieringMetrics.incRequestTableEmpty();
                 LOG.info("No available Tiering table found, will poll later.");
             }
         } else {
@@ -570,6 +588,7 @@ private void reportFailedTable(
             LOG.info("Report failed table to Fluss Coordinator success");
 
         } catch (Exception e) {
+            tieringMetrics.incHeartbeatFailure();
             LOG.error("Errors happens when report failed table to Fluss cluster.", e);
             throw new FlinkRuntimeException(
                     "Errors happens when report failed table to Fluss cluster.", e);
@@ -662,15 +681,20 @@ static LakeTieringHeartbeatRequest failedTableHeartBeat(
             }
             return lakeTieringHeartbeatRequest;
         }
+    }
 
-        static LakeTieringHeartbeatResponse waitHeartbeatResponse(
-                CompletableFuture<LakeTieringHeartbeatResponse> responseCompletableFuture) {
-            try {
-                return responseCompletableFuture.get(3, TimeUnit.MINUTES);
-            } catch (Exception e) {
-                LOG.error("Failed to wait heartbeat response due to ", e);
-                throw new FlinkRuntimeException("Failed to wait heartbeat response due to ", e);
-            }
+    /**
+     * Waits for the heartbeat response from the Fluss coordinator. Increments the heartbeat failure
+     * counter if the wait fails.
+     */
+    private LakeTieringHeartbeatResponse waitHeartbeatResponse(
+            CompletableFuture<LakeTieringHeartbeatResponse> responseCompletableFuture) {
+        try {
+            return responseCompletableFuture.get(3, TimeUnit.MINUTES);
+        } catch (Exception e) {
+            tieringMetrics.incHeartbeatFailure();
+            LOG.error("Failed to wait heartbeat response due to ", e);
+            throw new FlinkRuntimeException("Failed to wait heartbeat response due to ", e);
         }
     }
 

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

@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.flink.tiering.source.metrics;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.metrics.MetricNames;
+
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.SplitEnumeratorMetricGroup;
+
+/**
+ * A collection class for handling metrics in {@link
+ * org.apache.fluss.flink.tiering.source.enumerator.TieringSourceEnumerator}.
+ *
+ * <p>All metrics are registered under group "fluss.tieringService", which is a child group of
+ * {@link SplitEnumeratorMetricGroup}.
+ *
+ * <p>The following metrics are available:
+ *
+ * <ul>
+ *   <li>{@code fluss.tieringService.tiering.heartbeat.failure} - Counter: cumulative number of
+ *       times the heartbeat request to the Fluss coordinator failed.
+ *   <li>{@code fluss.tieringService.tiering.request-table.empty} - Counter: cumulative number of
+ *       times the enumerator requested a new tiering table but the coordinator returned none.
+ *   <li>{@code fluss.tieringService.tiering.request-table.failure} - Counter: cumulative number of
+ *       times the periodic {@code requestTieringTableSplitsViaHeartBeat} call failed.
+ * </ul>
+ */
+@Internal
+public class TieringEnumeratorMetrics {
+
+    public static final String FLUSS_METRIC_GROUP = "fluss";
+    public static final String TIERING_SERVICE_GROUP = "tieringService";
+
+    private final Counter heartbeatFailureCounter;
+    private final Counter requestTableEmptyCounter;
+    private final Counter requestTableFailureCounter;
+
+    public TieringEnumeratorMetrics(SplitEnumeratorMetricGroup enumeratorMetricGroup) {
+        MetricGroup tieringServiceGroup =
+                enumeratorMetricGroup.addGroup(FLUSS_METRIC_GROUP).addGroup(TIERING_SERVICE_GROUP);
+
+        this.heartbeatFailureCounter =
+                tieringServiceGroup.counter(MetricNames.TIERING_HEARTBEAT_FAILURE);
+        this.requestTableEmptyCounter =
+                tieringServiceGroup.counter(MetricNames.TIERING_REQUEST_TABLE_EMPTY);
+        this.requestTableFailureCounter =
+                tieringServiceGroup.counter(MetricNames.TIERING_REQUEST_TABLE_FAILURE);
+    }
+
+    /** Increments when a heartbeat request to the coordinator fails. */
+    public void incHeartbeatFailure() {
+        heartbeatFailureCounter.inc();
+    }
+
+    /** Increments when the coordinator has no tiering table to assign. */
+    public void incRequestTableEmpty() {
+        requestTableEmptyCounter.inc();
+    }
+
+    /** Increments when the periodic request for tiering table splits fails. */
+    public void incRequestTableFailure() {
+        requestTableFailureCounter.inc();
+    }
+
+    public long getHeartbeatFailureCount() {
+        return heartbeatFailureCounter.getCount();
+    }
+
+    public long getRequestTableEmptyCount() {
+        return requestTableEmptyCounter.getCount();
+    }
+
+    public long getRequestTableFailureCount() {
+        return requestTableFailureCounter.getCount();
+    }
+}

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

@@ -34,7 +34,9 @@
 import org.apache.fluss.metadata.TableChange;
 import org.apache.fluss.metadata.TableInfo;
 import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.rpc.gateway.CoordinatorGateway;
 import org.apache.fluss.rpc.messages.CommitLakeTableSnapshotRequest;
+import org.apache.fluss.rpc.messages.LakeTieringHeartbeatResponse;
 import org.apache.fluss.rpc.messages.PbLakeTableOffsetForBucket;
 import org.apache.fluss.rpc.messages.PbLakeTableSnapshotInfo;
 
@@ -49,12 +51,14 @@
 import javax.annotation.Nullable;
 
 import java.io.IOException;
+import java.lang.reflect.Proxy;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
@@ -937,4 +941,117 @@ void testTableReachMaxTieringDuration() throws Throwable {
                                             && !split.shouldSkipCurrentRound());
         }
     }
+
+    @Test
+    void testHeartbeatFailureMetricIncremented() throws Exception {
+        try (FlussMockSplitEnumeratorContext<TieringSplit> context =
+                new FlussMockSplitEnumeratorContext<>(1)) {
+            TieringSourceEnumerator enumerator = createTieringSourceEnumerator(flussConf, context);
+            enumerator.start();
+
+            // Register a reader so that the "request new table" path is taken
+            registerSingleReaderAndHandleSplitRequests(context, enumerator, 0, 0);
+
+            long heartbeatFailuresBefore =
+                    enumerator.getTieringMetrics().getHeartbeatFailureCount();
+
+            // Swap to a failing gateway that throws on lakeTieringHeartbeat
+            CoordinatorGateway failingGateway = createFailingHeartbeatGateway();
+            enumerator.setCoordinatorGateway(failingGateway);
+
+            // Call requestTieringTableSplitsViaHeartBeat; it should throw FlinkRuntimeException
+            assertThatThrownBy(enumerator::requestTieringTableSplitsViaHeartBeat)
+                    .isInstanceOf(FlinkRuntimeException.class);
+
+            // Verify heartbeat failure counter was incremented
+            assertThat(enumerator.getTieringMetrics().getHeartbeatFailureCount())
+                    .isGreaterThan(heartbeatFailuresBefore);
+        }
+    }
+
+    @Test
+    void testRequestTableEmptyMetricIncremented() throws Exception {
+        try (FlussMockSplitEnumeratorContext<TieringSplit> context =
+                new FlussMockSplitEnumeratorContext<>(1)) {
+            TieringSourceEnumerator enumerator = createTieringSourceEnumerator(flussConf, context);
+            enumerator.start();
+
+            // Register a reader so that the "request new table" path is taken
+            registerSingleReaderAndHandleSplitRequests(context, enumerator, 0, 0);
+
+            long emptyBefore = enumerator.getTieringMetrics().getRequestTableEmptyCount();
+
+            // Swap to a gateway that returns empty heartbeat response (no tiering table)
+            CoordinatorGateway emptyResponseGateway = createEmptyHeartbeatResponseGateway();
+            enumerator.setCoordinatorGateway(emptyResponseGateway);
+
+            // Call requestTieringTableSplitsViaHeartBeat; it should succeed but find no table
+            enumerator.requestTieringTableSplitsViaHeartBeat();
+
+            // Verify request-table.empty counter was incremented
+            assertThat(enumerator.getTieringMetrics().getRequestTableEmptyCount())
+                    .isGreaterThan(emptyBefore);
+        }
+    }
+
+    @Test
+    void testRequestTableFailureMetricIncrementedOnGenerateAndAssignSplits() throws Exception {
+        try (FlussMockSplitEnumeratorContext<TieringSplit> context =
+                new FlussMockSplitEnumeratorContext<>(1)) {
+            TieringSourceEnumerator enumerator = createTieringSourceEnumerator(flussConf, context);
+            enumerator.start();
+
+            long failuresBefore = enumerator.getTieringMetrics().getRequestTableFailureCount();
+
+            // generateAndAssignSplits with a non-null throwable should increment the counter
+            FlinkRuntimeException testException =
+                    new FlinkRuntimeException("test request-table failure");
+            assertThatThrownBy(() -> enumerator.generateAndAssignSplits(null, testException))
+                    .isSameAs(testException);
+
+            assertThat(enumerator.getTieringMetrics().getRequestTableFailureCount())
+                    .isEqualTo(failuresBefore + 1);
+        }
+    }
+
+    /**
+     * Creates a {@link CoordinatorGateway} proxy that always returns a failed future for {@code
+     * lakeTieringHeartbeat}.
+     */
+    private CoordinatorGateway createFailingHeartbeatGateway() {
+        return (CoordinatorGateway)
+                Proxy.newProxyInstance(
+                        CoordinatorGateway.class.getClassLoader(),
+                        new Class<?>[] {CoordinatorGateway.class},
+                        (proxy, method, args) -> {
+                            if (method.getName().equals("lakeTieringHeartbeat")) {
+                                CompletableFuture<LakeTieringHeartbeatResponse> failed =
+                                        new CompletableFuture<>();
+                                failed.completeExceptionally(
+                                        new RuntimeException("simulated heartbeat failure"));
+                                return failed;
+                            }
+                            throw new UnsupportedOperationException(
+                                    "Unexpected call: " + method.getName());
+                        });
+    }
+
+    /**
+     * Creates a {@link CoordinatorGateway} proxy that returns an empty {@link
+     * LakeTieringHeartbeatResponse} (no tiering table) for {@code lakeTieringHeartbeat}.
+     */
+    private CoordinatorGateway createEmptyHeartbeatResponseGateway() {
+        return (CoordinatorGateway)
+                Proxy.newProxyInstance(
+                        CoordinatorGateway.class.getClassLoader(),
+                        new Class<?>[] {CoordinatorGateway.class},
+                        (proxy, method, args) -> {
+                            if (method.getName().equals("lakeTieringHeartbeat")) {
+                                return CompletableFuture.completedFuture(
+                                        new LakeTieringHeartbeatResponse());
+                            }
+                            throw new UnsupportedOperationException(
+                                    "Unexpected call: " + method.getName());
+                        });
+    }
 }