KafkaBinderMetrics' metrics should be unregistered before it's thread…

[kurthong]

…pool is shutdown.

Hello,
Whenever I scale in my application cluster, some of the instances show the following error message. Upon investigation, I found that the problem is due to the metrics not being unregistered properly, even though they depend on the thread pool.
Thank you.

i.m.statsd.StatsdMeterRegistry - Failed to poll a meter 'spring.cloud.stream.binder.kafka.offset'. Note that subsequent logs will be logged at debug level.
java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask@255e8ffe[Not completed, task = org.springframework.cloud.stream.binder.kafka.KafkaBinderMetrics$$Lambda$2682/0x0000000801667c40@5b77f330] rejected from java.util.concurrent.ScheduledThreadPoolExecutor@6915670b[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 5060]
    at java.base/java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2055)
    at java.base/java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:825)
    at java.base/java.util.concurrent.ScheduledThreadPoolExecutor.delayedExecute(ScheduledThreadPoolExecutor.java:340)
    at java.base/java.util.concurrent.ScheduledThreadPoolExecutor.schedule(ScheduledThreadPoolExecutor.java:579)
    at java.base/java.util.concurrent.ScheduledThreadPoolExecutor.submit(ScheduledThreadPoolExecutor.java:731)
    at org.springframework.cloud.stream.binder.kafka.KafkaBinderMetrics.computeAndGetUnconsumedMessagesWithTimeout(KafkaBinderMetrics.java:173)
    at org.springframework.cloud.stream.binder.kafka.KafkaBinderMetrics.lambda$computeOffsetComputationFunction$1(KafkaBinderMetrics.java:165)
    at io.micrometer.statsd.StatsdGauge.value(StatsdGauge.java:54)
    at io.micrometer.statsd.StatsdGauge.poll(StatsdGauge.java:59)
    at io.micrometer.statsd.StatsdMeterRegistry.poll(StatsdMeterRegistry.java:183)
    at io.micrometer.statsd.StatsdMeterRegistry.close(StatsdMeterRegistry.java:323)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.base/java.lang.reflect.Method.invoke(Method.java:566)
    at org.springframework.beans.factory.support.DisposableBeanAdapter.invokeCustomDestroyMethod(DisposableBeanAdapter.java:325)
    at org.springframework.beans.factory.support.DisposableBeanAdapter.destroy(DisposableBeanAdapter.java:259)
    at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.destroyBean(DefaultSingletonBeanRegistry.java:587)
    at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.destroySingleton(DefaultSingletonBeanRegistry.java:559)
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.destroySingleton(DefaultListableBeanFactory.java:1163)
    at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.destroySingletons(DefaultSingletonBeanRegistry.java:520)
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.destroySingletons(DefaultListableBeanFactory.java:1156)
    at org.springframework.context.support.AbstractApplicationContext.destroyBeans(AbstractApplicationContext.java:1120)
    at org.springframework.context.support.AbstractApplicationContext.doClose(AbstractApplicationContext.java:1086)
    at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.doClose(ServletWebServerApplicationContext.java:174)
    at org.springframework.context.support.AbstractApplicationContext.close(AbstractApplicationContext.java:1032)
    at org.springframework.boot.SpringApplicationShutdownHook.closeAndWait(SpringApplicationShutdownHook.java:145)
    at java.base/java.lang.Iterable.forEach(Iterable.java:75)
    at org.springframework.boot.SpringApplicationShutdownHook.run(SpringApplicationShutdownHook.java:114)
    at java.base/java.lang.Thread.run(Thread.java:829)

[sobychacko]

@kurthong, Thanks for finding this issue and the PR. The changes look good. Could you add your name as an author to the classes you changed?

[kurthong]

@sobychacko
Authors and copyright years have been updated.
Thank you:-)

[sobychacko]

Merged the PR to main. Backported to 4.1.x. Thanks again for contributing the PR.

[sgtae]

It seems the code is registered normally in the metric. Thank you.