fix pipeline settings

credit-tracker.properties

@@ -75,19 +75,29 @@ ct.encoding=windows-1251
 
 
 #----------------------------Pipeline settings----------------------------#
-#???-?? ????????? ?? xml ?? ???? ????????????
+#batch size for saving in phone_fid_map
 ct.saver-subject-batch-size=15000
-ct.pipeline.subject-batch-size=5000
+#limit on select phone_fid_map
-#????????? ? PHONE ? NAME + ?????? ? ???? SIGNALS_PHONE_FID_MAP
+ct.pipeline.subject-batch-size=10000
+
+#STAGES SETTINGS
+#select NAME + FID BY PHONE
 ct.pipeline.name-enricher-stage-concurrency=20
-#????????? ? ID ? ?????? ????? xml ??? ??. ???????
+ct.pipeline.name-enricher-queue-capacity=20
+#select ID BY FID
 ct.pipeline.id-enricher-stage-concurrency=10
-#?????????? ??????? ??? ??????
+ct.pipeline.id-enricher-queue-capacity=20
+#Save part of signal xml file and merge; only one thread
+ct.pipeline.merger-queue-capacity=50
+#select and calc flags; instant run after name-enricher
+ct.pipeline.flag-enricher-stage-concurrency=10
+ct.pipeline.flag-enricher-queue-capacity=50
+
+#Backoff settings
 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
+
+#Pipeline global settings
+ct.pipeline.global-timeout=2h
+ct.pipeline.parallel-requests=3

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

