diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/Stage.java b/src/main/java/ru/nbch/credit_tracker/service/task/Stage.java
index 478fcaf..03dd6de 100644
--- a/src/main/java/ru/nbch/credit_tracker/service/task/Stage.java
+++ b/src/main/java/ru/nbch/credit_tracker/service/task/Stage.java
@@ -4,7 +4,6 @@ import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 import java.util.concurrent.LinkedBlockingQueue;
@@ -22,8 +21,11 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
     private final BlockingQueue<Z> out;
     private final Z POISON;
     private final ExecutorService ioPool;
-    private final CompletableFuture<Void> done = new CompletableFuture<>();
     private Consumer<Z> consumer = null;
+    private final String dispatcherName;
+
+    private static final ThreadPoolExecutor.CallerRunsPolicy callerRunsPolicy
+        = new ThreadPoolExecutor.CallerRunsPolicy();
 
     public Stage(Handler<X, Z> dispatcher,
                  BlockingQueue<X> in,
@@ -32,24 +34,24 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
         this.dispatcher = dispatcher;
         this.in = in;
         this.out = out;
+        this.dispatcherName = dispatcher.getClass().getSimpleName();
         this.ioPool = new ThreadPoolExecutor(10,
             10,
             0L, TimeUnit.MINUTES,
             new LinkedBlockingQueue<>(1),
             r -> {
                 Thread t = new Thread(r);
-                t.setName("pipe-handler-%s-%d".formatted(dispatcher.getClass().getSimpleName(), t.threadId()));
+                t.setName("pipe-handler-%s-%d".formatted(dispatcherName, t.threadId()));
                 return t;
             },
-            new ThreadPoolExecutor.CallerRunsPolicy()
+            (r, executor) -> {
+                log.info("stage {}: all threads are working; switch to CallerRunsPolicy", dispatcherName);
+                callerRunsPolicy.rejectedExecution(r, executor);
+            }
         );
         this.POISON = POISON;
     }
 
-    public CompletableFuture<Void> doneFuture() {
-        return done;
-    }
-
     public void startRunning() {
         ioPool.submit(this);
     }
@@ -78,37 +80,48 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
                     }
                     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);
+                        try {
+                            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 {} ended.", dispatcher.getClass().getSimpleName(), finalCounter);
+                            return p;
+                        } catch (Exception e) {
+                            log.error("{} ERROR: {}", dispatcherName, ExceptionUtils.getStackTrace(e));
                             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);
             }
         } catch (Exception e) {
-            log.error("error: {}", ExceptionUtils.getStackTrace(e));
+            log.error("Error for {}: {}; {}", dispatcherName, info(), ExceptionUtils.getStackTrace(e));
             putQuiet(out, POISON);
-            done.complete(null);
         }
     }
 
+    private String info() {
+        ThreadPoolExecutor casted = (ThreadPoolExecutor) ioPool;
+        return "executor active tasks: %d; queue size: %d, completed tasks: %d".formatted(
+            casted.getActiveCount(),
+            casted.getQueue().size(),
+            casted.getCompletedTaskCount()
+        );
+    }
+
     private boolean wasInterrupted() {
         if (Thread.currentThread().isInterrupted()) {
             log.warn("pipeline was interrupted...");
             putQuiet(out, POISON);
-            done.complete(null);
             return true;
         }
         return false;
@@ -117,9 +130,6 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
     public void close() {
         log.info("Stage {} is getting shutdown", dispatcher.getClass().getSimpleName());
         ioPool.shutdownNow(); // прерываем обработчики
-        if (!done.isDone()) {
-            done.completeExceptionally(new java.util.concurrent.CancellationException("Stage cancelled"));
-        }
     }
 
     void putQuiet(BlockingQueue<Z> q, Z b) {
@@ -129,7 +139,7 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
             }
             q.put(b);
         } catch (InterruptedException e) {
-            log.error("Error: {}", ExceptionUtils.getStackTrace(e));
+            log.error("Error for {}: {}; {}", dispatcherName, info(), ExceptionUtils.getStackTrace(e));
             Thread.currentThread().interrupt();
         }
     }
@@ -138,7 +148,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: {}", ExceptionUtils.getStackTrace(e));
+            log.error("Error for {}: {}; {}", dispatcherName, info(), ExceptionUtils.getStackTrace(e));
             Thread.currentThread().interrupt();
             return null;
         }
diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineImpl.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineImpl.java
index 150be44..9a89517 100644
--- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineImpl.java
+++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineImpl.java
@@ -15,6 +15,7 @@ import ru.nbch.credit_tracker.model.SubjectProfile;
 import ru.nbch.credit_tracker.model.impl.SubjectPayload;
 import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter;
 import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
+import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
 
 public class KickPipeLineImpl implements IPipeLineStarter<SubjectPayload> {
     private final Logger log = LoggerFactory.getLogger(getClass());
@@ -51,6 +52,7 @@ public class KickPipeLineImpl implements IPipeLineStarter<SubjectPayload> {
                         try {
                             firstQueue.put(new SubjectPayload(packageId, subjectProfileBatch));
                         } catch (InterruptedException e) {
+                            log.error("{}", ExceptionUtils.getStackTrace(e));
                             throw new RuntimeException(e);
                         }
                         request.getSubjects().getSubject().clear();
@@ -60,10 +62,14 @@ public class KickPipeLineImpl implements IPipeLineStarter<SubjectPayload> {
                 ,
                 BATCH_SIZE, "/MonitoringRequest/Subjects/Subject");
         } catch (IOException e) {
-            throw new RuntimeException("Error while parsing monitoring request", e);
+            log.error("{}", ExceptionUtils.getStackTrace(e));
         } finally {
             log.debug("Send poison pill");
-            firstQueue.add(new SubjectPayload(true));
+            try {
+                firstQueue.put(new SubjectPayload(true));
+            } catch (InterruptedException e) {
+                log.error("couldn't send poison: {}", ExceptionUtils.getStackTrace(e));
+            }
         }
     }
 }