Optional custom scheduler for KafkaMetrics

[vasiliy-sarzhynskyi]

Resolves gh-4976

[vasiliy-sarzhynskyi]

It's a proposed contribution for enhancement #4976 to extend KafkaMetrics with custom scheduler

[shakuzen]

Thank you for the pull request. This looks reasonable to me. Before merging, I'm trying to get an idea of the impact to other projects so this can be most easily used by end users. What would change in Spring Kafka and Spring Boot? I guess MicrometerConsumerListener in Spring Kafka would need to optionally take a ScheduledExecutorService and KafkaMetricsAutoConfiguration in Spring Boot would need a way to get a ScheduledExecutorService from user configuration (how would it do this?) to pass to the MicrometerConsumerListener it makes.

@artembilan do you have any concerns about this from a Spring Kafka perspective?

[artembilan]

This is fine with me since it does not bring any breaking changes.
In Spring we have a ThreadPoolTaskScheduler (which is auto-configured by Spring Boot) and that one comes with a property:

	/**
	 * Return the underlying ScheduledExecutorService for native access.
	 * @return the underlying ScheduledExecutorService (never {@code null})
	 * @throws IllegalStateException if the ThreadPoolTaskScheduler hasn't been initialized yet
	 */
	public ScheduledExecutorService getScheduledExecutor() throws IllegalStateException {

I'll be totally fine exposing that property on the MicrometerConsumerListener and MicrometerProducerListener.
However you are not done with the fixed here yet.
Pay attention to those ctors of this class which relies on the Producer and AdminClient, respectively.
Thanks

[shakuzen]

Thanks for the review @artembilan.
@vasiliy-sarzhynskyi any reason to not add similar constructors for producers and admin clients?

[vasiliy-sarzhynskyi]

@shakuzen I need to check how producers and admin clients are used by Spring-Kafka, and how many threads we might have with multiple produced topics (whether the number of threads increased or not). We could add there as well, but do we have use cases when producers and admin clients might need that? In my understanding, it's especially relevant for consumers as we might specify concurrency config value, and a single app instance will have hundreds of threads.

[artembilan]

I’m not sure how is this relevant to the use of those clients in the app, the scheduler we provide here is used for publishing metrics from those clients. See bindTo() impl.

[shakuzen]

This scheduler isn't for publishing; MeterRegistry implementations can be configured with a scheduler to publish metrics. This scheduler is for syncing the metrics from the Kafka client to Micrometer's registry periodically. The background is in #4976, but with many clients and the existing code, each client will have a thread for syncing between the Kafka client's metrics and Micrometer. With the changes here, a thread pool could be shared among the clients for this syncing.

[artembilan]

Thank you, @shakuzen , for clarification!
So, sounds like I was right anyway: this scheduler has nothing to do with Spring Kafka and has same effect for all the clients.
Therefore similar injection approach has to be applied for Producer and Admin.
Otherwise it is strange that we can customize a scheduler for Consumer client metrics synchronization, but that is not the case for Producer and Admin.

[vasiliy-sarzhynskyi]

Thank you. In that case, I will update the current PR for Producer and Admin as well

[vasiliy-sarzhynskyi]

@shakuzen I pushed my changes for Producer and AdminClient constructors as well. After I rebased on branch main, recently added test class VirtualThreadMetricsTests is failing, looks like it's flaky.
I could try to increase the upper bound threshold inside VirtualThreadMetricsTests pinnedEventsShouldBeRecorded(..), but would be better to contact owner of these changes to clarify the intention of asserted values (changes by Virtual threads metrics)

[shakuzen]

Pinging @jonatan-ivanov about the flakiness of VirtualThreadMetricsTests. I'll re-run the CI manually in the meantime.

[shakuzen]

Thanks for the pull request and quick response to review. This looks good to me. I'll polish a couple things when merging.

[shakuzen]

I've added the same changes for KafkaStreamsMetrics and updated the JavaDocs to specifically mention the purpose and responsibilities for the custom scheduler.