Allow user-provided custom scheduler for periodically binding KafkaMetrics

[vasiliy-sarzhynskyi]

Please describe the feature request.
Currently, KafkaClientMetrics does not provide the possibility to use a custom scheduler for binding metrics. Instead, it uses a default one:

ScheduledExecutorService scheduler = Executors
    .newSingleThreadScheduledExecutor(new NamedThreadFactory("micrometer-kafka-metrics"));

Ideally, it would be great to have the ability to pass an optional custom thread pool, so later Spring-Kafka will be able to propagate optional scheduler into KafkaClientMetrics.

Rationale
I have a Spring-Boot application with multiple consumers with the same consumer group using @KafkaListener from Spring-Kafka. After investigation, I discovered that per each consumer it creates additional Micrometer metrics daemon thread with the name micrometer-kafka-metrics, which is created by io.micrometer:micrometer.core inside KafkaMetrics.

The chain of invocations is the following. KafkaMetricsAutoConfiguration (from spring-boot-actuator-autoconfigure) creates a single instance of MicrometerConsumerListener (from Spring-Kafka) per application, and after that MicrometerConsumerListener has method consumerAdded(..) that is invoked per each Consumer. It creates KafkaClientMetrics (extended from KafkaMetrics). As a result, the number of consumers is the same as the number of threads micrometer-kafka-metrics.

If we have consumer factory concurrency as N (new ConcurrentKafkaListenerContainerFactory<>().setConcurrency(N);) and number of @KafkaListener as M (one per each topic, as we have M topics), as a result we have N * M threads with name micrometer-kafka-metrics. In most cases N * M is not a big number, but in some rare cases, it might be a few hundred (like in my case). I want to leave the number of concurrent consumers as is, but decrease the number of metrics threads, cause metrics threads are quite lightweight and scheduled with a fixed rate (1 minute by default).

So I'm looking at ways how to decrease the number of these metrics threads. Ideally, it would be great to have the ability to pass there optional custom thread pool, that will be shareable for all consumers and with the desired number of threads.

Additional context
I created a demo application to show this issue with the management API endpoint that returns current threads. You could execute the following integration test by your own to see the number of created threads micrometer-kafka-metrics.

References

• Clarification on StackOverflow
• Demo
• Spring-Kafka creates KafkaClientMetrics