treshenkov/credit_tracker
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Added env vars for 2 types of pipelines

Browse source
This commit is contained in:
Mikhail Trofimov 2026-07-03 15:55:51 +03:00
parent a93ed1ccb4
commit e6edf52d16
15 changed files with 106 additions and 64 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

48
credit-tracker.properties
View file

@ -1,4 +1,3 @@
# TODO delete unused env vars
server.port=${SERVER_PORT:8099}
server.tomcat.accesslog.enabled=true
server.tomcat.accesslog.directory=/opt/app/logs/access
@ -105,19 +104,31 @@ ct.encoding=windows-1251
#batch size for saving in phone_fid_map
ct.saver-subject-batch-size=${SAVER_SUBJECT_BATCH_SIZE:15000}
#limit on select phone_fid_map
ct.pipeline.subject-batch-size=${PIPELINE_SUBJECT_BATCH_SIZE:10000}
ct.np-pipeline.subject-batch-size=${NP_PIPELINE_SUBJECT_BATCH_SIZE:10000}
#limit on select signals_legal_map
ct.lp-pipeline.subject-batch-size=${LP_PIPELINE_SUBJECT_BATCH_SIZE:10000}
#STAGES SETTINGS
#select NAME + FID BY PHONE
ct.pipeline.name-enricher-stage-concurrency=${PIPELINE_NAME_ENRICHER_STAGE_CONCURRENCY:20}
ct.pipeline.name-enricher-queue-capacity=${PIPELINE_NAME_ENRICHER_QUEUE_CAPACITY:20}
ct.np-pipeline.name-enricher-stage-concurrency=${NP_PIPELINE_NAME_ENRICHER_STAGE_CONCURRENCY:20}
ct.np-pipeline.name-enricher-queue-capacity=${NP_PIPELINE_NAME_ENRICHER_QUEUE_CAPACITY:20}
#select FID BY INN+REG_NUM
ct.lp-pipeline.name-enricher-stage-concurrency=${LP_PIPELINE_NAME_ENRICHER_STAGE_CONCURRENCY:20}
ct.lp-pipeline.name-enricher-queue-capacity=${LP_PIPELINE_NAME_ENRICHER_QUEUE_CAPACITY:20}
#Save part of signal xml file and merge; only one thread
ct.pipeline.merger-queue-capacity=${PIPELINE_MERGER_QUEUE_CAPACITY:50}
ct.np-pipeline.merger-queue-capacity=${NP_PIPELINE_MERGER_QUEUE_CAPACITY:50}
ct.lp-pipeline.merger-queue-capacity=${LP_PIPELINE_MERGER_QUEUE_CAPACITY:50}
#STAGES SETTINGS
#select NAME + FID BY PHONE
ct.pipeline.name-updater-stage-concurrency=${PIPELINE_NAME_UPDATER_STAGE_CONCURRENCY:20}
ct.pipeline.name-updater-queue-capacity=${PIPELINE_NAME_UPDATER_QUEUE_CAPACITY:20}
ct.np-pipeline.name-updater-stage-concurrency=${NP_PIPELINE_NAME_UPDATER_STAGE_CONCURRENCY:20}
ct.np-pipeline.name-updater-queue-capacity=${NP_PIPELINE_NAME_UPDATER_QUEUE_CAPACITY:20}
#select FID BY INN+REG_NUM
ct.lp-pipeline.name-updater-stage-concurrency=${LP_PIPELINE_NAME_UPDATER_STAGE_CONCURRENCY:20}
ct.lp-pipeline.name-updater-queue-capacity=${LP_PIPELINE_NAME_UPDATER_QUEUE_CAPACITY:20}
#select and calc flags
@ -125,15 +136,22 @@ ct.flag-updater.concurrency=${FLAG_UPDATER_CONCURRENCY:10}
ct.flag-updater.queue-capacity=${FLAG_UPDATER_QUEUE_CAPACITY:100}
#select and calc last rejected date
ct.update-rejected-date-cron-expression=${UPDATE_REJECTED_DATE_CRON_EXPRESSION:0 25 18 * * *}
ct.rejected-date-updater.concurrency=${REJECTED_DATE_UPDATER_CONCURRENCY:10}
ct.rejected-date-updater.queue-capacity=${REJECTED_DATE_UPDATER_QUEUE_CAPACITY:100}
ct.update-rejected-date-cron-expression=${LP_UPDATE_REJECTED_DATE_CRON_EXPRESSION:0 25 18 * * *}
ct.rejected-date-updater.concurrency=${LP_REJECTED_DATE_UPDATER_CONCURRENCY:10}
ct.rejected-date-updater.queue-capacity=${LP_REJECTED_DATE_UPDATER_QUEUE_CAPACITY:100}
#Backoff settings
ct.pipeline.error-attempt-count=${PIPELINE_ERROR_ATTEMPT_COUNT:5}
ct.pipeline.error-attempt-delay=${PIPELINE_ERROR_ATTEMPT_DELAY:2S}
ct.pipeline.retryable-exceptions=${PIPELINE_RETRYABLE_EXCEPTIONS:org.springframework.dao.DataAccessException,org.springframework.dao.DataIntegrityViolationException}
ct.np-pipeline.error-attempt-count=${NP_PIPELINE_ERROR_ATTEMPT_COUNT:5}
ct.np-pipeline.error-attempt-delay=${NP_PIPELINE_ERROR_ATTEMPT_DELAY:2S}
ct.np-pipeline.retryable-exceptions=${NP_PIPELINE_RETRYABLE_EXCEPTIONS:org.springframework.dao.DataAccessException,org.springframework.dao.DataIntegrityViolationException}
ct.LP-pipeline.error-attempt-count=${LP_PIPELINE_ERROR_ATTEMPT_COUNT:5}
ct.LP-pipeline.error-attempt-delay=${LP_PIPELINE_ERROR_ATTEMPT_DELAY:2S}
ct.LP-pipeline.retryable-exceptions=${LP_PIPELINE_RETRYABLE_EXCEPTIONS:org.springframework.dao.DataAccessException,org.springframework.dao.DataIntegrityViolationException}
#Pipeline global settings
ct.pipeline.global-timeout=${PIPELINE_GLOBAL_TIMEOUT:2H}
ct.pipeline.parallel-requests=${PIPELINE_PARALLEL_REQUESTS:3}
ct.np-pipeline.global-timeout=${NP_PIPELINE_GLOBAL_TIMEOUT:2H}
ct.np-pipeline.parallel-requests=${NP_PIPELINE_PARALLEL_REQUESTS:3}
ct.lp-pipeline.global-timeout=${LP_PIPELINE_GLOBAL_TIMEOUT:2H}
ct.lp-pipeline.parallel-requests=${LP_PIPELINE_PARALLEL_REQUESTS:3}

