KafkaMonitoringService graceful shutdown

src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java

@@ -1,13 +1,18 @@
 package ru.nbch.credit_tracker.service.mode.monitoring;
 
 import jakarta.annotation.PreDestroy;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.Semaphore;
+import java.util.stream.Collectors;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.stereotype.Service;
-import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
 import ru.nbch.credit_tracker.entities.app.PhonesMap;
 import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage;
 import ru.nbch.credit_tracker.enums.EventType;
@@ -18,18 +23,13 @@ import ru.nbch.credit_tracker.service.signal.PackageRecordsService;
 import ru.nbch.credit_tracker.service.signal.SegmentMapService;
 import ru.nbch.credit_tracker.utils.CTUtil;
 
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.Semaphore;
-import java.util.stream.Collectors;
-
 @Slf4j
 @Service
 public class KafkaMonitoringService {
     private volatile boolean isShuttingdDown = false;
-    private final Semaphore processingPermits = new Semaphore(8); // количество равно concurency листенера
+    private final static String CONCURRENCY = "8";
+    private final static int PERMITS = Integer.parseInt(CONCURRENCY);
+    private final Semaphore processingPermits = new Semaphore(PERMITS); // количество равно concurency листенера
 
     private final BlockingQueue<SignalKafkaEvent> signalEventsQueue;
 
@@ -37,37 +37,33 @@ public class KafkaMonitoringService {
     private final SourceService sourceService;
     private final SegmentMapService segmentMapService;
 
-    private final CreditTrackerSettings config;
-
     public KafkaMonitoringService(@Qualifier("signalKafkaEventQueue") BlockingQueue<SignalKafkaEvent> signalEventsQueue,
                                   PackageRecordsService packageRecordsService,
                                   SourceService sourceService,
-                                  SegmentMapService segmentMapService,
+                                  SegmentMapService segmentMapService) {
-                                  CreditTrackerSettings config) {
         this.signalEventsQueue = signalEventsQueue;
         this.packageRecordsService = packageRecordsService;
         this.sourceService = sourceService;
         this.segmentMapService = segmentMapService;
-        this.config = config;
     }
 
     @KafkaListener(
             topics = "my-topic",
-            concurrency = "8"
+            concurrency = CONCURRENCY
     )
     public void listen(List<PackageMessage> records,
                        Acknowledgment acknowledgment) {
+        if (isShuttingdDown) {
+            log.debug("KafkaMonitoringService is shutting down, rejecting new messages");
+            return; // без ака, завершаем обработку сообщений, которые уже попали в очередь, новые не обрабатываем
+        }
+
         if (!processingPermits.tryAcquire()) {
-            log.warn("KafkaMonitoringService is shutting down, rejecting new messages");
+            log.warn("cannot acquire processing permits");
             return;
         }
 
         try {
-            if (isShuttingdDown) {
-                log.warn("KafkaMonitoringService is shutting down, rejecting new messages");
-                return; // без ака, завершаем обработку сообщений, которые уже попали в очередь, новые не обрабатываем
-            }
-
             acknowledgment.acknowledge();
             process(records);
             log.info("Processed {} messages from Kafka", records.size());
@@ -170,9 +166,7 @@ public class KafkaMonitoringService {
         log.info("Stopping KafkaMonitoringService...");
         isShuttingdDown = true;
 
-        for (int i = 0; i < 8; i++) {
+        processingPermits.acquire(PERMITS);
-            processingPermits.acquire();
-        }
 
         try {
             SignalKafkaEvent rotateEvent = new SignalKafkaEvent();
@@ -185,7 +179,7 @@ public class KafkaMonitoringService {
             signalEventsQueue.put(stopEvent);
             log.info("STOP event sent to queue");
         } finally {
-            processingPermits.release();
+            processingPermits.release(PERMITS);
         }
 
         log.info("KafkaMonitoringService stopped");