http://jira.mfd.msk:8088/browse/BKI-3343

credit-tracker.properties

@@ -93,6 +93,15 @@ ct.pipeline.id-enricher-queue-capacity=20
 #Save part of signal xml file and merge; only one thread
 ct.pipeline.merger-queue-capacity=50
 
+#STAGES SETTINGS
+#select NAME + FID BY PHONE
+ct.pipeline.name-updater-stage-concurrency=20
+ct.pipeline.name-updater-queue-capacity=20
+#select ID BY FID
+ct.pipeline.id-updater-stage-concurrency=10
+ct.pipeline.id-updater-queue-capacity=20
+
+
 #select and calc flags
 ct.flag-updater.concurrency=10
 ct.flag-updater.queue-capacity=100

pom.xml

@@ -233,7 +233,27 @@
             <plugin>
                 <groupId>org.springframework.boot</groupId>
                 <artifactId>spring-boot-maven-plugin</artifactId>
+
+                <executions>
+                    <execution>
+                        <id>repackage</id>
+                        <goals>
+                            <goal>repackage</goal>
+                        </goals>
                         <configuration>
+                            <skip>true</skip>
+                        </configuration>
+                    </execution>
+                    <execution>
+                        <id>build-common</id>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>repackage</goal>
+                        </goals>
+                        <configuration>
+                            <mainClass>ru.nbch.credit_tracker.service.mode.common.CTMainApplication</mainClass>
+                            <classifier>main</classifier>
+                            <attach>true</attach>
                             <excludes>
                                 <exclude>
                                     <groupId>org.projectlombok</groupId>
@@ -241,6 +261,46 @@
                                 </exclude>
                             </excludes>
                         </configuration>
+                    </execution>
+
+                    <execution>
+                        <id>build-monitoring</id>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>repackage</goal>
+                        </goals>
+                        <configuration>
+                            <mainClass>ru.nbch.credit_tracker.service.mode.monitoring.CTMonitoringApplication</mainClass>
+                            <classifier>monitoring</classifier>
+                            <attach>true</attach>
+                            <excludes>
+                                <exclude>
+                                    <groupId>org.projectlombok</groupId>
+                                    <artifactId>lombok</artifactId>
+                                </exclude>
+                            </excludes>
+                        </configuration>
+                    </execution>
+
+                    <execution>
+                        <id>build-migration</id>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>repackage</goal>
+                        </goals>
+                        <configuration>
+                            <mainClass>ru.nbch.credit_tracker.service.mode.migration.CTMigrationApplication</mainClass>
+                            <classifier>migration</classifier>
+                            <attach>true</attach>
+                            <excludes>
+                                <exclude>
+                                    <groupId>org.projectlombok</groupId>
+                                    <artifactId>lombok</artifactId>
+                                </exclude>
+                            </excludes>
+                        </configuration>
+                    </execution>
+                </executions>
             </plugin>
             <plugin>
                 <groupId>pl.project13.maven</groupId>

src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java

@@ -1,25 +0,0 @@
-package ru.nbch.credit_tracker;
-
-import org.springframework.boot.SpringApplication;
-import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.context.ConfigurableApplicationContext;
-import ru.nbch.credit_tracker.service.MonitoringService;
-import ru.nbch.credit_tracker.service.StartupRestoreService;
-
-@SpringBootApplication
-public class CreditTrackerApplication {
-
-	public static void main(String[] args) {
-		ConfigurableApplicationContext ctx = SpringApplication.run(CreditTrackerApplication.class, args);
-//		if (args.length > 0) {
-//			String arg = args[0];
-//			if (arg.equalsIgnoreCase("migration")) {
-//			}
-//		}
-		StartupRestoreService updateProcessingPackagesService = ctx.getBean(StartupRestoreService.class);
-		MonitoringService monitoringService = ctx.getBean(MonitoringService.class);
-		updateProcessingPackagesService.run();
-		monitoringService.run();
-	}
-
-}

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

