Get rid of InquiryDateDeserializer; Changed ack mode

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

@@ -1,5 +1,7 @@
 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;
@@ -35,12 +37,13 @@ public class KafkaConfig {
                 new ConcurrentKafkaListenerContainerFactory<>();
         factory.setConsumerFactory(consumerFactory);
         factory.setBatchListener(true);
-        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
+        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
 
         return factory;
     }
 
     @Bean
+    //todo надо проверить конфигурацию внимательно
     public ConsumerFactory<String, PackageMessage> consumerFactory(CreditTrackerSettings settings) {
         KafkaConsumerSettings kafkaSettings = settings.getKafkaConsumerSettings();
         Map<String, Object> props = new HashMap<>();
@@ -50,7 +53,12 @@ public class KafkaConfig {
         props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, kafkaSettings.maxPollRecords());
         props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, kafkaSettings.maxPollIntervalMs());
 
-        JsonDeserializer<PackageMessage> jsonDeserializer = new JsonDeserializer<>(PackageMessage.class);
+        ObjectMapper objectMapper = new ObjectMapper();
+        objectMapper.registerModule(new JavaTimeModule());
+
+        JsonDeserializer<PackageMessage> jsonDeserializer = new JsonDeserializer<>(
+                PackageMessage.class,
+                objectMapper);
         jsonDeserializer.addTrustedPackages("*");
         jsonDeserializer.setUseTypeHeaders(false);
 
@@ -61,7 +69,7 @@ public class KafkaConfig {
                 props,
                 new StringDeserializer(),
                 jsonDeserializer
-                );
+        );
     }
 
 }

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.getNpPipeline().parallelRequests(), r -> {
+        return Executors.newFixedThreadPool(creditTrackerSettings.getLpPipeline().parallelRequests(), r -> {
             Thread t = new Thread(r);
             t.setName("lp-subject-saver-%d".formatted(t.threadId()));
             return t;

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

@@ -1,30 +0,0 @@
-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;
-    }
-}

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

@@ -1,6 +1,7 @@
 package ru.nbch.credit_tracker.entities.external_request.kafka;
 
-import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.annotation.JsonFormat;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
@@ -13,10 +14,11 @@ import java.util.List;
 @NoArgsConstructor
 @Getter
 @Setter
+@JsonIgnoreProperties(ignoreUnknown = true)
 public class PackageMessage {
     private Integer fid;
     private String memberCode;
-    @JsonDeserialize(using = InquiryDateDeserializer.class)
+    @JsonFormat(shape = JsonFormat.Shape.ARRAY)
     private LocalDateTime inquiryDate;
     private String inquiryPurpose;
     private Long inquiryAmount;
@@ -24,10 +26,6 @@ public class PackageMessage {
     private Integer inquiryType;
     private List<Person> persons;
 
-    public boolean isNaturalPerson() {
-        return persons != null && !persons.isEmpty();
-    }
-
     @AllArgsConstructor
     @NoArgsConstructor
     @Getter

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 {
+            } else if (userPackage.getSubjectType() == 2) {
                 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.readAndSaveNpPackage(userPackage.getId(), xmlReqPath),
+                            () -> saver.readAndSaveLpPackage(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 {
+            } else if (userPackage.getSubjectType() == 2) {
                 try {
                     log.debug("Build lp pipeline for packageId {}", userPackage.getId());
                     Pipeline pipeline = lpPipelinePackageManager.build(new SetupParam(userPackage.getId(), 2, PipelineMode.DEFAULT));

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

@@ -105,6 +105,7 @@ 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()));
 
@@ -198,10 +199,6 @@ public class KafkaMonitoringService {
             }
 
             for (PhonesMap phonesMap : phonesMapList) {
-                // флаги просрочек
-                Integer flagDebt5000 = phonesMap.getFlagDebt5000();
-                Integer flag90days = phonesMap.getFlag90days();
-
                 // в userPackagePhoneDataMap только нужные номера, те что не вошли в userPackagePhoneDataList игнорируются
                 List<UserPackagePhoneData> userPackagePhoneDataList = userPackagePhoneDataMap.get(phonesMap.getPhone());
 
@@ -210,44 +207,30 @@ 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));
-                        event.setSegment(segment != null ? segment : -1);
+                        SignalKafkaEvent event = buildSignalKafkaEvent(
+                                true, message,
+                                phonesMap, userPackagePhoneData,
+                                null, null, // legal person fields
+                                signalType, busCategory, segment
+                        );
 
                         String reportKey = userPackagePhoneData.getUserId().substring(0, 6)
                                 + "_" + userPackagePhoneData.getPackageName();
-                        String reportFileName = resolveNameAndEnqueue(reportKey, event);
+                        String reportFileName = reportNameRegistry.resolveNameAndEnqueue(reportKey, event);
                         if (reportFileName == null) {
                             continue; // событие не встало в очередь — hit не пишем
                         }
-                        Hit hit = Hit.builder()
+                        Hit hit = buildHit(
-                                .packageId(userPackagePhoneData.getPackageId())
+                                true, message,
-                                .subjectId(subjectId)
+                                userPackagePhoneData,
-                                .signalType(signalType)
+                                null,
-                                .reportFileName(reportFileName.replace(".xml", ".zip"))
+                                signalType, busCategory,
-                                .inqT(message.getInquiryType())
+                                segment, reportFileName
-                                .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);
                     }
                 }
@@ -303,10 +286,6 @@ 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()));
 
