From a81093734eb4d85e00969c0700998d379984df97 Mon Sep 17 00:00:00 2001 From: aleksevi Date: Tue, 25 May 2021 11:09:59 +0200 Subject: [PATCH] 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 --- .../binder/kafka/KafkaBinderMetrics.java | 20 ++++++++++++++----- 1 file changed, 15 insertions(+), 5 deletions(-) diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java index a9c8a36b9..dc7ebb25b 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java +++ b/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 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 props = new HashMap<>(); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,