test prototype for kafka

pom.xml

@@ -38,6 +38,11 @@
             <artifactId>mybatis-spring-boot-starter</artifactId>
             <version>3.0.4</version>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.kafka</groupId>
+            <artifactId>spring-kafka</artifactId>
+            <version>3.3.15</version>
+        </dependency>
         <dependency>
             <groupId>ru.nbch</groupId>
             <artifactId>pipeline-engine</artifactId>

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

@@ -0,0 +1,71 @@
+package ru.nbch.credit_tracker.config.kafka;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.annotation.EnableKafka;
+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.TestData;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@EnableKafka
+@Configuration
+public class KafkaConfig {
+
+    @Bean
+    public ConcurrentKafkaListenerContainerFactory<String, TestData> kafkaListenerContainerFactory(ConsumerFactory<String, TestData> consumerFactory) {
+        ConcurrentKafkaListenerContainerFactory<String, TestData> factory =
+                new ConcurrentKafkaListenerContainerFactory<>();
+        factory.setConsumerFactory(consumerFactory);
+        factory.setBatchListener(true);
+        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
+
+        return factory;
+    }
+
+    @Bean
+    public ConsumerFactory<String, TestData> consumerFactory() {
+        Map<String, Object> props = new HashMap<>();
+        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:29092");
+        props.put(ConsumerConfig.GROUP_ID_CONFIG, "credit-tracker");
+        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 20);
+
+        JsonDeserializer<TestData> jsonDeserializer = new JsonDeserializer<>(TestData.class);
+        jsonDeserializer.addTrustedPackages("*");
+        jsonDeserializer.setUseTypeHeaders(false);
+
+
+        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
+
+        return new DefaultKafkaConsumerFactory<>(
+                props,
+                new StringDeserializer(),
+                jsonDeserializer
+                );
+    }
+
+    @Bean
+    public ProducerFactory<String, String> producerFactory() {
+        Map<String, Object> props = new HashMap<>();
+        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:29092");
+        props.put(ProducerConfig.LINGER_MS_CONFIG, 10);
+        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        return new DefaultKafkaProducerFactory<>(props);
+    }
+
+    @Bean
+    public KafkaTemplate<String, String> kafkaTemplate(ProducerFactory<String, String> producerFactory) {
+        return new KafkaTemplate<>(producerFactory);
+    }
+
+}

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

@@ -1,5 +1,6 @@
 package ru.nbch.credit_tracker.service.mode.monitoring;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.WebApplicationType;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