46
docker_compose/credit_tracker_config.env
View file

@ -102,22 +102,38 @@ NEW_FLAGS_CRON_EXPRESSION=0 25 18 * * *
UPDATE_FIDS_CRON_EXPRESSION=0 10 12 * * *
DAYS_TO_UPDATE_FIDS=30
# общие настройки
SAVER_SUBJECT_BATCH_SIZE=15000
PIPELINE_SUBJECT_BATCH_SIZE=10000
PIPELINE_NAME_ENRICHER_STAGE_CONCURRENCY=20
PIPELINE_NAME_ENRICHER_QUEUE_CAPACITY=20
PIPELINE_ID_ENRICHER_STAGE_CONCURRENCY=10
PIPELINE_ID_ENRICHER_QUEUE_CAPACITY=20
PIPELINE_MERGER_QUEUE_CAPACITY=50
PIPELINE_NAME_UPDATER_STAGE_CONCURRENCY=20
PIPELINE_NAME_UPDATER_QUEUE_CAPACITY=20
PIPELINE_ID_UPDATER_STAGE_CONCURRENCY=10
PIPELINE_ID_UPDATER_QUEUE_CAPACITY=20
FLAG_UPDATER_CONCURRENCY=10
FLAG_UPDATER_QUEUE_CAPACITY=100
PIPELINE_ERROR_ATTEMPT_COUNT=5
PIPELINE_ERROR_ATTEMPT_DELAY=2S
PIPELINE_RETRYABLE_EXCEPTIONS=org.springframework.dao.DataAccessException,org.springframework.dao.DataIntegrityViolationException}
PIPELINE_GLOBAL_TIMEOUT=2H
PIPELINE_PARALLEL_REQUESTS=3
# Физ лица
NP_PIPELINE_SUBJECT_BATCH_SIZE=10000
NP_PIPELINE_NAME_ENRICHER_STAGE_CONCURRENCY=20
NP_PIPELINE_NAME_ENRICHER_QUEUE_CAPACITY=20
NP_PIPELINE_MERGER_QUEUE_CAPACITY=50
NP_PIPELINE_NAME_UPDATER_STAGE_CONCURRENCY=20
NP_PIPELINE_NAME_UPDATER_QUEUE_CAPACITY=20
NP_PIPELINE_ERROR_ATTEMPT_COUNT=5
NP_PIPELINE_ERROR_ATTEMPT_DELAY=2S
NP_PIPELINE_RETRYABLE_EXCEPTIONS=org.springframework.dao.DataAccessException,org.springframework.dao.DataIntegrityViolationException}
NP_PIPELINE_GLOBAL_TIMEOUT=2H
NP_PIPELINE_PARALLEL_REQUESTS=3
# Юр. Лица
LP_PIPELINE_SUBJECT_BATCH_SIZE=10000
LP_PIPELINE_NAME_ENRICHER_STAGE_CONCURRENCY=20
LP_PIPELINE_NAME_ENRICHER_QUEUE_CAPACITY=20
LP_PIPELINE_MERGER_QUEUE_CAPACITY=50
LP_PIPELINE_NAME_UPDATER_STAGE_CONCURRENCY=20
LP_PIPELINE_NAME_UPDATER_QUEUE_CAPACITY=20
LP_PIPELINE_ERROR_ATTEMPT_COUNT=5
LP_PIPELINE_ERROR_ATTEMPT_DELAY=2S
LP_PIPELINE_RETRYABLE_EXCEPTIONS=org.springframework.dao.DataAccessException,org.springframework.dao.DataIntegrityViolationException}
LP_PIPELINE_GLOBAL_TIMEOUT=2H
LP_PIPELINE_PARALLEL_REQUESTS=3
LP_UPDATE_REJECTED_DATE_CRON_EXPRESSION=
LP_REJECTED_DATE_UPDATER_CONCURRENCY=10
LP_REJECTED_DATE_UPDATER_QUEUE_CAPACITY=100

