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

[XuQianJin-Stars]

Purpose

TieringSourceEnumerator in the fluss-flink-common module interacts with the
coordinator via coordinatorGateway.lakeTieringHeartbeat(...) — pulling new
tiering tasks, reporting in-progress / finished / failed / force-finished
tables. Today, when heartbeat transient errors happen or when requestTable
stays empty (i.e. no new tasks returned by the coordinator), operators have to
dig through logs to figure out what's wrong. This PR adds three counters to
make these failure / empty cases observable through standard Flink metrics.

Linked issue: N/A (new observability feature)

Brief change log

• Add 3 new metrics in MetricNames:

  • tiering.heartbeat.failure — number of lakeTieringHeartbeat calls that
    raised an exception.
  • tiering.request-table.empty — number of heartbeat rounds where the
    coordinator returned no available tiering table.
  • tiering.request-table.failure — number of times
    requestTieringTableSplitsViaHeartBeat ended up in the failure branch.

• Add new TieringEnumeratorMetrics class under
  fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/metrics/
  to own the three counters and expose getter accessors. The class reuses
  SplitEnumeratorMetricGroup, following the existing metrics-registration
  style in the same module.

• Refactor TieringSourceEnumerator:

  • Move waitHeartbeatResponse from a private static helper to an instance
    method on the new inner HeartBeatHelper so the helper can increment
    tiering.heartbeat.failure on exception.
  • Increment tiering.request-table.empty when
    LakeTieringHeartbeatResponse.hasTieringTable() returns false.
  • Increment tiering.request-table.failure inside generateAndAssignSplits
    on the failure path.

• Expose getTieringMetrics() and setCoordinatorGateway() for tests
  (package-private; not part of the public API).

Tests

• TieringSourceEnumeratorTest#testHeartbeatFailureMetricIncremented — uses
  a java.lang.reflect.Proxy CoordinatorGateway that always returns a
  failed CompletableFuture for lakeTieringHeartbeat, asserts the
  heartbeatFailure counter is incremented.
• TieringSourceEnumeratorTest#testRequestTableEmptyMetricIncremented —
  uses a proxy gateway that returns an empty LakeTieringHeartbeatResponse
  (no tieringTable), asserts the requestTableEmpty counter is incremented.
• TieringSourceEnumeratorTest#testRequestTableFailureMetricIncrementedOnGenerateAndAssignSplits
  — verifies the requestTableFailure counter is incremented when
  generateAndAssignSplits is invoked with a non-null Throwable.
• All 10 pre-existing tests in TieringSourceEnumeratorTest continue to pass
  (no regression). Run on the local Fluss cluster via
  FlussClusterExtension (3 tablet servers).

API and Format

• No public API changes. New methods (getTieringMetrics, setCoordinatorGateway,
  waitHeartbeatResponse) are package-private / @VisibleForTesting only.
• mvn spotless:apply clean.
• Java 8 compatible (no var, no Optional.isEmpty, no List.of).
• Uses AssertJ assertions in all new tests.
• Follows the project commit-message convention: [component] Brief description.

Documentation

• Metric names are added to org.apache.fluss.metrics.MetricNames next to the
  existing lake tiering metrics block; their javadoc / naming follows
  AGENTS.md §8 Configuration Patterns.
• The metrics will be surfaced through the standard Flink enumerator metric
  group once the Flink runtime exports them; no additional user-facing
  documentation is required beyond the metric names themselves.