@@ -315,45 +294,32 @@ public class KafkaMonitoringService {
                 }
 
                 for (UserPackageLegalData userPackageLegalData : userPackageLegalDataList) {
-                    String subjectId = userPackageLegalData.getSubjectId();
                     String signalTypes = userPackageLegalData.getSignalTypes();
 
                     if (signalTypes.contains(Integer.toString(signalType))) {
-                        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);
-
                         Integer segment = segmentMap.get(memberCode.substring(0, 6));
-                        event.setSegment(segment != null ? segment : -1);
+
+                        SignalKafkaEvent event = buildSignalKafkaEvent(
+                                false, message,
+                                null, null, // np fields
+                                legalsMap,
+                                userPackageLegalData,
+                                signalType, busCategory, segment
+                        );
 
                         String reportKey = userPackageLegalData.getUserId().substring(0, 6)
                                 + "_" + userPackageLegalData.getPackageName();
-                        String reportFileName = resolveNameAndEnqueue(reportKey, event);
+                        String reportFileName = reportNameRegistry.resolveNameAndEnqueue(reportKey, event);
                         if (reportFileName == null) {
                             continue; // событие не встало в очередь — hit не пишем
                         }
-                        Hit hit = Hit.builder()
+                        Hit hit = buildHit(
-                                .packageId(userPackageLegalData.getPackageId())
+                                false, message,
-                                .subjectId(subjectId)
+                                null,
-                                .signalType(signalType)
+                                userPackageLegalData,
-                                .reportFileName(reportFileName.replace(".xml", ".zip"))
+                                signalType, busCategory,
-                                .inqT(message.getInquiryType())
+                                segment, reportFileName
-                                .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);
                     }
                 }
@@ -370,18 +336,86 @@ public class KafkaMonitoringService {
                 (inquiryType == 3 && billingCode.equalsIgnoreCase("CHSP"));
     }
 
-    /**
+    private SignalKafkaEvent buildSignalKafkaEvent(boolean isNatural, PackageMessage message,
-     * Резолвит имя результирующего файла и ставит событие в очередь.
+                                                   PhonesMap phonesMap, UserPackagePhoneData userPackagePhoneData,
-     *
+                                                   LegalsMap legalsMap, UserPackageLegalData userPackageLegalData,
-     * @return имя xml-файла или null, если поток был прерван и событие не встало в очередь
+                                                   int signalType, String busCategory, Integer segment) {
-     */
+        if (isNatural) {
-    private String resolveNameAndEnqueue(String reportKey, SignalKafkaEvent event) {
+            // флаги просрочек
-        try {
+            Integer flagDebt5000 = phonesMap.getFlagDebt5000();
-            return reportNameRegistry.resolveNameAndEnqueue(reportKey, event);
+            Integer flag90days = phonesMap.getFlag90days();
-        } catch (InterruptedException e) {
+
-            log.error("Error while putting signal event", e);
+            String subjectId = userPackagePhoneData.getSubjectId();
-            Thread.currentThread().interrupt();
+
-            return null;
+            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();
         }
     }
 

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

@@ -1,5 +1,6 @@
 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;
@@ -21,6 +22,7 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
  * закрывающий эпоху N. Поэтому WRITE-события кладутся в очередь только через
  * {@link #resolveNameAndEnqueue}, а ROTATE — только через {@link #enqueueRotate}.
  */
+@Slf4j
 @Component
 public class ReportNameRegistry {
 
@@ -40,7 +42,7 @@ public class ReportNameRegistry {
      * @param key membercode_packageName
      * @return имя xml-файла, в который попадёт сигнал
      */
-    public String resolveNameAndEnqueue(String key, SignalKafkaEvent event) throws InterruptedException {
+    public String resolveNameAndEnqueue(String key, SignalKafkaEvent event) {
         lock.readLock().lock();
         try {
             String name = names.computeIfAbsent(key, k ->
@@ -48,6 +50,10 @@ 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();
         }