Get rid of MonitoringTimerService

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

@@ -11,7 +11,7 @@ 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.service.mode.monitoring.PackageMessage;
+import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage;
 
 import java.util.HashMap;
 import java.util.Map;

src/main/java/ru/nbch/credit_tracker/entities/external_request/kafka/PackageMessage.java

@@ -0,0 +1,15 @@
+package ru.nbch.credit_tracker.entities.external_request.kafka;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+@AllArgsConstructor
+@NoArgsConstructor
+@Getter
+@Setter
+public class PackageMessage {
+    private Integer fid;
+    private String memberCode;
+}

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

@@ -1,22 +1,12 @@
 package ru.nbch.credit_tracker.service.mode.monitoring;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.json.JsonMapper;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.support.SendResult;
+import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage;
 
-import java.io.IOException;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.nio.file.StandardOpenOption;
 import java.util.List;
-import java.util.Random;
-import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 
 @Slf4j
@@ -24,59 +14,28 @@ import java.util.concurrent.CompletableFuture;
 public class KafkaMonitoringService {
 
     private final KafkaTemplate<String, String> kafkaTemplate;
-    private final MonitoringTimerService monitoringTimerService;
 
 //    @Value("${ct.tmp.dir}")
     private String tmpDir;
 
-    public KafkaMonitoringService(KafkaTemplate<String, String> kafkaTemplate,
+    public KafkaMonitoringService(KafkaTemplate<String, String> kafkaTemplate) {
-                                  MonitoringTimerService monitoringTimerService) {
         this.kafkaTemplate = kafkaTemplate;
-        this.monitoringTimerService = monitoringTimerService;
     }
 
 //    @KafkaListener(
 //            topics = "my-topic",
 //            concurrency = "3"
 //    )
-    public void listen(List<ConsumerRecord<String, PackageMessage>> records,
+    public void listen(List<PackageMessage> records,
                        Acknowledgment acknowledgment) {
-        for (ConsumerRecord<String, PackageMessage> record : records) {
+        process(records);
-            try {
+
-                processRecord(record.value());
-            } catch (Exception e) {
-                log.error("Failed to process record: {}", record, e);
-            }
-        }
         // insert to db
         acknowledgment.acknowledge();
     }
 
-    private void processRecord(PackageMessage msg) throws IOException {
+    private void process(List<PackageMessage> messages) {
-        Path packageDir = Paths.get(tmpDir, String.valueOf(msg.packageId()));
-        Files.createDirectories(packageDir);
 
-        Path filePath = packageDir.resolve(msg.uuid() + ".xml");
-        Files.writeString(filePath, msg.message(), StandardOpenOption.CREATE_NEW);
-
-        monitoringTimerService.registerFile(msg.packageId(), filePath);
-    }
-
-    public void generateMessages() throws JsonProcessingException {
-        ObjectMapper mapper = JsonMapper.builder()
-                .build();
-        Random rand = new Random();
-        int max = 10;
-        int min = 1;
-        for (int i = 0; i < 500; i++) {
-            PackageMessage data = new PackageMessage(
-                    rand.nextInt((max - min) + 1) + min,
-                    "Message " + i,
-                    UUID.randomUUID()
-            );
-
-            send(data.uuid().toString(), mapper.writeValueAsString(data));
-        }
     }
 
     private void send(String id, String data) {

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

@@ -1,7 +1,5 @@
 package ru.nbch.credit_tracker.service.mode.monitoring;
 
-import lombok.Getter;
-import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.MDC;
@@ -11,6 +9,7 @@ import ru.nbch.credit_tracker.entities.app.Hit;
 import ru.nbch.credit_tracker.entities.app.PackagesRecordsMap;
 import ru.nbch.credit_tracker.entities.app.PhonesMap;
 import ru.nbch.credit_tracker.entities.app.UserPackage;
+import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage;
 import ru.nbch.credit_tracker.entities.external_response.online_download.Inquiry;
 import ru.nbch.credit_tracker.entities.external_response.online_download.Person;
 import ru.nbch.credit_tracker.entities.external_response.online_download.Persons;
@@ -75,30 +74,15 @@ public class MonitoringService {
     }
 
     public void run() {
-        List<MessageData> messages = new ArrayList<>();
+        List<PackageMessage> messages = new ArrayList<>();
 
         for (int i = 1000; i < 2000; i++) {
-            MessageData messageData = new MessageData();
+            PackageMessage messageData = new PackageMessage();
             messageData.setFid(i);
             messageData.setMemberCode("ZX01LL");
             messages.add(messageData);
         }
 
-        MessageData messageData1 = new MessageData();
-        messageData1.setFid(114885613);
-        messageData1.setMemberCode("ZX01LL");
-        messages.add(messageData1);
-
-        MessageData messageData2 = new MessageData();
-        messageData2.setFid(16445000);
-        messageData2.setMemberCode("ZX01LL");
-        messages.add(messageData2);
-
-        MessageData messageData3 = new MessageData();
-        messageData3.setFid(109837221);
-        messageData3.setMemberCode("ZX01LL");
-        messages.add(messageData3);
-
 
         long start = System.currentTimeMillis();
         log.info("Start selecting by one");
@@ -111,8 +95,8 @@ public class MonitoringService {
         log.info("Finish selecting by bacth. Time: {} ms", System.currentTimeMillis() - start);
     }
 
-    private void monitorByOne(List<MessageData> messages) {
+    private void monitorByOne(List<PackageMessage> messages) {
-        for (MessageData message : messages) {
+        for (PackageMessage message : messages) {
             Integer fid = message.getFid();
             List<PhonesMap> phonesMaps = packageRecordsService.findCachedByFid(fid);
             for (PhonesMap phonesMap : phonesMaps) {
@@ -136,8 +120,8 @@ public class MonitoringService {
         }
     }
 
-    private void monitorbyBatch(List<MessageData> messages) {
+    private void monitorbyBatch(List<PackageMessage> messages) {
-        List<Integer> fids = messages.stream().map(MessageData::getFid).collect(Collectors.toList());
+        List<Integer> fids = messages.stream().map(PackageMessage::getFid).collect(Collectors.toList());
         List<String> memberCodes = messages.stream()
                 .map(message -> message.getMemberCode().substring(0,6))
                 .toList();
@@ -161,7 +145,7 @@ public class MonitoringService {
         Map<Long, List<UserPackagePhoneData>> userPackagePhoneDataMap = packagePhoneData.stream()
                 .collect(Collectors.groupingBy(UserPackagePhoneData::getPhone));
 
-        for (MessageData message : messages) {
+        for (PackageMessage message : messages) {
             Integer fid = message.getFid();
             String memberCode = message.getMemberCode();
 
@@ -194,14 +178,6 @@ public class MonitoringService {
         }
     }
 
-    @Getter
-    @Setter
-    public static class MessageData {
-        private Integer fid;
-        private String memberCode;
-        // other data
-    }
-
     private void monitor() {
         try {
             log.info("Start monitoring");

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

@@ -1,112 +0,0 @@
-package ru.nbch.credit_tracker.service.mode.monitoring;
-
-import lombok.extern.slf4j.Slf4j;
-
-import java.io.BufferedWriter;
-import java.io.IOException;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.nio.file.StandardCopyOption;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Queue;
-import java.util.concurrent.*;
-
-//@Service
-@Slf4j
-public class MonitoringTimerService {
-
-    private static final long DELAY_MINUTES = 5;
-
-//    @Value("${ct.target.dir}")
-    private String targetDir;
-
-    private final ScheduledExecutorService scheduler =
-            Executors.newScheduledThreadPool(4);
-
-    private final ConcurrentHashMap<Integer, PackageBundle> bundles =
-            new ConcurrentHashMap<>();
-
-    public void registerFile(Integer packageId, Path filePath) {
-        bundles.compute(packageId, (id, existing) -> {
-            if (existing == null) {
-                PackageBundle bundle = new PackageBundle();
-                bundle.addFile(filePath);
-                scheduleFlush(packageId, bundle);
-                log.info("Started new timer for packageId={}", packageId);
-                return bundle;
-            }
-            existing.addFile(filePath);
-            return existing;
-        });
-    }
-
-    private void scheduleFlush(Integer packageId, PackageBundle bundle) {
-        scheduler.schedule(() -> flush(packageId, bundle), DELAY_MINUTES, TimeUnit.MINUTES);
-    }
-
-    private void flush(Integer packageId, PackageBundle bundle) {
-        // Атомарно забираем snapshot и удаляем из мапы
-        List<Path> snapshot = new ArrayList<>();
-        bundles.compute(packageId, (id, current) -> {
-            snapshot.addAll(bundle.drainFiles());
-            return null; // удаляем этот пакет
-        });
-
-        try {
-            if (!snapshot.isEmpty()) {
-                Path destDir = Paths.get(targetDir, String.valueOf(packageId));
-                Files.createDirectories(destDir);
-
-                List<Path> movedFiles = new ArrayList<>();
-                for (Path src : snapshot) {
-                    if (!Files.exists(src)) {
-                        log.warn("File missing, skipping: {}", src);
-                        continue;
-                    }
-                    Path dest = destDir.resolve(src.getFileName());
-                    Files.move(src, dest, StandardCopyOption.ATOMIC_MOVE);
-                    movedFiles.add(dest);
-                }
-                mergeXmlFiles(movedFiles, destDir, packageId);
-            }
-        } catch (IOException e) {
-            log.error("Flush failed for packageId={}", packageId, e);
-        } finally {
-            // Файлы, добавленные во время мержа — забираем и стартуем новый цикл
-            bundle.drainFiles().forEach(leftover -> registerFile(packageId, leftover));
-        }
-    }
-
-    private void mergeXmlFiles(List<Path> files, Path destDir, Integer packageId) throws IOException {
-        if (files.isEmpty()) return;
-        Path merged = destDir.resolve("merged_" + packageId + ".xml");
-        try (BufferedWriter writer = Files.newBufferedWriter(merged)) {
-            writer.write("<package id=\"" + packageId + "\">");
-            writer.newLine();
-            for (Path file : files) {
-                writer.write(Files.readString(file));
-                writer.newLine();
-            }
-            writer.write("</package>");
-        }
-        log.info("Merged {} files → {} for packageId={}", files.size(), merged, packageId);
-    }
-
-
-    private static class PackageBundle {
-        private final Queue<Path> files = new ConcurrentLinkedQueue<>();
-
-        void addFile(Path path) {
-            files.add(path);
-        }
-
-        List<Path> drainFiles() {
-            List<Path> result = new ArrayList<>();
-            Path p;
-            while ((p = files.poll()) != null) result.add(p);
-            return result;
-        }
-    }
-}

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

@@ -1,9 +0,0 @@
-package ru.nbch.credit_tracker.service.mode.monitoring;
-
-import java.util.UUID;
-
-public record PackageMessage(
-        Integer packageId,
-        String message,
-        UUID uuid
-) {}