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

@@ -1,132 +0,0 @@
-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,12 +321,8 @@ 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();