fresh-borzoni commented on code in PR #407:
URL: https://github.com/apache/fluss-rust/pull/407#discussion_r2886076678
##########
crates/fluss/src/rpc/server_connection.rs:
##########
@@ -386,12 +411,76 @@ where
ConnectionState::Poison(e) => return
Err(RpcError::Poisoned(Arc::clone(e)).into()),
}
- self.send_message(buf).await?;
+ // Guard to decrement in-flight on cancellation, panic, or any exit
without
+ // explicitly calling record_completion_metrics. Prevents gauge drift
when
+ // the request future is dropped (e.g. tokio::select! timeout).
+ let mut in_flight_guard = if let Some(label) = api_label {
+ metrics::counter!(
+ crate::metrics::CLIENT_REQUESTS_TOTAL,
+ crate::metrics::LABEL_API_KEY => label
+ )
+ .increment(1);
+ metrics::counter!(
+ crate::metrics::CLIENT_BYTES_SENT_TOTAL,
Review Comment:
you are right, checked in java, makes sense
--
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]
