KafkaSignalProcessor refactoring

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

@@ -0,0 +1,94 @@
+package ru.nbch.credit_tracker.service.mode.monitoring;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+import ru.nbch.credit_tracker.entities.app.Hit;
+import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage;
+import ru.nbch.credit_tracker.service.signal.HitsService;
+import ru.nbch.credit_tracker.service.signal.SegmentMapService;
+import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+@Slf4j
+@Component
+public class KafkaMonitoringBatchProcessor {
+
+    private final SegmentMapService segmentMapService;
+    private final HitsService hitsService;
+    private final KafkaMonitoringMessageFilter messageFilter;
+    private final NaturalPersonKafkaSignalProcessor naturalPersonSignalProcessor;
+    private final LegalPersonKafkaSignalProcessor legalPersonSignalProcessor;
+
+    public KafkaMonitoringBatchProcessor(SegmentMapService segmentMapService,
+                                         HitsService hitsService,
+                                         KafkaMonitoringMessageFilter messageFilter,
+                                         NaturalPersonKafkaSignalProcessor naturalPersonSignalProcessor,
+                                         LegalPersonKafkaSignalProcessor legalPersonSignalProcessor) {
+        this.segmentMapService = segmentMapService;
+        this.hitsService = hitsService;
+        this.messageFilter = messageFilter;
+        this.naturalPersonSignalProcessor = naturalPersonSignalProcessor;
+        this.legalPersonSignalProcessor = legalPersonSignalProcessor;
+    }
+
+    public void process(List<PackageMessage> messages,
+                        ExecutorService processingExecutor,
+                        ExecutorService dbExecutor) {
+        Map<Boolean, List<PackageMessage>> partition = messages.stream()
+                .filter(Objects::nonNull)
+                .filter(messageFilter::isSupported)
+                .collect(Collectors.partitioningBy(messageFilter::isNaturalPerson));
+
+        List<PackageMessage> npMessages = partition.get(true);
+        List<PackageMessage> lpMessages = partition.get(false);
+
+        Map<String, Integer> segmentMap = segmentMapService.loadSegmentMap();
+
+        CompletableFuture<List<Hit>> npFuture = CompletableFuture.supplyAsync(() -> {
+            if (npMessages != null && !npMessages.isEmpty()) {
+                return naturalPersonSignalProcessor.process(npMessages, segmentMap);
+            }
+            return Collections.emptyList();
+        }, processingExecutor);
+
+        CompletableFuture<List<Hit>> lpFuture = CompletableFuture.supplyAsync(() -> {
+            if (lpMessages != null && !lpMessages.isEmpty()) {
+                return legalPersonSignalProcessor.process(lpMessages, segmentMap);
+            }
+            return Collections.emptyList();
+        }, processingExecutor);
+
+        CompletableFuture<List<Hit>> combinedFuture = npFuture
+                .thenCombine(lpFuture, (npHits, lpHits) -> {
+                    List<Hit> allHits = new ArrayList<>();
+                    if (npHits != null) allHits.addAll(npHits);
+                    if (lpHits != null) allHits.addAll(lpHits);
+                    return allHits;
+                });
+
+        combinedFuture.thenAcceptAsync(hits -> {
+            if (!hits.isEmpty()) {
+                try {
+                    hitsService.insertHits(hits);
+                } catch (Exception e) {
+                    log.error("Failed to save hits. {}", ExceptionUtils.getStackTrace(e));
+                }
+            }
+        }, dbExecutor);
+
+        try {
+            combinedFuture.get(1, TimeUnit.MINUTES);
+        } catch (Exception e) {
+            log.error("Error processing messages", e);
+        }
+    }
+}

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