@@ -12,6 +12,10 @@ public record PipelineSettings(
     @DefaultValue("20") Integer nameEnricherQueueCapacity,
     @DefaultValue("10") Integer idEnricherStageConcurrency,
     @DefaultValue("15") Integer idEnricherQueueCapacity,
+    @DefaultValue("15") Integer nameUpdaterStageConcurrency,
+    @DefaultValue("20") Integer nameUpdaterQueueCapacity,
+    @DefaultValue("10") Integer idUpdaterStageConcurrency,
+    @DefaultValue("15") Integer idUpdaterQueueCapacity,
     @DefaultValue("50") Integer mergerQueueCapacity,
     @DefaultValue("5") Integer errorAttemptCount,
     @DefaultValue("2s") Duration errorAttemptDelay,

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

@@ -9,9 +9,9 @@ import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
 @Configuration
 public class ExecutorsConfig {
 
-    @Bean
+    @Bean(destroyMethod = "shutdown")
     public ExecutorService pipelineExecutorService() {
-        return Executors.newFixedThreadPool(10, r -> {
+        return Executors.newFixedThreadPool(15, r -> {
             Thread t = new Thread(r);
             t.setName("pipe-starter-%d".formatted(t.threadId()));
             return t;

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

@@ -17,7 +17,7 @@ public class StageConfig {
 
     @Bean("nameUpdaterQueueFactory")
     public StageQueueFactory<SubjectPayload> nameUpdaterQueueFactory(CreditTrackerSettings creditTrackerSettings) {
-        return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameEnricherQueueCapacity());
+        return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameUpdaterQueueCapacity());
     }
 
     @Bean("idEnricherQueueFactory")
@@ -27,7 +27,7 @@ public class StageConfig {
 
     @Bean("idUpdaterQueueFactory")
     public StageQueueFactory<SubjectPayload> idUpdaterQueueFactory(CreditTrackerSettings creditTrackerSettings) {
-        return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameEnricherQueueCapacity());
+        return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameUpdaterQueueCapacity());
     }
 
     @Bean("noOpQueueFactory")

src/main/java/ru/nbch/credit_tracker/service/mode/common/CTMainApplication.java

@@ -0,0 +1,30 @@
+package ru.nbch.credit_tracker.service.mode.common;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.ComponentScan;
+import org.springframework.context.annotation.FilterType;
+
+@SpringBootApplication
+@ComponentScan(
+    value = {"ru.nbch.credit_tracker"},
+    excludeFilters = {
+        @ComponentScan.Filter(
+            type = FilterType.REGEX,
+            pattern = {
+                "ru\\.nbch\\.credit_tracker\\.service\\.mode\\.migration..*",
+                "ru\\.nbch\\.credit_tracker\\.service\\.mode\\.monitoring..*"
+            }
+        )
+    }
+)
+public class CTMainApplication {
+
+    public static void main(String[] args) {
+        ConfigurableApplicationContext ctx = SpringApplication.run(CTMainApplication.class, args);
+        StartupRestoreService updateProcessingPackagesService = ctx.getBean(StartupRestoreService.class);
+        updateProcessingPackagesService.run();
+    }
+
+}

src/main/java/ru/nbch/credit_tracker/service/mode/common/StartupRestoreService.java

@@ -1,4 +1,4 @@
-package ru.nbch.credit_tracker.service;
+package ru.nbch.credit_tracker.service.mode.common;
 
 import java.nio.file.Path;
 import java.util.List;
@@ -10,6 +10,7 @@ import ru.nbch.credit_tracker.entities.app.UserPackage;
 import ru.nbch.credit_tracker.enums.Status;
 import ru.nbch.credit_tracker.mapper.signal.PackagesRecordsMapper;
 import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
+import ru.nbch.credit_tracker.service.MonitoringErrorProcesingService;
 import ru.nbch.credit_tracker.service.task.Pipeline;
 import ru.nbch.credit_tracker.service.task.PipelinePackageFabric;
 import ru.nbch.credit_tracker.service.task.PipelinePackageOrchestrator;

src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFidAndPdService.java

@@ -1,4 +1,4 @@
-package ru.nbch.credit_tracker.service;
+package ru.nbch.credit_tracker.service.mode.common;
 
 import java.util.List;
 import java.util.concurrent.CompletableFuture;

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

@@ -1,4 +1,4 @@
-package ru.nbch.credit_tracker.service;
+package ru.nbch.credit_tracker.service.mode.common;
 
 import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;

src/main/java/ru/nbch/credit_tracker/service/mode/migration/CTMigrationApplication.java

@@ -0,0 +1,32 @@
+package ru.nbch.credit_tracker.service.mode.migration;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.WebApplicationType;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.ComponentScan;
+import org.springframework.context.annotation.FilterType;
+
+@SpringBootApplication
+@ComponentScan(
+    value = {"ru.nbch.credit_tracker"},
+    excludeFilters = {
+        @ComponentScan.Filter(
+            type = FilterType.REGEX,
+            pattern = {
+                "ru\\.nbch\\.credit_tracker\\.service\\.mode\\.common..*",
+                "ru\\.nbch\\.credit_tracker\\.service\\.mode\\.monitoring..*"
+            }
+        )
+    }
+)
+public class CTMigrationApplication {
+    public static void main(String[] args) {
+        SpringApplication application = new SpringApplication(CTMigrationApplication.class);
+        application.setWebApplicationType(WebApplicationType.NONE);
+        ConfigurableApplicationContext ctx = application.run(args);
+        MigrationService migrationService = ctx.getBean(MigrationService.class);
+        migrationService.update();
+    }
+
+}

src/main/java/ru/nbch/credit_tracker/service/mode/migration/MigrationService.java

@@ -1,15 +1,13 @@
-package ru.nbch.credit_tracker.service;
+package ru.nbch.credit_tracker.service.mode.migration;
 
 import java.util.concurrent.CompletableFuture;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.SpringApplication;
+import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.scheduling.annotation.EnableScheduling;
 import org.springframework.stereotype.Service;
-import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
-import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
-import ru.nbch.credit_tracker.service.signal.PhonesMapService;
 import ru.nbch.credit_tracker.service.task.Pipeline;
 import ru.nbch.credit_tracker.service.task.PipelinePackageFabric;
-import ru.nbch.credit_tracker.service.task.PipelinePackageOrchestrator;
 import ru.nbch.credit_tracker.service.task.PipelineRecalcPDOrchestrator;
 import ru.nbch.credit_tracker.service.task.SetupParam;
 import ru.nbch.credit_tracker.service.task.impl.PipelineMode;
@@ -20,21 +18,14 @@ import ru.nbch.credit_tracker.service.task.impl.PipelineMode;
 public class MigrationService {
     private final PipelinePackageFabric pipelinePackageManager;
     private final PipelineRecalcPDOrchestrator pipelineRecalcPDOrchestrator;
-    private final PipelinePackageOrchestrator pipelinePackageOrchestrator;
+    private final ConfigurableApplicationContext context;
-    private final UserPackageMapper userPackageMapper;
-    private final PhonesMapService phonesMapService;
 
-    public MigrationService(CreditTrackerSettings config,
+    public MigrationService(PipelinePackageFabric pipelinePackageManager,
-                            PipelinePackageFabric pipelinePackageManager,
                             PipelineRecalcPDOrchestrator pipelineRecalcPDOrchestrator,
-                            PipelinePackageOrchestrator pipelinePackageOrchestrator,
+                            ConfigurableApplicationContext context) {
-                            UserPackageMapper userPackageMapper,
-                            PhonesMapService phonesMapService) {
         this.pipelinePackageManager = pipelinePackageManager;
         this.pipelineRecalcPDOrchestrator = pipelineRecalcPDOrchestrator;
-        this.pipelinePackageOrchestrator = pipelinePackageOrchestrator;
+        this.context = context;
-        this.userPackageMapper = userPackageMapper;
-        this.phonesMapService = phonesMapService;
     }
 
     public void update() {
@@ -51,6 +42,7 @@ public class MigrationService {
                     log.info("pipeline finished");
                     pipeline.close();
                     pipelineRecalcPDOrchestrator.release();
+                    new Thread(this::shutdown, "migration-app-exit").start();
                 }
             });
         } catch (Exception e) {
@@ -58,4 +50,12 @@ public class MigrationService {
         }
     }
 
+    private void shutdown() {
+        int exitCode = SpringApplication.exit(context, () -> {
+            // no errors
+            return 0;
+        });
+        System.exit(exitCode);
+    }
+
 }

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

@@ -0,0 +1,31 @@
+package ru.nbch.credit_tracker.service.mode.monitoring;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.WebApplicationType;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.ComponentScan;
+import org.springframework.context.annotation.FilterType;
+
+@SpringBootApplication
+@ComponentScan(
+    value = {"ru.nbch.credit_tracker"},
+    excludeFilters = {
+        @ComponentScan.Filter(
+            type = FilterType.REGEX,
+            pattern = {
+                "ru\\.nbch\\.credit_tracker\\.service\\.mode\\.common..*",
+                "ru\\.nbch\\.credit_tracker\\.service\\.mode\\.migration..*"
+            }
+        )
+    }
+)
+public class CTMonitoringApplication {
+    public static void main(String[] args) {
+        SpringApplication application = new SpringApplication(CTMonitoringApplication.class);
+        application.setWebApplicationType(WebApplicationType.NONE);
+        ConfigurableApplicationContext ctx = application.run(args);
+        MonitoringService monitoringService = ctx.getBean(MonitoringService.class);
+        monitoringService.run();
+    }
+}

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

@@ -1,4 +1,4 @@
-package ru.nbch.credit_tracker.service;
+package ru.nbch.credit_tracker.service.mode.monitoring;
 
 import java.io.BufferedInputStream;
 import java.io.IOException;

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

@@ -68,7 +68,7 @@ public class FidIdUpdater implements Handler<SubjectPayload, VoidPayload> {
 
     @Override
     public int preferredConcurrency() {
-        return creditTrackerSettings.getPipeline().idEnricherStageConcurrency();
+        return creditTrackerSettings.getPipeline().idUpdaterStageConcurrency();
     }
 
     @Override

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

@@ -92,7 +92,7 @@ public class FidNameUpdater implements Handler<SubjectPayload, SubjectPayload> {
 
     @Override
     public int preferredConcurrency() {
-        return creditTrackerSettings.getPipeline().nameEnricherStageConcurrency();
+        return creditTrackerSettings.getPipeline().nameUpdaterStageConcurrency();
     }
 
     @Override

src/main/resources/sql/v2.21.ddl

@@ -42,4 +42,15 @@ update signals_phone_fid_map set phone_l = phone::BIGINT where phone is not null
 update signals_user_packages set signal_types = '1' where signal_types is null;
 update signals_user_packages set external_package_name = concat(substring(membercode, 1, 6), '_', package_name) where membercode is not null and package_name is not null ;
 
+INSERT
+INTO signals_phones_map
+(fid, phone, flag_90_12, flag_debt5000, calculated_at)
+SELECT distinct on (phone::bigint) fid::int,
+                                   phone::bigint,
+                                   flag_90_12,
+                                   flag_debt5000,
+                                   '2025-01-01'
+FROM signals_phone_fid_map where fid is not null
+order by phone, calculated_at desc;
+
 insert into app_version values ('2.0')