GH-1081 Suspicious multiplication of ScheduledExecutorService in the method "bindTo" of class "KafkaBinderMetrics"

Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1081

The wrong consequence from the absense of scheduler.shutdown():
1) First of all we created the pool with 1 thread.
2) After we lost the reference on it and created the pool with 2 threads.
3) But, the first pool is not yet collected by GC and now you have 3 threads together. And so on.

Each thread does nothing, but it takes system memory and takes part in the scheduling process.
After 30 topics, for example, we potentially have (30+1)*15=465 threads.
It is already serious additional load on the switching contexts and the native memory.

Removed waiting after scheduler stop request.

checkstyle fixes

spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java

@@ -26,6 +26,7 @@ import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
@@ -92,9 +93,9 @@ public class KafkaBinderMetrics
 	Map<String, Long> unconsumedMessages = new ConcurrentHashMap<>();
 
 	public KafkaBinderMetrics(KafkaMessageChannelBinder binder,
-			KafkaBinderConfigurationProperties binderConfigurationProperties,
-			ConsumerFactory<?, ?> defaultConsumerFactory,
-			@Nullable MeterRegistry meterRegistry) {
+							KafkaBinderConfigurationProperties binderConfigurationProperties,
+							ConsumerFactory<?, ?> defaultConsumerFactory,
+							@Nullable MeterRegistry meterRegistry) {
 
 		this.binder = binder;
 		this.binderConfigurationProperties = binderConfigurationProperties;
@@ -104,7 +105,7 @@ public class KafkaBinderMetrics
 	}
 
 	public KafkaBinderMetrics(KafkaMessageChannelBinder binder,
-			KafkaBinderConfigurationProperties binderConfigurationProperties) {
+							KafkaBinderConfigurationProperties binderConfigurationProperties) {
 
 		this(binder, binderConfigurationProperties, null, null);
 	}
@@ -115,6 +116,15 @@ public class KafkaBinderMetrics
 
 	@Override
 	public void bindTo(MeterRegistry registry) {
+		/**
+		 * We can't just replace one scheduler with another.
+		 * Before and even after the old one is gathered by GC, it's threads still exist, consume memory and CPU resources to switch contexts.
+		 * Theoretically, as a result of processing n topics, there will be about (1+n)*n/2 threads simultaneously at the same time.
+		 */
+		if (this.scheduler != null) {
+			LOG.info("Try to shutdown the old scheduler with " + ((ScheduledThreadPoolExecutor) scheduler).getPoolSize() + " threads");
+			this.scheduler.shutdown();
+		}
 
 		this.scheduler = Executors.newScheduledThreadPool(this.binder.getTopicsInUse().size());
 
@@ -210,7 +220,7 @@ public class KafkaBinderMetrics
 		return lag;
 	}
 
-	private synchronized  ConsumerFactory<?, ?> createConsumerFactory() {
+	private synchronized ConsumerFactory<?, ?> createConsumerFactory() {
 		if (this.defaultConsumerFactory == null) {
 			Map<String, Object> props = new HashMap<>();
 			props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,