root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

MINOR: fixing updateBrokerContactTime (#19828) 

Fix `updateBrokerContactTime` so that existing brokers still have their
contact time updated when they are already tracked. Also, update the
unit test to test this case.

Reviewers: Kuan-Po Tseng <brandboat@gmail.com>, Yung
 <yungyung7654321@gmail.com>, TengYao Chi <frankvicky@apache.org>, Ken
 Huang <s7133700@gmail.com>
This commit is contained in:
Kevin Wu 2025-05-28 22:58:09 -05:00 committed by GitHub
parent 383a9ff9df
commit 8731c96122
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 4 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
metadata/src/main/java/org/apache/kafka/controller/metrics/QuorumControllerMetrics.java
View File

@ -267,7 +267,8 @@ public class QuorumControllerMetrics implements AutoCloseable {
}
public void updateBrokerContactTime(int brokerId) {
brokerContactTimesMs.putIfAbsent(brokerId, new AtomicLong(time.milliseconds()));
AtomicLong contactTime = brokerContactTimesMs.computeIfAbsent(brokerId, k -> new AtomicLong());
contactTime.set(time.milliseconds());
}
public int timeSinceLastHeartbeatMs(int brokerId) {

2
metadata/src/test/java/org/apache/kafka/controller/metrics/QuorumControllerMetricsTest.java
View File

@ -179,6 +179,8 @@ public class QuorumControllerMetricsTest {
metrics.updateBrokerContactTime(brokerId);
time.sleep(1000);
assertEquals(1000, timeSinceLastHeartbeatReceivedMs.value());
metrics.updateBrokerContactTime(brokerId);
assertEquals(0, timeSinceLastHeartbeatReceivedMs.value());
time.sleep(100000);
assertEquals(sessionTimeoutMs, timeSinceLastHeartbeatReceivedMs.value());
metrics.removeTimeSinceLastHeartbeatMetrics();