Simplify filtering for adding metrics

apache · Aug 28, 2024 · d390f0a · d390f0a
1 parent 9947822
commit d390f0a
Showing 1 changed file with 4 additions and 9 deletions.
diff --git a/...ain/java/org/apache/kafka/streams/internals/metrics/StreamsDelegatingMetricsReporter.java b/...ain/java/org/apache/kafka/streams/internals/metrics/StreamsDelegatingMetricsReporter.java
@@ -26,7 +26,6 @@
 
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 
 public class StreamsDelegatingMetricsReporter implements MetricsReporter {
 
@@ -49,24 +48,20 @@ public void init(final List<KafkaMetric> metrics) {
 
     @Override
     public void metricChange(final KafkaMetric metric) {
-        if (filteredMetric(metric).isPresent()) {
+        if (tagMatchesCurrentThread(metric)) {
             LOG.info("Registering metric {} for thread={}", metric.metricName().name(), threadId);
             consumer.registerMetric(metric);
         }
     }
 
-    Optional<KafkaMetric> filteredMetric(final KafkaMetric kafkaMetric) {
+    boolean tagMatchesCurrentThread(final KafkaMetric kafkaMetric) {
         final Map<String, String> tags = kafkaMetric.metricName().tags();
-        KafkaMetric maybeKafkaMetric = null;
-        if (tags.containsKey(THREAD_ID_TAG) && tags.get(THREAD_ID_TAG).equals(threadId)) {
-            maybeKafkaMetric = kafkaMetric;
-        }
-        return Optional.ofNullable(maybeKafkaMetric);
+        return tags.containsKey(THREAD_ID_TAG) && tags.get(THREAD_ID_TAG).equals(threadId);
     }
 
     @Override
     public void metricRemoval(final KafkaMetric metric) {
-        if (filteredMetric(metric).isPresent()) {
+        if (tagMatchesCurrentThread(metric)) {
             LOG.info("Unregistering metric {} for thread={}", metric.metricName().name(), threadId);
             consumer.unregisterMetric(metric);
         }