From 0cfe97667056ed49e4beb5cc232283e773327316 Mon Sep 17 00:00:00 2001 From: Mikhail Trofimov Date: Sun, 5 Jul 2026 14:15:41 +0300 Subject: [PATCH] Added ReportNameRegistry to store generated report names for hits writing --- .../model/SignalKafkaEvent.java | 2 + .../monitoring/KafkaMonitoringService.java | 53 +++++++++----- .../mode/monitoring/ReportNameRegistry.java | 72 +++++++++++++++++++ .../mode/monitoring/RotateSheduler.java | 14 ++-- .../SignalKafkaEventWriterService.java | 29 +++++--- 5 files changed, 132 insertions(+), 38 deletions(-) create mode 100644 src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/ReportNameRegistry.java diff --git a/src/main/java/ru/nbch/credit_tracker/model/SignalKafkaEvent.java b/src/main/java/ru/nbch/credit_tracker/model/SignalKafkaEvent.java index 19c91e5..4e546cd 100644 --- a/src/main/java/ru/nbch/credit_tracker/model/SignalKafkaEvent.java +++ b/src/main/java/ru/nbch/credit_tracker/model/SignalKafkaEvent.java @@ -5,6 +5,7 @@ import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage; import ru.nbch.credit_tracker.enums.EventType; import java.time.LocalDateTime; +import java.util.Map; @Getter @Setter @@ -13,6 +14,7 @@ import java.time.LocalDateTime; @AllArgsConstructor public class SignalKafkaEvent { private EventType eventType; + private Map reportNames; private UserPackagePhoneData userPackagePhoneData; private UserPackageLegalData userPackageLegalData; diff --git 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 index 8388569..16325b8 100644 --- 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 @@ -49,17 +49,20 @@ public class KafkaMonitoringService { private final SourceService sourceService; private final SegmentMapService segmentMapService; private final HitsService hitsService; + private final ReportNameRegistry reportNameRegistry; public KafkaMonitoringService(@Qualifier("signalKafkaEventQueue") BlockingQueue signalEventsQueue, PackageRecordsService packageRecordsService, SourceService sourceService, SegmentMapService segmentMapService, CreditTrackerSettings settings, - HitsService hitsService) { + HitsService hitsService, + ReportNameRegistry reportNameRegistry) { this.signalEventsQueue = signalEventsQueue; this.packageRecordsService = packageRecordsService; 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); @@ -133,14 +136,14 @@ public class KafkaMonitoringService { }); combinedFuture.thenAcceptAsync(hits -> { - if (!hits.isEmpty()) { - try { - hitsService.insertHits(hits); - } catch (Exception e) { - log.error("Failed to save hits. {}", ExceptionUtils.getStackTrace(e)); - } - } - }, dbExecutor); + 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); @@ -225,12 +228,17 @@ public class KafkaMonitoringService { Integer segment = segmentMap.get(memberCode.substring(0, 6)); event.setSegment(segment != null ? segment : -1); - putToQueue(event); + String reportKey = userPackagePhoneData.getUserId().substring(0, 6) + + "_" + userPackagePhoneData.getPackageName(); + String reportFileName = resolveNameAndEnqueue(reportKey, event); + if (reportFileName == null) { + continue; // событие не встало в очередь — hit не пишем + } Hit hit = Hit.builder() .packageId(userPackagePhoneData.getPackageId()) .subjectId(subjectId) .signalType(signalType) - .reportFileName("") // TODO + .reportFileName(reportFileName.replace(".xml", ".zip")) .inqT(message.getInquiryType()) .busCategory(busCategory) .inqDt(message.getInquiryDate()) @@ -325,12 +333,17 @@ public class KafkaMonitoringService { Integer segment = segmentMap.get(memberCode.substring(0, 6)); event.setSegment(segment != null ? segment : -1); - putToQueue(event); + String reportKey = userPackageLegalData.getUserId().substring(0, 6) + + "_" + userPackageLegalData.getPackageName(); + String reportFileName = resolveNameAndEnqueue(reportKey, event); + if (reportFileName == null) { + continue; // событие не встало в очередь — hit не пишем + } Hit hit = Hit.builder() .packageId(userPackageLegalData.getPackageId()) .subjectId(subjectId) .signalType(signalType) - .reportFileName("") // TODO + .reportFileName(reportFileName.replace(".xml", ".zip")) .inqT(message.getInquiryType()) .busCategory(busCategory) .inqDt(message.getInquiryDate()) @@ -357,12 +370,18 @@ public class KafkaMonitoringService { (inquiryType == 3 && billingCode.equalsIgnoreCase("CHSP")); } - private void putToQueue(SignalKafkaEvent event) { + /** + * Резолвит имя результирующего файла и ставит событие в очередь. + * + * @return имя xml-файла или null, если поток был прерван и событие не встало в очередь + */ + private String resolveNameAndEnqueue(String reportKey, SignalKafkaEvent event) { try { - signalEventsQueue.put(event); + return reportNameRegistry.resolveNameAndEnqueue(reportKey, event); } catch (InterruptedException e) { log.error("Error while putting signal event", e); Thread.currentThread().interrupt(); + return null; } } @@ -374,9 +393,7 @@ public class KafkaMonitoringService { processingPermits.acquire(permits); try { - SignalKafkaEvent rotateEvent = new SignalKafkaEvent(); - rotateEvent.setEventType(EventType.ROTATE); - signalEventsQueue.put(rotateEvent); + reportNameRegistry.enqueueRotate(); log.info("ROTATE event sent to queue"); SignalKafkaEvent stopEvent = new SignalKafkaEvent(); diff --git 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 new file mode 100644 index 0000000..f92b752 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/ReportNameRegistry.java @@ -0,0 +1,72 @@ +package ru.nbch.credit_tracker.service.mode.monitoring; + +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.stereotype.Component; +import ru.nbch.credit_tracker.enums.EventType; +import ru.nbch.credit_tracker.model.SignalKafkaEvent; +import ru.nbch.credit_tracker.utils.CTUtil; + +import java.time.LocalDateTime; +import java.util.HashMap; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.locks.ReadWriteLock; +import java.util.concurrent.locks.ReentrantReadWriteLock; + +/** + * Реестр имён результирующих файлов текущего цикла ротации. + * Имя фиксируется при первом сигнале цикла для пакета (timestamp — момент первого сигнала). + *

+ * событие, получившее имя эпохи N, встаёт в очередь раньше, чем ROTATE, + * закрывающий эпоху N. Поэтому WRITE-события кладутся в очередь только через + * {@link #resolveNameAndEnqueue}, а ROTATE — только через {@link #enqueueRotate}. + */ +@Component +public class ReportNameRegistry { + + // ключ: membercode_packageName; значение: финальное имя xml-файла + private final ConcurrentHashMap names = new ConcurrentHashMap<>(); + private final ReadWriteLock lock = new ReentrantReadWriteLock(); + + private final BlockingQueue signalEventsQueue; + + public ReportNameRegistry(@Qualifier("signalKafkaEventQueue") BlockingQueue signalEventsQueue) { + this.signalEventsQueue = signalEventsQueue; + } + + /** + * Резолвит имя файла для пакета и атомарно с этим ставит WRITE-событие в очередь. + * + * @param key membercode_packageName + * @return имя xml-файла, в который попадёт сигнал + */ + public String resolveNameAndEnqueue(String key, SignalKafkaEvent event) throws InterruptedException { + lock.readLock().lock(); + try { + String name = names.computeIfAbsent(key, k -> + String.format("%s_%s.xml", k, + LocalDateTime.now().format(CTUtil.responseReportDateTimeFormatter))); + signalEventsQueue.put(event); + return name; + } finally { + lock.readLock().unlock(); + } + } + + /** + * Закрывает текущую эпоху: снапшот имён уезжает внутри ROTATE-события, + * реестр очищается атомарно с постановкой события в очередь. + */ + public void enqueueRotate() throws InterruptedException { + lock.writeLock().lock(); + try { + SignalKafkaEvent rotateEvent = new SignalKafkaEvent(); + rotateEvent.setEventType(EventType.ROTATE); + rotateEvent.setReportNames(new HashMap<>(names)); + names.clear(); + signalEventsQueue.put(rotateEvent); + } finally { + lock.writeLock().unlock(); + } + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/RotateSheduler.java b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/RotateSheduler.java index 7ec5f73..1fe45b4 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/RotateSheduler.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/RotateSheduler.java @@ -3,14 +3,10 @@ package ru.nbch.credit_tracker.service.mode.monitoring; import jakarta.annotation.PostConstruct; import jakarta.annotation.PreDestroy; import lombok.extern.slf4j.Slf4j; -import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.scheduling.concurrent.CustomizableThreadFactory; import org.springframework.stereotype.Component; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; -import ru.nbch.credit_tracker.enums.EventType; -import ru.nbch.credit_tracker.model.SignalKafkaEvent; -import java.util.concurrent.BlockingQueue; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; @@ -23,12 +19,12 @@ public class RotateSheduler { ); private volatile boolean isShuttingdDown = false; - private final BlockingQueue signalEventsQueue; + private final ReportNameRegistry reportNameRegistry; private final CreditTrackerSettings config; - public RotateSheduler(@Qualifier("signalKafkaEventQueue") BlockingQueue signalEventsQueue, + public RotateSheduler(ReportNameRegistry reportNameRegistry, CreditTrackerSettings config) { - this.signalEventsQueue = signalEventsQueue; + this.reportNameRegistry = reportNameRegistry; this.config = config; } @@ -44,9 +40,7 @@ public class RotateSheduler { return; } try { - SignalKafkaEvent rotateEvent = new SignalKafkaEvent(); - rotateEvent.setEventType(EventType.ROTATE); - signalEventsQueue.put(rotateEvent); + reportNameRegistry.enqueueRotate(); } catch (InterruptedException e) { log.error("Error while putting signal event", e); Thread.currentThread().interrupt(); diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/SignalKafkaEventWriterService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/SignalKafkaEventWriterService.java index e63fdbf..e46932c 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/SignalKafkaEventWriterService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/SignalKafkaEventWriterService.java @@ -34,6 +34,7 @@ import java.time.format.DateTimeFormatter; import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.Map; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -85,7 +86,7 @@ public class SignalKafkaEventWriterService { } if (event.getEventType() == EventType.ROTATE) { - rotate(); + rotate(event.getReportNames()); } } } @@ -134,14 +135,16 @@ public class SignalKafkaEventWriterService { } /** - * Упаковывает все результирующие xml в архив + * Упаковывает все результирующие xml в архив. + * + * @param reportNames снапшот имён файлов закрываемой эпохи (ключ: membercode_packageName) */ - private void rotate() { + private void rotate(Map reportNames) { List packagesToUpdate = new ArrayList<>(); List userPackages = userPackageMapper.findAllPackagesByStatus(Status.ON_MONITORING.name()); for (UserPackage userPackage : userPackages) { try { - String reportName = movedAndZip(userPackage); + String reportName = movedAndZip(userPackage, reportNames); userPackage.setLastDownloadedReport(reportName); packagesToUpdate.add(userPackage); } catch (IOException e) { @@ -167,7 +170,7 @@ public class SignalKafkaEventWriterService { } } - private String movedAndZip(UserPackage userPackage) throws IOException { + private String movedAndZip(UserPackage userPackage, Map reportNames) throws IOException { String finalName = ""; String commonMembercode = userPackage.getUserId().substring(0, 6); Path tmpFilePath = Path.of(config.getLoadDir(), @@ -184,11 +187,17 @@ public class SignalKafkaEventWriterService { if (!Files.exists(rootReportPath)) { Files.createDirectories(rootReportPath); } - LocalDateTime newReportDate = LocalDateTime.now(); - finalName = String.format("%s_%s_%s.xml", - commonMembercode, - userPackage.getPackageName(), - newReportDate.format(CTUtil.responseReportDateTimeFormatter)); + + String reportKey = commonMembercode + "_" + userPackage.getPackageName(); + finalName = reportNames != null ? reportNames.get(reportKey) : null; + if (finalName == null) { + // tmp-файл есть, а имени в реестре нет — + // генерируем имя по времени ротации + finalName = String.format("%s_%s_%s.xml", + commonMembercode, + userPackage.getPackageName(), + LocalDateTime.now().format(CTUtil.responseReportDateTimeFormatter)); + } Path responsePath = rootReportPath.resolve(finalName);