charlesdong1991 commented on code in PR #407:
URL: https://github.com/apache/fluss-rust/pull/407#discussion_r2885913210
##########
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:
IMHO, this should be attempted payload bytes, not confirmed, that's why
increment it before sending, i think this is also aligned with how Java side
implements.
We do record completion with bytes received = 0 if request path observes an
error.
Which i think is right behaviour? Let me know how you think of it?
Thanks again for challenging! 👍 @fresh-borzoni
--
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]
