You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by sa...@apache.org on 2024/06/14 10:54:08 UTC
(kafka) branch 3.8 updated: KAFKA-16948: Reset tier lag metrics on becoming follower (#16321)
This is an automated email from the ASF dual-hosted git repository.
satishd pushed a commit to branch 3.8
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.8 by this push:
new af4ccc50aa2 KAFKA-16948: Reset tier lag metrics on becoming follower (#16321)
af4ccc50aa2 is described below
commit af4ccc50aa21c919c4dbbb7b574cf1186ab487f3
Author: Kamal Chandraprakash <kc...@uber.com>
AuthorDate: Fri Jun 14 09:50:12 2024 +0530
KAFKA-16948: Reset tier lag metrics on becoming follower (#16321)
When the node transitions from a leader to a follower for a partition, then the tier-lag metrics should be reset to zero. Otherwise, it would lead to false positive in metrics. Addressed the concurrency issue while emitting the metrics.
Reviewers: Satish Duggana <sa...@apache.org>, Francois Visconte <f....@gmail.com>,
---
.../java/kafka/log/remote/RemoteLogManager.java | 29 ++++++++++++++++++----
.../kafka/log/remote/RemoteLogManagerTest.java | 22 ++++++++++++++++
2 files changed, 46 insertions(+), 5 deletions(-)
diff --git a/core/src/main/java/kafka/log/remote/RemoteLogManager.java b/core/src/main/java/kafka/log/remote/RemoteLogManager.java
index 3eacbea475f..e31ef7faa29 100644
--- a/core/src/main/java/kafka/log/remote/RemoteLogManager.java
+++ b/core/src/main/java/kafka/log/remote/RemoteLogManager.java
@@ -431,12 +431,15 @@ public class RemoteLogManager implements Closeable {
leaderPartitions.forEach(this::cacheTopicPartitionIds);
followerPartitions.forEach(this::cacheTopicPartitionIds);
- followerPartitions.forEach(this::removeRemoteTopicPartitionMetrics);
remoteLogMetadataManager.onPartitionLeadershipChanges(leaderPartitions, followerPartitions);
followerPartitions.forEach(topicIdPartition ->
doHandleLeaderOrFollowerPartitions(topicIdPartition, RLMTask::convertToFollower));
+ // If this node was the previous leader for the partition, then the RLMTask might be running in the
+ // background thread and might emit metrics. So, removing the metrics after marking this node as follower.
+ followerPartitions.forEach(this::removeRemoteTopicPartitionMetrics);
+
leaderPartitionsWithLeaderEpoch.forEach((topicIdPartition, leaderEpoch) ->
doHandleLeaderOrFollowerPartitions(topicIdPartition,
rlmTask -> rlmTask.convertToLeader(leaderEpoch)));
@@ -665,6 +668,15 @@ public class RemoteLogManager implements Closeable {
}
}
+ // VisibleForTesting
+ RLMTask rlmTask(TopicIdPartition topicIdPartition) {
+ RLMTaskWithFuture task = leaderOrFollowerTasks.get(topicIdPartition);
+ if (task != null) {
+ return task.rlmTask;
+ }
+ return null;
+ }
+
class RLMTask extends CancellableRunnable {
private final TopicIdPartition topicIdPartition;
@@ -893,11 +905,18 @@ public class RemoteLogManager implements Closeable {
logger.info("Copied {} to remote storage with segment-id: {}", logFileName, copySegmentFinishedRlsm.remoteLogSegmentId());
long bytesLag = log.onlyLocalLogSegmentsSize() - log.activeSegment().size();
- String topic = topicIdPartition.topic();
- int partition = topicIdPartition.partition();
long segmentsLag = log.onlyLocalLogSegmentsCount();
- brokerTopicStats.recordRemoteCopyLagBytes(topic, partition, bytesLag);
- brokerTopicStats.recordRemoteCopyLagSegments(topic, partition, segmentsLag);
+ recordLagStats(bytesLag, segmentsLag);
+ }
+
+ // VisibleForTesting
+ void recordLagStats(long bytesLag, long segmentsLag) {
+ if (isLeader()) {
+ String topic = topicIdPartition.topic();
+ int partition = topicIdPartition.partition();
+ brokerTopicStats.recordRemoteCopyLagBytes(topic, partition, bytesLag);
+ brokerTopicStats.recordRemoteCopyLagSegments(topic, partition, segmentsLag);
+ }
}
private Path toPathIfExists(File file) {
diff --git a/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java b/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
index 3c9b8a48e90..f18658088b6 100644
--- a/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
+++ b/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
@@ -3014,6 +3014,28 @@ public class RemoteLogManagerTest {
assertEquals(99L, capturedValues.get(1).longValue());
}
+ @Test
+ public void testTierLagResetsToZeroOnBecomingFollower() {
+ remoteLogManager.startup();
+ remoteLogManager.onLeadershipChange(
+ Collections.singleton(mockPartition(leaderTopicIdPartition)), Collections.emptySet(), topicIds);
+ RemoteLogManager.RLMTask rlmTask = remoteLogManager.rlmTask(leaderTopicIdPartition);
+ assertNotNull(rlmTask);
+ rlmTask.recordLagStats(1024, 2);
+ assertEquals(1024, brokerTopicStats.topicStats(leaderTopicIdPartition.topic()).remoteCopyLagBytes());
+ assertEquals(2, brokerTopicStats.topicStats(leaderTopicIdPartition.topic()).remoteCopyLagSegments());
+ // The same node becomes follower now which was the previous leader
+ remoteLogManager.onLeadershipChange(Collections.emptySet(),
+ Collections.singleton(mockPartition(leaderTopicIdPartition)), topicIds);
+ assertEquals(0, brokerTopicStats.topicStats(leaderTopicIdPartition.topic()).remoteCopyLagBytes());
+ assertEquals(0, brokerTopicStats.topicStats(leaderTopicIdPartition.topic()).remoteCopyLagSegments());
+
+ // If the old task emits the tier-lag stats, then it should be discarded
+ rlmTask.recordLagStats(2048, 4);
+ assertEquals(0, brokerTopicStats.topicStats(leaderTopicIdPartition.topic()).remoteCopyLagBytes());
+ assertEquals(0, brokerTopicStats.topicStats(leaderTopicIdPartition.topic()).remoteCopyLagSegments());
+ }
+
private Partition mockPartition(TopicIdPartition topicIdPartition) {
TopicPartition tp = topicIdPartition.topicPartition();
Partition partition = mock(Partition.class);