.

src/main/java/ru/nbch/credit_tracker/service/task/BlockingRejectedExecutionHandler.java

@@ -0,0 +1,23 @@
+package ru.nbch.credit_tracker.service.task;
+
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.ThreadPoolExecutor;
+
+public class BlockingRejectedExecutionHandler implements RejectedExecutionHandler {
+
+    public static RejectedExecutionHandler INSTANCE = new BlockingRejectedExecutionHandler();
+
+    @Override
+    public void rejectedExecution(Runnable r, ThreadPoolExecutor exec) {
+        try {
+            if (exec.isShutdown()) {
+                throw new RejectedExecutionException("Executor is shut down");
+            }
+            exec.getQueue().put(r);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new RejectedExecutionException("Interrupted while enqueueing task", e);
+        }
+    }
+}

src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java

@@ -1,12 +1,10 @@
 package ru.nbch.credit_tracker.service.task;
 
-import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 import java.util.function.BiFunction;
-import java.util.function.Consumer;
 import java.util.function.Supplier;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Qualifier;
@@ -19,6 +17,7 @@ import ru.nbch.credit_tracker.model.impl.SubjectPayload;
 import ru.nbch.credit_tracker.service.signal.SignalRestService;
 import ru.nbch.credit_tracker.service.task.impl.FidIdEnricher;
 import ru.nbch.credit_tracker.service.task.impl.FlagsEnricher;
+import ru.nbch.credit_tracker.service.task.impl.FlagsEnricherV2;
 import ru.nbch.credit_tracker.service.task.impl.SaveSubjectHandler;
 import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter;
 import ru.nbch.credit_tracker.service.task.setup.IPipeLineTerminator;
