Async db writing

src/main/java/ru/nbch/credit_tracker/config/kafka/KafkaConfig.java

@@ -27,7 +27,6 @@ public class KafkaConfig {
 
     @Bean("signalKafkaEventQueue")
     public BlockingQueue<SignalKafkaEvent> signalKafkaEventQueue(CreditTrackerSettings settings) {
-        // TODO add capacity
         return new LinkedBlockingQueue<>();
     }
 

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.context.annotation.DependsOn;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
 import org.springframework.stereotype.Service;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
 import ru.nbch.credit_tracker.entities.app.Hit;
@@ -23,20 +24,21 @@ import ru.nbch.credit_tracker.service.signal.HitsService;
 import ru.nbch.credit_tracker.service.signal.PackageRecordsService;
 import ru.nbch.credit_tracker.service.signal.SegmentMapService;
 import ru.nbch.credit_tracker.utils.CTUtil;
+import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
 
 import java.time.LocalDateTime;
-import java.util.ArrayList;
+import java.util.*;
-import java.util.List;
+import java.util.concurrent.*;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.Semaphore;
 import java.util.stream.Collectors;
 
 @Slf4j
 @Service
 @DependsOn(value = "signalKafkaEventWriterService")
 public class KafkaMonitoringService {
+
+    private final ExecutorService processingExecutor;
+    private final ExecutorService dbExecutor;
+
     private volatile boolean isShuttingdDown = false;
     private final int permits;
     private final Semaphore processingPermits;
@@ -61,6 +63,15 @@ public class KafkaMonitoringService {
         this.permits = Integer.parseInt(settings.getKafkaConsumerSettings().concurency()); // количество равно concurency листенера
         this.hitsService = hitsService;
         this.processingPermits = new Semaphore(permits);
+
+        this.processingExecutor = Executors.newFixedThreadPool(
+                permits * 2,
+                new CustomizableThreadFactory("kafka-processor-")
+        );
+        this.dbExecutor = Executors.newFixedThreadPool(
+                permits,
+                new CustomizableThreadFactory("kafka-db-writer-")
+        );
     }
 
     @KafkaListener(
@@ -90,8 +101,6 @@ public class KafkaMonitoringService {
     }
 
     private void process(List<PackageMessage> messages) {
-        List<Hit> hits = new ArrayList<>();
-
         Map<Boolean, List<PackageMessage>> partition = messages.stream()
                 .filter(this::filterMessages)
                 .collect(Collectors.partitioningBy(m -> m.getPersons() != null && !m.getPersons().isEmpty()));
@@ -101,17 +110,43 @@ public class KafkaMonitoringService {
 
         Map<String, Integer> segmentMap = segmentMapService.loadSegmentMap();
 
+        CompletableFuture<List<Hit>> npFuture = CompletableFuture.supplyAsync(() -> {
             if (npMessages != null && !npMessages.isEmpty()) {
-            List<Hit> npHits = processNpMessages(npMessages, segmentMap);
+                return processNpMessages(npMessages, segmentMap);
-            hits.addAll(npHits);
             }
+            return Collections.emptyList();
+        }, processingExecutor);
 
+        CompletableFuture<List<Hit>> lpFuture = CompletableFuture.supplyAsync(() -> {
             if (lpMessages != null && !lpMessages.isEmpty()) {
-            List<Hit> lpHits = processLpMessages(lpMessages, segmentMap);
+                return processLpMessages(lpMessages, segmentMap);
-            hits.addAll(lpHits);
             }
+            return Collections.emptyList();
+        }, processingExecutor);
 
+        CompletableFuture<List<Hit>> combinedFuture = npFuture
+                .thenCombine(lpFuture, (npHits, lpHits) -> {
+                    List<Hit> allHits = new ArrayList<>();
+                    if (npHits != null) allHits.addAll(npHits);
+                    if (lpHits != null) allHits.addAll(lpHits);
+                    return allHits;
+                });
+
+        combinedFuture.thenAcceptAsync(hits -> {
+                    if (!hits.isEmpty()) {
+                        try {
                             hitsService.insertHits(hits);
+                        } catch (Exception e) {
+                            log.error("Failed to save hits. {}", ExceptionUtils.getStackTrace(e));
+                        }
+                    }
+                }, dbExecutor);
+
+        try {
+            combinedFuture.get(1, TimeUnit.MINUTES);
+        } catch (Exception e) {
+            log.error("Error processing messages", e);
+        }
     }
 
     private List<Hit> processNpMessages(List<PackageMessage> messages, Map<String, Integer> segmentMap) {