@@ -21,11 +22,13 @@ import org.springframework.context.annotation.FilterType;
     }
 )
 public class CTMonitoringApplication {
-    public static void main(String[] args) {
+    public static void main(String[] args) throws JsonProcessingException {
         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();
     }
 }

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

@@ -0,0 +1,94 @@
+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.beans.factory.annotation.Value;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.kafka.support.SendResult;
+import org.springframework.stereotype.Service;
+
+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.UUID;
+import java.util.concurrent.CompletableFuture;
+
+@Slf4j
+@Service
+public class KafkaMonitoringService {
+
+    private final KafkaTemplate<String, String> kafkaTemplate;
+    private final MonitoringTimerService monitoringTimerService;
+
+    @Value("${app.tmp-dir}")
+    private String tmpDir;
+
+    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,
+                       Acknowledgment acknowledgment) {
+        for (ConsumerRecord<String, PackageMessage> record : 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 {
+        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);
+    }
+
+    // TODO send PackageMessage
+    public void generateMessages() throws JsonProcessingException {
+        ObjectMapper mapper = JsonMapper.builder()
+                .build();
+        for (int i = 0; i < 500; i++) {
+            TestData data = new TestData();
+            data.setId(UUID.randomUUID());
+            data.setMessage("Message " + i);
+            send(data.getId().toString(), mapper.writeValueAsString(data));
+        }
+    }
+
+    private void send(String id, String data) {
+        CompletableFuture<SendResult<String, String>> future =
+                kafkaTemplate.send("test-topic", id, data);
+
+        future.whenComplete((result, exception) -> {
+            if (exception == null) {
+                log.info("Sent user event=[{}] with offset=[{}]",
+                        data, result.getRecordMetadata().offset());
+            } else {
+                log.error("Unable to send user event=[{}] due to: {}",
+                        data, exception.getMessage());
+            }
+        });
+    }
+
+}

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

@@ -3,7 +3,6 @@ package ru.nbch.credit_tracker.service.mode.monitoring;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.MDC;
-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.Hit;
@@ -37,9 +36,6 @@ import java.nio.file.Files;
 import java.nio.file.Path;
 import java.time.LocalDateTime;
 import java.util.*;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -72,9 +68,9 @@ public class MonitoringService {
     }
 
     public void run() {
-        ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(new CustomizableThreadFactory("monitoring-"));
+//        ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(new CustomizableThreadFactory("monitoring-"));
-        long period = config.getMonitoringInterval() != null ? config.getMonitoringInterval() : 15;
+//        long period = config.getMonitoringInterval() != null ? config.getMonitoringInterval() : 15;
-        scheduler.scheduleAtFixedRate(this::monitor, 0, period, TimeUnit.MINUTES);
+//        scheduler.scheduleAtFixedRate(this::monitor, 0, period, TimeUnit.MINUTES);
     }
 
     private void monitor() {

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

@@ -0,0 +1,137 @@
+package ru.nbch.credit_tracker.service.mode.monitoring;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+
+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("${app.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 || existing.isIdle()) {
+                PackageBundle bundle = new PackageBundle();
+                bundle.addFile(filePath);
+                scheduleFlush(packageId, bundle);
+                log.info("Started new timer for packageId={}", packageId);
+                return bundle;
+            }
+            existing.addFile(filePath);
+            log.debug("Queued file for packageId={} state={}", packageId, existing.getState());
+            return existing;
+        });
+    }
+
+    private void scheduleFlush(Integer packageId, PackageBundle bundle) {
+        bundle.setState(BundleState.SCHEDULED);
+        scheduler.schedule(() -> flush(packageId, bundle), DELAY_MINUTES, TimeUnit.MINUTES);
+    }
+
+    private void flush(Integer packageId, PackageBundle bundle) {
+        bundle.setState(BundleState.MERGING);
+        List<Path> snapshot = bundle.drainFiles();
+
+        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.setState(BundleState.IDLE);
+            bundles.remove(packageId);
+
+            // Файлы, добавленные во время мержа — забираем и стартуем новый цикл
+            List<Path> leftover = bundle.drainFiles();
+            if (!leftover.isEmpty()) {
+                log.info("Files arrived during merge, restarting timer for packageId={}, count={}",
+                        packageId, leftover.size());
+                PackageBundle next = new PackageBundle();
+                leftover.forEach(next::addFile);
+                bundles.put(packageId, next);
+                scheduleFlush(packageId, next);
+            }
+        }
+    }
+
+    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 enum BundleState {SCHEDULED, MERGING, IDLE}
+
+    private static class PackageBundle {
+        private final Queue<Path> files = new ConcurrentLinkedQueue<>();
+        private volatile BundleState state = BundleState.IDLE;
+
+        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;
+        }
+
+        BundleState getState() {
+            return state;
+        }
+
+        void setState(BundleState s) {
+            this.state = s;
+        }
+
+        boolean isIdle() {
+            return state == BundleState.IDLE;
+        }
+    }
+}

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

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

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

@@ -0,0 +1,17 @@
+package ru.nbch.credit_tracker.service.mode.monitoring;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+import java.util.UUID;
+
+@Getter
+@Setter
+@AllArgsConstructor
+@NoArgsConstructor
+public class TestData {
+    private UUID id;
+    private String message;
+}