SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

KafkaBinderMetrics Improvements

Browse Source 
- Calculate Kafka lag metric correctly even if there are no committed offsets for the group
 - Update authors and copyright years
 - Checkstyle fixes
This commit is contained in:
Lars Bilger
2022-04-19 16:26:32 +02:00
committed by Soby Chacko
parent 2cf0bd0040
commit b88d34db93
2 changed files with 32 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2021 the original author or authors.
* Copyright 2016-2022 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -61,6 +61,7 @@ import org.springframework.util.ObjectUtils;
* @author Jon Schneider
* @author Thomas Cheyney
* @author Gary Russell
* @author Lars Bilger
*/
public class KafkaBinderMetrics
implements MeterBinder, ApplicationListener<BindingCreatedEvent> {
@@ -209,14 +210,19 @@ public class KafkaBinderMetrics
.endOffsets(topicPartitions);
final Map<TopicPartition, OffsetAndMetadata> committedOffsets = metadataConsumer.committed(endOffsets.keySet());
final Map<TopicPartition, Long> beginningOffsets = metadataConsumer.beginningOffsets(endOffsets.keySet());
for (Map.Entry<TopicPartition, Long> endOffset : endOffsets
.entrySet()) {
OffsetAndMetadata current = committedOffsets.get(endOffset.getKey());
Long beginningOffset = beginningOffsets.get(endOffset.getKey());
lag += endOffset.getValue();
if (current != null) {
lag -= current.offset();
}
else if (beginningOffset != null) {
lag -= beginningOffset;
}
}
return lag;
}

25
binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java
View File

@@ -17,6 +17,7 @@
package org.springframework.cloud.stream.binder.kafka;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -49,6 +50,7 @@ import static org.mockito.Mockito.mock;
* @author Henryk Konsek
* @author Thomas Cheyney
* @author Soby Chacko
* @author Lars Bilger
*/
public class KafkaBinderMetricsTest {
@@ -254,6 +256,29 @@ public class KafkaBinderMetricsTest {
ArgumentMatchers.eq("group2-metrics"), ArgumentMatchers.any());
}
@Test
public void usesBeginningOffsetIfNoCommittedOffsetFound() {
org.mockito.BDDMockito
.given(consumer.committed(ArgumentMatchers.anySet()))
.willReturn(Collections.emptyMap());
final Map<TopicPartition, Long> beginnings = new HashMap<>();
TopicPartition topicPartition = new TopicPartition(TEST_TOPIC, 0);
beginnings.put(topicPartition, 500L);
org.mockito.BDDMockito
.given(consumer.beginningOffsets(ArgumentMatchers.anySet()))
.willReturn(beginnings);
List<PartitionInfo> partitions = partitions(new Node(0, null, 0));
topicsInUse.put(TEST_TOPIC,
new TopicInformation("group1-metrics", partitions, false));
org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC))
.willReturn(partitions);
metrics.bindTo(meterRegistry);
assertThat(meterRegistry.getMeters()).hasSize(1);
assertThat(meterRegistry.get(KafkaBinderMetrics.OFFSET_LAG_METRIC_NAME)
.tag("group", "group1-metrics").tag("topic", TEST_TOPIC).gauge().value())
.isEqualTo(500.0);
}
private List<PartitionInfo> partitions(Node... nodes) {
List<PartitionInfo> partitions = new ArrayList<>();
for (int i = 0; i < nodes.length; i++) {