37
docker_compose/docker-compose.yml
View file

@ -88,23 +88,30 @@ services:
- DAYS_TO_UPDATE_FIDS=${DAYS_TO_UPDATE_FIDS}
- DAYS_TO_UPDATE_FLAGS=${DAYS_TO_UPDATE_FLAGS}
- SAVER_SUBJECT_BATCH_SIZE=${SAVER_SUBJECT_BATCH_SIZE}
- PIPELINE_SUBJECT_BATCH_SIZE=${PIPELINE_SUBJECT_BATCH_SIZE}
- PIPELINE_NAME_ENRICHER_STAGE_CONCURRENCY=${PIPELINE_NAME_ENRICHER_STAGE_CONCURRENCY}
- PIPELINE_NAME_ENRICHER_QUEUE_CAPACITY=${PIPELINE_NAME_ENRICHER_QUEUE_CAPACITY}
- PIPELINE_ID_ENRICHER_STAGE_CONCURRENCY=${PIPELINE_ID_ENRICHER_STAGE_CONCURRENCY}
- PIPELINE_ID_ENRICHER_QUEUE_CAPACITY=${PIPELINE_ID_ENRICHER_QUEUE_CAPACITY}
- PIPELINE_MERGER_QUEUE_CAPACITY=${PIPELINE_MERGER_QUEUE_CAPACITY}
- PIPELINE_NAME_UPDATER_STAGE_CONCURRENCY=${PIPELINE_NAME_UPDATER_STAGE_CONCURRENCY}
- PIPELINE_NAME_UPDATER_QUEUE_CAPACITY=${PIPELINE_NAME_UPDATER_QUEUE_CAPACITY}
- PIPELINE_ID_UPDATER_STAGE_CONCURRENCY=${PIPELINE_ID_UPDATER_STAGE_CONCURRENCY}
- PIPELINE_ID_UPDATER_QUEUE_CAPACITY=${PIPELINE_ID_UPDATER_QUEUE_CAPACITY}
- FLAG_UPDATER_CONCURRENCY=${FLAG_UPDATER_CONCURRENCY}
- FLAG_UPDATER_QUEUE_CAPACITY=${FLAG_UPDATER_QUEUE_CAPACITY}
- PIPELINE_ERROR_ATTEMPT_COUNT=${PIPELINE_ERROR_ATTEMPT_COUNT}
- PIPELINE_ERROR_ATTEMPT_DELAY=${PIPELINE_ERROR_ATTEMPT_DELAY}
- PIPELINE_RETRYABLE_EXCEPTIONS=${PIPELINE_RETRYABLE_EXCEPTIONS}
- PIPELINE_GLOBAL_TIMEOUT=${PIPELINE_GLOBAL_TIMEOUT}
- PIPELINE_PARALLEL_REQUESTS=${PIPELINE_PARALLEL_REQUESTS}
- NP_PIPELINE_SUBJECT_BATCH_SIZE=${NP_PIPELINE_SUBJECT_BATCH_SIZE}
- NP_PIPELINE_NAME_ENRICHER_STAGE_CONCURRENCY=${NP_PIPELINE_NAME_ENRICHER_STAGE_CONCURRENCY}
- NP_PIPELINE_NAME_ENRICHER_QUEUE_CAPACITY=${NP_PIPELINE_NAME_ENRICHER_QUEUE_CAPACITY}
- NP_PIPELINE_MERGER_QUEUE_CAPACITY=${NP_PIPELINE_MERGER_QUEUE_CAPACITY}
- NP_PIPELINE_NAME_UPDATER_STAGE_CONCURRENCY=${NP_PIPELINE_NAME_UPDATER_STAGE_CONCURRENCY}
- NP_PIPELINE_NAME_UPDATER_QUEUE_CAPACITY=${NP_PIPELINE_NAME_UPDATER_QUEUE_CAPACITY}
- NP_PIPELINE_ERROR_ATTEMPT_COUNT=${NP_PIPELINE_ERROR_ATTEMPT_COUNT}
- NP_PIPELINE_ERROR_ATTEMPT_DELAY=${NP_PIPELINE_ERROR_ATTEMPT_DELAY}
- NP_PIPELINE_RETRYABLE_EXCEPTIONS=${NP_PIPELINE_RETRYABLE_EXCEPTIONS}
- NP_PIPELINE_GLOBAL_TIMEOUT=${NP_PIPELINE_GLOBAL_TIMEOUT}
- NP_PIPELINE_PARALLEL_REQUESTS=${NP_PIPELINE_PARALLEL_REQUESTS}
- LP_PIPELINE_SUBJECT_BATCH_SIZE=${LP_PIPELINE_SUBJECT_BATCH_SIZE}
- LP_PIPELINE_NAME_ENRICHER_STAGE_CONCURRENCY=${LP_PIPELINE_NAME_ENRICHER_STAGE_CONCURRENCY}
- LP_PIPELINE_NAME_ENRICHER_QUEUE_CAPACITY=${LP_PIPELINE_NAME_ENRICHER_QUEUE_CAPACITY}
- LP_PIPELINE_MERGER_QUEUE_CAPACITY=${LP_PIPELINE_MERGER_QUEUE_CAPACITY}
- LP_PIPELINE_NAME_UPDATER_STAGE_CONCURRENCY=${LP_PIPELINE_NAME_UPDATER_STAGE_CONCURRENCY}
- LP_PIPELINE_NAME_UPDATER_QUEUE_CAPACITY=${LP_PIPELINE_NAME_UPDATER_QUEUE_CAPACITY}
- LP_PIPELINE_ERROR_ATTEMPT_COUNT=${LP_PIPELINE_ERROR_ATTEMPT_COUNT}
- LP_PIPELINE_ERROR_ATTEMPT_DELAY=${LP_PIPELINE_ERROR_ATTEMPT_DELAY}
- LP_PIPELINE_RETRYABLE_EXCEPTIONS=${LP_PIPELINE_RETRYABLE_EXCEPTIONS}
- LP_PIPELINE_GLOBAL_TIMEOUT=${LP_PIPELINE_GLOBAL_TIMEOUT}
- LP_PIPELINE_PARALLEL_REQUESTS=${LP_PIPELINE_PARALLEL_REQUESTS}
command: ["/opt/app/start-credit-tracker.sh"]
credit-tracker-monitoring:

