interrupt

src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java

@@ -20,7 +20,7 @@ public class StageConfig {
 
     @Bean("partFileQueueFactory")
     public StageQueueFactory<PartFilePayload> partFileQueueFactory() {
-        return new StageQueueFactory<>(10);
+        return new StageQueueFactory<>(10000);
     }
 
 }

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

@@ -30,7 +30,11 @@ public class PipelinePackageManager {
     private final SaveSubjectHandler saveSubjectHandler;
     private final FidIdEnricher fidIdEnricherHandler;
     private final BiFunction<Long, String, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory;
-    private final ExecutorService executorService = Executors.newFixedThreadPool(10);
+    private final ExecutorService executorService = Executors.newFixedThreadPool(10, r -> {
+        Thread t = new Thread(r);
+        t.setName("pipe-starter-%d".formatted(t.threadId()));
+        return t;
+    });
     private final Semaphore semaphore;
     private final static int BATCH_SIZE = 10_000;
     private final static int MAX_PARALLEL_REQUESTS = 3;
@@ -57,9 +61,8 @@ public class PipelinePackageManager {
         if (!semaphore.tryAcquire()) {
             throw new RuntimeException("Too many concurrent requests");
         }
-        Pipeline pipeline = null;
         try {
-            pipeline = PipeLineBuilderV2.first(saveSubjectQueueFactory.getObject(), idEnricherQueueFactory.getObject())
+            Pipeline pipeline = PipeLineBuilderV2.first(saveSubjectQueueFactory.getObject(), idEnricherQueueFactory.getObject())
                     .handler(saveSubjectHandler)
                     .poison(SubjectPayload.POISON)
                     .andThen()
@@ -71,11 +74,14 @@ public class PipelinePackageManager {
             executorService.submit(pipeline.getStarter());
             pipeline.getLast()
                 .doneFuture()
-                .orTimeout(1, TimeUnit.MINUTES)
+                .orTimeout(5, TimeUnit.SECONDS)
                 .whenComplete((v, e) -> {
                     if (e != null) {
                         log.error(ExceptionUtils.getStackTrace(e));
+                    } else {
+                        log.info("packageId={}: pipeline finished", packageId);
                     }
+                    pipeline.getStages().forEach(Stage::close);
                     semaphore.release();
             });
         } catch (Exception e) {

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

@@ -2,15 +2,14 @@ package ru.nbch.credit_tracker.service.task;
 
 import java.time.Duration;
 import java.util.ArrayList;
-import java.util.Iterator;
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.exception.ExceptionUtils;
 import ru.nbch.credit_tracker.model.StagePayload;
@@ -31,7 +30,16 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
         this.dispatcher = dispatcher;
         this.in = in;
         this.out = out;
-        this.ioPool = Executors.newFixedThreadPool(10);
+        this.ioPool = new ThreadPoolExecutor(10,
+            10,
+            0L, TimeUnit.MINUTES,
+            new LinkedBlockingQueue<>(),
+            r -> {
+                Thread t = new Thread(r);
+                t.setName("pipe-handler-%s-%d".formatted(dispatcher.getClass().getSimpleName(), t.threadId()));
+                return t;
+            }
+        );
         this.POISON = POISON;
     }
 
@@ -46,68 +54,68 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
     @Override
     public void run() {
         List<Future<Z>> inflight = new ArrayList<>();
-        AtomicInteger inFlightCount = new AtomicInteger();
         try {
+            int counter = 0;
             while (true) {
                 X b = takeQuiet(in, Duration.ofSeconds(5));
+                if (wasInterrupted()) {
+                    break;
+                }
                 if (b == null) continue;
                 if (b.isPoison()) {
                     // дождаться всех in-flight задач
                     for (Future<Z> f : inflight) {
+                        if (wasInterrupted()) {
+                            log.info("stage {} was interrupted while waiting for tasks to finish",
+                                dispatcher.getClass().getSimpleName());
+                            break;
+                        }
                         f.get();
                     }
                     putQuiet(out, POISON);
+                    log.info("stage {} finished.", dispatcher.getClass().getSimpleName());
                     done.complete(null);
                     return;
                 }
-
+                counter++;
+                int finalCounter = counter;
                 Future<Z> f = ioPool.submit(() -> {
+                        if (wasInterrupted()) {
+                            log.info("stage {} task {} interrupted...", dispatcher.getClass().getSimpleName(), finalCounter);
+                            return null;
+                        }
+                        log.info("stage {} task {} started.", dispatcher.getClass().getSimpleName(), finalCounter);
                         Z p = dispatcher.handle(b);
                         putQuiet(out, p);
+                        log.info("stage {} task {} finished.", dispatcher.getClass().getSimpleName(), finalCounter);
                         return p;
                     }
                 );
                 inflight.add(f);
-
-                // чтобы не накапливать слишком много future в памяти, периодически сливаем
-                if (inFlightCount.incrementAndGet() % 8 == 0) {
-                    drainDone(inflight, out);
-                }
             }
         } catch (Exception e) {
-            log.error("Error: {}", ExceptionUtils.getStackTrace(e));
+            log.error("error: {}", ExceptionUtils.getStackTrace(e));
             putQuiet(out, POISON);
             done.complete(null);
-        } finally {
-            drainAll(inflight, out);
         }
     }
 
-    void drainDone(List<Future<Z>> fs, BlockingQueue<Z> out) throws Exception {
+    private boolean wasInterrupted() {
-        Iterator<Future<Z>> it = fs.iterator();
+        if (Thread.currentThread().isInterrupted()) {
-        while (it.hasNext()) {
+            log.warn("pipeline was interrupted...");
-            Future<Z> f = it.next();
+            putQuiet(out, POISON);
-            if (f.isDone()) {
+            done.complete(null);
-                putQuiet(out, f.get());
+            return true;
-                it.remove();
-            }
         }
+        return false;
     }
 
-    void drainAll(List<Future<Z>> fs, BlockingQueue<Z> out) {
+    public void close() {
-        for (Future<Z> f : fs) {
+        log.info("Stage {} is getting shutdown", dispatcher.getClass().getSimpleName());
-            try {
+        ioPool.shutdownNow(); // прерываем обработчики
-                putQuiet(out, f.get());
+        if (!done.isDone()) {
-            } catch (Exception e) {
+            done.completeExceptionally(new java.util.concurrent.CancellationException("Stage cancelled"));
-                log.error("Error: {}", ExceptionUtils.getStackTrace(e));
-            }
         }
-        fs.clear();
-    }
-
-    private ExecutorService newIoExecutor() {
-        // один carrier на ядро, задачи — виртуальные
-        return Executors.newThreadPerTaskExecutor(Thread.ofVirtual().name("vt-io-", 0).factory());
     }
 
     void putQuiet(BlockingQueue<Z> q, Z b) {
@@ -130,22 +138,4 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
     public BlockingQueue<X> getIn() {
         return in;
     }
-
-    static void sleep(long ms) {
-        try {
-            Thread.sleep(ms);
-        } catch (InterruptedException e) {
-            Thread.currentThread().interrupt();
-        }
-    }
-
-//    @Override
-    public void stop() {
-
-    }
-
-//    @Override
-    public boolean isRunning() {
-        return true;
-    }
 }

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

@@ -6,6 +6,7 @@ import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -33,6 +34,7 @@ public class KickPipeLineImpl implements IPipeLineStarter<SubjectPayload> {
     @Override
     public void startFillingQueue(BlockingQueue<SubjectPayload> firstQueue) {
         log.info("Start registering subjects. PackageId: {}", packageId);
+        AtomicInteger counter = new AtomicInteger(0);
         try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(Path.of(pkgPath)), 8 * 1024 * 1024)) {
             xmlProcessor.read(in, MonitoringRequest.class,
                 request -> {
@@ -44,8 +46,13 @@ public class KickPipeLineImpl implements IPipeLineStarter<SubjectPayload> {
                                 .build()
 
                         ).collect(Collectors.toList());
-                        log.trace("Reading subject's batch. Total count: {}. PackageId: {}", request.getSubjects().getSubject().size(), packageId);
+                        int count = counter.incrementAndGet();
-                        firstQueue.add(new SubjectPayload(packageId, subjectProfileBatch));
+                        log.trace("Reading subject's {} batch. Total count: {}. PackageId: {}", count, request.getSubjects().getSubject().size(), packageId);
+                        try {
+                            firstQueue.put(new SubjectPayload(packageId, subjectProfileBatch));
+                        } catch (InterruptedException e) {
+                            throw new RuntimeException(e);
+                        }
                         request.getSubjects().getSubject().clear();
                     }
                 },