Added RetryConfig to configure retries behavior

Author: alex268

topic/src/main/java/tech/ydb/topic/impl/GrpcStreamRetrier.java

@@ -11,6 +11,7 @@
 
 import org.slf4j.Logger;
 
+import tech.ydb.common.retry.RetryConfig;
 import tech.ydb.core.Status;
 
 /**
@@ -26,13 +27,15 @@ public abstract class GrpcStreamRetrier {
     private static final char[] ID_ALPHABET = "abcdefghijklmnopqrstuvwxyzABSDEFGHIJKLMNOPQRSTUVWXYZ1234567890"
             .toCharArray();
 
+    private final RetryConfig retryConfig;
     protected final String id;
     protected final AtomicBoolean isReconnecting = new AtomicBoolean(false);
     protected final AtomicBoolean isStopped = new AtomicBoolean(false);
     private final ScheduledExecutorService scheduler;
     protected final AtomicInteger reconnectCounter = new AtomicInteger(0);
 
-    protected GrpcStreamRetrier(ScheduledExecutorService scheduler) {
+    protected GrpcStreamRetrier(RetryConfig retryConfig, ScheduledExecutorService scheduler) {
+        this.retryConfig = retryConfig;
         this.scheduler = scheduler;
         this.id = generateRandomId(ID_LENGTH);
     }

topic/src/main/java/tech/ydb/topic/read/impl/ReaderImpl.java

@@ -55,7 +55,7 @@ public abstract class ReaderImpl extends GrpcStreamRetrier {
     private final String consumerName;
 
     public ReaderImpl(TopicRpc topicRpc, ReaderSettings settings) {
-        super(topicRpc.getScheduler());
+        super(settings.getRetryConfig(), topicRpc.getScheduler());
         this.topicRpc = topicRpc;
         this.settings = settings;
         this.session = new ReadSessionImpl();

topic/src/main/java/tech/ydb/topic/settings/ReaderSettings.java

@@ -8,6 +8,8 @@
 
 import com.google.common.collect.ImmutableList;
 
+import tech.ydb.common.retry.RetryConfig;
+
 /**
  * @author Nikolay Perfilov
  */
@@ -17,13 +19,15 @@ public class ReaderSettings {
     private final String consumerName;
     private final String readerName;
     private final List<TopicReadSettings> topics;
+    private final RetryConfig retryConfig;
     private final long maxMemoryUsageBytes;
     private final Executor decompressionExecutor;
 
     private ReaderSettings(Builder builder) {
         this.consumerName = builder.consumerName;
         this.readerName = builder.readerName;
         this.topics = ImmutableList.copyOf(builder.topics);
+        this.retryConfig = builder.retryConfig;
         this.maxMemoryUsageBytes = builder.maxMemoryUsageBytes;
         this.decompressionExecutor = builder.decompressionExecutor;
     }
@@ -37,6 +41,10 @@ public String getReaderName() {
         return readerName;
     }
 
+    public RetryConfig getRetryConfig() {
+        return retryConfig;
+    }
+
     public List<TopicReadSettings> getTopics() {
         return topics;
     }
@@ -61,6 +69,7 @@ public static class Builder {
         private boolean readWithoutConsumer = false;
         private String readerName = null;
         private List<TopicReadSettings> topics = new ArrayList<>();
+        private RetryConfig retryConfig = RetryConfig.idempotentRetryForever();
         private long maxMemoryUsageBytes = MAX_MEMORY_USAGE_BYTES_DEFAULT;
         private Executor decompressionExecutor = null;
 
@@ -81,6 +90,7 @@ public Builder withoutConsumer() {
 
         /**
          * Set reader name for debug purposes
+         * @param readerName name of reader
          * @return settings builder
          */
         public Builder setReaderName(String readerName) {
@@ -98,6 +108,16 @@ public Builder setTopics(List<TopicReadSettings> topics) {
             return this;
         }
 
+        /**
+         * Set {@link RetryConfig} to define behavior of the stream internal retries
+         * @param config retry mode
+         * @return settings builder
+         */
+        public Builder setRetryConfig(RetryConfig config) {
+            this.retryConfig = config;
+            return this;
+        }
+
         public Builder setMaxMemoryUsageBytes(long maxMemoryUsageBytes) {
             this.maxMemoryUsageBytes = maxMemoryUsageBytes;
             return this;

topic/src/main/java/tech/ydb/topic/settings/WriterSettings.java

@@ -1,5 +1,6 @@
 package tech.ydb.topic.settings;
 
+import tech.ydb.common.retry.RetryConfig;
 import tech.ydb.topic.description.Codec;
 
 /**
@@ -14,6 +15,7 @@ public class WriterSettings {
     private final String messageGroupId;
     private final Long partitionId;
     private final Codec codec;
+    private final RetryConfig retryConfig;
     private final long maxSendBufferMemorySize;
     private final int maxSendBufferMessagesCount;
 
@@ -23,6 +25,7 @@ private WriterSettings(Builder builder) {
         this.messageGroupId = builder.messageGroupId;
         this.partitionId = builder.partitionId;
         this.codec = builder.codec;
+        this.retryConfig = builder.retryConfig;
         this.maxSendBufferMemorySize = builder.maxSendBufferMemorySize;
         this.maxSendBufferMessagesCount = builder.maxSendBufferMessagesCount;
     }
@@ -51,6 +54,10 @@ public Codec getCodec() {
         return codec;
     }
 
+    public RetryConfig getRetryConfig() {
+        return retryConfig;
+    }
+
     public long getMaxSendBufferMemorySize() {
         return maxSendBufferMemorySize;
     }
@@ -68,6 +75,7 @@ public static class Builder {
         private String messageGroupId = null;
         private Long partitionId = null;
         private Codec codec = Codec.GZIP;
+        private RetryConfig retryConfig = RetryConfig.idempotentRetryForever();
         private long maxSendBufferMemorySize = MAX_MEMORY_USAGE_BYTES_DEFAULT;
         private int maxSendBufferMessagesCount = MAX_IN_FLIGHT_COUNT_DEFAULT;
 
@@ -125,6 +133,16 @@ public Builder setCodec(Codec codec) {
             return this;
         }
 
+        /**
+         * Set {@link RetryConfig} to define behavior of the stream internal retries
+         * @param config retry mode
+         * @return settings builder
+         */
+        public Builder setRetryConfig(RetryConfig config) {
+            this.retryConfig = config;
+            return this;
+        }
+
         /**
          * Set memory usage limit for send buffer.
          * Writer will not accept new messages if memory usage exceeds this limit.

topic/src/main/java/tech/ydb/topic/write/impl/WriterImpl.java

@@ -66,7 +66,7 @@ public abstract class WriterImpl extends GrpcStreamRetrier {
     private CompletableFuture<WriteAck> lastAcceptedMessageFuture;
 
     public WriterImpl(TopicRpc topicRpc, WriterSettings settings, Executor compressionExecutor) {
-        super(topicRpc.getScheduler());
+        super(settings.getRetryConfig(), topicRpc.getScheduler());
         this.topicRpc = topicRpc;
         this.settings = settings;
         this.session = new WriteSessionImpl();