@@ -6,9 +6,13 @@ import org.springframework.boot.context.properties.bind.DefaultValue;
 
 public record PipelineSettings(
     @DefaultValue("5000") Integer subjectBatchSize,
-    @DefaultValue("3") Integer saveSubjectStageConcurrency,
     @DefaultValue("15") Integer nameEnricherStageConcurrency,
+    @DefaultValue("20") Integer nameEnricherQueueCapacity,
     @DefaultValue("10") Integer idEnricherStageConcurrency,
+    @DefaultValue("15") Integer idEnricherQueueCapacity,
+    @DefaultValue("10") Integer flagEnricherStageConcurrency,
+    @DefaultValue("100") Integer flagEnricherQueueCapacity,
+    @DefaultValue("50") Integer mergerQueueCapacity,
     @DefaultValue("5") Integer errorAttemptCount,
     @DefaultValue("2s") Duration errorAttemptDelay,
     @DefaultValue("5m") Duration globalTimeout,

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

@@ -2,30 +2,26 @@ package ru.nbch.credit_tracker.config.task;
 
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
 import ru.nbch.credit_tracker.model.impl.PartFilePayload;
 import ru.nbch.credit_tracker.model.impl.SubjectPayload;
 
 @Configuration
 public class StageConfig {
 
-    @Bean("saveSubjectQueueFactory")
-    public StageQueueFactory<SubjectPayload> saveSubjectQueueFactory() {
-        return new StageQueueFactory<>(50);
-    }
-
     @Bean("nameEnricherQueueFactory")
-    public StageQueueFactory<SubjectPayload> nameEnricherQueueFactory() {
+    public StageQueueFactory<SubjectPayload> nameEnricherQueueFactory(CreditTrackerSettings creditTrackerSettings) {
-        return new StageQueueFactory<>(10);
+        return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameEnricherQueueCapacity());
     }
 
     @Bean("idEnricherQueueFactory")
-    public StageQueueFactory<SubjectPayload> idEnricherQueueFactory() {
+    public StageQueueFactory<SubjectPayload> idEnricherQueueFactory(CreditTrackerSettings creditTrackerSettings) {
-        return new StageQueueFactory<>(10);
+        return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameEnricherQueueCapacity());
     }
 
     @Bean("partFileQueueFactory")
-    public StageQueueFactory<PartFilePayload> partFileQueueFactory() {
+    public StageQueueFactory<PartFilePayload> partFileQueueFactory(CreditTrackerSettings creditTrackerSettings) {
-        return new StageQueueFactory<>(100);
+        return new StageQueueFactory<>(creditTrackerSettings.getPipeline().mergerQueueCapacity());
     }
 
 }

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

@@ -9,6 +9,7 @@ import java.util.concurrent.TimeUnit;
 import java.util.function.BiFunction;
 import java.util.function.Supplier;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.InitializingBean;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.stereotype.Component;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
@@ -30,7 +31,7 @@ import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
 
 @Component
 @Slf4j
-public class PipelinePackageManager {
+public class PipelinePackageManager implements InitializingBean {
     private final SignalRestService signalRestService;
     private final FlagsEnricher flagsEnricherV2;
     private final StageQueueFactory<SubjectPayload> idEnricherQueueFactory;
@@ -48,11 +49,7 @@ public class PipelinePackageManager {
         return t;
     });
 
-    private final ExecutorService subjectSaverExecutor = Executors.newFixedThreadPool(3, r -> {
+    private final ExecutorService subjectSaverExecutor;
-        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;
@@ -80,9 +77,13 @@ public class PipelinePackageManager {
         this.fidNameEnricher = fidNameEnricher;
         this.creditTrackerSettings = creditTrackerSettings;
         this.semaphore = new Semaphore(creditTrackerSettings.getPipeline().parallelRequests());
-        log.info("pipeline settings: {}", creditTrackerSettings.getPipeline());
         this.pipelineBackoff = pipelineBackoff;
         this.saver = saver;
+        this.subjectSaverExecutor = Executors.newFixedThreadPool(creditTrackerSettings.getPipeline().parallelRequests(), r -> {
+            Thread t = new Thread(r);
+            t.setName("subject-saver-%d".formatted(t.threadId()));
+            return t;
+        });
     }
 
     public void buildAndRun(Long packageId, Path pkgPath) throws Exception {
@@ -147,4 +148,9 @@ public class PipelinePackageManager {
             throw e;
         }
     }
+
+    @Override
+    public void afterPropertiesSet() throws Exception {
+        log.info("pipeline settings: {}", creditTrackerSettings.getPipeline());
+    }
 }

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

@@ -12,6 +12,7 @@ import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
+import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
 import ru.nbch.credit_tracker.entities.app.FlagDetail;
 import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper;
 import ru.nbch.credit_tracker.model.SubjectProfile;
@@ -26,12 +27,12 @@ public class FlagsEnricher implements Consumer<SubjectPayload> {
     private final ExecutorService ioPool;
 
     public FlagsEnricher(FlagsService2 flagsService,
-                         PhoneFidMapMapper phoneFidMapMapper) {
+                         CreditTrackerSettings creditTrackerSettings) {
         this.flagsService = flagsService;
-        this.ioPool = new ThreadPoolExecutor(10,
+        this.ioPool = new ThreadPoolExecutor(creditTrackerSettings.getPipeline().flagEnricherStageConcurrency(),
             10,
             0L, TimeUnit.MINUTES,
-            new LinkedBlockingQueue<>(100),
+            new LinkedBlockingQueue<>(creditTrackerSettings.getPipeline().flagEnricherQueueCapacity()),
             r -> {
                 Thread t = new Thread(r);
                 t.setName("flags-enricher-%d".formatted(t.threadId()));

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

@@ -1,67 +0,0 @@
-package ru.nbch.credit_tracker.service.task.impl;
-
-import java.util.List;
-import lombok.extern.slf4j.Slf4j;
-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.enums.Status;
-import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
-import ru.nbch.credit_tracker.model.SubjectProfile;
-import ru.nbch.credit_tracker.model.impl.SubjectPayload;
-import ru.nbch.credit_tracker.service.indic.PhoneService2;
-import ru.nbch.credit_tracker.service.task.Handler;
-
-/**
- * Стадия для сопоставления FID к номеру телефона и регистрации пакета в базе данных.
- * Так же выполняется поиск и обогащение перс.данными из NAME для дальнейшей обработки.
- */
-@Service
-@Slf4j
-public class SaveSubjectHandler implements Handler<SubjectPayload, SubjectPayload> {
-    private final PhoneService2 phoneService2;
-    private final UserPackageMapper userPackageMapper;
-    private final CreditTrackerSettings creditTrackerSettings;
-
-    public SaveSubjectHandler(PhoneService2 phoneService2,
-                              UserPackageMapper userPackageMapper,
-                              CreditTrackerSettings creditTrackerSettings) {
-        this.phoneService2 = phoneService2;
-        this.userPackageMapper = userPackageMapper;
-        this.creditTrackerSettings = creditTrackerSettings;
-    }
-
-    @Override
-    public SubjectPayload handle(SubjectPayload taskPayload) {
-        List<SubjectProfile> subjects = taskPayload.getSubjects();
-        Long packageId = taskPayload.getPackageId();
-        log.debug("SaveSubjectStage handle by packageId: {}, size of bucket for save: {}", taskPayload.getPackageId(), subjects.size());
-        List<PhoneFidMap> subjectForSave = subjects.stream().map(subject -> PhoneFidMap.builder()
-            .subjectId(subject.getIdentifier())
-            .packageId(packageId)
-            .phone(subject.getPhone())
-            .build()).toList();
-
-        List<Long> ids = phoneService2.registerSubjects(subjectForSave);
-        for (int i = 0; i < subjects.size(); i++) {
-            subjects.get(i).setMapId(ids.get(i));
-        }
-        return taskPayload;
-    }
-
-    @Override
-    public void handleOnPoison(SubjectPayload poison) {
-        log.trace("Got poison; set package on PROCESSING by packageId: {}", poison.getPackageId());
-        userPackageMapper.updatePackageStatus(poison.getPackageId(), Status.PROCESSING.name());
-    }
-
-    @Override
-    public int preferredConcurrency() {
-         return creditTrackerSettings.getPipeline().saveSubjectStageConcurrency();
-    }
-
-    @Override
-    public SubjectPayload cookPoison() {
-        return SubjectPayload.POISON;
-    }
-}