From 7f3a7f856fa44f27e08127d067f73066fa7bfb38 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Mon, 5 Oct 2020 13:19:01 -0400 Subject: [PATCH] Kafka binder metrics improvements (#965) * Kafka binder metrics improvements KafkaBinderMetrics has a blocking call in which it waits for the default timeout of 60 seconds if Kafka broker is down. This happens for each topic within a consumer group. Refactor this code, so that we have this check performed in a periodic task and if the runtime check fails to return within a smaller timewindow (5 seconds), return immediately by providing the latest value from the periodic task results. Periodic task for computing the lags is run every 60 seconds. Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/809 * Addressing PR review --- .../binder/kafka/KafkaBinderMetrics.java | 95 ++++++++++++------- 1 file changed, 62 insertions(+), 33 deletions(-) diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java index 6385448ef..6449be335 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java @@ -25,6 +25,7 @@ import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -62,7 +63,9 @@ import org.springframework.util.ObjectUtils; public class KafkaBinderMetrics implements MeterBinder, ApplicationListener { - private static final int DEFAULT_TIMEOUT = 60; + private static final int DEFAULT_TIMEOUT = 5; + + private static final int DELAY_BETWEEN_TASK_EXECUTION = 60; private static final Log LOG = LogFactory.getLog(KafkaBinderMetrics.class); @@ -80,6 +83,10 @@ public class KafkaBinderMetrics private int timeout = DEFAULT_TIMEOUT; + ScheduledExecutorService scheduler; + + Map unconsumedMessages = new ConcurrentHashMap<>(); + public KafkaBinderMetrics(KafkaMessageChannelBinder binder, KafkaBinderConfigurationProperties binderConfigurationProperties, ConsumerFactory defaultConsumerFactory, @@ -104,6 +111,9 @@ public class KafkaBinderMetrics @Override public void bindTo(MeterRegistry registry) { + + this.scheduler = Executors.newScheduledThreadPool(this.binder.getTopicsInUse().size()); + for (Map.Entry topicInfo : this.binder .getTopicsInUse().entrySet()) { @@ -114,45 +124,37 @@ public class KafkaBinderMetrics String topic = topicInfo.getKey(); String group = topicInfo.getValue().getConsumerGroup(); + //Schedule a task to compute the unconsumed messages for this group/topic every minute. + this.scheduler.scheduleWithFixedDelay(computeUnconsumedMessagesRunnable(topic, group, this.metadataConsumers), + 10, DELAY_BETWEEN_TASK_EXECUTION, TimeUnit.SECONDS); + Gauge.builder(METRIC_NAME, this, - (o) -> computeUnconsumedMessages(topic, group)).tag("group", group) + (o) -> computeAndGetUnconsumedMessages(topic, group)).tag("group", group) .tag("topic", topic) .description("Unconsumed messages for a particular group and topic") .register(registry); } } - private long computeUnconsumedMessages(String topic, String group) { - ExecutorService exec = Executors.newSingleThreadExecutor(); + private Runnable computeUnconsumedMessagesRunnable(String topic, String group, Map> metadataConsumers) { + return () -> { + try { + long lag = findTotalTopicGroupLag(topic, group, this.metadataConsumers); + this.unconsumedMessages.put(topic + "-" + group, lag); + } + catch (Exception ex) { + LOG.debug("Cannot generate metric for topic: " + topic, ex); + } + }; + } + + private long computeAndGetUnconsumedMessages(String topic, String group) { + ExecutorService exec = Executors.newCachedThreadPool(); Future future = exec.submit(() -> { long lag = 0; try { - Consumer metadataConsumer = this.metadataConsumers.computeIfAbsent( - group, - (g) -> createConsumerFactory().createConsumer(g, "monitoring")); - synchronized (metadataConsumer) { - List partitionInfos = metadataConsumer - .partitionsFor(topic); - List topicPartitions = new LinkedList<>(); - for (PartitionInfo partitionInfo : partitionInfos) { - topicPartitions.add(new TopicPartition(partitionInfo.topic(), - partitionInfo.partition())); - } - - Map endOffsets = metadataConsumer - .endOffsets(topicPartitions); - - for (Map.Entry endOffset : endOffsets - .entrySet()) { - OffsetAndMetadata current = metadataConsumer - .committed(endOffset.getKey()); - lag += endOffset.getValue(); - if (current != null) { - lag -= current.offset(); - } - } - } + lag = findTotalTopicGroupLag(topic, group, this.metadataConsumers); } catch (Exception ex) { LOG.debug("Cannot generate metric for topic: " + topic, ex); @@ -164,16 +166,44 @@ public class KafkaBinderMetrics } catch (InterruptedException ex) { Thread.currentThread().interrupt(); - return 0L; + return this.unconsumedMessages.getOrDefault(topic + "-" + group, 0L); } catch (ExecutionException | TimeoutException ex) { - return 0L; + return this.unconsumedMessages.getOrDefault(topic + "-" + group, 0L); } finally { exec.shutdownNow(); } } + private long findTotalTopicGroupLag(String topic, String group, Map> metadataConsumers) { + long lag = 0; + Consumer metadataConsumer = metadataConsumers.computeIfAbsent( + group, + (g) -> createConsumerFactory().createConsumer(g, "monitoring")); + List partitionInfos = metadataConsumer + .partitionsFor(topic); + List topicPartitions = new LinkedList<>(); + for (PartitionInfo partitionInfo : partitionInfos) { + topicPartitions.add(new TopicPartition(partitionInfo.topic(), + partitionInfo.partition())); + } + + Map endOffsets = metadataConsumer + .endOffsets(topicPartitions); + + for (Map.Entry endOffset : endOffsets + .entrySet()) { + OffsetAndMetadata current = metadataConsumer + .committed(endOffset.getKey()); + lag += endOffset.getValue(); + if (current != null) { + lag -= current.offset(); + } + } + return lag; + } + private synchronized ConsumerFactory createConsumerFactory() { if (this.defaultConsumerFactory == null) { Map props = new HashMap<>(); @@ -200,8 +230,7 @@ public class KafkaBinderMetrics @Override public void onApplicationEvent(BindingCreatedEvent event) { if (this.meterRegistry != null) { - // meters are idempotent when called with the same arguments so safe to call - // it multiple times + // It is safe to call bindTo multiple times, since meters are idempotent when called with the same arguments this.bindTo(this.meterRegistry); } }