7 changed files with 121 additions and 137 deletions
--- a/src/main/java/ru/nbch/credit_tracker/config/kafka/KafkaConfig.java
+++ b/src/main/java/ru/nbch/credit_tracker/config/kafka/KafkaConfig.java
@ -1,7 +1,5 @@
 package ru.nbch.credit_tracker.config.kafka;

-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.springframework.context.annotation.Bean;
@ -37,13 +35,12 @@ public class KafkaConfig {
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setBatchListener(true);
-        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
+        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);

        return factory;
    }

    @Bean
-    //todo надо проверить конфигурацию внимательно
    public ConsumerFactory<String, PackageMessage> consumerFactory(CreditTrackerSettings settings) {
        KafkaConsumerSettings kafkaSettings = settings.getKafkaConsumerSettings();
        Map<String, Object> props = new HashMap<>();
@ -53,12 +50,7 @@ public class KafkaConfig {
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, kafkaSettings.maxPollRecords());
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, kafkaSettings.maxPollIntervalMs());

-        ObjectMapper objectMapper = new ObjectMapper();
-        objectMapper.registerModule(new JavaTimeModule());
-
-        JsonDeserializer<PackageMessage> jsonDeserializer = new JsonDeserializer<>(
-                PackageMessage.class,
-                objectMapper);
+        JsonDeserializer<PackageMessage> jsonDeserializer = new JsonDeserializer<>(PackageMessage.class);
        jsonDeserializer.addTrustedPackages("*");
        jsonDeserializer.setUseTypeHeaders(false);

