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

KafkaBinderMetrics' metrics should be unregistered before it's thread

Browse Source 
* KafkaBinderMetrics' metrics should be unregistered before it's threadpool is shutdown.
* update authors and copyright years
This commit is contained in:
Kurt Hong
2024-08-24 02:36:41 +09:00
committed by GitHub
parent 8e5f67a1a6
commit fe1392efb0
2 changed files with 21 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
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-2022 the original author or authors.
* Copyright 2016-2024 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.
@@ -67,6 +67,7 @@ import org.springframework.util.ObjectUtils;
* @author Lars Bilger
* @author Tomek Szmytka
* @author Nico Heller
* @author Kurt Hong
*/
public class KafkaBinderMetrics
implements MeterBinder, ApplicationListener<BindingCreatedEvent>, AutoCloseable {
@@ -263,6 +264,9 @@ public class KafkaBinderMetrics
@Override
public void close() throws Exception {
if (this.meterRegistry != null) {
this.meterRegistry.find(OFFSET_LAG_METRIC_NAME).meters().forEach(this.meterRegistry::remove);
}
Optional.ofNullable(scheduler).ifPresent(ExecutorService::shutdown);
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2023 the original author or authors.
* Copyright 2016-2024 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.
@@ -56,6 +56,7 @@ import static org.mockito.Mockito.mock;
* @author Lars Bilger
* @author Tomek Szmytka
* @author Nico Heller
* @author Kurt Hong
*/
class KafkaBinderMetricsTest {
@@ -91,7 +92,7 @@ class KafkaBinderMetricsTest {
org.mockito.BDDMockito.given(kafkaBinderConfigurationProperties.getMetrics().getOffsetLagMetricsInterval())
.willReturn(Duration.ofSeconds(60));
metrics = new KafkaBinderMetrics(binder, kafkaBinderConfigurationProperties,
consumerFactory, null
consumerFactory, meterRegistry
);
org.mockito.BDDMockito
.given(consumer.endOffsets(ArgumentMatchers.anyCollection()))
@@ -351,6 +352,19 @@ class KafkaBinderMetricsTest {
assertThat(metrics.scheduler.isShutdown()).isTrue();
}
@Test
public void shouldUnregisterMetersOnClose() throws Exception {
final List<PartitionInfo> partitions = partitions(new Node(0, null, 0));
topicsInUse.put(
TEST_TOPIC,
new TopicInformation("group4-metrics", partitions, false)
);
metrics.bindTo(meterRegistry);
assertThat(meterRegistry.find(KafkaBinderMetrics.OFFSET_LAG_METRIC_NAME).meters()).hasSize(1);
metrics.close();
assertThat(meterRegistry.find(KafkaBinderMetrics.OFFSET_LAG_METRIC_NAME).meters()).isEmpty();
}
private List<PartitionInfo> partitions(Node... nodes) {
List<PartitionInfo> partitions = new ArrayList<>();
for (int i = 0; i < nodes.length; i++) {