@@ -0,0 +1,19 @@
+package ru.nbch.credit_tracker.service.mode.monitoring;
+
+import org.springframework.stereotype.Component;
+import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage;
+
+@Component
+public class KafkaMonitoringMessageFilter {
+
+    public boolean isSupported(PackageMessage message) {
+        Integer inquiryType = message.getInquiryType();
+        String billingCode = message.getBillingCode();
+        return inquiryType == 1 || inquiryType == 2 ||
+                (inquiryType == 3 && billingCode.equalsIgnoreCase("CHSP"));
+    }
+
+    public boolean isNaturalPerson(PackageMessage message) {
+        return message.getPersons() != null && !message.getPersons().isEmpty();
+    }
+}

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

@@ -2,8 +2,6 @@ package ru.nbch.credit_tracker.service.mode.monitoring;
 
 import jakarta.annotation.PreDestroy;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.commons.lang3.tuple.Pair;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.annotation.DependsOn;
 import org.springframework.kafka.annotation.KafkaListener;
@@ -11,25 +9,12 @@ import org.springframework.kafka.support.Acknowledgment;
 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;
-import ru.nbch.credit_tracker.entities.app.LegalsMap;
-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.UserPackageLegalData;
-import ru.nbch.credit_tracker.model.UserPackagePhoneData;
-import ru.nbch.credit_tracker.service.nbch.SourceService;
-import ru.nbch.credit_tracker.service.signal.HitsService;
-import ru.nbch.credit_tracker.service.signal.PackageRecordsService;
-import ru.nbch.credit_tracker.service.signal.SegmentMapService;
-import ru.nbch.credit_tracker.utils.CTUtil;
-import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
 
-import java.time.LocalDateTime;
-import java.util.*;
 import java.util.concurrent.*;
-import java.util.stream.Collectors;
+import java.util.List;
 
 @Slf4j
 @Service
