add new tables

credit-tracker.properties

@@ -11,7 +11,7 @@ ct.indic-data-source-settings.driver=com.ibm.db2.jcc.DB2Driver
 ct.indic-data-source-settings.url=${DB_URL_INDIC:jdbc:db2://10.230.227.100:2668/cprosd22:currentSchema=INDIC;}
 ct.indic-data-source-settings.username=${USERNAME_INDIC:ros9appl}
 ct.indic-data-source-settings.password=${PASSWORD_INDIC:R0s9Appl}
-ct.indic-data-source-settings.maximum-pool-size=${MAXIMUM_POOL_SIZE_INDIC:10}
+ct.indic-data-source-settings.maximum-pool-size=${MAXIMUM_POOL_SIZE_INDIC:30}
 ct.indic-data-source-settings.minimum-idle=${MINIMUM_IDLE_INDIC:10}
 ct.indic-data-source-settings.connection-timeout=${CONNECTION_TIMEOUT_INDIC:10000}
 ct.indic-data-source-settings.validation-timeout=${VALIDATION_TIMEOUT_INDIC:5000}
@@ -25,7 +25,7 @@ ct.signal-data-source-settings.driver=org.postgresql.Driver
 ct.signal-data-source-settings.url=${DB_URL_SIGNAL:jdbc:postgresql://10.200.200.133:5432/credit_tracker}
 ct.signal-data-source-settings.username=${USERNAME_SIGNAL:credit_tracker_user}
 ct.signal-data-source-settings.password=${PASSWORD_SIGNAL:-}
-ct.signal-data-source-settings.maximum-pool-size=${MAXIMUM_POOL_SIZE_SIGNAL:10}
+ct.signal-data-source-settings.maximum-pool-size=${MAXIMUM_POOL_SIZE_SIGNAL:40}
 ct.signal-data-source-settings.minimum-idle=${MINIMUM_IDLE_SIGNAL:10}
 ct.signal-data-source-settings.connection-timeout=${CONNECTION_TIMEOUT_SIGNAL:10000}
 ct.signal-data-source-settings.validation-timeout=${VALIDATION_TIMEOUT_SIGNAL:5000}
@@ -39,7 +39,7 @@ ct.scoring-data-source-settings.driver=com.ibm.db2.jcc.DB2Driver
 ct.scoring-data-source-settings.url=${DB_URL_SCORING:jdbc:db2://10.230.227.101:2668/cprosd22:currentSchema=TEST_2_INDIC;}
 ct.scoring-data-source-settings.username=${USERNAME_SCORING:tester}
 ct.scoring-data-source-settings.password=${PASSWORD_SCORING:tester}
-ct.scoring-data-source-settings.maximum-pool-size=${MAXIMUM_POOL_SIZE_SCORING:10}
+ct.scoring-data-source-settings.maximum-pool-size=${MAXIMUM_POOL_SIZE_SCORING:30}
 ct.scoring-data-source-settings.minimum-idle=${MINIMUM_IDLE_SCORING:10}
 ct.scoring-data-source-settings.connection-timeout=${CONNECTION_TIMEOUT_SCORING:10000}
 ct.scoring-data-source-settings.validation-timeout=${VALIDATION_TIMEOUT_SCORING:5000}
@@ -56,7 +56,7 @@ spring.servlet.multipart.max-request-size=${ACCEPTED_ARCHIVE_SIZE:2048MB}
 ct.xsd-validation.monitoring-request-path=monitoring-request.xsd
 ct.xsd-validation.signal-package-path=signal-package.xsd
 
-ct.signals-root-uri=${SIGNALS_ROOT_URI:http://10.230.227.102:11094/signalRUalfa/}
+ct.signals-root-uri=${SIGNALS_ROOT_URI:http://127.0.01:11094/signalRUalfa/}
 ct.signals-token=${SIGNALS_TOKEN:gttcHxwcGojoAvZK}
 ct.signals-user=${SIGNALS_USER:0001ZZ000001}
 
@@ -73,7 +73,21 @@ ct.db-version=${DB_VERSION:1234}
 
 ct.encoding=windows-1251
 
-#Batch size and global pipeline timeout
+
-ct.batchSize=1000
+#----------------------------Pipeline settings----------------------------#
-#in minute
+#???-?? ????????? ?? xml ?? ???? ????????????
-ct.timeout=30
+ct.saver-subject-batch-size=15000
+ct.pipeline.subject-batch-size=5000
+#????????? ? PHONE ? NAME + ?????? ? ???? SIGNALS_PHONE_FID_MAP
+ct.pipeline.name-enricher-stage-concurrency=20
+#????????? ? ID ? ?????? ????? xml ??? ??. ???????
+ct.pipeline.id-enricher-stage-concurrency=10
+#?????????? ??????? ??? ??????
+ct.pipeline.error-attempt-count=5
+#????? ???????? ????? ????????? ???????
+ct.pipeline.error-attempt-delay=2s
+#??????? ?? ???? ???????
+ct.pipeline.global-timeout=2h
+#???-?? ??????????? ???????????? ?????????????? ????????
+ct.pipeline.parallel-requests=3
+ct.pipeline.retryable-exceptions=org.springframework.dao.DataAccessException,org.springframework.dao.DataIntegrityViolationException 

src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java

@@ -35,4 +35,5 @@ public class CreditTrackerSettings {
     private Long daysToUpdateFlags;
     private String encoding;
     private String dbVersion;
+    private Integer saverSubjectBatchSize;
 }

src/main/java/ru/nbch/credit_tracker/entities/app/UserPackage.java

@@ -17,5 +17,5 @@ public class UserPackage {
     private String status;
     private LocalDateTime deactivatedAt;
     private String lastDownloadedReport;
-    private LocalDateTime monitoringAt;
+//    private LocalDateTime monitoringAt;
 }

src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java

@@ -128,7 +128,7 @@ public class SignalFacadeV3 {
 
             Long packageId = signalService.registerPackage(userId, packageName);
             LogUtil.addMdcLogKey("%s %d".formatted(packageName, packageId));
-            log.info("Package registered. PackageId {}. Status: {}", packageId, Status.PROCESSING.name());
+            log.info("Package registered. PackageId {}. Status: {}", packageId, Status.ACCEPTED.name());
             pipelinePackageManager.buildAndRun(packageId, parsedFilePath);
         } catch (Exception e) {
             try {

src/main/java/ru/nbch/credit_tracker/mapper/signal/PhoneFidMapMapper.java

@@ -17,7 +17,7 @@ public interface PhoneFidMapMapper {
     void setFidJson(@Param("json") String json);
     void registerWithId(@Param("phoneFidMap") List<PhoneFidMap> phoneFidMap);
 
-    List<PhoneFidMap> findFullPhoneFidMap(@Param("packageId") Long packageId, @Param("offset") int offset, @Param("limit") int limit);
+    List<PhoneFidMap> findFullPhoneFidMap(@Param("packageId") Long packageId, @Param("limit") int limit, @Param("offset") int offset);
 
     void updatePhoneFidMaps(@Param("subjects") Collection<PhoneFidMap> subjects);
     void updateOne(@Param("subject") PhoneFidMap subject);

src/main/java/ru/nbch/credit_tracker/repository/PhoneRepository.java

@@ -37,15 +37,8 @@ public class PhoneRepository {
         SELECT
             pn.number,
             pn.fid,
-            pn.file_since_dt,
+            pn.file_since_dt
-            nf.first,
-            nf.middle,
-            nf.name_1,
-            nf.birth_dt,
-            nf.file_since_dt as file_since_dt_name
         FROM PHONE_NORM pn
-                 JOIN NAME nf
-                      ON nf.fid = pn.fid
         WHERE number in (:phones) WITH UR
         """;
 

src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java

@@ -69,9 +69,9 @@ public class UpdatingFlagsService {
             SignalService.UpdateResult updateResult = signalService.updateSubjects(packageId, filter);
             if (updateResult.isUpdated()) {
                 if (signalRestService.setupMonitoring(packageId)) {
-                    LocalDateTime monitoringAt = userPackage.getMonitoringAt() != null ? userPackage.getMonitoringAt() : LocalDateTime.now();
+//                    LocalDateTime monitoringAt = userPackage.getMonitoringAt() != null ? userPackage.getMonitoringAt() : LocalDateTime.now();
-                    log.info("Flags were updated successfully. PackageId: {} PackageName: {} MonitoringAt: {}. Package status changed to ON_MONITORING", packageId, packageName, monitoringAt);
+                    log.info("Flags were updated successfully. PackageId: {} PackageName: {}. Package status changed to ON_MONITORING", packageId, packageName);
-                    signalService.updatePackageOnMonitoringSuccess(packageId, monitoringAt);
+                    signalService.updatePackageOnMonitoringSuccess(packageId, LocalDateTime.now());
                     monitoringErrorProcesingService.findErrors(packageId);
                 } else {
                     log.info("Updated package registration was rejected by signal service. PackageId: {} PackageName: {}", packageId, packageName);

src/main/java/ru/nbch/credit_tracker/service/indic/PhoneService2.java

@@ -90,4 +90,8 @@ public class PhoneService2 {
             throw new RuntimeException(e);
         }
     }
+
+    public List<PhoneFidMap> nextPage(Long packageId, Integer limit, Integer offset) {
+        return phoneFidMapMapper.findFullPhoneFidMap(packageId, limit, offset);
+    }
 }

src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java

@@ -1,6 +1,7 @@
 package ru.nbch.credit_tracker.service.task;
 
 import java.nio.file.Path;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Semaphore;
@@ -21,6 +22,7 @@ import ru.nbch.credit_tracker.service.task.impl.FidIdEnricher;
 import ru.nbch.credit_tracker.service.task.impl.FidNameEnricher;
 import ru.nbch.credit_tracker.service.task.impl.FlagsEnricher;
 import ru.nbch.credit_tracker.service.task.impl.SaveSubjectHandler;
+import ru.nbch.credit_tracker.service.task.impl.Saver;
 import ru.nbch.credit_tracker.service.task.impl.XmlMergeCoordinator;
 import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter;
 import ru.nbch.credit_tracker.service.task.setup.IPipeLineTerminator;
@@ -48,8 +50,15 @@ public class PipelinePackageManager {
         t.setName("pipe-starter-%d".formatted(t.threadId()));
         return t;
     });
+
+    private final ExecutorService subjectSaverExecutor = Executors.newFixedThreadPool(3, r -> {
+        Thread t = new Thread(r);
+        t.setName("subject-saver-%d".formatted(t.threadId()));
+        return t;
+    });
     private final Semaphore semaphore;
     private final PipelineBackoff pipelineBackoff;
+    private final Saver saver;
     private final static int MAX_PARALLEL_REQUESTS = 3;
 
     public PipelinePackageManager(SignalRestService signalRestService,
@@ -64,7 +73,7 @@ public class PipelinePackageManager {
                                   BiFunction<Long, Path, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory,
                                   FlagsEnricher flagsEnricherV2, FidNameEnricher fidNameEnricher,
                                   CreditTrackerSettings creditTrackerSettings,
-                                  PipelineBackoff pipelineBackoff) {
+                                  PipelineBackoff pipelineBackoff, Saver saver) {
         this.signalRestService = signalRestService;
         this.saveSubjectQueueFactory = saveSubjectQueueFactory;
         this.idEnricherQueueFactory = idEnricherQueueFactory;
@@ -81,6 +90,7 @@ public class PipelinePackageManager {
         this.semaphore = new Semaphore(creditTrackerSettings.getPipeline().parallelRequests());
         log.info("pipeline settings: {}", creditTrackerSettings.getPipeline());
         this.pipelineBackoff = pipelineBackoff;
+        this.saver = saver;
     }
 
     public void buildAndRun(Long packageId, Path pkgPath) throws Exception {
@@ -90,10 +100,7 @@ public class PipelinePackageManager {
         log.info("throttler permits after acquire: {}", semaphore.availablePermits());
         long startTime = System.currentTimeMillis();
         try {
-            Pipeline pipeline = PipeLineBuilderV2.first(saveSubjectQueueFactory.getObject(), idEnricherQueueFactory.getObject())
+            Pipeline pipeline = PipeLineBuilderV2.first(nameEnricherQueueFactory.getObject(), idEnricherQueueFactory.getObject())
-                    .handler(saveSubjectHandler)
-                    .andThen()
-                .stage(nameEnricherQueueFactory.getObject())
                     .handler(fidNameEnricher)
                     .handlerResult(flagsEnricherV2)
                     .andThen()
@@ -106,7 +113,6 @@ public class PipelinePackageManager {
                 .backoff(pipelineBackoff)
                 .build();
 
-            pipeline.start(executorService);
             pipeline.getTerminator()
                 .doneFuture()
                 .orTimeout(creditTrackerSettings.getPipeline().globalTimeout().toMillis(), TimeUnit.MILLISECONDS)
@@ -133,6 +139,17 @@ public class PipelinePackageManager {
                         log.info("semaphore amount: {}", semaphore.availablePermits());
                     }
                 });
+
+            CompletableFuture
+                .runAsync(() -> {
+                    try {
+                        saver.readAndSave(packageId, pkgPath);
+                    } catch (Throwable t) {
+                        log.error(ExceptionUtils.getStackTrace(t));
+                        pipeline.close();
+                    }
+                }, subjectSaverExecutor)
+                .thenRun(() -> pipeline.start(executorService));
         } catch (Exception e) {
             semaphore.release();
             throw e;

src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineImpl.java

@@ -11,6 +11,7 @@ import java.util.stream.Collectors;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
+import ru.nbch.credit_tracker.entities.app.PhoneFidMap;
 import ru.nbch.credit_tracker.entities.app.UserPackage;
 import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest;
 import ru.nbch.credit_tracker.exceptions.SignalClientException;
@@ -56,7 +57,47 @@ public class KickPipeLineImpl implements IPipeLineStarter<SubjectPayload> {
 
     @Override
     public void startFillingQueue(BlockingQueue<SubjectPayload> firstQueue) {
-        log.info("Start registering subjects. PackageId: {}", packageId);
+        try {
+            log.info("Start registering subjects. PackageId: {};", packageId);
+            int offset = 0;
+            int limit = BATCH_SIZE;
+            UserPackage userPackage = userPackageMapper.findPackage(packageId);
+            List<PhoneFidMap> page = phoneService2.nextPage(packageId, limit, offset);
+            while (!page.isEmpty()) {
+                log.debug("Get page {}", page.size());
+                List<SubjectProfile> subjectProfiles = page.stream().map(phoneFidMap -> SubjectProfile.builder()
+                    .identifier(phoneFidMap.getSubjectId())
+                    .phone(phoneFidMap.getPhone())
+                    .build()).toList();
+                try {
+                    firstQueue.put(new SubjectPayload(packageId, userPackage.getUserId(), userPackage.getPackageName(), subjectProfiles));
+                } catch (InterruptedException e) {
+                    isInterrupted = true;
+                    log.error("Pipeline starter was interrupted!");
+                    throw new PipelineStarterException();
+                }
+                offset += limit;
+                page = phoneService2.nextPage(packageId, limit, offset);
+            }
+        } catch (PipelineStarterException e) {
+            log.info("Pipeline starter shitting down...");
+        } finally {
+            if (isInterrupted) {
+                //noinspection ReturnInsideFinallyBlock
+                return;
+            }
+            log.debug("Send poison pill");
+            try {
+                SubjectPayload poison = new SubjectPayload(packageId, PoisonType.NORMALLY);
+                firstQueue.put(poison);
+            } catch (InterruptedException e) {
+                log.error("couldn't send poison: {}", ExceptionUtils.getStackTrace(e));
+            }
+        }
+    }
+//    @Override
+    public void startFillingQueueXml(BlockingQueue<SubjectPayload> firstQueue) {
+        log.info("Start registering subjects. PackageId: {}; pkg path: {}", packageId, pkgPath);
         UserPackage userPackage = userPackageMapper.findPackage(packageId);
         AtomicInteger counter = new AtomicInteger(0);
         try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(pkgPath), 8 * 1024 * 1024)) {

src/main/java/ru/nbch/credit_tracker/service/task/impl/Saver.java

@@ -0,0 +1,72 @@
+package ru.nbch.credit_tracker.service.task.impl;
+
+import java.io.BufferedInputStream;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
+import ru.nbch.credit_tracker.entities.app.PhoneFidMap;
+import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest;
+import ru.nbch.credit_tracker.enums.Status;
+import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
+import ru.nbch.credit_tracker.model.SubjectProfile;
+import ru.nbch.credit_tracker.service.indic.PhoneService2;
+import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
+import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
+
+@Service
+public class Saver {
+    private final Logger log = LoggerFactory.getLogger(getClass());
+    private final StaxXmlProcessor xmlProcessor;
+    private final PhoneService2 phoneService2;
+    private final UserPackageMapper userPackageMapper;
+    private final int BATCH_SIZE;
+
+
+    public Saver(StaxXmlProcessor xmlProcessor,
+                 PhoneService2 phoneService2,
+                 UserPackageMapper userPackageMapper,
+                 CreditTrackerSettings creditTrackerSettings) {
+        this.xmlProcessor = xmlProcessor;
+        this.phoneService2 = phoneService2;
+        this.userPackageMapper = userPackageMapper;
+        this.BATCH_SIZE = creditTrackerSettings.getSaverSubjectBatchSize();
+    }
+
+    public void readAndSave(Long packageId, Path pkgPath) {
+        log.info("Start registering subjects. PackageId: {}; pkg path: {}", packageId, pkgPath);
+        try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(pkgPath), 8 * 1024 * 1024)) {
+            xmlProcessor.read(in, MonitoringRequest.class,
+                request -> {
+                    if (!request.getSubjects().getSubject().isEmpty()) {
+                        List<SubjectProfile> subjects = request.getSubjects().getSubject().stream().map(subject ->
+                            SubjectProfile.builder()
+                                .identifier(subject.getId())
+                                .phone(subject.getClientId())
+                                .build()
+
+                        ).toList();
+                        log.trace("Reading subject's {} batch. PackageId: {}", request.getSubjects().getSubject().size(), packageId);
+                        List<PhoneFidMap> subjectForSave = subjects.stream().map(subject -> PhoneFidMap.builder()
+                            .subjectId(subject.getIdentifier())
+                            .packageId(packageId)
+                            .phone(subject.getPhone())
+                            .build()).toList();
+                        request.getSubjects().getSubject().clear();
+                        phoneService2.registerSubjects(subjectForSave);
+                    }
+                },
+                request -> request.getSubjects().getSubject().isEmpty(),
+                BATCH_SIZE, "/MonitoringRequest/Subjects/Subject"
+            );
+        } catch (IOException e) {
+            log.error("{}", ExceptionUtils.getStackTrace(e));
+        }
+        userPackageMapper.updatePackageStatus(packageId, Status.PROCESSING.name());
+        log.info("Complete registering subjects. PackageId: {}; pkg path: {}", packageId, pkgPath);
+    }
+}

src/main/java/ru/nbch/credit_tracker/utils/LogUtil.java

@@ -68,6 +68,8 @@ public class LogUtil {
                 }
 
                 task.run();
+            } catch (Throwable e) {
+                log.error(ExceptionUtils.getStackTrace(e));
             } finally {
                 // восстанавливаем предыдущий (может быть null)
                 try {

src/main/resources/mybatis/mapper/signal/UserPackageMapper.xml

@@ -13,7 +13,7 @@
         <result property="status" column="status"/>
         <result property="deactivatedAt" column="deactivated_at"/>
         <result property="lastDownloadedReport" column="last_downloaded_report"/>
-        <result property="monitoringAt" column="monitoring_at"/>
+<!--        <result property="monitoringAt" column="monitoring_at"/>-->
     </resultMap>
 
     <select id="isPackageNotUnique" resultType="boolean">
@@ -90,7 +90,7 @@
 
     <select id="findPackage" resultMap="packageDataResultMap"
             resultType="ru.nbch.credit_tracker.entities.app.UserPackage">
-        SELECT id, membercode, package_name, created_at, status, deactivated_at, last_downloaded_report, monitoring_at
+        SELECT id, membercode, package_name, created_at, status, deactivated_at, last_downloaded_report
         FROM signals_user_packages
         WHERE id = #{id}
         LIMIT 1