@@ -41,6 +40,7 @@ public class PipelinePackageManager {
     private final Supplier<IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload>> pipelineTerminatorFactory;
     private final Supplier<FlagsEnricher> flagEnricherFactory;
     private final BiFunction<Long, String, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory;
+    private final FlagsEnricherV2 flagsEnricherV2;
     private final ExecutorService executorService = Executors.newFixedThreadPool(10, r -> {
         Thread t = new Thread(r);
         t.setName("pipe-starter-%d".formatted(t.threadId()));
@@ -60,7 +60,7 @@ public class PipelinePackageManager {
                                   SaveSubjectHandler saveSubjectHandler, FidIdEnricher fidIdEnricherHandler,
                                   Supplier<IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload>> pipelineTerminatorFactory,
                                   Supplier<FlagsEnricher> flagEnricherFactory,
-                                  BiFunction<Long, String, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory) {
+                                  BiFunction<Long, String, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory, FlagsEnricherV2 flagsEnricherV2) {
         this.applicationContext = applicationContext;
         this.xmlProcessor = xmlProcessor;
         this.signalRestService = signalRestService;
@@ -73,6 +73,7 @@ public class PipelinePackageManager {
         this.pipelineTerminatorFactory = pipelineTerminatorFactory;
         this.flagEnricherFactory = flagEnricherFactory;
         this.pipelineStarterFactory = pipelineStarterFactory;
+        this.flagsEnricherV2 = flagsEnricherV2;
         this.semaphore = new Semaphore(MAX_PARALLEL_REQUESTS);
     }
 
@@ -80,19 +81,12 @@ public class PipelinePackageManager {
         if (!semaphore.tryAcquire()) {
             throw new RuntimeException("Too many concurrent requests");
         }
+        log.info("throttler permits after acquire: {}", semaphore.availablePermits());
         try {
-            BlockingQueue<SubjectPayload> flagQueue = flagQueueFactory.getObject();
-            Consumer<SubjectPayload> consumer = subjectPayload -> {
-                try {
-                    flagQueue.put(subjectPayload);
-                } catch (InterruptedException e) {
-                    throw new RuntimeException(e);
-                }
-            };
             Pipeline pipeline = PipeLineBuilderV2.first(saveSubjectQueueFactory.getObject(), idEnricherQueueFactory.getObject())
                     .handler(saveSubjectHandler)
                     .poison(SubjectPayload.POISON)
-                    .handlerResult(consumer)
+                    .handlerResult(flagsEnricherV2)
                     .andThen()
                 .stage(sendSignalQueueFactory.getObject())
                     .handler(fidIdEnricherHandler)
@@ -102,10 +96,6 @@ public class PipelinePackageManager {
                 .pollFinishingQueue(pipelineTerminatorFactory.get())
                 .build();
 
-            FlagsEnricher flagEnricher = flagEnricherFactory.get();
-            flagEnricher.setPollingQueue(flagQueue);
-
-            executorService.submit(flagEnricher);
             executorService.submit(pipeline.getStarter());
             executorService.submit(pipeline.getTerminator());
 
@@ -113,42 +103,25 @@ public class PipelinePackageManager {
                 .doneFuture()
                 .orTimeout(5, TimeUnit.MINUTES)
                 .whenComplete((v, e) -> {
-                    if (e != null) {
-                        log.error(ExceptionUtils.getStackTrace(e));
-                    }
-                    SignalXmlFilePayload res = (SignalXmlFilePayload) v;
-                    log.trace("Signal file by path: {} is completed", res.getSignalXmlFilePath());
                     try {
+                        if (e != null) {
+                            log.error(ExceptionUtils.getStackTrace(e));
+                            return;
+                        }
+                        SignalXmlFilePayload res = (SignalXmlFilePayload) v;
+                        log.trace("Signal file by path: {} is completed", res.getSignalXmlFilePath());
                         signalRestService.setupMonitoring(res.getSignalXmlFilePath().toString());
                     } catch (Exception ex) {
-                        throw new RuntimeException(ex);
+                        log.error("pipeline finished with error{}", ExceptionUtils.getStackTrace(ex));
+                    } finally {
+                        log.info("packageId={}: pipeline finished", packageId);
+                        pipeline.getStages().forEach(Stage::close);
+                        semaphore.release();
                     }
-                    log.info("packageId={}: pipeline finished", packageId);
-                    pipeline.getStages().forEach(Stage::close);
-                    semaphore.release();
             });
-
-            flagEnricher.doneFuture()
-                .orTimeout(10, TimeUnit.MINUTES)
-                .whenComplete((v, e) -> {
-                    if (e != null) {
-                        log.error(ExceptionUtils.getStackTrace(e));
-                    }
-                    log.debug("Calculation flags completed");
-                });
         } catch (Exception e) {
             semaphore.release();
             throw e;
         }
-
-//        try {
-//            pipeline.getLast().doneFuture().get(5, TimeUnit.SECONDS);
-//            log.info("Subjects were registered. PackageId: {}", packageId);
-//        } catch (InterruptedException e) {
-//            Thread.currentThread().interrupt();
-//            return;
-//        } catch (ExecutionException|TimeoutException e) {
-//            log.error(ExceptionUtils.getStackTrace(e));
-//        }
     }
 }

src/main/java/ru/nbch/credit_tracker/service/task/Stage.java

@@ -35,12 +35,13 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
         this.ioPool = new ThreadPoolExecutor(10,
             10,
             0L, TimeUnit.MINUTES,
-            new LinkedBlockingQueue<>(),
+            new LinkedBlockingQueue<>(1),
             r -> {
                 Thread t = new Thread(r);
                 t.setName("pipe-handler-%s-%d".formatted(dispatcher.getClass().getSimpleName(), t.threadId()));
                 return t;
-            }
+            },
+            BlockingRejectedExecutionHandler.INSTANCE
         );
         this.POISON = POISON;
     }
@@ -152,4 +153,5 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
     public void setConsumer(Consumer<Z> consumer) {
         this.consumer = consumer;
     }
+
 }

src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricherV2.java

@@ -0,0 +1,112 @@
+package ru.nbch.credit_tracker.service.task.impl;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper;
+import ru.nbch.credit_tracker.model.SubjectProfile;
+import ru.nbch.credit_tracker.model.impl.SubjectPayload;
+import ru.nbch.credit_tracker.service.indic.FlagsService2;
+import ru.nbch.credit_tracker.service.task.BlockingRejectedExecutionHandler;
+
+@Slf4j
+@Service
+public class FlagsEnricherV2 implements Consumer<SubjectPayload> {
+
+    private final FlagsService2 flagsService;
+    private final PhoneFidMapMapper phoneFidMapMapper;
+    private final ExecutorService ioPool;
+    private final ObjectMapper mapper = new ObjectMapper()
+        .registerModule(new JavaTimeModule())
+        .disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS)
+        .setSerializationInclusion(JsonInclude.Include.NON_NULL);
+
+    public FlagsEnricherV2(FlagsService2 flagsService,
+                           PhoneFidMapMapper phoneFidMapMapper) {
+        this.flagsService = flagsService;
+        this.phoneFidMapMapper = phoneFidMapMapper;
+        this.ioPool = new ThreadPoolExecutor(10,
+            10,
+            0L, TimeUnit.MINUTES,
+            new LinkedBlockingQueue<>(100),
+            r -> {
+                Thread t = new Thread(r);
+                t.setName("flags-enricher-%d".formatted(t.threadId()));
+                return t;
+            },
+            BlockingRejectedExecutionHandler.INSTANCE
+        );
+    }
+
+    public void process(SubjectPayload subjectPayload) {
+        List<SubjectProfile> subjects = subjectPayload.getSubjects();
+        log.debug("FlagsEnricher handle by packageId: {}, size: {}", subjectPayload.getPackageId(), subjects.size());
+
+        List<Integer> fids = subjects.stream()
+            .map(SubjectProfile::getFid)
+            .filter(Objects::nonNull)
+            .toList();
+        LocalDate yearAgo = LocalDate.now().minusYears(1);
+
+        Map<Integer, Integer> flagsDebt5000 = flagsService.findFlagsDebt5000(fids);
+        Map<Integer, Integer> flag90ByPastDueArrear = flagsService.findFlag90ByPastDueArrear(fids, yearAgo);
+        Map<Integer, Integer> flag90ByRuPaymt = flagsService.findFlag90ByRuPaymt(fids, yearAgo);
+
+        String json;
+        try {
+            json = mapper.writeValueAsString(subjects.stream().map(subject -> {
+                Integer debt5000 = null;
+                Integer flagDays90 = null;
+                LocalDateTime calculatedAt = null;
+
+                if (subject.getFid() != null) {
+                    debt5000 = flagsDebt5000.getOrDefault(subject.getFid(), 0);
+                    int pastDue90 = flag90ByPastDueArrear.getOrDefault(subject.getFid(), 0);
+                    int paymt90 = flag90ByRuPaymt.getOrDefault(subject.getFid(), 0);
+                    flagDays90 = Math.max(pastDue90, paymt90);
+                    calculatedAt = LocalDateTime.now();
+                }
+
+                Integer finalDebt500 = debt5000;
+                Integer finalFlagDays9 = flagDays90;
+                LocalDateTime finalCalculatedAt = calculatedAt;
+                return new HashMap<String, Object>() {{
+                    put("subject_id", subject.getIdentifier());
+                    put("fid", subject.getFid());
+                    put("phone", subject.getPhone());
+                    put("package_id", subjectPayload.getPackageId());
+                    put("flag_debt5000", finalDebt500);
+                    put("flag_90_12", finalFlagDays9);
+                    put("calculated_at", finalCalculatedAt);
+                }};
+            }).toList());
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException(e);
+        }
+        phoneFidMapMapper.registerJson(json);
+    }
+
+    @Override
+    public void accept(SubjectPayload subjectPayload) {
+        ThreadPoolExecutor casted = (ThreadPoolExecutor) ioPool;
+        log.info("submitting payload. active tasks number: {}; queue size: {}", casted.getActiveCount(), casted.getQueue().size());
+        this.ioPool.submit(() -> {
+            process(subjectPayload);
+        });
+    }
+}