pipeline interrupt [2]

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

@@ -1,16 +1,52 @@
 package ru.nbch.credit_tracker.service.task;
 
 import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import ru.nbch.credit_tracker.service.task.setup.IPipeLineTerminator;
+import ru.nbch.credit_tracker.utils.LogUtil;
 
 public class Pipeline {
+    private final Logger log = LoggerFactory.getLogger(getClass());
     private Runnable starter;
     private IPipeLineTerminator<?,?> terminator;
     private List<Stage<?, ?>> stages;
+    private Semaphore semaphore;
+    private Future<?> starterFuture;
+    private Future<?> terminatorFuture;
+    private final AtomicBoolean closed = new AtomicBoolean(false);
 
     public Pipeline() {
     }
 
+    public void start(ExecutorService executorService) {
+        this.starterFuture = executorService.submit(LogUtil.wrap(LogUtil.extractMdcMap(), starter));
+        this.terminatorFuture = executorService.submit(LogUtil.wrap(LogUtil.extractMdcMap(), terminator));
+    }
+
+    public void close() {
+        if (closed.compareAndSet(false, true)) {
+            log.info("closing the pipeline...");
+            try {
+                stages.forEach(Stage::close);
+                this.starterFuture.cancel(true);
+                this.terminatorFuture.cancel(true);
+            } finally {
+                semaphore.release();
+            }
+        } else {
+            log.info("pipeline already closed.");
+        }
+    }
+
+    public void setSemaphore(Semaphore semaphore) {
+        this.semaphore = semaphore;
+    }
+
     public Stage<?, ?> getLast() {
         return stages.getLast();
     }

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

@@ -18,10 +18,10 @@ 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.FlagsEnricherV2;
 import ru.nbch.credit_tracker.service.task.impl.SaveSubjectHandler;
+import ru.nbch.credit_tracker.service.task.impl.XmlMergeCoordinator;
 import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter;
 import ru.nbch.credit_tracker.service.task.setup.IPipeLineTerminator;
 import ru.nbch.credit_tracker.service.task.setup.PipeLineBuilderV2;
-import ru.nbch.credit_tracker.utils.LogUtil;
 import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
 
 @Component
@@ -84,17 +84,19 @@ public class PipelinePackageManager {
                     .andThen()
                 .fillInitialQueue(pipelineStarterFactory.apply(packageId, pkgPath))
                 .pollFinishingQueue(pipelineTerminatorFactory.get())
+                .semaphore(semaphore)
                 .build();
 
-            executorService.submit(LogUtil.wrap(LogUtil.extractMdcMap(), pipeline.getStarter()));
+            pipeline.start(executorService);
-            executorService.submit(LogUtil.wrap(LogUtil.extractMdcMap(), pipeline.getTerminator()));
-
             pipeline.getTerminator()
                 .doneFuture()
                 .orTimeout(creditTrackerSettings.getTimeout(), TimeUnit.MINUTES)
                 .whenComplete((v, e) -> {
                     try {
-                        if (e != null) {
+                        if (e instanceof XmlMergeCoordinator.PipelineTerminatorException) {
+                            log.info("when complete: terminator was interrupted.");
+                            return;
+                        } else if (e != null) {
                             log.error(ExceptionUtils.getStackTrace(e));
                             return;
                         }
@@ -105,8 +107,8 @@ public class PipelinePackageManager {
                         log.error("pipeline finished with error{}", ExceptionUtils.getStackTrace(ex));
                     } finally {
                         log.info("packageId={}: pipeline finished", packageId);
-                        pipeline.getStages().forEach(Stage::close);
+                        pipeline.close();
-                        semaphore.release();
+                        log.info("semaphore amount: {}", semaphore.availablePermits());
                     }
                 });
         } catch (Exception e) {

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

@@ -9,9 +9,11 @@ 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 java.util.function.Consumer;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.mybatis.spring.MyBatisSystemException;
 import ru.nbch.credit_tracker.model.StagePayload;
 import ru.nbch.credit_tracker.utils.LogUtil;
 
@@ -24,6 +26,7 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
     private final ExecutorService ioPool;
     private Consumer<Z> consumer = null;
     private final String dispatcherName;
+    private Pipeline pipeline;
 
     private static final ThreadPoolExecutor.CallerRunsPolicy callerRunsPolicy
         = new ThreadPoolExecutor.CallerRunsPolicy();
@@ -54,6 +57,10 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
         this.POISON = POISON;
     }
 
+    public void setPipeline(Pipeline pipeline) {
+        this.pipeline = pipeline;
+    }
+
     public void startRunning() {
         ioPool.submit(this);
     }
@@ -61,6 +68,7 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
     @Override
     public void run() {
         List<Future<Z>> inflight = new ArrayList<>();
+        log.info("this is main thread for {}", dispatcherName);
         try {
             int counter = 0;
             while (true) {
@@ -74,34 +82,47 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
                     // дождаться всех in-flight задач
                     for (Future<Z> f : inflight) {
                         if (wasInterrupted()) {
-                            log.info("stage {} was interrupted while waiting for tasks to finish",
+                            log.info("stage {} was interrupted while waiting for tasks to finish", dispatcherName);
-                                dispatcher.getClass().getSimpleName());
                             break;
                         }
                         f.get();
                     }
                     putQuiet(out, POISON);
-                    log.info("stage {} finished.", dispatcher.getClass().getSimpleName());
+                    log.info("stage {} finished.", dispatcherName);
                     return;
                 }
                 counter++;
                 int finalCounter = counter;
+                AtomicInteger retryForSingleTask = new AtomicInteger(0);
                 Future<Z> f = ioPool.submit(() -> {
+                        while (retryForSingleTask.incrementAndGet() < 3) {
                             try {
-                            if (wasInterrupted()) {
+                                if (Thread.currentThread().isInterrupted()) {
-                                log.info("stage {} task {} interrupted...", dispatcher.getClass().getSimpleName(), finalCounter);
+                                    log.info("stage {} was interrupted. skipping task {}...", dispatcherName, finalCounter);
                                     return null;
                                 }
                                 log.info("stage {} task {} started.", dispatcher.getClass().getSimpleName(), finalCounter);
                                 Z p = dispatcher.handle(b);
+                                if (Thread.currentThread().isInterrupted()) {
+                                    log.info("after {} task {} finished, worker has been interrupted. not putting result in queue.",
+                                        dispatcherName, finalCounter);
+                                    return null;
+                                }
                                 putQuiet(out, p);
                                 log.info("stage {} task {} ended.", dispatcher.getClass().getSimpleName(), finalCounter);
                                 return p;
+                            } catch (MyBatisSystemException e) {
+                                log.warn("{} RETRY AFTER ERROR: {}", dispatcherName, ExceptionUtils.getStackTrace(e));
                             } catch (Exception e) {
                                 log.error("{} ERROR: {}", dispatcherName, ExceptionUtils.getStackTrace(e));
+                                pipeline.close();
                                 return null;
                             }
                         }
+                        log.error("tried for {} times, failed batch...", retryForSingleTask.get());
+                        pipeline.close();
+                        return null;
+                    }
                 );
                 inflight.add(f);
                 if ((counter % 20) == 0) {
@@ -124,9 +145,9 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
     }
 
     private boolean wasInterrupted() {
-        if (Thread.currentThread().isInterrupted()) {
+        if (Thread.interrupted()) {
             log.warn("pipeline was interrupted...");
-            putQuiet(out, POISON);
+            //putQuiet(out, POISON);
             return true;
         }
         return false;
@@ -153,7 +174,7 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
         try {
             return q.poll(timeout.toMillis(), TimeUnit.MILLISECONDS);
         } catch (InterruptedException e) {
-            log.error("Error for {}: {}; {}", dispatcherName, info(), ExceptionUtils.getStackTrace(e));
+            log.error("Main thread for {}: {}; {}", dispatcherName, info(), ExceptionUtils.getStackTrace(e));
             Thread.currentThread().interrupt();
             return null;
         }

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

@@ -12,6 +12,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
 import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest;
+import ru.nbch.credit_tracker.exceptions.SignalClientException;
 import ru.nbch.credit_tracker.model.SubjectProfile;
 import ru.nbch.credit_tracker.model.impl.SubjectPayload;
 import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter;
@@ -35,6 +36,12 @@ public class KickPipeLineImpl implements IPipeLineStarter<SubjectPayload> {
         this.BATCH_SIZE = creditTrackerSettings.getBatchSize();
     }
 
+    private static class PipelineStarterException extends SignalClientException {
+        public PipelineStarterException() {
+            super(null, null, null);
+        }
+    }
+
     @Override
     public void startFillingQueue(BlockingQueue<SubjectPayload> firstQueue) {
         log.info("Start registering subjects. PackageId: {}", packageId);
@@ -55,8 +62,8 @@ public class KickPipeLineImpl implements IPipeLineStarter<SubjectPayload> {
                         try {
                             firstQueue.put(new SubjectPayload(packageId, subjectProfileBatch));
                         } catch (InterruptedException e) {
-                            log.error("{}", ExceptionUtils.getStackTrace(e));
+                            log.error("Pipeline starter was interrupted!");
-                            throw new RuntimeException(e);
+                            throw new PipelineStarterException();
                         }
                         request.getSubjects().getSubject().clear();
                     }
@@ -66,6 +73,8 @@ public class KickPipeLineImpl implements IPipeLineStarter<SubjectPayload> {
                 BATCH_SIZE, "/MonitoringRequest/Subjects/Subject");
         } catch (IOException e) {
             log.error("{}", ExceptionUtils.getStackTrace(e));
+        } catch (PipelineStarterException e) {
+            log.info("Pipeline starter shitting down...");
         } finally {
             log.debug("Send poison pill");
             try {

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

@@ -59,11 +59,16 @@ public final class XmlMergeCoordinator implements IPipeLineTerminator<PartFilePa
                 }
                 mergeBatch(Collections.singletonList(part));
             } catch (InterruptedException ie) {
-                Thread.currentThread().interrupt();
+                log.info("pipeline terminator was interrupted!");
+                done.completeExceptionally(new PipelineTerminatorException());
             } catch (Exception e) {
                 log.error("Error: {}", ExceptionUtils.getStackTrace(e));
             }
         }
+        done.completeExceptionally(new PipelineTerminatorException());
+    }
+
+    public static class PipelineTerminatorException extends RuntimeException {
     }
 
     private void mergeBatch(List<PartFilePayload> inputs) throws Exception {

src/main/java/ru/nbch/credit_tracker/service/task/setup/PipeLineBuilderV2.java

@@ -4,6 +4,7 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.Semaphore;
 import ru.nbch.credit_tracker.model.StagePayload;
 import ru.nbch.credit_tracker.service.task.Pipeline;
 import ru.nbch.credit_tracker.service.task.Stage;
@@ -15,6 +16,7 @@ public class PipeLineBuilderV2<FIRST extends StagePayload, LAST extends StagePay
     private BlockingQueue<LAST> lastQueue;
     private IPipeLineStarter<FIRST> pipeLineStarter;
     private IPipeLineTerminator<LAST, ?> pipeLineTerminator;
+    private Semaphore semaphore;
 
     private PipeLineBuilderV2() {
         this.stages = new ArrayList<>();
@@ -53,10 +55,19 @@ public class PipeLineBuilderV2<FIRST extends StagePayload, LAST extends StagePay
         return this;
     }
 
+    public PipeLineBuilderV2<FIRST, LAST> semaphore(Semaphore semaphore) {
+        this.semaphore = semaphore;
+        return this;
+    }
+
     public Pipeline build() {
         Pipeline p = new Pipeline();
+        p.setSemaphore(semaphore);
         p.setStarter(() -> {
-            stages.forEach(Stage::startRunning);
+            stages.forEach(stage -> {
+                stage.setPipeline(p);
+                stage.startRunning();
+            });
             pipeLineStarter.startFillingQueue(firstStage.getIn());
         });
         pipeLineTerminator.setPollingQueue(lastQueue);