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

Reuse Kafka consumer Metrics

Browse Source 
Polishing
This commit is contained in:
Thomas Cheyney
2018-04-24 18:30:32 +01:00
committed by Soby Chacko
parent d2012c287a
commit 74689bf007
2 changed files with 42 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -50,6 +50,7 @@ import org.springframework.util.ObjectUtils;
* @author Artem Bilan
* @author Oleg Zhurakousky
* @author Jon Schneider
* @author Thomas Cheyney
*/
public class KafkaBinderMetrics implements MeterBinder, ApplicationListener<BindingCreatedEvent> {
@@ -65,6 +66,8 @@ public class KafkaBinderMetrics implements MeterBinder, ApplicationListener<Bind
private final MeterRegistry meterRegistry;
private Consumer<?, ?> metadataConsumer;
public KafkaBinderMetrics(KafkaMessageChannelBinder binder,
KafkaBinderConfigurationProperties binderConfigurationProperties,
ConsumerFactory<?, ?> defaultConsumerFactory, @Nullable MeterRegistry meterRegistry) {
@@ -104,7 +107,10 @@ public class KafkaBinderMetrics implements MeterBinder, ApplicationListener<Bind
private double calculateConsumerLagOnTopic(String topic, String group) {
long lag = 0;
try (Consumer<?, ?> metadataConsumer = createConsumerFactory(group).createConsumer()) {
try {
if (metadataConsumer == null) {
metadataConsumer = createConsumerFactory(group).createConsumer();
}
List<PartitionInfo> partitionInfos = metadataConsumer.partitionsFor(topic);
List<TopicPartition> topicPartitions = new LinkedList<>();
for (PartitionInfo partitionInfo : partitionInfos) {

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

@@ -23,9 +23,11 @@ import java.util.Map;
import java.util.concurrent.TimeUnit;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.TimeGauge;
import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
@@ -33,6 +35,7 @@ import org.junit.Before;
import org.junit.Test;
import org.mockito.ArgumentMatchers;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.MockitoAnnotations;
import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.TopicInformation;
@@ -43,6 +46,7 @@ import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Henryk Konsek
* @author Thomas Cheyney
*/
public class KafkaBinderMetricsTest {
@@ -123,6 +127,37 @@ public class KafkaBinderMetricsTest {
assertThat(meterRegistry.getMeters()).isEmpty();
}
@Test
public void createsConsumerOnceWhenInvokedMultipleTimes() {
final List<PartitionInfo> partitions = partitions(new Node(0, null, 0));
topicsInUse.put(TEST_TOPIC, new TopicInformation("group", partitions));
metrics.bindTo(meterRegistry);
TimeGauge gauge = meterRegistry.get(KafkaBinderMetrics.METRIC_NAME).tag("group", "group").tag("topic", TEST_TOPIC).timeGauge();
gauge.value(TimeUnit.MILLISECONDS);
assertThat(gauge.value(TimeUnit.MILLISECONDS)).isEqualTo(1000.0);
org.mockito.Mockito.verify(this.consumerFactory).createConsumer();
}
@Test
public void consumerCreationFailsFirstTime() {
org.mockito.BDDMockito.given(consumerFactory.createConsumer()).willThrow(KafkaException.class)
.willReturn(consumer);
final List<PartitionInfo> partitions = partitions(new Node(0, null, 0));
topicsInUse.put(TEST_TOPIC, new TopicInformation("group", partitions));
metrics.bindTo(meterRegistry);
TimeGauge gauge = meterRegistry.get(KafkaBinderMetrics.METRIC_NAME).tag("group", "group").tag("topic", TEST_TOPIC).timeGauge();
assertThat(gauge.value(TimeUnit.MILLISECONDS)).isEqualTo(0);
assertThat(gauge.value(TimeUnit.MILLISECONDS)).isEqualTo(1000.0);
org.mockito.Mockito.verify(this.consumerFactory, Mockito.times(2)).createConsumer();
}
private List<PartitionInfo> partitions(Node... nodes) {
List<PartitionInfo> partitions = new ArrayList<>();
for (int i = 0; i < nodes.length; i++) {