From ec93c4b72625c884f81563567a88cdfb428dc43c Mon Sep 17 00:00:00 2001
From: Mikhail Trofimov <mtrofimov@mfd.ru>
Date: Tue, 23 Jun 2026 15:35:35 +0300
Subject: [PATCH] Use semaphore instead of AtomicInteger counter

---
 .../monitoring/KafkaMonitoringService.java    | 33 +++++++++----------
 1 file changed, 16 insertions(+), 17 deletions(-)

diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java
index 6418209..8bb19e6 100644
--- a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java
+++ b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java
@@ -21,14 +21,14 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.Semaphore;
 import java.util.stream.Collectors;
 
 @Slf4j
 @Service
 public class KafkaMonitoringService {
     private volatile boolean isShuttingdDown = false;
-    private final AtomicInteger activeProcessCount = new AtomicInteger(0);
+    private final Semaphore processingPermits = new Semaphore(8); // количество равно concurency листенера
 
     private final BlockingQueue<SignalKafkaEvent> signalEventsQueue;
 
@@ -53,17 +53,22 @@ public class KafkaMonitoringService {
     )
     public void listen(List<PackageMessage> records,
                        Acknowledgment acknowledgment) {
-        if (isShuttingdDown) {
+        if (!processingPermits.tryAcquire()) {
             log.warn("KafkaMonitoringService is shutting down, rejecting new messages");
-            return; // без ака, завершаем обработку сообщений, которые уже попали в очередь, новые не обрабатываем
+            return;
         }
+
         try {
+            if (isShuttingdDown) {
+                log.warn("KafkaMonitoringService is shutting down, rejecting new messages");
+                return; // без ака, завершаем обработку сообщений, которые уже попали в очередь, новые не обрабатываем
+            }
+
             acknowledgment.acknowledge();
-            activeProcessCount.incrementAndGet();
             process(records);
             log.info("Processed {} messages from Kafka", records.size());
         } finally {
-            activeProcessCount.decrementAndGet();
+            processingPermits.release();
         }
     }
 
@@ -149,17 +154,12 @@ public class KafkaMonitoringService {
     }
 
     @PreDestroy
-    public void stopKafkaMonitoringService() {
+    public void stopKafkaMonitoringService() throws InterruptedException {
         log.info("Stopping KafkaMonitoringService...");
         isShuttingdDown = true;
 
-        while (activeProcessCount.get() > 0) {
-            try {
-                Thread.sleep(1000);
-            } catch (InterruptedException e) {
-                Thread.currentThread().interrupt();
-                break;
-            }
+        for (int i = 0; i < 8; i++) {
+            processingPermits.acquire();
         }
 
         try {
@@ -172,9 +172,8 @@ public class KafkaMonitoringService {
             stopEvent.setEventType(EventType.STOP);
             signalEventsQueue.put(stopEvent);
             log.info("STOP event sent to queue");
-        } catch (InterruptedException e) {
-            log.error("Error while sending ROTATE and STOP event", e);
-            Thread.currentThread().interrupt();
+        } finally {
+            processingPermits.release();
         }
 
         log.info("KafkaMonitoringService stopped");