Block Kafka event writer on empty queue

src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaSignalProcessorSupport.java

@@ -0,0 +1,132 @@
+package ru.nbch.credit_tracker.service.mode.monitoring;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.stereotype.Component;
+import ru.nbch.credit_tracker.entities.app.Hit;
+import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage;
+import ru.nbch.credit_tracker.model.IUserPackageData;
+import ru.nbch.credit_tracker.model.SignalKafkaEvent;
+import ru.nbch.credit_tracker.service.nbch.SourceService;
+import ru.nbch.credit_tracker.utils.CTUtil;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+@Slf4j
+@Component
+public class KafkaSignalProcessorSupport {
+
+    private final SourceService sourceService;
+    private final SignalKafkaEventPublisher eventPublisher;
+
+    public KafkaSignalProcessorSupport(SourceService sourceService,
+                                       SignalKafkaEventPublisher eventPublisher) {
+        this.sourceService = sourceService;
+        this.eventPublisher = eventPublisher;
+    }
+
+    public <M, D extends IUserPackageData, K> List<Hit> process(List<PackageMessage> messages,
+                                                               Map<String, Integer> segmentMap,
+                                                               Function<List<Integer>, List<M>> mappingLoader,
+                                                               Function<M, Integer> fidExtractor,
+                                                               Function<M, K> keyExtractor,
+                                                               Function<List<M>, Map<K, List<D>>> packageDataLoader,
+                                                               SignalEventBuilder<M, D> eventBuilder,
+                                                               SignalHitBuilder<D> hitBuilder) {
+        List<Hit> hits = new ArrayList<>();
+
+        List<Integer> fids = messages.stream().map(PackageMessage::getFid).collect(Collectors.toList());
+        List<String> memberCodes = messages.stream()
+                .map(message -> message.getMemberCode().substring(0, 6))
+                .toList();
+
+        List<M> mappings = mappingLoader.apply(fids);
+        Map<Integer, List<M>> mappingsByFid = mappings.stream()
+                .collect(Collectors.groupingBy(fidExtractor));
+        Map<K, List<D>> packageDataByKey = packageDataLoader.apply(mappings);
+
+        // прелоад по memberCode категорий
+        Map<String, String> busCategories = sourceService.findBusCategories(memberCodes);
+
+        for (PackageMessage message : messages) {
+            Integer fid = message.getFid();
+            String commonMemberCode = message.getMemberCode().substring(0, 6);
+
+            String busCategory = busCategories.get(commonMemberCode);
+            if (StringUtils.isBlank(busCategory)) {
+                log.error("Для memberсode = {} отсутствует bus_category или не найден memberсode", commonMemberCode);
+                continue;
+            }
+
+            int signalType = CTUtil.checkBusCategory(busCategory);
+
+            List<M> messageMappings = mappingsByFid.get(fid);
+            if (messageMappings == null) {
+                continue;
+            }
+
+            for (M mapping : messageMappings) {
+                List<D> packageDataList = packageDataByKey.get(keyExtractor.apply(mapping));
+                if (packageDataList == null) {
+                    continue;
+                }
+
+                for (D packageData : packageDataList) {
+                    if (!hasSignalType(packageData.getSignalTypes(), signalType)) {
+                        continue;
+                    }
+
+                    Integer segment = segmentMap.get(commonMemberCode);
+                    SignalKafkaEvent event = eventBuilder.build(message, mapping, packageData, signalType, busCategory, segment);
+
+                    String reportKey = packageData.getUserId().substring(0, 6)
+                            + "_" + packageData.getPackageName();
+                    String reportFileName = eventPublisher.resolveNameAndEnqueue(reportKey, event);
+                    if (reportFileName == null) {
+                        continue; // событие не встало в очередь — hit не пишем
+                    }
+
+                    hits.add(hitBuilder.build(message, packageData, signalType, busCategory, segment, reportFileName));
+                }
+            }
+        }
+
+        return hits;
+    }
+
+    private boolean hasSignalType(String signalTypes, int signalType) {
+        if (StringUtils.isBlank(signalTypes)) {
+            return false;
+        }
+
+        String expectedSignalType = Integer.toString(signalType);
+        return Arrays.stream(signalTypes.split(","))
+                .map(String::trim)
+                .anyMatch(expectedSignalType::equals);
+    }
+
+    @FunctionalInterface
+    public interface SignalEventBuilder<M, D extends IUserPackageData> {
+        SignalKafkaEvent build(PackageMessage message,
+                               M mapping,
+                               D packageData,
+                               int signalType,
+                               String busCategory,
+                               Integer segment);
+    }
+
+    @FunctionalInterface
+    public interface SignalHitBuilder<D extends IUserPackageData> {
+        Hit build(PackageMessage message,
+                  D packageData,
+                  int signalType,
+                  String busCategory,
+                  Integer segment,
+                  String reportFileName);
+    }
+}

src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/SignalKafkaEventWriterService.java

@@ -321,8 +321,12 @@ public class SignalKafkaEventWriterService {
     private List<SignalKafkaEvent> drain() {
         try {
             List<SignalKafkaEvent> batch = new ArrayList<>(8000);
+            batch.add(signalEventsQueue.take());
             signalEventsQueue.drainTo(batch, 8000);
             return batch;
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            return Collections.emptyList();
         } catch (Exception e) {
             log.error("Error while drain signal events", e);
             return Collections.emptyList();