Added errorThreshold to cdc config

Author: alex268

src/main/java/tech/ydb/app/CdcConfig.java

@@ -16,4 +16,6 @@ public interface CdcConfig {
     int getThreadsCount();
 
     int getTimeoutSeconds();
+
+    int getErrorThreshold();
 }

src/main/java/tech/ydb/app/XmlConfig.java

@@ -36,6 +36,8 @@ public static class Cdc implements CdcConfig {
         private Integer threadsCount;
         @XmlAttribute(name = "timeoutSeconds")
         private Integer timeoutSeconds;
+        @XmlAttribute(name = "errorThreshold")
+        private Integer errorThreshold;
 
         @XmlValue
         private String query;
@@ -78,5 +80,13 @@ public int getTimeoutSeconds() {
             }
             return timeoutSeconds;
         }
+
+        @Override
+        public int getErrorThreshold() {
+            if (errorThreshold == null) {
+                return 0;
+            }
+            return errorThreshold;
+        }
     }
 }

src/main/java/tech/ydb/app/YqlWriter.java

@@ -39,6 +39,7 @@ public class YqlWriter implements AutoCloseable {
     private final YdbService ydb;
     private final String queryYql;
     private final int timeoutSeconds;
+    private final int errorThreshold;
 
     private final List<Writer> writers;
 
@@ -52,6 +53,7 @@ private YqlWriter(YdbService ydb, CdcConfig config, String prmName, StructType t
         this.ydb = ydb;
         this.queryYql = config.getQuery();
         this.timeoutSeconds = config.getTimeoutSeconds();
+        this.errorThreshold = config.getErrorThreshold();
 
         this.lastWrited = null;
         this.lastReaded = null;
@@ -152,6 +154,12 @@ public void run() {
                 Random rnd = new Random();
 
                 while (!Thread.interrupted()) {
+                    Message msg = queue.poll();
+                    if (msg == null) {
+                        Thread.sleep(1000);
+                        continue;
+                    }
+
                     long now = System.currentTimeMillis();
                     long printedAt = lastPrinted.get();
                     if ((now - printedAt > 1000) && lastPrinted.compareAndSet(printedAt, now)) {
@@ -161,12 +169,6 @@ public void run() {
                         logger.debug("writed {} rows, {} rps", written, String.format("%.2f", avg));
                     }
 
-                    Message msg = queue.poll();
-                    if (msg == null) {
-                        Thread.sleep(1000);
-                        continue;
-                    }
-
                     DeferredCommitter committer = DeferredCommitter.newInstance();
                     Instant last = msg.getCreatedAt();
 
@@ -198,8 +200,14 @@ public void run() {
                         retry++;
                         long delay = 25 << Math.min(retry, 8);
                         delay = delay + rnd.nextLong(delay);
-                        logger.warn("got error {} after {} ms", lastStatus, ms);
-                        logger.warn("retry #{} in {} ms", retry, delay);
+                        if (retry > errorThreshold) {
+                            logger.warn("got error {} after {} ms", lastStatus, ms);
+                            logger.warn("retry #{} in {} ms", retry, delay);
+                        } else {
+                            logger.trace("got error {} after {} ms", lastStatus, ms);
+                            logger.trace("retry #{} in {} ms", retry, delay);
+                        }
+
                         Thread.sleep(delay);
 
                         now = System.currentTimeMillis();