2
src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java
View file

@ -24,7 +24,7 @@ public class CreditTrackerSettings {
@NestedConfigurationProperty
private XsdValidation xsdValidation;
@NestedConfigurationProperty
private PipelineSettings pipeline;
private PipelineSettings npPipeline;
@NestedConfigurationProperty
private PipelineSettings lpPipeline;
@NestedConfigurationProperty

7
src/main/java/ru/nbch/credit_tracker/config/task/ExecutorsConfig.java
View file

@ -1,11 +1,12 @@
package ru.nbch.credit_tracker.config.task;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@Configuration
public class ExecutorsConfig {
@ -20,7 +21,7 @@ public class ExecutorsConfig {
@Bean
public ExecutorService saverExecutorService(CreditTrackerSettings creditTrackerSettings) {
return Executors.newFixedThreadPool(creditTrackerSettings.getPipeline().parallelRequests(), r -> {
return Executors.newFixedThreadPool(creditTrackerSettings.getNpPipeline().parallelRequests(), r -> {
Thread t = new Thread(r);
t.setName("subject-saver-%d".formatted(t.threadId()));
return t;

6
src/main/java/ru/nbch/credit_tracker/config/task/PipelineStarterConfig.java
View file

@ -60,9 +60,9 @@ public class PipelineStarterConfig {
@Bean
public PipelineBackoff npPipelineBackoff(CreditTrackerSettings creditTrackerSettings) {
return new PipelineBackoff(
creditTrackerSettings.getPipeline().errorAttemptCount(),
creditTrackerSettings.getPipeline().errorAttemptDelay(),
creditTrackerSettings.getPipeline().retryableExceptions()
creditTrackerSettings.getNpPipeline().errorAttemptCount(),
creditTrackerSettings.getNpPipeline().errorAttemptDelay(),
creditTrackerSettings.getNpPipeline().retryableExceptions()
);
}

8
src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java
View file

@ -12,22 +12,22 @@ public class StageConfig {
@Bean("npFidEnricherQueueFactory")
public StageQueueFactory<SubjectPayload> npFidEnricherQueueFactory(CreditTrackerSettings creditTrackerSettings) {
return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameEnricherQueueCapacity());
return new StageQueueFactory<>(creditTrackerSettings.getNpPipeline().nameEnricherQueueCapacity());
}
@Bean("npFidUpdaterQueueFactoryUpdater")
public StageQueueFactory<SubjectPayload> npFidUpdaterQueueFactoryUpdater(CreditTrackerSettings creditTrackerSettings) {
return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameUpdaterQueueCapacity());
return new StageQueueFactory<>(creditTrackerSettings.getNpPipeline().nameUpdaterQueueCapacity());
}
@Bean("npNoOpQueueFactory")
public StageQueueFactory<VoidPayload> npNoOpQueueFactory(CreditTrackerSettings creditTrackerSettings) {
return new StageQueueFactory<>(creditTrackerSettings.getPipeline().mergerQueueCapacity());
return new StageQueueFactory<>(creditTrackerSettings.getNpPipeline().mergerQueueCapacity());
}
@Bean("npFinishSetupQueueFactory")
public StageQueueFactory<SubjectPayload> npFinishSetupQueueFactory(CreditTrackerSettings creditTrackerSettings) {
return new StageQueueFactory<>(creditTrackerSettings.getPipeline().mergerQueueCapacity());
return new StageQueueFactory<>(creditTrackerSettings.getNpPipeline().mergerQueueCapacity());
}
// LP

2
src/main/java/ru/nbch/credit_tracker/service/task/LpPipelinePackageOrchestrator.java
View file

@ -21,6 +21,6 @@ public class LpPipelinePackageOrchestrator extends PackageOrchestrator {
@Override
protected PipelineSettings getPipeline() {
return creditTrackerSettings.getPipeline();
return creditTrackerSettings.getLpPipeline();
}
}

2
src/main/java/ru/nbch/credit_tracker/service/task/NpPipelinePackageFabric.java
View file

@ -81,6 +81,6 @@ public class NpPipelinePackageFabric implements InitializingBean {
@Override
public void afterPropertiesSet() throws Exception {
log.info("pipeline settings: {}", creditTrackerSettings.getPipeline());
log.info("pipeline settings: {}", creditTrackerSettings.getNpPipeline());
}
}

2
src/main/java/ru/nbch/credit_tracker/service/task/NpPipelinePackageOrchestrator.java
View file

@ -21,6 +21,6 @@ public class NpPipelinePackageOrchestrator extends PackageOrchestrator {
@Override
protected PipelineSettings getPipeline() {
return creditTrackerSettings.getPipeline();
return creditTrackerSettings.getNpPipeline();
}
}

2
src/main/java/ru/nbch/credit_tracker/service/task/NpPipelineRecalcOrchestrator.java
View file

@ -18,6 +18,6 @@ public class NpPipelineRecalcOrchestrator extends RecalcOrchestrator {
@Override
protected PipelineSettings getPipelineSettings() {
return creditTrackerSettings.getPipeline();
return creditTrackerSettings.getNpPipeline();
}
}

2
src/main/java/ru/nbch/credit_tracker/service/task/impl/NpFidEnricher.java
View file

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

2
src/main/java/ru/nbch/credit_tracker/service/task/impl/NpFidUpdater.java
View file

@ -90,7 +90,7 @@ public class NpFidUpdater implements Handler<SubjectPayload, VoidPayload> {
@Override
public int preferredConcurrency() {
return creditTrackerSettings.getPipeline().nameUpdaterStageConcurrency();
return creditTrackerSettings.getNpPipeline().nameUpdaterStageConcurrency();
}
@Override

2
src/main/java/ru/nbch/credit_tracker/service/task/impl/NpKickPipeLine.java
View file

@ -42,7 +42,7 @@ public abstract class NpKickPipeLine implements IPipeLineStarter<SubjectPayload>
this.userPackageMapper = userPackageMapper;
this.packageRecordsService = packageRecordsService;
this.creditTrackerSettings = creditTrackerSettings;
this.BATCH_SIZE = creditTrackerSettings.getPipeline().subjectBatchSize();
this.BATCH_SIZE = creditTrackerSettings.getNpPipeline().subjectBatchSize();
this.mode = setupParam.mode();
}

2
src/main/java/ru/nbch/credit_tracker/service/task/impl/NpKickRecalcPipeLine.java
View file

@ -43,7 +43,7 @@ public class NpKickRecalcPipeLine implements IPipeLineStarter<SubjectPayload> {
this.phonesMapMapper = phonesMapMapper;
this.packageRecordsService = packageRecordsService;
this.creditTrackerSettings = creditTrackerSettings;
this.BATCH_SIZE = creditTrackerSettings.getPipeline().subjectBatchSize();
this.BATCH_SIZE = creditTrackerSettings.getNpPipeline().subjectBatchSize();
this.daysToUpdateFids = creditTrackerSettings.getDaysToUpdateNpFids() != null ? creditTrackerSettings.getDaysToUpdateNpFids() : 1;
}