TaskConsumerV2 with poison pill

pom.xml

@@ -10,12 +10,12 @@
 	</parent>
 	<groupId>ru.nbch</groupId>
 	<artifactId>credit-tracker</artifactId>
-	<version>1.15</version>
+	<version>2.0</version>
 	<name>credit_tracker</name>
 	<description>Credit Tracker Application</description>
 
 	<properties>
-		<java.version>17</java.version>
+		<java.version>21</java.version>
 		<git.info.prefix>git_info</git.info.prefix>
 		<git.info.filename>git.properties</git.info.filename>
 		<registry.url>rh8-vm3.nbki.msk:5000</registry.url>

src/main/java/ru/nbch/credit_tracker/builder/PersonFactoryBuilderV2.java

@@ -1,8 +1,7 @@
 package ru.nbch.credit_tracker.builder;
 
-import java.text.DateFormat;
-import java.text.SimpleDateFormat;
 import java.time.LocalDate;
+import java.time.format.DateTimeFormatter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import ru.nbch.credit_tracker.entities.dto.IdProfile;
@@ -13,7 +12,7 @@ import ru.nbch.credit_tracker.utils.CTUtil;
 
 @Slf4j
 public class PersonFactoryBuilderV2 {
-    private final static DateFormat DATE_FORMAT = new SimpleDateFormat("dd.MM.yyyy");
+    private final static DateTimeFormatter DATE_FORMAT = DateTimeFormatter.ofPattern("dd.MM.yyyy");
 
     private final static String DEFAULT_DATE = "01.01.2000";
     private final static String IP = "0";
@@ -58,10 +57,10 @@ public class PersonFactoryBuilderV2 {
         person.setS(nameData.getLastName());
         person.setN(nameData.getFirstName());
         person.setM(StringUtils.isNotBlank(nameData.getMiddleName()) ? nameData.getMiddleName() : null);
-        person.setD(DATE_FORMAT.format(CTUtil.localDatetoDate(nameData.getBirthDate())));
+        person.setD(DATE_FORMAT.format(nameData.getBirthDate()));
 
         person.setPn(getPn(passportData.getSerNum(), passportData.getIdNum()));
-        person.setPd(DATE_FORMAT.format(CTUtil.localDatetoDate(LocalDate.now()))); // Do not use passport iss_date. Signal service could reject the package
+        person.setPd(DATE_FORMAT.format(LocalDate.now())); // Do not use passport iss_date. Signal service could reject the package
         person.setDt(getDt(passportData.getIdType(), passportData.getIdType2()));
         if (person.getDt() != null && person.getDt().equals("999")) {
             person.setOdt(ODT);
@@ -70,12 +69,12 @@ public class PersonFactoryBuilderV2 {
         person.setIp(IP);
         person.setUid(uid);
         person.setOksm(OKSM);
-        person.setConsentDate(DATE_FORMAT.format(CTUtil.localDatetoDate(LocalDate.now())));
+        person.setConsentDate(DATE_FORMAT.format(LocalDate.now()));
         person.setConsentPurpose(CONSENT_PURPOSE);
         person.setReportUser(REPORT_USER);
         person.setLiability(LIABILITY);
         person.setConsentPeriod(CONSENT_PERIOD);
-        person.setAgrDate(DATE_FORMAT.format(CTUtil.localDatetoDate(LocalDate.now())));
+        person.setAgrDate(DATE_FORMAT.format(LocalDate.now()));
         person.setReportUserRegNum(REPORT_USER_REG_NUM);
         person.setReportUserTaxID(REPORT_USER_TAX_ID);
         person.setConsentHash(CONSENT_HASH);

src/main/java/ru/nbch/credit_tracker/config/datasource/DatabaseConfig.java

@@ -1,10 +1,9 @@
 package ru.nbch.credit_tracker.config.datasource;
 
+import javax.sql.DataSource;
 import org.springframework.boot.jdbc.DataSourceBuilder;
 import ru.nbch.credit_tracker.config.settings.DataSourceSettings;
 
-import javax.sql.DataSource;
-
 public class DatabaseConfig {
 
     protected DataSource createDataSource(DataSourceSettings settings) {

src/main/java/ru/nbch/credit_tracker/config/datasource/IndicDataSourceConfig.java

@@ -1,5 +1,6 @@
 package ru.nbch.credit_tracker.config.datasource;
 
+import com.zaxxer.hikari.HikariDataSource;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.ibatis.session.SqlSessionFactory;
 import org.mybatis.spring.SqlSessionFactoryBean;
@@ -29,7 +30,8 @@ public class IndicDataSourceConfig extends DatabaseConfig {
 
     @Bean(name = "indicDataSource")
     public DataSource dataSource(CreditTrackerSettings config) throws SQLException {
-        DataSource dataSource = createDataSource(config.getIndicDataSourceSettings());
+        HikariDataSource dataSource = (HikariDataSource) createDataSource(config.getIndicDataSourceSettings());
+        dataSource.setMaximumPoolSize(50);
         try (Connection connection = dataSource.getConnection()) {
             connection.isValid(3);
         } catch (SQLException e) {

src/main/java/ru/nbch/credit_tracker/config/datasource/ScoringDataSourceConfig.java

@@ -1,5 +1,9 @@
 package ru.nbch.credit_tracker.config.datasource;
 
+import com.zaxxer.hikari.HikariDataSource;
+import java.sql.Connection;
+import java.sql.SQLException;
+import javax.sql.DataSource;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.ibatis.session.SqlSessionFactory;
 import org.mybatis.spring.SqlSessionFactoryBean;
@@ -11,10 +15,6 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.core.io.Resource;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
 
-import javax.sql.DataSource;
-import java.sql.Connection;
-import java.sql.SQLException;
-
 @Slf4j
 @Configuration
 @MapperScan(
@@ -28,7 +28,8 @@ public class ScoringDataSourceConfig extends DatabaseConfig {
 
     @Bean(name = "scoringDataSource")
     public DataSource dataSource(CreditTrackerSettings config) throws SQLException {
-        DataSource dataSource = createDataSource(config.getScoringDataSourceSettings());
+        HikariDataSource dataSource = (HikariDataSource) createDataSource(config.getScoringDataSourceSettings());
+        dataSource.setMaximumPoolSize(10);
         try (Connection connection = dataSource.getConnection()) {
             connection.isValid(3);
         } catch (SQLException e) {

src/main/java/ru/nbch/credit_tracker/config/datasource/SignalDataSourceConfig.java

@@ -1,7 +1,13 @@
 package ru.nbch.credit_tracker.config.datasource;
 
+import com.zaxxer.hikari.HikariDataSource;
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.util.UUID;
+import javax.sql.DataSource;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.ibatis.session.SqlSessionFactory;
+import org.apache.ibatis.type.TypeHandlerRegistry;
 import org.mybatis.spring.SqlSessionFactoryBean;
 import org.mybatis.spring.annotation.MapperScan;
 import org.springframework.beans.factory.annotation.Qualifier;
@@ -9,12 +15,9 @@ import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.core.io.Resource;
+import ru.nbch.credit_tracker.config.datasource.handlers.UUIDTypeHandler;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
 
-import javax.sql.DataSource;
-import java.sql.Connection;
-import java.sql.SQLException;
-
 @Slf4j
 @Configuration
 @MapperScan(
@@ -27,7 +30,8 @@ public class SignalDataSourceConfig extends DatabaseConfig {
 
     @Bean(name = "signalDataSource")
     public DataSource dataSource(CreditTrackerSettings config) throws SQLException {
-        DataSource dataSource = createDataSource(config.getSignalDataSourceSettings());
+        HikariDataSource dataSource = (HikariDataSource) createDataSource(config.getSignalDataSourceSettings());
+        dataSource.setMaximumPoolSize(100);
         try (Connection connection = dataSource.getConnection()) {
             connection.isValid(3);
         } catch (SQLException e) {
@@ -43,6 +47,11 @@ public class SignalDataSourceConfig extends DatabaseConfig {
         SqlSessionFactoryBean sqlSessionFactoryBean = new SqlSessionFactoryBean();
         sqlSessionFactoryBean.setDataSource(dataSource);
         sqlSessionFactoryBean.setMapperLocations(resources);
+
+        SqlSessionFactory sqlSessionFactory = sqlSessionFactoryBean.getObject();
+        TypeHandlerRegistry typeHandlerRegistry = sqlSessionFactory.getConfiguration().getTypeHandlerRegistry();
+        typeHandlerRegistry.register(UUID.class, UUIDTypeHandler.class);
+
         return sqlSessionFactoryBean.getObject();
     }
 }

src/main/java/ru/nbch/credit_tracker/config/datasource/handlers/UUIDTypeHandler.java

@@ -0,0 +1,53 @@
+package ru.nbch.credit_tracker.config.datasource.handlers;
+
+import java.sql.CallableStatement;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Types;
+import java.util.UUID;
+import org.apache.ibatis.type.JdbcType;
+import org.apache.ibatis.type.MappedTypes;
+import org.apache.ibatis.type.TypeHandler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@MappedTypes({UUID.class})
+public class UUIDTypeHandler implements TypeHandler<UUID> {
+    private static final Logger log = LoggerFactory.getLogger(UUIDTypeHandler.class);
+
+    @Override
+    public void setParameter(PreparedStatement ps, int i, UUID parameter, JdbcType jdbcType) throws SQLException {
+        if (parameter == null) {
+            ps.setObject(i, null, Types.OTHER);
+        } else {
+            ps.setObject(i, parameter.toString(), Types.OTHER);
+        }
+    }
+
+    @Override
+    public UUID getResult(ResultSet rs, String columnName) throws SQLException {
+        return toUUID(rs.getString(columnName));
+    }
+
+    @Override
+    public UUID getResult(ResultSet rs, int columnIndex) throws SQLException {
+        return toUUID(rs.getString(columnIndex));
+    }
+
+    @Override
+    public UUID getResult(CallableStatement cs, int columnIndex) throws SQLException {
+        return toUUID(cs.getString(columnIndex));
+    }
+
+    private static UUID toUUID(String val) {
+        if (val != null && !val.isEmpty() && !val.isBlank()) {
+            try {
+                return UUID.fromString(val);
+            } catch (IllegalArgumentException e) {
+                log.warn("Bad UUID found: {}", val);
+            }
+        }
+        return null;
+    }
+}

src/main/java/ru/nbch/credit_tracker/config/task/ConsumerConfigV2.java

@@ -0,0 +1,55 @@
+package ru.nbch.credit_tracker.config.task;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import ru.nbch.credit_tracker.model.PhoneFidPayloadV2;
+import ru.nbch.credit_tracker.model.SendSignalsPayload;
+import ru.nbch.credit_tracker.model.SubjectsProfilePayloadV2;
+import ru.nbch.credit_tracker.service.task.FidIdEnricherV2;
+import ru.nbch.credit_tracker.service.task.FidNameEnricherV2;
+import ru.nbch.credit_tracker.service.task.TaskConsumerV2;
+import ru.nbch.credit_tracker.service.task.TaskPayloadV2;
+
+@Configuration
+public class ConsumerConfigV2 {
+
+    @Bean
+    public BlockingQueue<PhoneFidPayloadV2> savingRequest2() {
+        return new LinkedBlockingQueue<>();
+    }
+    @Bean
+    public BlockingQueue<SubjectsProfilePayloadV2> enricherQueue2() {
+        return new LinkedBlockingQueue<>();
+    }
+
+    @Bean
+    public BlockingQueue<SendSignalsPayload> enricherIdQueue2() {
+        return new LinkedBlockingQueue<>();
+    }
+
+    @Bean
+    public TaskPayloadV2 POISON() {
+        return () -> true;
+    }
+
+    @SuppressWarnings("unchecked")
+    @Bean
+    public Integer consumerConfigurer(BlockingQueue<PhoneFidPayloadV2> savingRequest2,
+                                     BlockingQueue<SubjectsProfilePayloadV2> enricherQueue2,
+                                     BlockingQueue<SendSignalsPayload> enricherIdQueue2,
+                                     FidNameEnricherV2 fidNameEnricherV2,
+                                     FidIdEnricherV2 fidIdEnricherV2) {
+
+        SubjectsProfilePayloadV2 POISON = new SubjectsProfilePayloadV2(true);
+        SendSignalsPayload FID_ID_ENRICHER_POISON = new SendSignalsPayload(true);
+        TaskConsumerV2 fidNameEnricher = new TaskConsumerV2(fidNameEnricherV2, savingRequest2, enricherQueue2, POISON);
+        TaskConsumerV2 fidIdEnricher = new TaskConsumerV2(fidIdEnricherV2, enricherQueue2, enricherIdQueue2, FID_ID_ENRICHER_POISON);
+        Thread thread1 = new Thread(fidNameEnricher::start);
+        Thread thread2 = new Thread(fidIdEnricher::start);
+        thread1.start();
+        thread2.start();
+        return 0;
+    }
+}

src/main/java/ru/nbch/credit_tracker/entities/dto/PhoneFidMap.java

@@ -1,5 +1,6 @@
 package ru.nbch.credit_tracker.entities.dto;
 
+import java.util.UUID;
 import lombok.*;
 
 import java.time.LocalDate;
@@ -12,6 +13,7 @@ import java.time.LocalDateTime;
 @Builder
 public class PhoneFidMap {
     private Integer id;
+    private UUID recUuid;
     private Integer packageId;
     private String subjectId;
     private String phone;

src/main/java/ru/nbch/credit_tracker/model/PackageToSavePayloadV2.java

@@ -0,0 +1,33 @@
+package ru.nbch.credit_tracker.model;
+
+import ru.nbch.credit_tracker.service.task.TaskPayloadV2;
+
+public class PackageToSavePayloadV2 implements TaskPayloadV2 {
+    private final Integer packageId;
+    private final String filePath;
+    private final boolean poison;
+
+    public PackageToSavePayloadV2(boolean poison) {
+        this.packageId = null;
+        this.filePath = null;
+        this.poison = true;
+    }
+    public PackageToSavePayloadV2(Integer packageId, String filePath) {
+        this.packageId = packageId;
+        this.filePath = filePath;
+        this.poison = false;
+    }
+
+    public Integer getPackageId() {
+        return packageId;
+    }
+
+    public String getFilePath() {
+        return filePath;
+    }
+
+    @Override
+    public boolean isPoison() {
+        return poison;
+    }
+}

src/main/java/ru/nbch/credit_tracker/model/PhoneFidPayloadV2.java

@@ -0,0 +1,36 @@
+package ru.nbch.credit_tracker.model;
+
+import java.util.List;
+import ru.nbch.credit_tracker.entities.dto.PhoneFidMap;
+import ru.nbch.credit_tracker.service.task.TaskPayloadV2;
+
+public class PhoneFidPayloadV2 implements TaskPayloadV2 {
+    private final Integer packageId;
+    private final List<PhoneFidMap> phoneFidsBatch;
+    private final boolean poison;
+
+    public PhoneFidPayloadV2(boolean poison) {
+        this.packageId = null;
+        this.phoneFidsBatch = null;
+        this.poison = poison;
+    }
+
+    public PhoneFidPayloadV2(Integer packageId, List<PhoneFidMap> phoneFidsBatch) {
+        this.packageId = packageId;
+        this.phoneFidsBatch = phoneFidsBatch;
+        this.poison = false;
+    }
+
+    public Integer getPackageId() {
+        return packageId;
+    }
+
+    public List<PhoneFidMap> getPhoneFidsBatch() {
+        return phoneFidsBatch;
+    }
+
+    @Override
+    public boolean isPoison() {
+        return poison;
+    }
+}

src/main/java/ru/nbch/credit_tracker/model/SendSignalsPayload.java

@@ -0,0 +1,17 @@
+package ru.nbch.credit_tracker.model;
+
+import ru.nbch.credit_tracker.service.task.TaskPayloadV2;
+
+public class SendSignalsPayload implements TaskPayloadV2 {
+    private String signalXmlPath;
+    private boolean poison;
+
+    public SendSignalsPayload(boolean poison) {
+        this.poison = poison;
+    }
+
+    @Override
+    public boolean isPoison() {
+        return poison;
+    }
+}

src/main/java/ru/nbch/credit_tracker/model/SubjectsProfilePayloadV2.java

@@ -0,0 +1,36 @@
+package ru.nbch.credit_tracker.model;
+
+import java.util.List;
+import ru.nbch.credit_tracker.entities.dto.SubjectProfile;
+import ru.nbch.credit_tracker.service.task.TaskPayloadV2;
+
+public class SubjectsProfilePayloadV2 implements TaskPayloadV2 {
+    private final Integer packageId;
+    private final List<SubjectProfile> subjectProfiles;
+    private final boolean poison;
+
+    public SubjectsProfilePayloadV2(boolean poison) {
+        this.packageId = null;
+        this.subjectProfiles = null;
+        this.poison = poison;
+    }
+
+    public SubjectsProfilePayloadV2(Integer packageId, List<SubjectProfile> subjectProfiles) {
+        this.packageId = packageId;
+        this.subjectProfiles = subjectProfiles;
+        this.poison = false;
+    }
+
+    public Integer getPackageId() {
+        return packageId;
+    }
+
+    public List<SubjectProfile> getSubjectProfiles() {
+        return subjectProfiles;
+    }
+
+    @Override
+    public boolean isPoison() {
+        return poison;
+    }
+}

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

@@ -1,5 +1,9 @@
 package ru.nbch.credit_tracker.repository;
 
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.time.LocalDate;
+import java.util.List;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
@@ -9,48 +13,45 @@ import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
 import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
 import org.springframework.stereotype.Repository;
 
-import java.time.LocalDate;
-import java.util.List;
-
 @Repository
 public class PhoneRepository {
 
     private static final String SELCT_FIDS = """
-            SELECT
+        SELECT
-                pn.number,
+            pn.number,
-                pn.fid,
+            pn.fid,
-                pn.file_since_dt
+            pn.file_since_dt
-            FROM PHONE_NORM pn
+        FROM PHONE_NORM pn
-                JOIN (SELECT DISTINCT fid FROM NAME) nf
+            JOIN (SELECT DISTINCT fid FROM NAME) nf
-                ON nf.fid = pn.fid
+            ON nf.fid = pn.fid
-            WHERE number in (:phones)
+        WHERE number in (:phones)
-            ORDER BY
+        ORDER BY
-                pn.number ASC,
+            pn.number ASC,
-                pn.file_since_dt DESC,
+            pn.file_since_dt DESC,
-                pn.fid DESC
+            pn.fid DESC
-            WITH UR
+        WITH UR
-            """;
+        """;
 
     private static final String SELCT_FIDS_WITH_NAME = """
-            SELECT
+        SELECT
-                pn.number,
+            pn.number,
-                pn.fid,
+            pn.fid,
-                pn.file_since_dt,
+            pn.file_since_dt,
-                nf.first,
+            nf.first,
-                nf.middle,
+            nf.middle,
-                nf.name_1,
+            nf.name_1,
-                nf.birth_dt,
+            nf.birth_dt,
-                nf.file_since_dt as file_since_dt_name
+            nf.file_since_dt as file_since_dt_name
-            FROM PHONE_NORM pn
+        FROM PHONE_NORM pn
-                     JOIN NAME nf
+                 JOIN NAME nf
-                          ON nf.fid = pn.fid
+                      ON nf.fid = pn.fid
-            WHERE number in (:phones)
+        WHERE number in (:phones)
-            ORDER BY
+        ORDER BY
-                pn.number ASC,
+            pn.number ASC,
-                pn.file_since_dt DESC,
+            pn.file_since_dt DESC,
-                pn.fid DESC
+            pn.fid DESC
-            WITH UR
+        WITH UR
-            """;
+        """;
 
     private final NamedParameterJdbcTemplate jdbcTemplate;
 
@@ -64,7 +65,7 @@ public class PhoneRepository {
         params.addValue("phones", phones);
 
         return jdbcTemplate.query(SELCT_FIDS, params, (rs, rowNum) ->
-                new PhoneRepository.FidsData(rs.getInt("fid"), rs.getString("number"), rs.getObject("file_since_dt", LocalDate.class))
+            new PhoneRepository.FidsData(rs.getInt("fid"), rs.getString("number"), rs.getObject("file_since_dt", LocalDate.class))
         );
     }
 
@@ -73,16 +74,21 @@ public class PhoneRepository {
         params.addValue("phones", phones);
 
         return jdbcTemplate.query(SELCT_FIDS_WITH_NAME, params, (rs, rowNum) ->
-                new PhoneRepository.FidsDataWithName(
+            new PhoneRepository.FidsDataWithName(
-                        rs.getInt("FID"),
+                rs.getInt("FID"),
-                        rs.getString("NUMBER"),
+                rs.getString("NUMBER"),
-                        rs.getObject("FILE_SINCE_DT", LocalDate.class),
+                getLocalDate(rs, "FILE_SINCE_DT"),
-                        rs.getString("FIRST"),
+                rs.getString("FIRST"),
-                        rs.getString("NAME_1"),
+                rs.getString("NAME_1"),
-                        rs.getString("MIDDLE"),
+                rs.getString("MIDDLE"),
-                        rs.getObject("BIRTH_DT", LocalDate.class),
+                getLocalDate(rs, "BIRTH_DT"),
-                        rs.getObject("FILE_SINCE_DT_NAME", LocalDate.class)
+                getLocalDate(rs, "FILE_SINCE_DT_NAME")
-                ));
+            ));
+    }
+
+    private static LocalDate getLocalDate(ResultSet rs, String col) throws SQLException {
+        java.sql.Date d = rs.getDate(col);
+        return d != null ? d.toLocalDate() : null;
     }
 
     public List<FidsDataLong> findFidsLong(List<String> phones) {
@@ -90,7 +96,7 @@ public class PhoneRepository {
         params.addValue("phones", phones);
 
         return jdbcTemplate.query(SELCT_FIDS, params, (rs, rowNum) ->
-                new PhoneRepository.FidsDataLong(rs.getInt("fid"), Long.parseLong(rs.getString("number")), rs.getObject("file_since_dt", LocalDate.class))
+            new PhoneRepository.FidsDataLong(rs.getInt("fid"), Long.parseLong(rs.getString("number")), rs.getObject("file_since_dt", LocalDate.class))
         );
     }
 

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

@@ -0,0 +1,5 @@
+package ru.nbch.credit_tracker.service.task;
+
+public interface DispatcherV2<X, Z> {
+    Z handle(X taskPayload);
+}

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

@@ -42,13 +42,14 @@ public class FidIdEnricher implements Dispatcher<SubjectsProfilePayload>{
 
     @Override
     public void handle(SubjectsProfilePayload taskPayload) {
-        log.debug("FidIdEnricher handle by packageId: {}", taskPayload.getPackageId());
         List<SubjectProfile> subjectsProfile = taskPayload.getSubjectProfiles();
+        log.debug("FidIdEnricher handle by packageId: {}; size: {}", taskPayload.getPackageId(), subjectsProfile.size());
         personService.filledPassportsData(subjectsProfile);
         List<Person> personOnProcessing = subjectsProfile.stream()
             .filter(SubjectProfile::isFilled)
             .map(subjectProfile -> PersonFactoryBuilderV2.builder().personData(subjectProfile).build())
             .toList();
+        log.debug("personOnProcessing size: {}", personOnProcessing.size());
         writeSignals(personOnProcessing, taskPayload.getPackageId());
     }
 

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

@@ -0,0 +1,89 @@
+package ru.nbch.credit_tracker.service.task;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
+import java.nio.file.StandardOpenOption;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.BlockingQueue;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import ru.nbch.credit_tracker.builder.PersonFactoryBuilderV2;
+import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
+import ru.nbch.credit_tracker.entities.dto.SubjectProfile;
+import ru.nbch.credit_tracker.entities.external_request.signals.Person;
+import ru.nbch.credit_tracker.model.PartFilePayload;
+import ru.nbch.credit_tracker.model.SendSignalsPayload;
+import ru.nbch.credit_tracker.model.SubjectsProfilePayloadV2;
+import ru.nbch.credit_tracker.service.SignalService;
+import ru.nbch.credit_tracker.service.indic.PersonService;
+import ru.nbch.credit_tracker.service.xml.stax.SignalsStaxXmlWriterV2;
+
+@Service
+@Slf4j
+public class FidIdEnricherV2 implements DispatcherV2<SubjectsProfilePayloadV2, SendSignalsPayload>{
+    private final PersonService personService;
+    private final SignalService signalService;
+    private final CreditTrackerSettings config;
+    private final BlockingQueue<PartFilePayload> mergerQueue;
+    public FidIdEnricherV2(PersonService personService,
+                           SignalService signalService,
+                           CreditTrackerSettings config,
+                           BlockingQueue<PartFilePayload> mergerQueue) {
+        this.personService = personService;
+        this.signalService = signalService;
+        this.config = config;
+        this.mergerQueue = mergerQueue;
+    }
+
+    @Override
+    public SendSignalsPayload handle(SubjectsProfilePayloadV2 taskPayload) {
+        List<SubjectProfile> subjectsProfile = taskPayload.getSubjectProfiles();
+        log.debug("FidIdEnricher handle by packageId: {}; size: {}", taskPayload.getPackageId(), subjectsProfile.size());
+        personService.filledPassportsData(subjectsProfile);
+        List<Person> personOnProcessing = subjectsProfile.stream()
+            .filter(SubjectProfile::isFilled)
+            .map(subjectProfile -> PersonFactoryBuilderV2.builder().personData(subjectProfile).build())
+            .toList();
+        log.debug("personOnProcessing size: {}", personOnProcessing.size());
+        writeSignals(personOnProcessing, taskPayload.getPackageId());
+        return new SendSignalsPayload(false);
+    }
+
+    private void writeSignals(List<Person> persons, Integer packageId){
+        Path dir = Paths.get(config.getLoadDir(), "tmp_folder");
+        try {
+            Files.createDirectories(dir);
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+
+        String base = UUID.randomUUID() + "_output.xml";
+        log.debug("Write tmp xml by packageId: {}", packageId);
+        Path tmp = dir.resolve(base + ".part");
+        Path done = dir.resolve(base);
+
+//        Signals signals = signalService.buildSignals(packageId);
+
+        try (OutputStream os = Files.newOutputStream(tmp, StandardOpenOption.CREATE_NEW)) {
+            SignalsStaxXmlWriterV2 writer = new SignalsStaxXmlWriterV2(persons, null, config.getEncoding());
+            writer.writeWithStax(os); // перегрузи на OutputStream, если нужно
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+        try {
+            Files.move(tmp, done, StandardCopyOption.ATOMIC_MOVE);
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+        try {
+            mergerQueue.put(new PartFilePayload(done, packageId, null));
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        }
+    }
+}

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

@@ -0,0 +1,56 @@
+package ru.nbch.credit_tracker.service.task;
+
+import java.util.List;
+import java.util.Map;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import ru.nbch.credit_tracker.entities.dto.NameProfile;
+import ru.nbch.credit_tracker.entities.dto.PhoneFidMap;
+import ru.nbch.credit_tracker.entities.dto.SubjectProfile;
+import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper;
+import ru.nbch.credit_tracker.model.PhoneFidPayloadV2;
+import ru.nbch.credit_tracker.model.SubjectsProfilePayloadV2;
+import ru.nbch.credit_tracker.service.indic.PhoneService2;
+import ru.nbch.credit_tracker.service.signal.PhoneFidMapService;
+
+@Service
+@Slf4j
+public class FidNameEnricherV2 implements DispatcherV2<PhoneFidPayloadV2, SubjectsProfilePayloadV2> {
+    private final PhoneService2 phoneService2;
+    private final PhoneFidMapService phoneFidMapService;
+    private final PhoneFidMapMapper mapper;
+
+    public FidNameEnricherV2(PhoneService2 phoneService2,
+                             PhoneFidMapService phoneFidMapService,
+                             PhoneFidMapMapper mapper) {
+        this.phoneService2 = phoneService2;
+        this.phoneFidMapService = phoneFidMapService;
+        this.mapper = mapper;
+    }
+
+    @Override
+    public SubjectsProfilePayloadV2 handle(PhoneFidPayloadV2 taskPayload) {
+        List<PhoneFidMap> phoneFidMap = taskPayload.getPhoneFidsBatch();
+        log.debug("FidNameEnricher handle by packageId: {}, size: {}", taskPayload.getPackageId(), phoneFidMap.size());
+//        phoneFidMapService.register(phoneFidMap, taskPayload.getPackageId());
+        Map<Integer, PhoneFidMap> fidsMapByBatch = phoneService2.defineFidsAndNameByPhone(phoneFidMap);
+        log.debug("Found {} fids by batch", fidsMapByBatch.size());
+        //todo по phoneFidMap может быть найден не полный набор фидов, поэтому стоит это контролировать
+//        mapper.updateFid(fidsMapByBatch.values());
+
+        List<SubjectProfile> subjectProfiles = fidsMapByBatch.values().stream().map(pf -> {
+            SubjectProfile subjectProfile = new SubjectProfile();
+            subjectProfile.setFid(pf.getFid());
+            subjectProfile.setNameProfile(NameProfile.builder()
+                .firstName(pf.getFirstName())
+                .middleName(pf.getMiddleName())
+                .lastName(pf.getLastName())
+                .birthDate(pf.getBirthDate())
+                .fileSinceDate(pf.getFileSinceDate())
+                .build());
+            return subjectProfile;
+        }).toList();
+
+        return new SubjectsProfilePayloadV2(taskPayload.getPackageId(), subjectProfiles);
+    }
+}

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

@@ -7,6 +7,8 @@ import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.List;
 import java.util.Queue;
+import java.util.UUID;
+import java.util.concurrent.BlockingQueue;
 import java.util.stream.Collectors;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
@@ -14,6 +16,7 @@ import ru.nbch.credit_tracker.entities.dto.PhoneFidMap;
 import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest;
 import ru.nbch.credit_tracker.model.PhoneFidPayload;
 import ru.nbch.credit_tracker.model.PackageToSavePayload;
+import ru.nbch.credit_tracker.model.PhoneFidPayloadV2;
 import ru.nbch.credit_tracker.service.signal.PhoneFidMapService;
 import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
 
@@ -23,14 +26,17 @@ public class PackageSaver implements Dispatcher<PackageToSavePayload> {
     private final StaxXmlProcessor xmlProcessor;
     private final PhoneFidMapService phoneFidMapService;
     private final Queue<PhoneFidPayload> fidNameEnricherQueue;
+    private final BlockingQueue<PhoneFidPayloadV2> savingRequest2;
     private final static int BATCH_SIZE = 10_000;
 
     public PackageSaver(StaxXmlProcessor xmlProcessor,
                         PhoneFidMapService phoneFidMapService,
-                        Queue<PhoneFidPayload> fidNameEnricherQueue) {
+                        Queue<PhoneFidPayload> fidNameEnricherQueue,
+                        BlockingQueue<PhoneFidPayloadV2> savingRequest2) {
         this.xmlProcessor = xmlProcessor;
         this.phoneFidMapService = phoneFidMapService;
         this.fidNameEnricherQueue = fidNameEnricherQueue;
+        this.savingRequest2 = savingRequest2;
     }
     @Override
     public void handle(PackageToSavePayload taskPayload) {
@@ -43,11 +49,12 @@ public class PackageSaver implements Dispatcher<PackageToSavePayload> {
                     if (!request.getSubjects().getSubject().isEmpty()) {
                         List<PhoneFidMap> phoneFidMap = request.getSubjects().getSubject().stream().map(subject -> PhoneFidMap.builder()
                             .subjectId(subject.getId())
+                            .recUuid(UUID.randomUUID())
                             .phone(subject.getClientId())
                             .packageId(packageId)
                             .build()).collect(Collectors.toList());
                         log.trace("Reading subject's batch. Total count: {}. PackageId: {}", request.getSubjects().getSubject().size(), packageId);
-                        fidNameEnricherQueue.add(new PhoneFidPayload(packageId, phoneFidMap));
+                        savingRequest2.add(new PhoneFidPayloadV2(packageId, phoneFidMap));
                         request.getSubjects().getSubject().clear();
                     }
                 },
@@ -56,6 +63,8 @@ public class PackageSaver implements Dispatcher<PackageToSavePayload> {
                 BATCH_SIZE, "/MonitoringRequest/Subjects/Subject");
         } catch (IOException e) {
             throw new RuntimeException("Error while parsing monitoring request", e);
+        } finally {
+            savingRequest2.add(new PhoneFidPayloadV2(true));
         }
         log.info("Subjects were registered. PackageId: {}", packageId);
     }

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

@@ -30,7 +30,13 @@ public class TaskConsumer<X extends TaskPayload> implements SmartLifecycle {
             while (running && !Thread.currentThread().isInterrupted()) {
                 try {
                     X job = taskQueue.take();
-                    worker.submit(() -> dispatcher.handle(job));
+                    worker.submit(() -> {
+                        try {
+                            dispatcher.handle(job);
+                        } catch (Exception e) {
+                            log.error("Handle Error: {}", ExceptionUtils.getStackTrace(e));
+                        }
+                    });
                 } catch (InterruptedException e) {
                     Thread.currentThread().interrupt();
                 } catch (Exception e) {

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

@@ -0,0 +1,131 @@
+package ru.nbch.credit_tracker.service.task;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+
+@Slf4j
+public class TaskConsumerV2<X extends TaskPayloadV2, Z extends TaskPayloadV2>
+//    implements SmartLifecycle
+{
+    private final DispatcherV2<X, Z> dispatcher;
+    private final BlockingQueue<X> in;
+    private final BlockingQueue<Z> out;
+    private Z POISON;
+    private final ExecutorService ioPool;
+
+    public TaskConsumerV2(DispatcherV2<X, Z> dispatcher,
+                          BlockingQueue<X> in,
+                          BlockingQueue<Z> out,
+                          Z POISON) {
+        this.dispatcher = dispatcher;
+        this.in = in;
+        this.out = out;
+        this.ioPool = newIoExecutor();
+        this.POISON = POISON;
+    }
+
+//    @Override
+    public void start() {
+        List<Future<Z>> inflight = new ArrayList<>();
+        AtomicInteger inFlightCount = new AtomicInteger();
+
+        try {
+            while (true) {
+                X b = takeQuiet(in, Duration.ofSeconds(5));
+                if (b == null) continue;
+                if (b.isPoison()) {
+                    // дождаться всех in-flight задач
+                    for (Future<Z> f : inflight) {
+                        putQuiet(out, f.get());
+                    }
+                    putQuiet(out, POISON);
+                    return;
+                }
+                Future<Z> f = ioPool.submit(() -> dispatcher.handle(b));
+                inflight.add(f);
+
+                // чтобы не накапливать слишком много future в памяти, периодически сливаем
+                if (inFlightCount.incrementAndGet() % 8 == 0) {
+                    drainDone(inflight, out);
+                }
+            }
+        } catch (Exception e) {
+            log.error("Error: {}", ExceptionUtils.getStackTrace(e));
+            putQuiet(out, POISON);
+        } finally {
+            drainAll(inflight, out);
+        }
+    }
+
+    void drainDone(List<Future<Z>> fs, BlockingQueue<Z> out) throws Exception {
+        Iterator<Future<Z>> it = fs.iterator();
+        while (it.hasNext()) {
+            Future<Z> f = it.next();
+            if (f.isDone()) {
+                putQuiet(out, f.get());
+                it.remove();
+            }
+        }
+    }
+
+    void drainAll(List<Future<Z>> fs, BlockingQueue<Z> out) {
+        for (Future<Z> f : fs) {
+            try {
+                putQuiet(out, f.get());
+            } catch (Exception e) {
+                log.error("Error: {}", ExceptionUtils.getStackTrace(e));
+            }
+        }
+        fs.clear();
+    }
+
+    private ExecutorService newIoExecutor() {
+        // один carrier на ядро, задачи — виртуальные
+        return Executors.newThreadPerTaskExecutor(Thread.ofVirtual().name("vt-io-", 0).factory());
+    }
+
+    void putQuiet(BlockingQueue<Z> q, Z b) {
+        try {
+            q.put(b);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
+    }
+
+    X takeQuiet(BlockingQueue<X> q, Duration timeout) {
+        try {
+            return q.poll(timeout.toMillis(), TimeUnit.MILLISECONDS);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            return null;
+        }
+    }
+
+    static void sleep(long ms) {
+        try {
+            Thread.sleep(ms);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
+    }
+
+//    @Override
+    public void stop() {
+
+    }
+
+//    @Override
+    public boolean isRunning() {
+        return true;
+    }
+}

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

@@ -0,0 +1,5 @@
+package ru.nbch.credit_tracker.service.task;
+
+public interface TaskPayloadV2 {
+    boolean isPoison();
+}

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

@@ -11,7 +11,6 @@ import java.nio.file.StandardCopyOption;
 import java.nio.file.StandardOpenOption;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.UUID;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -91,7 +90,7 @@ public final class XmlMergeCoordinator implements InitializingBean {
 
     private void mergeBatch(List<PartFilePayload> inputs) throws Exception {
         Integer packageId = inputs.stream().findFirst().get().getPackageId();
-        String outName = "merged_" + UUID.randomUUID() + ".xml";
+        String outName = "merged_" + packageId + ".xml";
         Path tmp = mergedDir.resolve(outName + ".part");
         Path out = mergedDir.resolve(outName);
 
@@ -102,6 +101,7 @@ public final class XmlMergeCoordinator implements InitializingBean {
             w.writeStartElement("Persons");
 
             for (PartFilePayload in : inputs) {
+                log.debug("Copy person to file {}", in.getPartPath());
                 copyPersonsChildren(in.getPartPath(), w);
             }
 

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

@@ -25,16 +25,12 @@
         </foreach>
     </insert>
 
-    <insert id="register" parameterType="java.util.List"
+    <insert id="register" parameterType="java.util.List">
-            useGeneratedKeys="true"
+        INSERT INTO signals_phone_fid_map (rec_uuid, subject_id, phone, package_id)
-            keyProperty="phoneFidMap.id"
-            keyColumn="id">
-        INSERT INTO signals_phone_fid_map (subject_id, phone, package_id)
         values
         <foreach collection="phoneFidMap" item="phoneFid" separator=",">
-            (#{phoneFid.subjectId}, #{phoneFid.phone}, #{packageId})
+            (#{phoneFid.recUuid}, #{phoneFid.subjectId}, #{phoneFid.phone}, #{packageId})
         </foreach>
-        RETURNING id
     </insert>
 
 
@@ -73,7 +69,7 @@
         UPDATE signals_phone_fid_map
         SET
             fid = #{subject.fid}
-        WHERE id = #{subject.id}
+        WHERE rec_uuid = #{subject.id}
         </foreach>
     </update>
 
@@ -82,7 +78,7 @@
             UPDATE signals_phone_fid_map
             SET
                 error_text = #{subject.errorText}
-            WHERE id = #{subject.id}
+            WHERE id = #{subject.recUuid}
         </foreach>
     </update>