--- a/src/main/java/ru/nbch/credit_tracker/config/task/ExecutorsConfig.java
+++ b/src/main/java/ru/nbch/credit_tracker/config/task/ExecutorsConfig.java
@ -41,7 +41,7 @@ public class ExecutorsConfig {

    @Bean("lpSaverExecutorService")
    public ExecutorService lpSaverExecutorService(CreditTrackerSettings creditTrackerSettings) {
-        return Executors.newFixedThreadPool(creditTrackerSettings.getLpPipeline().parallelRequests(), r -> {
+        return Executors.newFixedThreadPool(creditTrackerSettings.getNpPipeline().parallelRequests(), r -> {
            Thread t = new Thread(r);
            t.setName("lp-subject-saver-%d".formatted(t.threadId()));
            return t;
--- a/src/main/java/ru/nbch/credit_tracker/entities/external_request/kafka/InquiryDateDeserializer.java
+++ b/src/main/java/ru/nbch/credit_tracker/entities/external_request/kafka/InquiryDateDeserializer.java
@ -0,0 +1,30 @@
+package ru.nbch.credit_tracker.entities.external_request.kafka;
+
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonToken;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+
+import java.io.IOException;
+import java.time.LocalDateTime;
+import java.util.List;
+
+public class InquiryDateDeserializer extends JsonDeserializer<LocalDateTime> {
+    @Override
+    public LocalDateTime deserialize(JsonParser p, DeserializationContext ctxt) throws IOException {
+        if (p.currentToken() == JsonToken.START_ARRAY) {
+            List<Integer> dateParts = p.readValueAs(List.class);
+            if (dateParts.size() >= 6) {
+                int year = dateParts.get(0);
+                int month = dateParts.get(1);
+                int day = dateParts.get(2);
+                int hour = dateParts.get(3);
+                int minute = dateParts.get(4);
+                int second = dateParts.get(5);
+                int nano = dateParts.size() > 6 ? dateParts.get(6) : 0;
+                return LocalDateTime.of(year, month, day, hour, minute, second, nano);
+            }
+        }
+        return null;
+    }
+}
--- a/src/main/java/ru/nbch/credit_tracker/entities/external_request/kafka/PackageMessage.java
+++ b/src/main/java/ru/nbch/credit_tracker/entities/external_request/kafka/PackageMessage.java
@ -1,7 +1,6 @@
 package ru.nbch.credit_tracker.entities.external_request.kafka;

-import com.fasterxml.jackson.annotation.JsonFormat;
-import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
@ -14,11 +13,10 @@ import java.util.List;
@NoArgsConstructor
@Getter
@Setter
-@JsonIgnoreProperties(ignoreUnknown = true)
 public class PackageMessage {
    private Integer fid;
    private String memberCode;
-    @JsonFormat(shape = JsonFormat.Shape.ARRAY)
+    @JsonDeserialize(using = InquiryDateDeserializer.class)
    private LocalDateTime inquiryDate;
    private String inquiryPurpose;
    private Long inquiryAmount;
@ -26,6 +24,10 @@ public class PackageMessage {
    private Integer inquiryType;
    private List<Person> persons;

+    public boolean isNaturalPerson() {
+        return persons != null && !persons.isEmpty();
+    }
+
    @AllArgsConstructor
    @NoArgsConstructor
    @Getter
--- a/src/main/java/ru/nbch/credit_tracker/service/mode/common/StartupRestoreService.java
+++ b/src/main/java/ru/nbch/credit_tracker/service/mode/common/StartupRestoreService.java
@ -81,7 +81,7 @@ public class StartupRestoreService {
                } catch (Exception e) {
                    log.error("Error build default np pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e));
                }
-            } else if (userPackage.getSubjectType() == 2) {
+            } else {
                try {
                    Long count = legalsFidMapMapper.deleteByPackageId(userPackage.getId());
                    log.trace("Deleted {} lp rows", count);
@ -89,7 +89,7 @@ public class StartupRestoreService {
                    log.debug("Try to read and build lp pipeline for packageId {}; req xml path: {}", userPackage.getId(), xmlReqPath);
                    Pipeline pipeline = lpPipelinePackageManager.build(new SetupParam(userPackage.getId(), 2, PipelineMode.DEFAULT));
                    lpPipelineOrchestrator.startOrDelay(
-                            () -> saver.readAndSaveLpPackage(userPackage.getId(), xmlReqPath),
+                            () -> saver.readAndSaveNpPackage(userPackage.getId(), xmlReqPath),
                            pipeline, userPackage.getId()
                    );
                } catch (Exception e) {
@ -111,7 +111,7 @@ public class StartupRestoreService {
                } catch (Exception e) {
                    log.error("Error build restore np pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e));
                }
-            } else if (userPackage.getSubjectType() == 2) {
+            } else {
                try {
                    log.debug("Build lp pipeline for packageId {}", userPackage.getId());
                    Pipeline pipeline = lpPipelinePackageManager.build(new SetupParam(userPackage.getId(), 2, PipelineMode.DEFAULT));
--- a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java
+++ b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java
@ -105,7 +105,6 @@ public class KafkaMonitoringService {

    private void process(List<PackageMessage> messages) {
        Map<Boolean, List<PackageMessage>> partition = messages.stream()
-                .filter(Objects::nonNull)
                .filter(this::filterMessages)
                .collect(Collectors.partitioningBy(m -> m.getPersons() != null && !m.getPersons().isEmpty()));

@ -199,6 +198,10 @@ public class KafkaMonitoringService {
            }

            for (PhonesMap phonesMap : phonesMapList) {
+                // флаги просрочек
+                Integer flagDebt5000 = phonesMap.getFlagDebt5000();
+                Integer flag90days = phonesMap.getFlag90days();
+
                // в userPackagePhoneDataMap только нужные номера, те что не вошли в userPackagePhoneDataList игнорируются
                List<UserPackagePhoneData> userPackagePhoneDataList = userPackagePhoneDataMap.get(phonesMap.getPhone());

@ -207,30 +210,44 @@ public class KafkaMonitoringService {
                }

                for (UserPackagePhoneData userPackagePhoneData : userPackagePhoneDataList) {
+                    String subjectId = userPackagePhoneData.getSubjectId();
                    String signalTypes = userPackagePhoneData.getSignalTypes();

                    if (signalTypes.contains(Integer.toString(signalType))) {
+                        SignalKafkaEvent event = new SignalKafkaEvent();
+                        event.setEventType(EventType.WRITE);
+                        event.setNatural(true);
+                        event.setUserPackagePhoneData(userPackagePhoneData);
+                        event.setPackageMessage(message);
+                        event.setSubjectId(subjectId);
+                        event.setFlagDebt5000(flagDebt5000);
+                        event.setFlag90days(flag90days);
+                        event.setSignalType(Integer.toString(signalType));
+                        event.setBusCategory(signalType == 3 ? busCategory : null);
+
                        Integer segment = segmentMap.get(memberCode.substring(0, 6));
-                        SignalKafkaEvent event = buildSignalKafkaEvent(
-                                true, message,
-                                phonesMap, userPackagePhoneData,
-                                null, null, // legal person fields
-                                signalType, busCategory, segment
-                        );
+                        event.setSegment(segment != null ? segment : -1);

                        String reportKey = userPackagePhoneData.getUserId().substring(0, 6)
                                + "_" + userPackagePhoneData.getPackageName();
-                        String reportFileName = reportNameRegistry.resolveNameAndEnqueue(reportKey, event);
+                        String reportFileName = resolveNameAndEnqueue(reportKey, event);
                        if (reportFileName == null) {
                            continue; // событие не встало в очередь — hit не пишем
                        }
-                        Hit hit = buildHit(
-                                true, message,
-                                userPackagePhoneData,
-                                null,
-                                signalType, busCategory,
-                                segment, reportFileName
-                        );
+                        Hit hit = Hit.builder()
+                                .packageId(userPackagePhoneData.getPackageId())
+                                .subjectId(subjectId)
+                                .signalType(signalType)
+                                .reportFileName(reportFileName.replace(".xml", ".zip"))
+                                .inqT(message.getInquiryType())
+                                .busCategory(busCategory)
+                                .inqDt(message.getInquiryDate())
+                                .inqP(message.getInquiryPurpose())
+                                .inqA(message.getInquiryAmount())
+                                .segment(segment != null ? segment : -1)
+                                .recordedAt(LocalDateTime.now())
+                                .phone(userPackagePhoneData.getPhone())
+                                .build();
                        hits.add(hit);
                    }
                }
@ -286,6 +303,10 @@ public class KafkaMonitoringService {
            }

            for (LegalsMap legalsMap : legalsMapList) {
+                // флаги просрочек
+                Integer flag90days = legalsMap.getFlag90days();
+                LocalDateTime lastRejectedDate = legalsMap.getLastRejectedDate();
+
                // в userPackageLegalDataMap только нужные inn, regnum, те что не вошли в userPackageLegalDataList игнорируются
                List<UserPackageLegalData> userPackageLegalDataList = userPackageLegalDataMap.get(Pair.of(legalsMap.getInn(), legalsMap.getRegNum()));

@ -294,32 +315,45 @@ public class KafkaMonitoringService {
                }

                for (UserPackageLegalData userPackageLegalData : userPackageLegalDataList) {
+                    String subjectId = userPackageLegalData.getSubjectId();
                    String signalTypes = userPackageLegalData.getSignalTypes();

                    if (signalTypes.contains(Integer.toString(signalType))) {
-                        Integer segment = segmentMap.get(memberCode.substring(0, 6));
+                        SignalKafkaEvent event = new SignalKafkaEvent();
+                        event.setEventType(EventType.WRITE);
+                        event.setNatural(false);
+                        event.setUserPackageLegalData(userPackageLegalData);
+                        event.setPackageMessage(message);
+                        event.setSubjectId(subjectId);
+                        event.setFlag90days(flag90days);
+                        event.setSignalType(Integer.toString(signalType));
+                        event.setBusCategory(signalType == 3 ? busCategory : null);
+                        event.setLastRejectedDate(lastRejectedDate);

-                        SignalKafkaEvent event = buildSignalKafkaEvent(
-                                false, message,
-                                null, null, // np fields
-                                legalsMap,
-                                userPackageLegalData,
-                                signalType, busCategory, segment
-                        );
+                        Integer segment = segmentMap.get(memberCode.substring(0, 6));
+                        event.setSegment(segment != null ? segment : -1);

                        String reportKey = userPackageLegalData.getUserId().substring(0, 6)
                                + "_" + userPackageLegalData.getPackageName();
-                        String reportFileName = reportNameRegistry.resolveNameAndEnqueue(reportKey, event);
+                        String reportFileName = resolveNameAndEnqueue(reportKey, event);
                        if (reportFileName == null) {
                            continue; // событие не встало в очередь — hit не пишем
                        }
-                        Hit hit = buildHit(
-                                false, message,
-                                null,
-                                userPackageLegalData,
-                                signalType, busCategory,
-                                segment, reportFileName
-                        );
+                        Hit hit = Hit.builder()
+                                .packageId(userPackageLegalData.getPackageId())
+                                .subjectId(subjectId)
+                                .signalType(signalType)
+                                .reportFileName(reportFileName.replace(".xml", ".zip"))
+                                .inqT(message.getInquiryType())
+                                .busCategory(busCategory)
+                                .inqDt(message.getInquiryDate())
+                                .inqP(message.getInquiryPurpose())
+                                .inqA(message.getInquiryAmount())
+                                .segment(segment != null ? segment : -1)
+                                .recordedAt(LocalDateTime.now())
+                                .inn(userPackageLegalData.getInn())
+                                .regNum(userPackageLegalData.getRegNum())
+                                .build();
                        hits.add(hit);
                    }
                }
@ -336,86 +370,18 @@ public class KafkaMonitoringService {
                (inquiryType == 3 && billingCode.equalsIgnoreCase("CHSP"));
    }

-    private SignalKafkaEvent buildSignalKafkaEvent(boolean isNatural, PackageMessage message,
-                                                   PhonesMap phonesMap, UserPackagePhoneData userPackagePhoneData,
-                                                   LegalsMap legalsMap, UserPackageLegalData userPackageLegalData,
-                                                   int signalType, String busCategory, Integer segment) {
-        if (isNatural) {
-            // флаги просрочек
-            Integer flagDebt5000 = phonesMap.getFlagDebt5000();
-            Integer flag90days = phonesMap.getFlag90days();
-
-            String subjectId = userPackagePhoneData.getSubjectId();
-
-            return SignalKafkaEvent.builder()
-                    .eventType(EventType.WRITE)
-                    .natural(true)
-                    .userPackagePhoneData(userPackagePhoneData)
-                    .packageMessage(message)
-                    .subjectId(subjectId)
-                    .flagDebt5000(flagDebt5000)
-                    .flag90days(flag90days)
-                    .signalType(Integer.toString(signalType))
-                    .busCategory(signalType == 3 ? busCategory : null)
-                    .segment(segment != null ? segment : -1)
-                    .build();
-        } else {
-            // флаги просрочек
-            Integer flag90days = legalsMap.getFlag90days();
-            LocalDateTime lastRejectedDate = legalsMap.getLastRejectedDate();
-
-            String subjectId = userPackageLegalData.getSubjectId();
-            return SignalKafkaEvent.builder()
-                    .eventType(EventType.WRITE)
-                    .natural(false)
-                    .userPackageLegalData(userPackageLegalData)
-                    .packageMessage(message)
-                    .subjectId(subjectId)
-                    .flag90days(flag90days)
-                    .signalType(Integer.toString(signalType))
-                    .busCategory(signalType == 3 ? busCategory : null)
-                    .lastRejectedDate(lastRejectedDate)
-                    .segment(segment != null ? segment : -1)
-                    .build();
-        }
-    }
-
-    private Hit buildHit(boolean isNatural, PackageMessage message,
-                         UserPackagePhoneData userPackagePhoneData,
-                         UserPackageLegalData userPackageLegalData,
-                         int signalType, String busCategory,
-                         Integer segment, String reportFileName) {
-        if (isNatural) {
-            return Hit.builder()
-                    .packageId(userPackagePhoneData.getPackageId())
-                    .subjectId(userPackagePhoneData.getSubjectId())
-                    .signalType(signalType)
-                    .reportFileName(reportFileName.replace(".xml", ".zip"))
-                    .inqT(message.getInquiryType())
-                    .busCategory(busCategory)
-                    .inqDt(message.getInquiryDate())
-                    .inqP(message.getInquiryPurpose())
-                    .inqA(message.getInquiryAmount())
-                    .segment(segment != null ? segment : -1)
-                    .recordedAt(LocalDateTime.now())
-                    .phone(userPackagePhoneData.getPhone())
-                    .build();
-        } else {
-           return Hit.builder()
-                    .packageId(userPackageLegalData.getPackageId())
-                    .subjectId(userPackageLegalData.getSubjectId())
-                    .signalType(signalType)
-                    .reportFileName(reportFileName.replace(".xml", ".zip"))
-                    .inqT(message.getInquiryType())
-                    .busCategory(busCategory)
-                    .inqDt(message.getInquiryDate())
-                    .inqP(message.getInquiryPurpose())
-                    .inqA(message.getInquiryAmount())
-                    .segment(segment != null ? segment : -1)
-                    .recordedAt(LocalDateTime.now())
-                    .inn(userPackageLegalData.getInn())
-                    .regNum(userPackageLegalData.getRegNum())
-                    .build();
+    /**
+     * Резолвит имя результирующего файла и ставит событие в очередь.
+     *
+     * @return имя xml-файла или null, если поток был прерван и событие не встало в очередь
+     */
+    private String resolveNameAndEnqueue(String reportKey, SignalKafkaEvent event) {
+        try {
+            return reportNameRegistry.resolveNameAndEnqueue(reportKey, event);
+        } catch (InterruptedException e) {
+            log.error("Error while putting signal event", e);
+            Thread.currentThread().interrupt();
+            return null;
        }
    }

--- a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/ReportNameRegistry.java
+++ b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/ReportNameRegistry.java
@ -1,6 +1,5 @@
 package ru.nbch.credit_tracker.service.mode.monitoring;

-import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.stereotype.Component;
 import ru.nbch.credit_tracker.enums.EventType;
@ -22,7 +21,6 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
 * закрывающий эпоху N. Поэтому WRITE-события кладутся в очередь только через
 * {@link #resolveNameAndEnqueue}, а ROTATE — только через {@link #enqueueRotate}.
 */
-@Slf4j
@Component
 public class ReportNameRegistry {

@ -42,7 +40,7 @@ public class ReportNameRegistry {
     * @param key membercode_packageName
     * @return имя xml-файла, в который попадёт сигнал
     */
-    public String resolveNameAndEnqueue(String key, SignalKafkaEvent event) {
+    public String resolveNameAndEnqueue(String key, SignalKafkaEvent event) throws InterruptedException {
        lock.readLock().lock();
        try {
            String name = names.computeIfAbsent(key, k ->
@ -50,10 +48,6 @@ public class ReportNameRegistry {
                            LocalDateTime.now().format(CTUtil.responseReportDateTimeFormatter)));
            signalEventsQueue.put(event);
            return name;
-        } catch (InterruptedException e) {
-            log.error("Error while putting signal event", e);
-            Thread.currentThread().interrupt();
-            return null;
        } finally {
            lock.readLock().unlock();
        }