diff --git a/credit-tracker.properties b/credit-tracker.properties index 5703426..7d10cbc 100644 --- a/credit-tracker.properties +++ b/credit-tracker.properties @@ -22,9 +22,9 @@ ct.indic-data-source-settings.leak-detection-threshold=${LEAK_DETECTION_THRESHOL # signal 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?currentSchema=test_2} +ct.signal-data-source-settings.url=${DB_URL_SIGNAL:jdbc:postgresql://10.230.227.125:5432/credit_tracker?currentSchema=test} ct.signal-data-source-settings.username=${USERNAME_SIGNAL:credit_tracker_user} -ct.signal-data-source-settings.password=${PASSWORD_SIGNAL:-} +ct.signal-data-source-settings.password=${PASSWORD_SIGNAL:password} 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} @@ -80,7 +80,7 @@ ct.tmp.dir=D:\\folder_name\\signals\\tmp ct.target.dir=D:\\folder_name\\signals\\tmpt # min -ct.monitoring-interval=${MONITORING_INTERVAL:15} +ct.monitoring-interval=${MONITORING_INTERVAL:5} # days ct.update-flags-cron-expression=${UPDATE_FLAGS_CRON_EXPRESSION:0 25 18 * * *} ct.new-flags-cron-expression=${NEW_FLAGS_CRON_EXPRESSION:0 25 18 * * *} @@ -88,6 +88,15 @@ ct.days-to-update-flags=${DAYS_TO_UPDATE_FLAGS:30} ct.update-fid-cron-expression=${UPDATE_FIDS_CRON_EXPRESSION:0 10 12 * * *} ct.days-to-update-fids=${DAYS_TO_UPDATE_FIDS:30} +#kafka +ct.kafka-consumer-settings.topic=${CONSUMER_TOPIC:my-topic} +ct.kafka-consumer-settings.concurency=${CONSUMER_CONCURENCY:8} +ct.kafka-consumer-settings.bootstrap-servers=${CONSUMER_BOOTSTRAP_SERVERS:localhost:29092} +ct.kafka-consumer-settings.group-id=${CONSUMER_GROUP_ID:credit-tracker} +ct.kafka-consumer-settings.auto-offset-reset=${CONSUMER_AUTO_OFFSET_RESET:latest} +ct.kafka-consumer-settings.max-poll-records=${CONSUMER_MAX_POLL_RECORDS:125} +ct.kafka-consumer-settings.max-poll-interval-ms=${CONSUMER_MAX_POLL_INTERVAL_MS:300000} + ct.db-version=${DB_VERSION:1234} ct.encoding=windows-1251 diff --git a/docker_compose/credit_tracker_config.env b/docker_compose/credit_tracker_config.env index 11ab8ed..3c8b915 100644 --- a/docker_compose/credit_tracker_config.env +++ b/docker_compose/credit_tracker_config.env @@ -75,6 +75,15 @@ MAX_LIFETIME_NBCH=1800000 KEEPALIVE_TIME_NBCH=0 LEAK_DETECTION_THRESHOLD_NBCH=0 +#kafka consumer +CONSUMER_TOPIC=my-topic +CONSUMER_CONCURENCY=8 +CONSUMER_BOOTSTRAP_SERVERS=localhost:29092 +CONSUMER_GROUP_ID=credit-tracker +CONSUMER_AUTO_OFFSET_RESET=latest +CONSUMER_MAX_POLL_RECORDS=125 +CONSUMER_MAX_POLL_INTERVAL_MS=300000 + # размер в mb ACCEPTED_XML_SIZE=1024 ACCEPTED_ARCHIVE_SIZE=2048MB diff --git a/docker_compose/docker-compose.yml b/docker_compose/docker-compose.yml index 225a794..50058cf 100644 --- a/docker_compose/docker-compose.yml +++ b/docker_compose/docker-compose.yml @@ -69,6 +69,13 @@ services: - MAX_LIFETIME_NBCH=${MAX_LIFETIME_NBCH} - KEEPALIVE_TIME_NBCH=${KEEPALIVE_TIME_NBCH} - LEAK_DETECTION_THRESHOLD_NBCH=${LEAK_DETECTION_THRESHOLD_NBCH} + - CONSUMER_TOPIC=${CONSUMER_TOPIC} + - CONSUMER_CONCURENCY=${CONSUMER_CONCURENCY} + - CONSUMER_BOOTSTRAP_SERVERS=${CONSUMER_BOOTSTRAP_SERVERS} + - CONSUMER_GROUP_ID=${CONSUMER_GROUP_ID} + - CONSUMER_AUTO_OFFSET_RESET=${CONSUMER_AUTO_OFFSET_RESET} + - CONSUMER_MAX_POLL_RECORDS=${CONSUMER_MAX_POLL_RECORDS} + - CONSUMER_MAX_POLL_INTERVAL_MS=${CONSUMER_MAX_POLL_INTERVAL_MS} - ACCEPTED_XML_SIZE=${ACCEPTED_XML_SIZE} - ACCEPTED_ARCHIVE_SIZE=${ACCEPTED_ARCHIVE_SIZE} - SIGNALS_ROOT_URI=${SIGNALS_ROOT_URI} @@ -164,6 +171,13 @@ services: - MAX_LIFETIME_NBCH=${MAX_LIFETIME_NBCH} - KEEPALIVE_TIME_NBCH=${KEEPALIVE_TIME_NBCH} - LEAK_DETECTION_THRESHOLD_NBCH=${LEAK_DETECTION_THRESHOLD_NBCH} + - CONSUMER_TOPIC=${CONSUMER_TOPIC} + - CONSUMER_CONCURENCY=${CONSUMER_CONCURENCY} + - CONSUMER_BOOTSTRAP_SERVERS=${CONSUMER_BOOTSTRAP_SERVERS} + - CONSUMER_GROUP_ID=${CONSUMER_GROUP_ID} + - CONSUMER_AUTO_OFFSET_RESET=${CONSUMER_AUTO_OFFSET_RESET} + - CONSUMER_MAX_POLL_RECORDS=${CONSUMER_MAX_POLL_RECORDS} + - CONSUMER_MAX_POLL_INTERVAL_MS=${CONSUMER_MAX_POLL_INTERVAL_MS} - SIGNALS_ROOT_URI=${SIGNALS_ROOT_URI} - SIGNALS_TOKEN=${SIGNALS_TOKEN} - SIGNALS_USER=${SIGNALS_USER} diff --git a/src/main/java/ru/nbch/credit_tracker/config/kafka/KafkaConfig.java b/src/main/java/ru/nbch/credit_tracker/config/kafka/KafkaConfig.java index 3630601..c09eba7 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/kafka/KafkaConfig.java +++ b/src/main/java/ru/nbch/credit_tracker/config/kafka/KafkaConfig.java @@ -12,6 +12,7 @@ import org.springframework.kafka.core.*; import org.springframework.kafka.listener.ContainerProperties; import org.springframework.kafka.support.serializer.JsonDeserializer; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.config.settings.KafkaConsumerSettings; import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage; import ru.nbch.credit_tracker.model.SignalKafkaEvent; @@ -42,12 +43,14 @@ public class KafkaConfig { } @Bean - public ConsumerFactory consumerFactory() { + public ConsumerFactory consumerFactory(CreditTrackerSettings settings) { + KafkaConsumerSettings kafkaSettings = settings.getKafkaConsumerSettings(); Map props = new HashMap<>(); - props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:29092"); - props.put(ConsumerConfig.GROUP_ID_CONFIG, "credit-tracker"); - props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); - props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 20); + props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaSettings.bootstrapServers()); + props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaSettings.groupId()); + props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaSettings.autoOffsetReset()); + props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, kafkaSettings.maxPollRecords()); + props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, kafkaSettings.maxPollIntervalMs()); JsonDeserializer jsonDeserializer = new JsonDeserializer<>(PackageMessage.class); jsonDeserializer.addTrustedPackages("*"); diff --git a/src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java b/src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java index 80a202e..da48710 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java +++ b/src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java @@ -27,6 +27,8 @@ public class CreditTrackerSettings { private PipelineSettings pipeline; @NestedConfigurationProperty private FlagUpdaterSettings flagUpdater; + @NestedConfigurationProperty + private KafkaConsumerSettings kafkaConsumerSettings; private String signalsRootUri; private String signalsToken; diff --git a/src/main/java/ru/nbch/credit_tracker/config/settings/KafkaConsumerSettings.java b/src/main/java/ru/nbch/credit_tracker/config/settings/KafkaConsumerSettings.java new file mode 100644 index 0000000..ab32372 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/config/settings/KafkaConsumerSettings.java @@ -0,0 +1,11 @@ +package ru.nbch.credit_tracker.config.settings; + +public record KafkaConsumerSettings( + String topic, + String concurency, + String bootstrapServers, + String groupId, + String autoOffsetReset, + Integer maxPollRecords, + Integer maxPollIntervalMs +) {} diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java index 1fe2c87..ae8567c 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java @@ -8,6 +8,7 @@ import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.kafka.support.Acknowledgment; import org.springframework.stereotype.Service; +import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.entities.app.Hit; import ru.nbch.credit_tracker.entities.app.LegalsMap; import ru.nbch.credit_tracker.entities.app.PhonesMap; @@ -34,9 +35,8 @@ import java.util.stream.Collectors; @Service public class KafkaMonitoringService { private volatile boolean isShuttingdDown = false; - private final static String CONCURRENCY = "8"; - private final static int PERMITS = Integer.parseInt(CONCURRENCY); - private final Semaphore processingPermits = new Semaphore(PERMITS); // количество равно concurency листенера + private final int permits; + private final Semaphore processingPermits; private final BlockingQueue signalEventsQueue; @@ -47,16 +47,19 @@ public class KafkaMonitoringService { public KafkaMonitoringService(@Qualifier("signalKafkaEventQueue") BlockingQueue signalEventsQueue, PackageRecordsService packageRecordsService, SourceService sourceService, - SegmentMapService segmentMapService) { + SegmentMapService segmentMapService, + CreditTrackerSettings settings) { this.signalEventsQueue = signalEventsQueue; this.packageRecordsService = packageRecordsService; this.sourceService = sourceService; this.segmentMapService = segmentMapService; + this.permits = Integer.parseInt(settings.getKafkaConsumerSettings().concurency()); // количество равно concurency листенера + this.processingPermits = new Semaphore(permits); } @KafkaListener( - topics = "my-topic", - concurrency = CONCURRENCY + topics = "#{@creditTrackerSettings.kafkaConsumerSettings.topic}", + concurrency = "#{@creditTrackerSettings.kafkaConsumerSettings.concurency}" ) public void listen(List records, Acknowledgment acknowledgment) { @@ -72,8 +75,9 @@ public class KafkaMonitoringService { try { acknowledgment.acknowledge(); + long startTime = System.currentTimeMillis(); process(records); - log.info("Processed {} messages from Kafka", records.size()); + log.info("Processed {} messages from Kafka. Time {} ms", records.size(), System.currentTimeMillis() - startTime); } finally { processingPermits.release(); } @@ -294,7 +298,7 @@ public class KafkaMonitoringService { log.info("Stopping KafkaMonitoringService..."); isShuttingdDown = true; - processingPermits.acquire(PERMITS); + processingPermits.acquire(permits); try { SignalKafkaEvent rotateEvent = new SignalKafkaEvent(); @@ -307,7 +311,7 @@ public class KafkaMonitoringService { signalEventsQueue.put(stopEvent); log.info("STOP event sent to queue"); } finally { - processingPermits.release(PERMITS); + processingPermits.release(permits); } log.info("KafkaMonitoringService stopped"); diff --git a/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml b/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml index b160321..a53c95b 100644 --- a/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml @@ -58,19 +58,11 @@