This is an automated email from the ASF dual-hosted git repository.
fresh-borzoni pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 4ca24ea0 [metrics] Add scanner poll-timing metrics (#555)
4ca24ea0 is described below
commit 4ca24ea03492c412bc3f7912dae081acf8e00b22
Author: Kaiqi Dong <[email protected]>
AuthorDate: Tue May 26 16:42:56 2026 +0200
[metrics] Add scanner poll-timing metrics (#555)
* Add scanner poll timing metrics
* cargo fmt
* improve overlapping polls scenario
* address feedback
---
crates/fluss/src/client/table/scanner.rs | 287 ++++++++++++++++++++++++++
crates/fluss/src/metrics.rs | 48 +++++
website/docs/user-guide/rust/api-reference.md | 4 +-
3 files changed, 338 insertions(+), 1 deletion(-)
diff --git a/crates/fluss/src/client/table/scanner.rs
b/crates/fluss/src/client/table/scanner.rs
index 0167af02..5fa8bb92 100644
--- a/crates/fluss/src/client/table/scanner.rs
+++ b/crates/fluss/src/client/table/scanner.rs
@@ -29,6 +29,7 @@ use crate::error::{ApiError, Error, FlussError, Result};
use crate::metadata::{LogFormat, PhysicalTablePath, RowType, TableBucket,
TableInfo, TablePath};
use crate::metrics::{
SCANNER_BYTES_PER_REQUEST, SCANNER_FETCH_LATENCY_MS,
SCANNER_FETCH_REQUESTS_TOTAL,
+ SCANNER_POLL_IDLE_RATIO, SCANNER_TIME_BETWEEN_POLL_MS,
};
use crate::proto::{
ErrorResponse, FetchLogRequest, FetchLogResponse, PbFetchLogReqForBucket,
PbFetchLogReqForTable,
@@ -281,6 +282,64 @@ struct LogScannerInner {
/// Guards against subscription changes while a
/// [`crate::client::RecordBatchLogReader`] is iterating.
reader_active: std::sync::atomic::AtomicBool,
+ /// Holds the snapshot fields used by [`PollGuard`] to derive the
+ /// scanner poll-timing metrics. The mutex makes the state updates
+ /// in `record_poll_start` / `record_poll_end` atomic; metric
+ /// emission and `log::warn!` calls happen after the lock is
+ /// released. The start↔end pairing depends on the single-consumer
+ /// contract documented on [`LogScanner::poll`] and
+ /// [`RecordBatchLogScanner::poll`] (mirrors Java's
+ /// `LogScannerImpl.acquire()`). Overlapping polls on the same
+ /// scanner trip a `debug_assert!` in `record_poll_start` (debug
+ /// builds) or emit a `log::warn!` (release builds).
+ poll_state: Mutex<PollState>,
+}
+
+/// Snapshot state used to derive the scanner poll-timing metrics.
+///
+/// The mutex makes the state updates in `record_poll_start` /
+/// `record_poll_end` atomic with respect to themselves; metric
+/// emission (`metrics::gauge!(...).set(...)`) and `log::warn!` calls
+/// happen after the lock is released so a user-installed recorder or
+/// logger cannot stall the critical section. The mutex does **not** by
+/// itself preserve start↔end pairing across overlapping `poll()` calls
+/// — that invariant relies on the single-consumer contract that
+/// mirrors Java's `LogScannerImpl.acquire()`. Concurrent polls on the
+/// same scanner are detected by a `debug_assert!` in
+/// `record_poll_start` (panics in debug / tests) and a `log::warn!` on
+/// both anomalous paths (`record_poll_start` sees a stale `Some`;
+/// `record_poll_end` sees `None`) for release-build observability.
+#[derive(Default, Debug)]
+struct PollState {
+ /// Instant captured at the most recent `record_poll_start()`. `None`
+ /// before the first poll.
+ last_poll_at: Option<Instant>,
+ /// Instant captured at the start of the in-flight poll. `None` after
+ /// the last `record_poll_end()`.
+ poll_start_at: Option<Instant>,
+ /// Cached ms between the two most recent poll starts, used to compute
+ /// `poll_idle_ratio` in `record_poll_end`.
+ time_between_poll_ms: f64,
+}
+
+/// Pairs `record_poll_start` with `record_poll_end`. Created
+/// at the top of `poll_records` / `poll_batches`; `record_poll_end` runs on
+/// drop, including the cancellation path (caller drops the future).
+struct PollGuard<'a> {
+ inner: &'a LogScannerInner,
+}
+
+impl<'a> PollGuard<'a> {
+ fn new(inner: &'a LogScannerInner) -> Self {
+ inner.record_poll_start();
+ Self { inner }
+ }
+}
+
+impl Drop for PollGuard<'_> {
+ fn drop(&mut self) {
+ self.inner.record_poll_end();
+ }
}
impl LogScannerInner {
@@ -322,6 +381,7 @@ impl LogScannerInner {
)?,
arrow_schema,
reader_active: std::sync::atomic::AtomicBool::new(false),
+ poll_state: Mutex::new(PollState::default()),
})
}
@@ -340,6 +400,10 @@ impl LogScannerInner {
}
async fn poll_records(&self, timeout: Duration) -> Result<ScanRecords> {
+ // Pairs record_poll_start (now) with record_poll_end
+ // (drop). Runs on every exit, including the cancellation path
+ // where the caller drops this future.
+ let _poll_guard = PollGuard::new(self);
let start = Instant::now();
let deadline = start + timeout;
@@ -378,6 +442,90 @@ impl LogScannerInner {
}
}
+ /// Records the start of a `poll()` call and emits
+ /// `SCANNER_TIME_BETWEEN_POLL_MS`. The first poll emits `0.0`,
+ /// matching Java's `ScannerMetricGroup.recordPollStart`
+ /// (`timeMsBetweenPoll = lastPollMs != 0L ? pollStartMs - lastPollMs :
0L`).
+ ///
+ /// Single-consumer contract: a previous poll must have recorded its
+ /// end before the next start. Java enforces this with
+ /// `LogScannerImpl.acquire()` (throws `ConcurrentModificationException`).
+ /// Rust surfaces violations as:
+ /// - debug builds: `debug_assert!` panics (caught by tests),
+ /// - release builds: `log::warn!` + the in-flight `poll_start_at` is
+ /// overwritten so the metric series keeps moving; the resulting
+ /// `time_between_poll_ms` / `poll_idle_ratio` values for the
+ /// overlapping polls are not meaningful until the overlap clears.
+ fn record_poll_start(&self) {
+ let now = Instant::now();
+ // Compute under the lock; emit the metric outside the critical
+ // section so a user-installed recorder cannot stall the next poll.
+ let (between_ms, overlap) = {
+ let mut state = self.poll_state.lock();
+ let overlap = state.poll_start_at.is_some();
+ debug_assert!(
+ !overlap,
+ "concurrent poll() detected on the same scanner; \
+ LogScanner / RecordBatchLogScanner are single-consumer \
+ (see LogScannerImpl.acquire() for Java parity)"
+ );
+ let between_ms = match state.last_poll_at {
+ Some(prev) => now.duration_since(prev).as_secs_f64() * 1000.0,
+ None => 0.0,
+ };
+ state.time_between_poll_ms = between_ms;
+ state.last_poll_at = Some(now);
+ state.poll_start_at = Some(now);
+ (between_ms, overlap)
+ };
+ if overlap {
+ warn!(
+ "concurrent poll() detected on scanner; single-consumer \
+ contract violated, poll-timing metrics will be inaccurate \
+ until the overlap clears"
+ );
+ }
+ metrics::gauge!(SCANNER_TIME_BETWEEN_POLL_MS).set(between_ms);
+ }
+
+ /// Computes `poll_idle_ratio = poll_time / (poll_time + between_time)`.
+ /// On the first poll, `between_time` is 0 so the ratio is 1.0
+ /// (poll-bound).
+ ///
+ /// Orphan call: if no matching `record_poll_start` is in flight,
+ /// emits a `log::warn!` (single-consumer contract may have been
+ /// violated, e.g. in release builds where the start-side
+ /// `debug_assert!` is compiled out) and skips the metric update.
+ fn record_poll_end(&self) {
+ let now = Instant::now();
+ // Compute under the lock; emit metric / warn outside the critical
+ // section so neither the user-installed recorder nor the logger
+ // can stall the next poll.
+ let (orphan, ratio) = {
+ let mut state = self.poll_state.lock();
+ match state.poll_start_at.take() {
+ None => (true, None),
+ Some(start) => {
+ let poll_time_ms = now.duration_since(start).as_secs_f64()
* 1000.0;
+ let total = poll_time_ms + state.time_between_poll_ms;
+ let r = (total > 0.0).then_some(poll_time_ms / total);
+ (false, r)
+ }
+ }
+ };
+ if orphan {
+ warn!(
+ "record_poll_end called without a matching record_poll_start; \
+ single-consumer contract may have been violated, idle ratio \
+ for this poll is not emitted"
+ );
+ return;
+ }
+ if let Some(r) = ratio {
+ metrics::gauge!(SCANNER_POLL_IDLE_RATIO).set(r);
+ }
+ }
+
async fn subscribe(&self, bucket: i32, offset: i64) -> Result<()> {
self.check_no_active_reader()?;
if self.is_partitioned_table {
@@ -520,6 +668,7 @@ impl LogScannerInner {
}
async fn poll_batches(&self, timeout: Duration) -> Result<Vec<ScanBatch>> {
+ let _poll_guard = PollGuard::new(self);
let start = Instant::now();
let deadline = start + timeout;
@@ -2232,6 +2381,144 @@ mod tests {
Ok(())
}
+ /// Builds a self-contained `LogScannerInner` for poll-timing tests
+ /// inside a `current_thread` runtime so callers can drive `PollGuard`
+ /// lifecycles synchronously.
+ fn with_test_log_scanner_inner<F: FnOnce(&LogScannerInner)>(body: F) {
+ let rt = tokio::runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .expect("build current_thread runtime");
+ rt.block_on(async {
+ let table_path = TablePath::new("db".to_string(),
"tbl".to_string());
+ let table_info = build_table_info(table_path.clone(), 1, 1);
+ let cluster = build_cluster_arc(&table_path, 1, 1);
+ let metadata = Arc::new(Metadata::new_for_test(cluster));
+ let inner = LogScannerInner::new(
+ &table_info,
+ metadata,
+ Arc::new(RpcClient::new()),
+ &Config::default(),
+ None,
+ )
+ .expect("build LogScannerInner");
+ body(&inner);
+ });
+ }
+
+ fn snapshot_gauge(
+ snapshotter: &metrics_util::debugging::Snapshotter,
+ name: &str,
+ ) -> Option<f64> {
+ use metrics_util::debugging::DebugValue;
+ snapshotter
+ .snapshot()
+ .into_vec()
+ .into_iter()
+ .find_map(|(key, _, _, val)| {
+ if key.key().name() == name {
+ if let DebugValue::Gauge(g) = val {
+ return Some(g.into_inner());
+ }
+ }
+ None
+ })
+ }
+
+ /// Exercises the `PollGuard` lifecycle across two consecutive
+ /// `record_poll_start` calls. Asserts both poll-timing gauges are
+ /// emitted at the right moments and `record_poll_end` runs on guard
+ /// drop (also the cancellation-safety path, since dropping the
+ /// `poll()` future drops the guard).
+ #[test]
+ fn poll_guard_emits_time_between_poll_and_idle_ratio() {
+ use crate::metrics::{SCANNER_POLL_IDLE_RATIO,
SCANNER_TIME_BETWEEN_POLL_MS};
+ use metrics_util::debugging::DebuggingRecorder;
+
+ let recorder = DebuggingRecorder::new();
+ let snapshotter = recorder.snapshotter();
+
+ metrics::with_local_recorder(&recorder, || {
+ with_test_log_scanner_inner(|inner| {
+ // First poll: emits time_between_poll_ms=0 (Java parity:
+ // ScannerMetricGroup.recordPollStart emits 0 when there is
+ // no previous poll). Idle ratio is also emitted as 1.0
+ // on drop (poll_time / (poll_time + 0) = 1.0).
+ {
+ let _g = PollGuard::new(inner);
+ std::thread::sleep(std::time::Duration::from_millis(5));
+ }
+
+ // Brief gap so time_between_poll_ms is observably > 0.
+ std::thread::sleep(std::time::Duration::from_millis(5));
+
+ // Second poll: refreshes both time_between_poll_ms (>0)
+ // and a fresh idle ratio.
+ {
+ let _g = PollGuard::new(inner);
+ std::thread::sleep(std::time::Duration::from_millis(5));
+ }
+ });
+ });
+
+ let between = snapshot_gauge(&snapshotter,
SCANNER_TIME_BETWEEN_POLL_MS)
+ .expect("time_between_poll_ms must be emitted on every poll");
+ assert!(
+ between > 0.0,
+ "second-poll time_between_poll_ms must be positive, got {between}"
+ );
+
+ let ratio = snapshot_gauge(&snapshotter, SCANNER_POLL_IDLE_RATIO)
+ .expect("poll_idle_ratio must be emitted on poll end");
+ assert!(
+ (0.0..=1.0).contains(&ratio),
+ "poll_idle_ratio must be in [0, 1], got {ratio}"
+ );
+ }
+
+ /// Java parity: `ScannerMetricGroup.recordPollStart` emits
+ /// `timeMsBetweenPoll = 0` on the very first poll. The Rust gauge
+ /// must do the same so dashboards see the metric series from poll #1.
+ #[test]
+ fn time_between_poll_ms_emits_zero_on_first_poll() {
+ use crate::metrics::SCANNER_TIME_BETWEEN_POLL_MS;
+ use metrics_util::debugging::DebuggingRecorder;
+
+ let recorder = DebuggingRecorder::new();
+ let snapshotter = recorder.snapshotter();
+
+ metrics::with_local_recorder(&recorder, || {
+ with_test_log_scanner_inner(|inner| {
+ let _g = PollGuard::new(inner);
+ // Drop at end of scope completes the poll; the value of
+ // SCANNER_TIME_BETWEEN_POLL_MS was emitted at start, not end.
+ });
+ });
+
+ let between = snapshot_gauge(&snapshotter,
SCANNER_TIME_BETWEEN_POLL_MS)
+ .expect("time_between_poll_ms must be emitted on the first poll");
+ assert_eq!(
+ between, 0.0,
+ "first-poll time_between_poll_ms must be 0.0 (Java parity), got
{between}"
+ );
+ }
+
+ /// Pins the single-consumer contract: overlapping `PollGuard`s on the
+ /// same scanner trip the `debug_assert!` in `record_poll_start`.
+ /// Release builds skip the check, so the test is gated on
+ /// `debug_assertions`.
+ #[cfg(debug_assertions)]
+ #[test]
+ #[should_panic(expected = "concurrent poll() detected")]
+ fn overlapping_polls_panic_in_debug_builds() {
+ with_test_log_scanner_inner(|inner| {
+ let _g1 = PollGuard::new(inner);
+ // _g1 has not been dropped → poll_start_at is still Some,
+ // so the second start must panic.
+ let _g2 = PollGuard::new(inner);
+ });
+ }
+
/// Drives `handle_fetch_response` against a local metrics recorder and
/// asserts that latency + bytes-per-request histograms are emitted with
/// values that mirror what Java would record. This complements the unit
diff --git a/crates/fluss/src/metrics.rs b/crates/fluss/src/metrics.rs
index a533abd5..540e991f 100644
--- a/crates/fluss/src/metrics.rs
+++ b/crates/fluss/src/metrics.rs
@@ -49,6 +49,34 @@ pub const CLIENT_BYTES_RECEIVED_TOTAL: &str =
"fluss.client.bytes_received.total
pub const CLIENT_REQUEST_LATENCY_MS: &str = "fluss.client.request_latency_ms";
pub const CLIENT_REQUESTS_IN_FLIGHT: &str = "fluss.client.requests_in_flight";
+// ---------------------------------------------------------------------------
+// Scanner poll-timing metrics
+//
+// Java reference: ScannerMetricGroup.java, LogScannerImpl.java
+//
+// These track consumer liveness and processing efficiency at the `poll()`
+// boundary. Java records via `volatile long` fields read by gauge suppliers;
+// Rust snapshots the values at poll start/end.
+//
+// Java's `lastPollSecondsAgo` gauge is intentionally NOT ported. Java
+// implements it as a gauge supplier evaluated at scrape time, which the
+// `metrics` crate facade has no equivalent for. A snapshot-at-poll-start
+// port would just duplicate `time_between_poll_ms / 1000` and would not
+// advance while a consumer is hung — defeating the metric's purpose
+// (detecting a stuck consumer). Revisit if the `metrics` crate gains a
+// supplier abstraction or we add a background liveness task.
+// ---------------------------------------------------------------------------
+
+/// Gauge: milliseconds between the start of consecutive `poll()` calls. A
+/// large value usually means the consumer's downstream processing is slow.
+pub const SCANNER_TIME_BETWEEN_POLL_MS: &str =
"fluss.client.scanner.time_between_poll_ms";
+
+/// Gauge: fraction of wall-clock time spent inside `poll()` —
+/// `poll_time_ms / (poll_time_ms + time_between_poll_ms)`. A value near 1.0
+/// means the scanner is starved for data; a low value means the consumer is
+/// the bottleneck.
+pub const SCANNER_POLL_IDLE_RATIO: &str =
"fluss.client.scanner.poll_idle_ratio";
+
// ---------------------------------------------------------------------------
// Scanner fetch + remote download metrics
//
@@ -305,6 +333,26 @@ mod tests {
assert_eq!(counter_by_api_key.get("fetch_log"), Some(&3));
}
+ #[test]
+ fn scanner_poll_timing_metrics_emit_correctly() {
+ let recorder = DebuggingRecorder::new();
+ let snapshotter = recorder.snapshotter();
+
+ metrics::with_local_recorder(&recorder, || {
+ metrics::gauge!(SCANNER_TIME_BETWEEN_POLL_MS).set(200.0);
+ metrics::gauge!(SCANNER_POLL_IDLE_RATIO).set(0.8);
+ });
+
+ let snapshot = snapshotter.snapshot();
+ let entries: Vec<_> = snapshot.into_vec();
+
+ assert_eq!(
+ find_gauge!(entries, SCANNER_TIME_BETWEEN_POLL_MS),
+ Some(200.0)
+ );
+ assert_eq!(find_gauge!(entries, SCANNER_POLL_IDLE_RATIO), Some(0.8));
+ }
+
#[test]
fn scanner_fetch_metrics_emit_correctly() {
let recorder = DebuggingRecorder::new();
diff --git a/website/docs/user-guide/rust/api-reference.md
b/website/docs/user-guide/rust/api-reference.md
index 52999e8c..3f94841e 100644
--- a/website/docs/user-guide/rust/api-reference.md
+++ b/website/docs/user-guide/rust/api-reference.md
@@ -139,6 +139,8 @@ Complete API reference for the Fluss Rust client.
## `LogScanner`
+Single-consumer: do not call `poll` concurrently on the same scanner (e.g.
from `tokio::join!` or two tasks sharing an `Arc`). Mirrors Java's
`LogScannerImpl.acquire()` guard. Debug builds surface overlapping calls via a
`debug_assert!`; release builds skip the check for performance and produce
skewed poll-timing metrics (`fluss.client.scanner.time_between_poll_ms`,
`fluss.client.scanner.poll_idle_ratio`) if the contract is violated.
+
| Method
| Description
|
|-----------------------------------------------------------------------------------------------------------|----------------------------------------------------------|
| `async fn subscribe(&self, bucket_id: i32, start_offset: i64) -> Result<()>`
| Subscribe to a bucket
|
@@ -151,7 +153,7 @@ Complete API reference for the Fluss Rust client.
## `RecordBatchLogScanner`
-Overlapping `poll` calls on clones that share state, or `poll` concurrent with
`RecordBatchLogReader::next_batch`, are not supported. Use one active
polling/consumption call at a time per underlying scanner state.
+Single-consumer: overlapping `poll` calls on handles that share state, or
`poll` concurrent with `RecordBatchLogReader::next_batch`, are not supported —
use one active polling/consumption call at a time per underlying scanner state.
Mirrors Java's `LogScannerImpl.acquire()` guard. Debug builds surface
overlapping calls via a `debug_assert!`; release builds skip the check for
performance and produce skewed poll-timing metrics
(`fluss.client.scanner.time_between_poll_ms`, `fluss.client.sca [...]
| Method
| Description
|
|-----------------------------------------------------------------------------------------------------------|----------------------------------------------------------|
