[FIX_KAFKA_EXECUTOR] fix executor service

kiegroup · Feb 29, 2024 · 550927d · 550927d
1 parent 5dad97a
commit 550927d
Showing 1 changed file with 2 additions and 4 deletions.
diff --git a/...-services-kafka/src/main/java/org/kie/server/services/jbpm/kafka/KafkaServerConsumer.java b/...-services-kafka/src/main/java/org/kie/server/services/jbpm/kafka/KafkaServerConsumer.java
@@ -23,6 +23,7 @@
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -112,10 +113,7 @@ private void registrationUpdated(Set<String> topics2Register, boolean shouldInit
                     consumer = consumerSupplier.get();
                     consumer.subscribe(topics2Register);
                     logger.debug("Created kafka consumer with these topics registered {}", topics2Register);
-                    notifyService.set(
-                            new ThreadPoolExecutor(1, Integer.getInteger(KAFKA_EXTENSION_PREFIX + "maxNotifyThreads", 10),
-                                60L,
-                                TimeUnit.SECONDS, new LinkedBlockingQueue<>()));
+                    notifyService.set(Executors.newFixedThreadPool(Integer.getInteger(KAFKA_EXTENSION_PREFIX + "maxNotifyThreads", 10)));
                     new Thread(this).start();
                 }
             } else {