From 29b47b8b96d3d4a507bff49c8ea6fda5020a73db Mon Sep 17 00:00:00 2001
From: AKurakin <akurakin@mfd.ru>
Date: Wed, 3 May 2023 19:02:00 +0300
Subject: [PATCH] =?UTF-8?q?QueueConsumer=20=D0=B7=D0=B0=D1=89=D0=B8=D1=82?=
 =?UTF-8?q?=D0=B0=20=D0=BE=D1=82=20=D0=BF=D0=B5=D1=80=D0=B5=D0=BF=D0=BE?=
 =?UTF-8?q?=D0=BB=D0=BD=D0=B5=D0=BD=D0=B8=D1=8F=20=D1=81=D0=BF=D0=B0=D0=BC?=
 =?UTF-8?q?=D0=BE=D0=BC=20=D0=BB=D0=BE=D0=B3=D0=BE=D0=B2?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../clearing/platform/messaging/domain/Consts.java    |  1 -
 .../platform/messaging/service/QueueConsumer.java     | 11 +++++++++++
 2 files changed, 11 insertions(+), 1 deletion(-)

diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java
index 51410a656..c6c9f0d28 100644
--- a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java
+++ b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java
@@ -99,7 +99,6 @@ public interface Consts {
     String SDF11_PROCESS = "sdf11-process";
     String EXPORT_PROCESS = "export-process";
     String S_TRADES_IMPORTED = "s_trades-imported";
-    String ACCOUNT_NEW = "account-new";
     String ACCOUNT_TERMINATION = "account-termination";
     String BALANCE_ACCOUNT_NEW = "balance-account-new";
     String BALANCE_ACCOUNT_UPDATE = "balance-account-update";
diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/QueueConsumer.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/QueueConsumer.java
index 83a2d7ca5..0b26d5c03 100644
--- a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/QueueConsumer.java
+++ b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/QueueConsumer.java
@@ -80,6 +80,7 @@ public class QueueConsumer implements AutoCloseable {
                     consumer.subscribe(callbacks.keySet());
                 }
                 Object o = null;
+                int lastErrors = 0;
                 while (!closed.get()) {
                     try {
                         ConsumerRecords<String, Object> records = consumer.poll(Duration.of(10, ChronoUnit.SECONDS));
@@ -95,11 +96,21 @@ public class QueueConsumer implements AutoCloseable {
                                 }
                             }
                         }
+                        lastErrors = 0;
                     } catch (Throwable e) {
                         log.error(ExceptionUtils.getStackTrace(e));
                         if (producer != null && o != null) {
                             sendErrorResponse((BaseRequest<?>) o);
                         }
+                        if (lastErrors++ > 20) {
+                            log.warn("Too many error at row, {}. Sleep.", lastErrors);
+                            try {
+                                Thread.sleep(1000L);
+                            } catch (InterruptedException ie) {
+                                log.info("Thread interrupted. {}", ExceptionUtils.getStackTrace(ie));
+                                break;
+                            }
+                        }
                     }
                 }
             } catch (WakeupException e) {