Moved RotateSheduler as separate component

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

@@ -28,8 +28,5 @@ public class CTMonitoringApplication {
         ConfigurableApplicationContext ctx = application.run(args);
 //        MonitoringService monitoringService = ctx.getBean(MonitoringService.class);
 //        monitoringService.run();
-
-        SignalKafkaEventWriterService signalKafkaEventWriterService = ctx.getBean(SignalKafkaEventWriterService.class);
-        signalKafkaEventWriterService.startWriting();
     }
 }

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

@@ -1,13 +1,11 @@
 package ru.nbch.credit_tracker.service.mode.monitoring;
 
-import jakarta.annotation.PostConstruct;
 import jakarta.annotation.PreDestroy;
 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.scheduling.concurrent.CustomizableThreadFactory;
 import org.springframework.stereotype.Service;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
 import ru.nbch.credit_tracker.entities.app.PhonesMap;
@@ -23,19 +21,12 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 
 @Slf4j
 @Service
 public class KafkaMonitoringService {
-
-    private final ScheduledExecutorService rotateScheduler = Executors.newSingleThreadScheduledExecutor(
-            new CustomizableThreadFactory("monitoring-rotate-")
-    );
     private volatile boolean isShuttingdDown = false;
     private final AtomicInteger activeProcessCount = new AtomicInteger(0);
 
@@ -157,25 +148,8 @@ public class KafkaMonitoringService {
         }
     }
 
-    /**
-     * Таймер, который отсчитывает {@code monitoring-interval:5} минут с момента начала накопления сигналов.
-     * По истечении таймера отправляет Event на ROTATE файлов
-     */
-    @PostConstruct
-    public void startRotateTimer() {
-        long period = config.getMonitoringInterval() != null ? config.getMonitoringInterval() : 5;
-        rotateScheduler.scheduleAtFixedRate(() -> {
-            if (isShuttingdDown) {
-                return;
-            }
-            SignalKafkaEvent rotateEvent = new SignalKafkaEvent();
-            rotateEvent.setEventType(EventType.ROTATE);
-            putToQueue(rotateEvent);
-        }, 0, period, TimeUnit.MINUTES);
-    }
-
     @PreDestroy
-    public void stopRotateTimer() {
+    public void stopKafkaMonitoringService() {
         log.info("Stopping KafkaMonitoringService...");
         isShuttingdDown = true;
 
@@ -203,16 +177,6 @@ public class KafkaMonitoringService {
             Thread.currentThread().interrupt();
         }
 
-        rotateScheduler.shutdown();
-        try {
-            if (!rotateScheduler.awaitTermination(5, TimeUnit.SECONDS)) {
-                rotateScheduler.shutdownNow();
-            }
-        } catch (InterruptedException e) {
-            rotateScheduler.shutdownNow();
-            Thread.currentThread().interrupt();
-        }
-
         log.info("KafkaMonitoringService stopped");
     }
 }

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

@@ -0,0 +1,74 @@
+package ru.nbch.credit_tracker.service.mode.monitoring;
+
+import jakarta.annotation.PostConstruct;
+import jakarta.annotation.PreDestroy;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
+import org.springframework.stereotype.Component;
+import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
+import ru.nbch.credit_tracker.enums.EventType;
+import ru.nbch.credit_tracker.model.SignalKafkaEvent;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+@Slf4j
+@Component
+public class RotateSheduler {
+    private final ScheduledExecutorService rotateScheduler = Executors.newSingleThreadScheduledExecutor(
+            new CustomizableThreadFactory("monitoring-rotate-")
+    );
+    private volatile boolean isShuttingdDown = false;
+
+    private final BlockingQueue<SignalKafkaEvent> signalEventsQueue;
+    private final CreditTrackerSettings config;
+
+    public RotateSheduler(@Qualifier("signalKafkaEventQueue") BlockingQueue<SignalKafkaEvent> signalEventsQueue,
+                          CreditTrackerSettings config) {
+        this.signalEventsQueue = signalEventsQueue;
+        this.config = config;
+    }
+
+    /**
+     * Таймер, который отсчитывает {@code monitoring-interval:5} минут с момента начала накопления сигналов.
+     * По истечении таймера отправляет Event на ROTATE файлов
+     */
+    @PostConstruct
+    public void startRotateTimer() {
+        long period = config.getMonitoringInterval() != null ? config.getMonitoringInterval() : 5;
+        rotateScheduler.scheduleAtFixedRate(() -> {
+            if (isShuttingdDown) {
+                return;
+            }
+            try {
+                SignalKafkaEvent rotateEvent = new SignalKafkaEvent();
+                rotateEvent.setEventType(EventType.ROTATE);
+                signalEventsQueue.put(rotateEvent);
+            } catch (InterruptedException e) {
+                log.error("Error while putting signal event", e);
+                Thread.currentThread().interrupt();
+            }
+        }, 0, period, TimeUnit.MINUTES);
+    }
+
+    @PreDestroy
+    public void stopRotateTimer() {
+        log.info("Stopping RotateSheduler...");
+        isShuttingdDown = true;
+
+        rotateScheduler.shutdown();
+        try {
+            if (!rotateScheduler.awaitTermination(5, TimeUnit.SECONDS)) {
+                rotateScheduler.shutdownNow();
+            }
+        } catch (InterruptedException e) {
+            rotateScheduler.shutdownNow();
+            Thread.currentThread().interrupt();
+        }
+
+        log.info("RotateSheduler stopped");
+    }
+}

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

@@ -1,7 +1,10 @@
 package ru.nbch.credit_tracker.service.mode.monitoring;
 
+import jakarta.annotation.PostConstruct;
+import jakarta.annotation.PreDestroy;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
 import org.springframework.stereotype.Service;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
 import ru.nbch.credit_tracker.entities.app.UserPackage;
@@ -16,10 +19,16 @@ import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 
 @Slf4j
 @Service
 public class SignalKafkaEventWriterService {
+    private final ExecutorService executorService = Executors.newSingleThreadExecutor(
+            new CustomizableThreadFactory("signal-kafka-event-writer-")
+    );
 
     private final BlockingQueue<SignalKafkaEvent> signalEventsQueue;
     private final UserPackageMapper userPackageMapper;
@@ -33,6 +42,11 @@ public class SignalKafkaEventWriterService {
         this.config = config;
     }
 
+    @PostConstruct
+    public void start() {
+        executorService.submit(this::startWriting);
+    }
+
     public void startWriting() {
         while (true) {
             SignalKafkaEvent event = take();
@@ -103,4 +117,21 @@ public class SignalKafkaEventWriterService {
             return null;
         }
     }
+
+    @PreDestroy
+    public void stop() {
+        log.info("Stopping SignalKafkaEventWriterService...");
+
+        executorService.shutdown();
+        try {
+            if (!executorService.awaitTermination(5, TimeUnit.MINUTES)) {
+                executorService.shutdownNow();
+            }
+        } catch (InterruptedException e) {
+            executorService.shutdownNow();
+            Thread.currentThread().interrupt();
+        }
+
+        log.info("SignalKafkaEventWriterService stopped");
+    }
 }