Added queue for kafka signals; Added SignalKafkaEventWriterService

src/main/java/ru/nbch/credit_tracker/config/kafka/KafkaConfig.java

@@ -11,15 +11,25 @@ import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
 import org.springframework.kafka.core.*;
 import org.springframework.kafka.listener.ContainerProperties;
 import org.springframework.kafka.support.serializer.JsonDeserializer;
+import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
 import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage;
+import ru.nbch.credit_tracker.model.SignalKafkaEvent;
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
 
 @EnableKafka
 @Configuration
 public class KafkaConfig {
 
+    @Bean("signalKafkaEventQueue")
+    public BlockingQueue<SignalKafkaEvent> signalKafkaEventQueue(CreditTrackerSettings settings) {
+        // TODO add capacity
+        return new LinkedBlockingQueue<>();
+    }
+
     @Bean
     public ConcurrentKafkaListenerContainerFactory<String, PackageMessage> kafkaListenerContainerFactory(ConsumerFactory<String, PackageMessage> consumerFactory) {
         ConcurrentKafkaListenerContainerFactory<String, PackageMessage> factory =

src/main/java/ru/nbch/credit_tracker/enums/EventType.java

@@ -0,0 +1,7 @@
+package ru.nbch.credit_tracker.enums;
+
+public enum EventType {
+    WRITE,
+    ROTATE,
+    STOP // используется для остановки
+}

src/main/java/ru/nbch/credit_tracker/model/SignalKafkaEvent.java

@@ -0,0 +1,16 @@
+package ru.nbch.credit_tracker.model;
+
+import lombok.*;
+import ru.nbch.credit_tracker.enums.EventType;
+
+@Getter
+@Setter
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class SignalKafkaEvent {
+    private EventType eventType;
+
+    private Long packageId;
+    // other data for signal
+}

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

@@ -26,9 +26,10 @@ public class CTMonitoringApplication {
         SpringApplication application = new SpringApplication(CTMonitoringApplication.class);
         application.setWebApplicationType(WebApplicationType.NONE);
         ConfigurableApplicationContext ctx = application.run(args);
-        MonitoringService monitoringService = ctx.getBean(MonitoringService.class);
+//        MonitoringService monitoringService = ctx.getBean(MonitoringService.class);
-        monitoringService.run();
+//        monitoringService.run();
 
-//        ctx.getBean(KafkaMonitoringService.class).generateMessages();
+        SignalKafkaEventWriterService signalKafkaEventWriterService = ctx.getBean(SignalKafkaEventWriterService.class);
+        signalKafkaEventWriterService.startWriting();
     }
 }

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

@@ -1,12 +1,19 @@
 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.kafka.core.KafkaTemplate;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.support.Acknowledgment;
-import org.springframework.kafka.support.SendResult;
+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;
 import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage;
+import ru.nbch.credit_tracker.enums.EventType;
+import ru.nbch.credit_tracker.model.SignalKafkaEvent;
 import ru.nbch.credit_tracker.model.UserPackagePhoneData;
 import ru.nbch.credit_tracker.service.nbch.SourceService;
 import ru.nbch.credit_tracker.service.signal.PackageRecordsService;
@@ -15,41 +22,64 @@ import ru.nbch.credit_tracker.utils.CTUtil;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.CompletableFuture;
+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
+@Service
 public class KafkaMonitoringService {
 
-    private final KafkaTemplate<String, String> kafkaTemplate;
+    private final ScheduledExecutorService rotateScheduler = Executors.newSingleThreadScheduledExecutor(
+            new CustomizableThreadFactory("monitoring-rotate-")
+    );
+    private volatile boolean isShuttingdDown = false;
+    private final AtomicInteger activeProcessCount = new AtomicInteger(0);
+
+    private final BlockingQueue<SignalKafkaEvent> signalEventsQueue;
+
     private final PackageRecordsService packageRecordsService;
     private final SourceService sourceService;
 
-    public KafkaMonitoringService(KafkaTemplate<String, String> kafkaTemplate,
+    private final CreditTrackerSettings config;
+
+    public KafkaMonitoringService(@Qualifier("signalKafkaEventQueue") BlockingQueue<SignalKafkaEvent> signalEventsQueue,
                                   PackageRecordsService packageRecordsService,
-                                  SourceService sourceService) {
+                                  SourceService sourceService,
-        this.kafkaTemplate = kafkaTemplate;
+                                  CreditTrackerSettings config) {
+        this.signalEventsQueue = signalEventsQueue;
         this.packageRecordsService = packageRecordsService;
         this.sourceService = sourceService;
+        this.config = config;
     }
 
-//    @KafkaListener(
+    @KafkaListener(
-//            topics = "my-topic",
+            topics = "my-topic",
-//            concurrency = "3"
+            concurrency = "8"
-//    )
+    )
     public void listen(List<PackageMessage> records,
                        Acknowledgment acknowledgment) {
-        process(records);
+        if (isShuttingdDown) {
-
+            log.warn("KafkaMonitoringService is shutting down, rejecting new messages");
-        // insert to db
+            return; // без ака, завершаем обработку сообщений, которые уже попали в очередь, новые не обрабатываем
-        acknowledgment.acknowledge();
+        }
+        try {
+            acknowledgment.acknowledge();
+            activeProcessCount.incrementAndGet();
+            process(records);
+            log.info("Processed {} messages from Kafka", records.size());
+        } finally {
+            activeProcessCount.decrementAndGet();
+        }
     }
 
     private void process(List<PackageMessage> messages) {
         List<Integer> fids = messages.stream().map(PackageMessage::getFid).collect(Collectors.toList());
         List<String> memberCodes = messages.stream()
-                .map(message -> message.getMemberCode().substring(0,6))
+                .map(message -> message.getMemberCode().substring(0, 6))
                 .toList();
 
         List<PhonesMap> phonesMaps = packageRecordsService.findCachedByFids(fids);
@@ -102,30 +132,88 @@ public class KafkaMonitoringService {
                 }
 
                 for (UserPackagePhoneData userPackagePhoneData : userPackagePhoneDataList) {
+                    Long packageId = userPackagePhoneData.getPackageId();
                     String subjectId = userPackagePhoneData.getSubjectId();
                     String signalTypes = userPackagePhoneData.getSignalTypes();
 
                     if (signalTypes.contains(signalType.toString())) {
-                        // TODO put to queue for process
+                        SignalKafkaEvent event = new SignalKafkaEvent();
+                        event.setEventType(EventType.WRITE);
+                        event.setPackageId(packageId);
+                        putToQueue(event);
                     }
                 }
             }
         }
+
+        // TODO insert hits
     }
 
-    private void send(String id, String data) {
+    private void putToQueue(SignalKafkaEvent event) {
-        CompletableFuture<SendResult<String, String>> future =
+        try {
-                kafkaTemplate.send("my-topic", id, data);
+            signalEventsQueue.put(event);
+        } catch (InterruptedException e) {
+            log.error("Error while putting signal event", e);
+            Thread.currentThread().interrupt();
+        }
+    }
 
-        future.whenComplete((result, exception) -> {
+    /**
-            if (exception == null) {
+     * Таймер, который отсчитывает {@code monitoring-interval:5} минут с момента начала накопления сигналов.
-                log.info("Sent user event=[{}] with offset=[{}]",
+     * По истечении таймера отправляет Event на ROTATE файлов
-                        data, result.getRecordMetadata().offset());
+     */
-            } else {
+    @PostConstruct
-                log.error("Unable to send user event=[{}] due to: {}",
+    public void startRotateTimer() {
-                        data, exception.getMessage());
+        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() {
+        log.info("Stopping KafkaMonitoringService...");
+        isShuttingdDown = true;
+
+        while (activeProcessCount.get() > 0) {
+            try {
+                Thread.sleep(1000);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                break;
+            }
+        }
+
+        try {
+            SignalKafkaEvent rotateEvent = new SignalKafkaEvent();
+            rotateEvent.setEventType(EventType.ROTATE);
+            signalEventsQueue.put(rotateEvent);
+            log.info("ROTATE event sent to queue");
+
+            SignalKafkaEvent stopEvent = new SignalKafkaEvent();
+            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();
+        }
+
+        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/SignalKafkaEventWriterService.java

@@ -0,0 +1,67 @@
+package ru.nbch.credit_tracker.service.mode.monitoring;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.stereotype.Service;
+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;
+
+@Slf4j
+@Service
+public class SignalKafkaEventWriterService {
+
+    private final BlockingQueue<SignalKafkaEvent> signalEventsQueue;
+    private final CreditTrackerSettings config;
+
+    public SignalKafkaEventWriterService(@Qualifier("signalKafkaEventQueue") BlockingQueue<SignalKafkaEvent> signalEventsQueue,
+                                         CreditTrackerSettings config) {
+        this.signalEventsQueue = signalEventsQueue;
+        this.config = config;
+    }
+
+    // TODO вызвать в main
+    public void startWriting() {
+        while (true) {
+            SignalKafkaEvent event = take();
+            if (event.getEventType() == EventType.STOP) {
+                break;
+            }
+
+            if (event.getEventType() == EventType.WRITE) {
+                writeSignal(event);
+            }
+
+            if (event.getEventType() == EventType.ROTATE) {
+                rotate();
+            }
+        }
+    }
+
+    /**
+     * Записывает сигнал в xml для конкретного пакета
+     */
+    private void writeSignal(SignalKafkaEvent event) {
+        // TODO записать сигнал в существующий xml/создать новый и записать
+        // TODO добавить пакет в сет
+    }
+
+    /**
+     * Упаковывает все результирующие xml в архив
+     */
+    private void rotate() {
+        // TODO пройтись по сету пакетов и финализировать запись. Очистить сет
+    }
+
+    private SignalKafkaEvent take() {
+        try {
+            return signalEventsQueue.take();
+        } catch (InterruptedException e) {
+            log.error("Error while taking signal event", e);
+            Thread.currentThread().interrupt();
+            throw new RuntimeException(e);
+        }
+    }
+}