Added kafka settings to props

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

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

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}

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<String, PackageMessage> consumerFactory() {
+    public ConsumerFactory<String, PackageMessage> consumerFactory(CreditTrackerSettings settings) {
+        KafkaConsumerSettings kafkaSettings = settings.getKafkaConsumerSettings();
         Map<String, Object> props = new HashMap<>();
-        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:29092");
+        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaSettings.bootstrapServers());
-        props.put(ConsumerConfig.GROUP_ID_CONFIG, "credit-tracker");
+        props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaSettings.groupId());
-        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaSettings.autoOffsetReset());
-        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 20);
+        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, kafkaSettings.maxPollRecords());
+        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, kafkaSettings.maxPollIntervalMs());
 
         JsonDeserializer<PackageMessage> jsonDeserializer = new JsonDeserializer<>(PackageMessage.class);
         jsonDeserializer.addTrustedPackages("*");

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;

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
+) {}

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 int permits;
-    private final static int PERMITS = Integer.parseInt(CONCURRENCY);
+    private final Semaphore processingPermits;
-    private final Semaphore processingPermits = new Semaphore(PERMITS); // количество равно concurency листенера
 
     private final BlockingQueue<SignalKafkaEvent> signalEventsQueue;
 
@@ -47,16 +47,19 @@ public class KafkaMonitoringService {
     public KafkaMonitoringService(@Qualifier("signalKafkaEventQueue") BlockingQueue<SignalKafkaEvent> 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",
+            topics = "#{@creditTrackerSettings.kafkaConsumerSettings.topic}",
-            concurrency = CONCURRENCY
+            concurrency = "#{@creditTrackerSettings.kafkaConsumerSettings.concurency}"
     )
     public void listen(List<PackageMessage> 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");

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

@@ -58,19 +58,11 @@
     </select>
 
     <select id="findByFids" resultMap="PhoneMap">
-        select id,
+        select
-               fid,
+        fid,
-               phone,
+        phone,
-               name_1,
+        flag_debt5000,
-               first,
+        flag_90_12
-               middle,
-               birth_dt,
-               ser_num,
-               id_type,
-               flag_debt5000,
-               flag_90_12,
-               searched_at,
-               calculated_at
         FROM signals_phones_map
         WHERE fid in
         <foreach item='fid' collection='fids' open='(' separator=',' close=')'>