From a3a1debe45ac94304a7b78bfc85c90659e2f7a3a Mon Sep 17 00:00:00 2001
From: Mikhail Trofimov <mtrofimov@mfd.ru>
Date: Mon, 29 Jun 2026 15:45:42 +0300
Subject: [PATCH] Drain ivents from queue

---
 .../SignalKafkaEventWriterService.java        | 43 +++++++++++--------
 1 file changed, 24 insertions(+), 19 deletions(-)

diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/SignalKafkaEventWriterService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/SignalKafkaEventWriterService.java
index 50cff8a..91109a9 100644
--- a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/SignalKafkaEventWriterService.java
+++ b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/SignalKafkaEventWriterService.java
@@ -30,6 +30,8 @@ import java.nio.file.StandardCopyOption;
 import java.nio.file.StandardOpenOption;
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutorService;
@@ -40,7 +42,7 @@ import java.util.concurrent.TimeUnit;
 @Service
 public class SignalKafkaEventWriterService {
     private final static Charset encoding = Charset.forName("windows-1251");
-    private final static DateTimeFormatter formatter = DateTimeFormatter.ofPattern("dd.MM.yyyy");
+    private final static DateTimeFormatter formatter = DateTimeFormatter.ofPattern("dd.MM.yyyy HH:mm:ss"); // TODO убрать время
 
     private final ExecutorService executorService = Executors.newSingleThreadExecutor(
             new CustomizableThreadFactory("signal-kafka-event-writer-")
@@ -65,20 +67,22 @@ public class SignalKafkaEventWriterService {
 
     public void startWriting() {
         while (true) {
-            SignalKafkaEvent event = take();
-            if (event == null) {
-                continue;
-            }
-            if (event.getEventType() == EventType.STOP) {
-                break;
-            }
+            List<SignalKafkaEvent> events = drain();
+            for (SignalKafkaEvent event : events) {
+                if (event == null) {
+                    continue;
+                }
+                if (event.getEventType() == EventType.STOP) {
+                    break;
+                }
 
-            if (event.getEventType() == EventType.WRITE) {
-                writeSignal(event);
-            }
+                if (event.getEventType() == EventType.WRITE) {
+                    writeSignal(event);
+                }
 
-            if (event.getEventType() == EventType.ROTATE) {
-                rotate();
+                if (event.getEventType() == EventType.ROTATE) {
+                    rotate();
+                }
             }
         }
     }
@@ -295,13 +299,14 @@ public class SignalKafkaEventWriterService {
                 .replace("'", "&apos;");
     }
 
-    private SignalKafkaEvent take() {
+    private List<SignalKafkaEvent> drain() {
         try {
-            return signalEventsQueue.take();
-        } catch (InterruptedException e) {
-            log.error("Error while taking signal event", e);
-            Thread.currentThread().interrupt();
-            return null;
+            List<SignalKafkaEvent> batch = new ArrayList<>(1000);
+            signalEventsQueue.drainTo(batch, 1000);
+            return batch;
+        } catch (Exception e) {
+            log.error("Error while drain signal events", e);
+            return Collections.emptyList();
         }
     }