fresh-borzoni commented on code in PR #555:
URL: https://github.com/apache/fluss-rust/pull/555#discussion_r3298119886
##########
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";
Review Comment:
with multiple scanners these collide last-writer-wins, mb we need to
consider table-path labels(Java scopes via ScannerMetricGroup(tablePath))
Can be a followup
##########
crates/fluss/src/client/table/scanner.rs:
##########
@@ -378,6 +436,51 @@ 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`).
+ ///
+ /// In debug builds, panics if a previous poll has not yet recorded
+ /// its end — that indicates a concurrent `poll()` on the same scanner,
+ /// which violates the single-consumer contract (Java enforces this
+ /// with `LogScannerImpl.acquire()` and throws
+ /// `ConcurrentModificationException`).
+ fn record_poll_start(&self) {
+ let now = Instant::now();
+ let mut state = self.poll_state.lock();
+ debug_assert!(
+ state.poll_start_at.is_none(),
+ "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;
+ metrics::gauge!(SCANNER_TIME_BETWEEN_POLL_MS).set(between_ms);
Review Comment:
Mutex held across metrics::gauge!(...).set(...). Shall we drop the lock
before emitting?
##########
crates/fluss/src/client/table/scanner.rs:
##########
@@ -378,6 +436,51 @@ 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`).
+ ///
+ /// In debug builds, panics if a previous poll has not yet recorded
+ /// its end — that indicates a concurrent `poll()` on the same scanner,
+ /// which violates the single-consumer contract (Java enforces this
+ /// with `LogScannerImpl.acquire()` and throws
+ /// `ConcurrentModificationException`).
+ fn record_poll_start(&self) {
+ let now = Instant::now();
+ let mut state = self.poll_state.lock();
+ debug_assert!(
+ state.poll_start_at.is_none(),
+ "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;
+ metrics::gauge!(SCANNER_TIME_BETWEEN_POLL_MS).set(between_ms);
+ state.last_poll_at = Some(now);
+ state.poll_start_at = Some(now);
+ }
+
+ /// 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).
+ fn record_poll_end(&self) {
+ let now = Instant::now();
+ let mut state = self.poll_state.lock();
+ let Some(start) = state.poll_start_at.take() else {
+ return;
Review Comment:
On overlap, take() returns None and the function returns silently. No log,
no signal that metrics are wrong. Should we warn at least?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
