This is an automated email from the ASF dual-hosted git repository.
fanningpj pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new ed4f49ef44 Include roundtrip latency and seq nr in heartbeat debug log
(#2736)
ed4f49ef44 is described below
commit ed4f49ef44564734195cd5d934f138f3ddc86965
Author: Philippus Baalman <[email protected]>
AuthorDate: Mon Mar 16 13:15:32 2026 +0100
Include roundtrip latency and seq nr in heartbeat debug log (#2736)
---
.../org/apache/pekko/cluster/ClusterHeartbeat.scala | 17 +++++++++++------
1 file changed, 11 insertions(+), 6 deletions(-)
diff --git
a/cluster/src/main/scala/org/apache/pekko/cluster/ClusterHeartbeat.scala
b/cluster/src/main/scala/org/apache/pekko/cluster/ClusterHeartbeat.scala
index c4136c8643..e75a23fe95 100644
--- a/cluster/src/main/scala/org/apache/pekko/cluster/ClusterHeartbeat.scala
+++ b/cluster/src/main/scala/org/apache/pekko/cluster/ClusterHeartbeat.scala
@@ -54,8 +54,7 @@ private[cluster] final class
ClusterHeartbeatReceiver(getCluster: () => Cluster)
def receive: Receive = {
case hb: Heartbeat =>
- // TODO log the sequence nr once serializer is enabled
- if (verboseHeartbeat) clusterLogger.logDebug("Heartbeat from [{}]",
hb.from)
+ if (verboseHeartbeat) clusterLogger.logDebug("Heartbeat #{} from [{}]",
hb.sequenceNr, hb.from)
sender() ! HeartbeatRsp(cluster.selfUniqueAddress, hb.sequenceNr,
hb.creationTimeNanos)
}
@@ -220,9 +219,9 @@ private[cluster] class ClusterHeartbeatSender extends Actor
{
val nextHB = selfHeartbeat()
state.activeReceivers.foreach { to =>
if (failureDetector.isMonitoring(to.address)) {
- if (verboseHeartbeat) logDebug("Heartbeat to [{}]", to.address)
+ if (verboseHeartbeat) logDebug("Heartbeat #{} to [{}]",
nextHB.sequenceNr, to.address)
} else {
- if (verboseHeartbeat) logDebug("First Heartbeat to [{}]", to.address)
+ if (verboseHeartbeat) logDebug("First Heartbeat #{} to [{}]",
nextHB.sequenceNr, to.address)
// schedule the expected first heartbeat for later, which will give the
// other side a chance to reply, and also trigger some resends if
needed
scheduler.scheduleOnce(HeartbeatExpectedResponseAfter, self,
ExpectedFirstHeartbeat(to))
@@ -248,8 +247,14 @@ private[cluster] class ClusterHeartbeatSender extends
Actor {
}
def heartbeatRsp(response: HeartbeatRsp): Unit = {
- // TODO: log response time and validate sequence nrs once serialisation of
sendTime is released
- if (verboseHeartbeat) logDebug("Heartbeat response from [{}]",
response.from.address)
+ if (verboseHeartbeat) {
+ val latencyMs = (System.nanoTime() - response.creationTimeNanos) /
1000000
+ logDebug(
+ "Heartbeat #{} response from [{}] (round trip [{}] ms)",
+ response.sequenceNr,
+ response.from.address,
+ latencyMs)
+ }
state = state.heartbeatRsp(response.from)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