@@ -45,26 +30,17 @@ public class KafkaMonitoringService {
 
     private final BlockingQueue<SignalKafkaEvent> signalEventsQueue;
 
-    private final PackageRecordsService packageRecordsService;
+    private final KafkaMonitoringBatchProcessor batchProcessor;
-    private final SourceService sourceService;
-    private final SegmentMapService segmentMapService;
-    private final HitsService hitsService;
     private final ReportNameRegistry reportNameRegistry;
 
     public KafkaMonitoringService(@Qualifier("signalKafkaEventQueue") BlockingQueue<SignalKafkaEvent> signalEventsQueue,
-                                  PackageRecordsService packageRecordsService,
-                                  SourceService sourceService,
-                                  SegmentMapService segmentMapService,
                                   CreditTrackerSettings settings,
-                                  HitsService hitsService,
+                                  KafkaMonitoringBatchProcessor batchProcessor,
                                   ReportNameRegistry reportNameRegistry) {
         this.signalEventsQueue = signalEventsQueue;
-        this.packageRecordsService = packageRecordsService;
+        this.batchProcessor = batchProcessor;
-        this.sourceService = sourceService;
-        this.segmentMapService = segmentMapService;
         this.reportNameRegistry = reportNameRegistry;
         this.permits = Integer.parseInt(settings.getKafkaConsumerSettings().concurency()); // количество равно concurency листенера
-        this.hitsService = hitsService;
         this.processingPermits = new Semaphore(permits);
 
         this.processingExecutor = Executors.newFixedThreadPool(
@@ -96,329 +72,13 @@ public class KafkaMonitoringService {
         try {
             acknowledgment.acknowledge();
             long startTime = System.currentTimeMillis();
-            process(records);
+            batchProcessor.process(records, processingExecutor, dbExecutor);
             log.info("Processed {} messages from Kafka. Time {} ms", records.size(), System.currentTimeMillis() - startTime);
         } finally {
             processingPermits.release();
         }
     }
 
-    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()));
-
-        List<PackageMessage> npMessages = partition.get(true);
-        List<PackageMessage> lpMessages = partition.get(false);
-
-        Map<String, Integer> segmentMap = segmentMapService.loadSegmentMap();
-
-        CompletableFuture<List<Hit>> npFuture = CompletableFuture.supplyAsync(() -> {
-            if (npMessages != null && !npMessages.isEmpty()) {
-                return processNpMessages(npMessages, segmentMap);
-            }
-            return Collections.emptyList();
-        }, processingExecutor);
-
-        CompletableFuture<List<Hit>> lpFuture = CompletableFuture.supplyAsync(() -> {
-            if (lpMessages != null && !lpMessages.isEmpty()) {
-                return processLpMessages(lpMessages, segmentMap);
-            }
-            return Collections.emptyList();
-        }, processingExecutor);
-
-        CompletableFuture<List<Hit>> combinedFuture = npFuture
-                .thenCombine(lpFuture, (npHits, lpHits) -> {
-                    List<Hit> allHits = new ArrayList<>();
-                    if (npHits != null) allHits.addAll(npHits);
-                    if (lpHits != null) allHits.addAll(lpHits);
-                    return allHits;
-                });
-
-        combinedFuture.thenAcceptAsync(hits -> {
-            if (!hits.isEmpty()) {
-                try {
-                    hitsService.insertHits(hits);
-                } catch (Exception e) {
-                    log.error("Failed to save hits. {}", ExceptionUtils.getStackTrace(e));
-                }
-            }
-        }, dbExecutor);
-
-        try {
-            combinedFuture.get(1, TimeUnit.MINUTES);
-        } catch (Exception e) {
-            log.error("Error processing messages", e);
-        }
-    }
-
-    private List<Hit> processNpMessages(List<PackageMessage> messages, Map<String, Integer> segmentMap) {
-        List<Hit> hits = new ArrayList<>();
-
-        List<Integer> fids = messages.stream().map(PackageMessage::getFid).collect(Collectors.toList());
-        List<String> memberCodes = messages.stream()
-                .map(message -> message.getMemberCode().substring(0, 6))
-                .toList();
-
-        List<PhonesMap> phonesMaps = packageRecordsService.findCachedByFids(fids);
-
-        // key: fid; value: список флагов просрочек (у одного фида может быть несколько номеров)
-        Map<Integer, List<PhonesMap>> phonesDataMap = phonesMaps.stream()
-                .collect(Collectors.groupingBy(PhonesMap::getFid));
-
-        Set<Long> uniquePhones = phonesMaps.stream().map(PhonesMap::getPhone).collect(Collectors.toSet());
-
-        // прелоад по memberCode категорий
-        Map<String, String> busCategories = sourceService.findBusCategories(memberCodes);
-
-        // subjectId, packageId, signalTypes, phone только для статуса ON_MONITORING
-        List<UserPackagePhoneData> packagePhoneData = packageRecordsService
-                .selectUserPackagePhoneData(uniquePhones);
-
-        // 1 номер может быть у нескольких пакетов
-        Map<Long, List<UserPackagePhoneData>> userPackagePhoneDataMap = packagePhoneData.stream()
-                .collect(Collectors.groupingBy(UserPackagePhoneData::getPhone));
-
-        for (PackageMessage message : messages) {
-            Integer fid = message.getFid();
-            String memberCode = message.getMemberCode();
-
-            String busCategory = busCategories.get(memberCode.substring(0, 6));
-            if (StringUtils.isBlank(busCategory)) {
-                log.error("Для memberсode = {} отсутствует bus_category или не найден memberсode", memberCode.substring(0, 6));
-                continue;
-            }
-
-            int signalType = CTUtil.checkBusCategory(busCategory);
-
-            List<PhonesMap> phonesMapList = phonesDataMap.get(fid);
-
-            if (phonesMapList == null) {
-                continue;
-            }
-
-            for (PhonesMap phonesMap : phonesMapList) {
-                // в userPackagePhoneDataMap только нужные номера, те что не вошли в userPackagePhoneDataList игнорируются
-                List<UserPackagePhoneData> userPackagePhoneDataList = userPackagePhoneDataMap.get(phonesMap.getPhone());
-
-                if (userPackagePhoneDataList == null) {
-                    continue;
-                }
-
-                for (UserPackagePhoneData userPackagePhoneData : userPackagePhoneDataList) {
-                    String signalTypes = userPackagePhoneData.getSignalTypes();
-
-                    if (signalTypes.contains(Integer.toString(signalType))) {
-                        Integer segment = segmentMap.get(memberCode.substring(0, 6));
-                        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 = reportNameRegistry.resolveNameAndEnqueue(reportKey, event);
-                        if (reportFileName == null) {
-                            continue; // событие не встало в очередь — hit не пишем
-                        }
-                        Hit hit = buildHit(
-                                true, message,
-                                userPackagePhoneData,
-                                null,
-                                signalType, busCategory,
-                                segment, reportFileName
-                        );
-                        hits.add(hit);
-                    }
-                }
-            }
-        }
-
-        return hits;
-    }
-
-    private List<Hit> processLpMessages(List<PackageMessage> messages, Map<String, Integer> segmentMap) {
-        List<Hit> hits = new ArrayList<>();
-
-        List<Integer> fids = messages.stream().map(PackageMessage::getFid).collect(Collectors.toList());
-        List<String> memberCodes = messages.stream()
-                .map(message -> message.getMemberCode().substring(0, 6))
-                .toList();
-
-        List<LegalsMap> legalsMaps = packageRecordsService.findCachedLegalsMapByFids(fids);
-
-        Map<Integer, List<LegalsMap>> legalsDataMap = legalsMaps.stream()
-                .collect(Collectors.groupingBy(LegalsMap::getFid));
-
-        Set<Pair<Long, Long>> uniqueInnRegnumPairs = legalsMaps.stream()
-                .map(l -> Pair.of(l.getInn(), l.getRegNum()))
-                .collect(Collectors.toSet());
-
-        // прелоад по memberCode категорий
-        Map<String, String> busCategories = sourceService.findBusCategories(memberCodes);
-
-        // subjectId, packageId, signalTypes, inn, regNum только для статуса ON_MONITORING
-        List<UserPackageLegalData> packageLegalData = packageRecordsService
-                .selectUserPackageLegalData(uniqueInnRegnumPairs);
-
-        Map<Pair<Long, Long>, List<UserPackageLegalData>> userPackageLegalDataMap = packageLegalData.stream()
-                .collect(Collectors.groupingBy(p -> Pair.of(p.getInn(), p.getRegNum())));
-
-        for (PackageMessage message : messages) {
-            Integer fid = message.getFid();
-            String memberCode = message.getMemberCode();
-
-            String busCategory = busCategories.get(memberCode.substring(0, 6));
-            if (StringUtils.isBlank(busCategory)) {
-                log.error("Для memberсode = {} отсутствует bus_category или не найден memberсode", memberCode.substring(0, 6));
-                continue;
-            }
-
-            int signalType = CTUtil.checkBusCategory(busCategory);
-
-            List<LegalsMap> legalsMapList = legalsDataMap.get(fid);
-
-            if (legalsMapList == null) {
-                continue;
-            }
-
-            for (LegalsMap legalsMap : legalsMapList) {
-                // в userPackageLegalDataMap только нужные inn, regnum, те что не вошли в userPackageLegalDataList игнорируются
-                List<UserPackageLegalData> userPackageLegalDataList = userPackageLegalDataMap.get(Pair.of(legalsMap.getInn(), legalsMap.getRegNum()));
-
-                if (userPackageLegalDataList == null) {
-                    continue;
-                }
-
-                for (UserPackageLegalData userPackageLegalData : userPackageLegalDataList) {
-                    String signalTypes = userPackageLegalData.getSignalTypes();
-
-                    if (signalTypes.contains(Integer.toString(signalType))) {
-                        Integer segment = segmentMap.get(memberCode.substring(0, 6));
-
-                        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 = reportNameRegistry.resolveNameAndEnqueue(reportKey, event);
-                        if (reportFileName == null) {
-                            continue; // событие не встало в очередь — hit не пишем
-                        }
-                        Hit hit = buildHit(
-                                false, message,
-                                null,
-                                userPackageLegalData,
-                                signalType, busCategory,
-                                segment, reportFileName
-                        );
-                        hits.add(hit);
-                    }
-                }
-            }
-        }
-
-        return hits;
-    }
-
-    private boolean filterMessages(PackageMessage message) {
-        Integer inquiryType = message.getInquiryType();
-        String billingCode = message.getBillingCode();
-        return inquiryType == 1 || inquiryType == 2 ||
-                (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();
-        }
-    }
-
     @PreDestroy
     public void stopKafkaMonitoringService() throws InterruptedException {
         log.info("Stopping KafkaMonitoringService...");

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

@@ -0,0 +1,102 @@
+package ru.nbch.credit_tracker.service.mode.monitoring;
+
+import org.apache.commons.lang3.tuple.Pair;
+import org.springframework.stereotype.Component;
+import ru.nbch.credit_tracker.entities.app.Hit;
+import ru.nbch.credit_tracker.entities.app.LegalsMap;
+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.UserPackageLegalData;
+import ru.nbch.credit_tracker.service.signal.PackageRecordsService;
+
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+@Component
+public class LegalPersonKafkaSignalProcessor {
+
+    private final PackageRecordsService packageRecordsService;
+    private final KafkaSignalProcessorSupport processorSupport;
+
+    public LegalPersonKafkaSignalProcessor(PackageRecordsService packageRecordsService,
+                                           KafkaSignalProcessorSupport processorSupport) {
+        this.packageRecordsService = packageRecordsService;
+        this.processorSupport = processorSupport;
+    }
+
+    public List<Hit> process(List<PackageMessage> messages, Map<String, Integer> segmentMap) {
+        return processorSupport.process(
+                messages,
+                segmentMap,
+                packageRecordsService::findCachedLegalsMapByFids,
+                LegalsMap::getFid,
+                this::mappingKey,
+                this::loadPackageData,
+                this::buildEvent,
+                this::buildHit
+        );
+    }
+
+    private Pair<Long, Long> mappingKey(LegalsMap legalsMap) {
+        return Pair.of(legalsMap.getInn(), legalsMap.getRegNum());
+    }
+
+    private Map<Pair<Long, Long>, List<UserPackageLegalData>> loadPackageData(List<LegalsMap> legalsMaps) {
+        Set<Pair<Long, Long>> uniqueInnRegnumPairs = legalsMaps.stream()
+                .map(this::mappingKey)
+                .collect(Collectors.toSet());
+
+        // subjectId, packageId, signalTypes, inn, regNum только для статуса ON_MONITORING
+        List<UserPackageLegalData> packageLegalData = packageRecordsService.selectUserPackageLegalData(uniqueInnRegnumPairs);
+
+        return packageLegalData.stream()
+                .collect(Collectors.groupingBy(p -> Pair.of(p.getInn(), p.getRegNum())));
+    }
+
+    private SignalKafkaEvent buildEvent(PackageMessage message,
+                                        LegalsMap legalsMap,
+                                        UserPackageLegalData packageData,
+                                        int signalType,
+                                        String busCategory,
+                                        Integer segment) {
+        SignalKafkaEvent event = new SignalKafkaEvent();
+        event.setEventType(EventType.WRITE);
+        event.setNatural(false);
+        event.setUserPackageLegalData(packageData);
+        event.setPackageMessage(message);
+        event.setSubjectId(packageData.getSubjectId());
+        event.setFlag90days(legalsMap.getFlag90days());
+        event.setSignalType(Integer.toString(signalType));
+        event.setBusCategory(signalType == 3 ? busCategory : null);
+        event.setLastRejectedDate(legalsMap.getLastRejectedDate());
+        event.setSegment(segment != null ? segment : -1);
+        return event;
+    }
+
+    private Hit buildHit(PackageMessage message,
+                         UserPackageLegalData packageData,
+                         int signalType,
+                         String busCategory,
+                         Integer segment,
+                         String reportFileName) {
+        return Hit.builder()
+                .packageId(packageData.getPackageId())
+                .subjectId(packageData.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(packageData.getInn())
+                .regNum(packageData.getRegNum())
+                .build();
+    }
+}

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

@@ -0,0 +1,97 @@
+package ru.nbch.credit_tracker.service.mode.monitoring;
+
+import org.springframework.stereotype.Component;
+import ru.nbch.credit_tracker.entities.app.Hit;
+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.signal.PackageRecordsService;
+
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+@Component
+public class NaturalPersonKafkaSignalProcessor {
+
+    private final PackageRecordsService packageRecordsService;
+    private final KafkaSignalProcessorSupport processorSupport;
+
+    public NaturalPersonKafkaSignalProcessor(PackageRecordsService packageRecordsService,
+                                             KafkaSignalProcessorSupport processorSupport) {
+        this.packageRecordsService = packageRecordsService;
+        this.processorSupport = processorSupport;
+    }
+
+    public List<Hit> process(List<PackageMessage> messages, Map<String, Integer> segmentMap) {
+        return processorSupport.process(
+                messages,
+                segmentMap,
+                packageRecordsService::findCachedByFids,
+                PhonesMap::getFid,
+                PhonesMap::getPhone,
+                this::loadPackageData,
+                this::buildEvent,
+                this::buildHit
+        );
+    }
+
+    private Map<Long, List<UserPackagePhoneData>> loadPackageData(List<PhonesMap> phonesMaps) {
+        Set<Long> uniquePhones = phonesMaps.stream()
+                .map(PhonesMap::getPhone)
+                .collect(Collectors.toSet());
+
+        // subjectId, packageId, signalTypes, phone только для статуса ON_MONITORING
+        List<UserPackagePhoneData> packagePhoneData = packageRecordsService.selectUserPackagePhoneData(uniquePhones);
+
+        // 1 номер может быть у нескольких пакетов
+        return packagePhoneData.stream()
+                .collect(Collectors.groupingBy(UserPackagePhoneData::getPhone));
+    }
+
+    private SignalKafkaEvent buildEvent(PackageMessage message,
+                                        PhonesMap phonesMap,
+                                        UserPackagePhoneData packageData,
+                                        int signalType,
+                                        String busCategory,
+                                        Integer segment) {
+        SignalKafkaEvent event = new SignalKafkaEvent();
+        event.setEventType(EventType.WRITE);
+        event.setNatural(true);
+        event.setUserPackagePhoneData(packageData);
+        event.setPackageMessage(message);
+        event.setSubjectId(packageData.getSubjectId());
+        event.setFlagDebt5000(phonesMap.getFlagDebt5000());
+        event.setFlag90days(phonesMap.getFlag90days());
+        event.setSignalType(Integer.toString(signalType));
+        event.setBusCategory(signalType == 3 ? busCategory : null);
+        event.setSegment(segment != null ? segment : -1);
+        return event;
+    }
+
+    private Hit buildHit(PackageMessage message,
+                         UserPackagePhoneData packageData,
+                         int signalType,
+                         String busCategory,
+                         Integer segment,
+                         String reportFileName) {
+        return Hit.builder()
+                .packageId(packageData.getPackageId())
+                .subjectId(packageData.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(packageData.getPhone())
+                .build();
+    }
+}

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

@@ -0,0 +1,23 @@
+package ru.nbch.credit_tracker.service.mode.monitoring;
+
+import org.springframework.stereotype.Component;
+import ru.nbch.credit_tracker.model.SignalKafkaEvent;
+
+@Component
+public class SignalKafkaEventPublisher {
+
+    private final ReportNameRegistry reportNameRegistry;
+
+    public SignalKafkaEventPublisher(ReportNameRegistry reportNameRegistry) {
+        this.reportNameRegistry = reportNameRegistry;
+    }
+
+    /**
+     * Резолвит имя результирующего файла и ставит событие в очередь.
+     *
+     * @return имя xml-файла или null, если поток был прерван и событие не встало в очередь
+     */
+    public String resolveNameAndEnqueue(String reportKey, SignalKafkaEvent event) {
+        return reportNameRegistry.resolveNameAndEnqueue(reportKey, event);
+    }
+}