@Test public void createsConsumerOnceWhenInvokedMultipleTimes() { final List<PartitionInfo> partitions = partitions(new Node(0, null, 0)); topicsInUse.put(TEST_TOPIC, new TopicInformation("group4-metrics", partitions, false)); metrics.bindTo(meterRegistry); Gauge gauge = meterRegistry.get(KafkaBinderMetrics.METRIC_NAME) .tag("group", "group4-metrics").tag("topic", TEST_TOPIC).gauge(); gauge.value(); assertThat(gauge.value()).isEqualTo(1000.0); org.mockito.Mockito.verify(this.consumerFactory) .createConsumer(ArgumentMatchers.any(), ArgumentMatchers.any()); }
@Override public void bindTo(MeterRegistry registry) { this.scheduler = Executors.newScheduledThreadPool(this.binder.getTopicsInUse().size()); for (Map.Entry<String, KafkaMessageChannelBinder.TopicInformation> topicInfo : this.binder .getTopicsInUse().entrySet()) { if (!topicInfo.getValue().isConsumerTopic()) { continue; } String topic = topicInfo.getKey(); String group = topicInfo.getValue().getConsumerGroup(); this.scheduler.scheduleWithFixedDelay(computeUnconsumedMessagesRunnable(topic, group, this.metadataConsumers), 10, DELAY_BETWEEN_TASK_EXECUTION, TimeUnit.SECONDS); Gauge.builder(METRIC_NAME, this, (o) -> computeAndGetUnconsumedMessages(topic, group)).tag("group", group) .tag("topic", topic) .description("Unconsumed messages for a particular group and topic") .register(registry); } }
KafkaBinderMetrics implements MeterBinder, ApplicationListener<BindingCreatedEvent> { @Override public void bindTo(MeterRegistry registry) { this.scheduler = Executors.newScheduledThreadPool(this.binder.getTopicsInUse().size()); for (Map.Entry<String, KafkaMessageChannelBinder.TopicInformation> topicInfo : this.binder .getTopicsInUse().entrySet()) { if (!topicInfo.getValue().isConsumerTopic()) { continue; } String topic = topicInfo.getKey(); String group = topicInfo.getValue().getConsumerGroup(); this.scheduler.scheduleWithFixedDelay(computeUnconsumedMessagesRunnable(topic, group, this.metadataConsumers), 10, DELAY_BETWEEN_TASK_EXECUTION, TimeUnit.SECONDS); Gauge.builder(METRIC_NAME, this, (o) -> computeAndGetUnconsumedMessages(topic, group)).tag("group", group) .tag("topic", topic) .description("Unconsumed messages for a particular group and topic") .register(registry); } } }
KafkaBinderMetrics implements MeterBinder, ApplicationListener<BindingCreatedEvent> { @Override public void bindTo(MeterRegistry registry) { this.scheduler = Executors.newScheduledThreadPool(this.binder.getTopicsInUse().size()); for (Map.Entry<String, KafkaMessageChannelBinder.TopicInformation> topicInfo : this.binder .getTopicsInUse().entrySet()) { if (!topicInfo.getValue().isConsumerTopic()) { continue; } String topic = topicInfo.getKey(); String group = topicInfo.getValue().getConsumerGroup(); this.scheduler.scheduleWithFixedDelay(computeUnconsumedMessagesRunnable(topic, group, this.metadataConsumers), 10, DELAY_BETWEEN_TASK_EXECUTION, TimeUnit.SECONDS); Gauge.builder(METRIC_NAME, this, (o) -> computeAndGetUnconsumedMessages(topic, group)).tag("group", group) .tag("topic", topic) .description("Unconsumed messages for a particular group and topic") .register(registry); } } KafkaBinderMetrics(KafkaMessageChannelBinder binder,
			KafkaBinderConfigurationProperties binderConfigurationProperties,
			ConsumerFactory<?, ?> defaultConsumerFactory,
			@Nullable MeterRegistry meterRegistry);  KafkaBinderMetrics(KafkaMessageChannelBinder binder,
			KafkaBinderConfigurationProperties binderConfigurationProperties); }
KafkaBinderMetrics implements MeterBinder, ApplicationListener<BindingCreatedEvent> { @Override public void bindTo(MeterRegistry registry) { this.scheduler = Executors.newScheduledThreadPool(this.binder.getTopicsInUse().size()); for (Map.Entry<String, KafkaMessageChannelBinder.TopicInformation> topicInfo : this.binder .getTopicsInUse().entrySet()) { if (!topicInfo.getValue().isConsumerTopic()) { continue; } String topic = topicInfo.getKey(); String group = topicInfo.getValue().getConsumerGroup(); this.scheduler.scheduleWithFixedDelay(computeUnconsumedMessagesRunnable(topic, group, this.metadataConsumers), 10, DELAY_BETWEEN_TASK_EXECUTION, TimeUnit.SECONDS); Gauge.builder(METRIC_NAME, this, (o) -> computeAndGetUnconsumedMessages(topic, group)).tag("group", group) .tag("topic", topic) .description("Unconsumed messages for a particular group and topic") .register(registry); } } KafkaBinderMetrics(KafkaMessageChannelBinder binder,
			KafkaBinderConfigurationProperties binderConfigurationProperties,
			ConsumerFactory<?, ?> defaultConsumerFactory,
			@Nullable MeterRegistry meterRegistry);  KafkaBinderMetrics(KafkaMessageChannelBinder binder,
			KafkaBinderConfigurationProperties binderConfigurationProperties); void setTimeout(int timeout); @Override void bindTo(MeterRegistry registry); @Override void onApplicationEvent(BindingCreatedEvent event); }
KafkaBinderMetrics implements MeterBinder, ApplicationListener<BindingCreatedEvent> { @Override public void bindTo(MeterRegistry registry) { this.scheduler = Executors.newScheduledThreadPool(this.binder.getTopicsInUse().size()); for (Map.Entry<String, KafkaMessageChannelBinder.TopicInformation> topicInfo : this.binder .getTopicsInUse().entrySet()) { if (!topicInfo.getValue().isConsumerTopic()) { continue; } String topic = topicInfo.getKey(); String group = topicInfo.getValue().getConsumerGroup(); this.scheduler.scheduleWithFixedDelay(computeUnconsumedMessagesRunnable(topic, group, this.metadataConsumers), 10, DELAY_BETWEEN_TASK_EXECUTION, TimeUnit.SECONDS); Gauge.builder(METRIC_NAME, this, (o) -> computeAndGetUnconsumedMessages(topic, group)).tag("group", group) .tag("topic", topic) .description("Unconsumed messages for a particular group and topic") .register(registry); } } KafkaBinderMetrics(KafkaMessageChannelBinder binder,
			KafkaBinderConfigurationProperties binderConfigurationProperties,
			ConsumerFactory<?, ?> defaultConsumerFactory,
			@Nullable MeterRegistry meterRegistry);  KafkaBinderMetrics(KafkaMessageChannelBinder binder,
			KafkaBinderConfigurationProperties binderConfigurationProperties); void setTimeout(int timeout); @Override void bindTo(MeterRegistry registry); @Override void onApplicationEvent(BindingCreatedEvent event);  }