Added RetryMode to configure retries behavior

Author: alex268

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

@@ -12,6 +12,7 @@
 import org.slf4j.Logger;
 
 import tech.ydb.core.Status;
+import tech.ydb.topic.settings.RetryMode;
 
 /**
  * @author Nikolay Perfilov
@@ -26,13 +27,15 @@ public abstract class GrpcStreamRetrier {
     private static final char[] ID_ALPHABET = "abcdefghijklmnopqrstuvwxyzABSDEFGHIJKLMNOPQRSTUVWXYZ1234567890"
             .toCharArray();
 
+    private final RetryMode retryMode;
     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(RetryMode retryMode, ScheduledExecutorService scheduler) {
+        this.retryMode = retryMode;
         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.getRetryMode(), topicRpc.getScheduler());
         this.topicRpc = topicRpc;
         this.settings = settings;
         this.session = new ReadSessionImpl();

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

@@ -17,13 +17,15 @@ public class ReaderSettings {
     private final String consumerName;
     private final String readerName;
     private final List<TopicReadSettings> topics;
+    private final RetryMode retryMode;
     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.retryMode = builder.retryMode;
         this.maxMemoryUsageBytes = builder.maxMemoryUsageBytes;
         this.decompressionExecutor = builder.decompressionExecutor;
     }
@@ -37,6 +39,10 @@ public String getReaderName() {
         return readerName;
     }
 
+    public RetryMode getRetryMode() {
+        return retryMode;
+    }
+
     public List<TopicReadSettings> getTopics() {
         return topics;
     }
@@ -61,6 +67,7 @@ public static class Builder {
         private boolean readWithoutConsumer = false;
         private String readerName = null;
         private List<TopicReadSettings> topics = new ArrayList<>();
+        private RetryMode retryMode = RetryMode.ALWAYS;
         private long maxMemoryUsageBytes = MAX_MEMORY_USAGE_BYTES_DEFAULT;
         private Executor decompressionExecutor = null;
 
@@ -81,6 +88,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 +106,16 @@ public Builder setTopics(List<TopicReadSettings> topics) {
             return this;
         }
 
+        /**
+         * Set {@link RetryMode} to define behavior of the stream internal retries
+         * @param mode retry mode
+         * @return settings builder
+         */
+        public Builder setRetryMode(RetryMode mode) {
+            this.retryMode = mode;
+            return this;
+        }
+
         public Builder setMaxMemoryUsageBytes(long maxMemoryUsageBytes) {
             this.maxMemoryUsageBytes = maxMemoryUsageBytes;
             return this;

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

@@ -0,0 +1,16 @@
+package tech.ydb.topic.settings;
+
+/**
+ *
+ * @author Aleksandr Gorshenin
+ */
+public enum RetryMode {
+    /** Never retry stream. After any error reader/writer will be stopped */
+    NONE,
+
+    /** Don't retry first attempt to establish a stream, but after successful connection retry any error until stop */
+    RECOVER,
+
+    /** Retry any error until reader/writer was stopped by client */
+    ALWAYS,
+}

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

@@ -14,6 +14,7 @@ public class WriterSettings {
     private final String messageGroupId;
     private final Long partitionId;
     private final Codec codec;
+    private final RetryMode retryMode;
     private final long maxSendBufferMemorySize;
     private final int maxSendBufferMessagesCount;
 
@@ -23,6 +24,7 @@ private WriterSettings(Builder builder) {
         this.messageGroupId = builder.messageGroupId;
         this.partitionId = builder.partitionId;
         this.codec = builder.codec;
+        this.retryMode = builder.retryMode;
         this.maxSendBufferMemorySize = builder.maxSendBufferMemorySize;
         this.maxSendBufferMessagesCount = builder.maxSendBufferMessagesCount;
     }
@@ -51,6 +53,10 @@ public Codec getCodec() {
         return codec;
     }
 
+    public RetryMode getRetryMode() {
+        return retryMode;
+    }
+
     public long getMaxSendBufferMemorySize() {
         return maxSendBufferMemorySize;
     }
@@ -68,6 +74,7 @@ public static class Builder {
         private String messageGroupId = null;
         private Long partitionId = null;
         private Codec codec = Codec.GZIP;
+        private RetryMode retryMode = RetryMode.ALWAYS;
         private long maxSendBufferMemorySize = MAX_MEMORY_USAGE_BYTES_DEFAULT;
         private int maxSendBufferMessagesCount = MAX_IN_FLIGHT_COUNT_DEFAULT;
 
@@ -125,6 +132,16 @@ public Builder setCodec(Codec codec) {
             return this;
         }
 
+        /**
+         * Set {@link RetryMode} to define behavior of the stream internal retries
+         * @param mode retry mode
+         * @return settings builder
+         */
+        public Builder setRetryMode(RetryMode mode) {
+            this.retryMode = mode;
+            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.getRetryMode(), topicRpc.getScheduler());
         this.topicRpc = topicRpc;
         this.settings = settings;
         this.session = new WriteSessionImpl();