You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tz...@apache.org on 2023/03/24 04:57:58 UTC
[flink-connector-kafka] 01/02: [FLINK-31600] Fix a producer leak in KafkaMetricMutableWrapperTest
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
commit b25dfe63ff35a6bee76eb30135c859df4ec07f94
Author: Alex Sorokoumov <al...@gmail.com>
AuthorDate: Thu Mar 23 14:07:28 2023 -0700
[FLINK-31600] Fix a producer leak in KafkaMetricMutableWrapperTest
This closes #13
---
.../metrics/KafkaMetricMutableWrapperTest.java | 19 ++++++++++++-------
1 file changed, 12 insertions(+), 7 deletions(-)
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/internals/metrics/KafkaMetricMutableWrapperTest.java b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/internals/metrics/KafkaMetricMutableWrapperTest.java
index 952f0fd..269a155 100644
--- a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/internals/metrics/KafkaMetricMutableWrapperTest.java
+++ b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/internals/metrics/KafkaMetricMutableWrapperTest.java
@@ -73,14 +73,19 @@ class KafkaMetricMutableWrapperTest {
private static void testOnlyMeasurableMetricsAreRegistered(
Function<Metric, Gauge<Double>> wrapperFactory) {
final Collection<Gauge<Double>> metricWrappers = new ArrayList<>();
- final KafkaConsumer<?, ?> consumer = new KafkaConsumer<>(getKafkaClientConfiguration());
- final KafkaProducer<?, ?> producer = new KafkaProducer<>(getKafkaClientConfiguration());
- Stream.concat(consumer.metrics().values().stream(), producer.metrics().values().stream())
- .map(wrapperFactory::apply)
- .forEach(metricWrappers::add);
+ try (final KafkaConsumer<?, ?> consumer =
+ new KafkaConsumer<>(getKafkaClientConfiguration());
+ final KafkaProducer<?, ?> producer =
+ new KafkaProducer<>(getKafkaClientConfiguration())) {
+ Stream.concat(
+ consumer.metrics().values().stream(),
+ producer.metrics().values().stream())
+ .map(wrapperFactory::apply)
+ .forEach(metricWrappers::add);
- // Ensure that all values are accessible and return valid double values
- metricWrappers.forEach(Gauge::getValue);
+ // Ensure that all values are accessible and return valid double values
+ metricWrappers.forEach(Gauge::getValue);
+ }
}
private static Properties getKafkaClientConfiguration() {