GrpcStreamRetrier was made configurable

Author: Igor Melnichenko

topic/src/main/java/tech/ydb/topic/TopicClient.java

@@ -125,7 +125,20 @@ default CompletableFuture<Result<ConsumerDescription>> describeConsumer(String p
      * @param settings  reader settings
      * @return topic {@link SyncReader}
      */
-    SyncReader createSyncReader(ReaderSettings settings);
+    default SyncReader createSyncReader(ReaderSettings settings) {
+        return createSyncReader(null, settings);
+    }
+
+    /**
+     * Create sync topic reader.
+     *
+     * @param id an ID of a reader to be used for logging, tracing, etc. If is {@code null}, an autogenerated value will
+     *           be used
+     * @param settings reader settings
+     *
+     * @return topic {@link SyncReader}
+     */
+    SyncReader createSyncReader(String id, ReaderSettings settings);
 
     /**
      * Create async topic reader.
@@ -134,7 +147,21 @@ default CompletableFuture<Result<ConsumerDescription>> describeConsumer(String p
      * @param handlersSettings  settings for read event handling
      * @return topic {@link AsyncReader}
      */
-    AsyncReader createAsyncReader(ReaderSettings settings, ReadEventHandlersSettings handlersSettings);
+    default AsyncReader createAsyncReader(ReaderSettings settings, ReadEventHandlersSettings handlersSettings) {
+        return createAsyncReader(null, settings, handlersSettings);
+    }
+
+    /**
+     * Create async topic reader.
+     *
+     * @param id an ID of a reader to be used for logging, tracing, etc. If is {@code null}, an autogenerated value will
+     *           be used
+     * @param settings reader settings
+     * @param handlersSettings settings for read event handling
+     *
+     * @return topic {@link AsyncReader}
+     */
+    AsyncReader createAsyncReader(String id, ReaderSettings settings, ReadEventHandlersSettings handlersSettings);
 
     /**
      * Commit offset to topic
@@ -151,15 +178,41 @@ default CompletableFuture<Result<ConsumerDescription>> describeConsumer(String p
      * @param settings  {@link WriterSettings}
      * @return topic {@link SyncWriter}
      */
-    SyncWriter createSyncWriter(WriterSettings settings);
+    default SyncWriter createSyncWriter(WriterSettings settings) {
+        return createSyncWriter(null, settings);
+    }
+
+    /**
+     * Create sync topic writer.
+     *
+     * @param id an ID of a writer to be used for logging, tracing, etc. If is {@code null}, an autogenerated value will
+     *           be used
+     * @param settings  {@link WriterSettings}
+     *
+     * @return topic {@link SyncWriter}
+     */
+    SyncWriter createSyncWriter(String id, WriterSettings settings);
 
     /**
      * Create async topic writer.
      *
      * @param settings  {@link WriterSettings}
      * @return topic {@link AsyncWriter}
      */
-    AsyncWriter createAsyncWriter(WriterSettings settings);
+    default AsyncWriter createAsyncWriter(WriterSettings settings) {
+        return createAsyncWriter(null, settings);
+    }
+
+    /**
+     * Create async topic writer.
+     *
+     * @param id an ID of a writer to be used for logging, tracing, etc. If is {@code null}, an autogenerated value will
+     *           be used
+     * @param settings  {@link WriterSettings}
+     *
+     * @return topic {@link AsyncWriter}
+     */
+    AsyncWriter createAsyncWriter(String id, WriterSettings settings);
 
     @Override
     void close();

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

@@ -35,9 +35,13 @@ public abstract class GrpcStreamRetrier {
     private final ScheduledExecutorService scheduler;
     private final BiConsumer<Status, Throwable> errorsHandler;
 
-    protected GrpcStreamRetrier(ScheduledExecutorService scheduler, BiConsumer<Status, Throwable> errorsHandler) {
+    protected GrpcStreamRetrier(
+            String id,
+            ScheduledExecutorService scheduler,
+            BiConsumer<Status, Throwable> errorsHandler
+    ) {
         this.scheduler = scheduler;
-        this.id = generateRandomId(ID_LENGTH);
+        this.id = id==null ? generateRandomId(ID_LENGTH):id;
         this.errorsHandler = errorsHandler;
     }
 

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

@@ -362,13 +362,13 @@ private TopicDescription mapDescribeTopic(YdbTopic.DescribeTopicResult result) {
     }
 
     @Override
-    public SyncReader createSyncReader(ReaderSettings settings) {
-        return new SyncReaderImpl(topicRpc, settings);
+    public SyncReader createSyncReader(String id, ReaderSettings settings) {
+        return new SyncReaderImpl(id, topicRpc, settings);
     }
 
     @Override
-    public AsyncReader createAsyncReader(ReaderSettings settings, ReadEventHandlersSettings handlersSettings) {
-        return new AsyncReaderImpl(topicRpc, settings, handlersSettings);
+    public AsyncReader createAsyncReader(String id, ReaderSettings settings, ReadEventHandlersSettings handlersSettings) {
+        return new AsyncReaderImpl(id, topicRpc, settings, handlersSettings);
     }
 
     @Override
@@ -389,13 +389,13 @@ public CompletableFuture<Status> commitOffset(String path, CommitOffsetSettings
     }
 
     @Override
-    public SyncWriter createSyncWriter(WriterSettings settings) {
-        return new SyncWriterImpl(topicRpc, settings, compressionExecutor);
+    public SyncWriter createSyncWriter(String id, WriterSettings settings) {
+        return new SyncWriterImpl(id, topicRpc, settings, compressionExecutor);
     }
 
     @Override
-    public AsyncWriter createAsyncWriter(WriterSettings settings) {
-        return new AsyncWriterImpl(topicRpc, settings, compressionExecutor);
+    public AsyncWriter createAsyncWriter(String id, WriterSettings settings) {
+        return new AsyncWriterImpl(id, topicRpc, settings, compressionExecutor);
     }
 
     private static YdbTopic.MeteringMode toProto(MeteringMode meteringMode) {

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

@@ -47,10 +47,19 @@ public class AsyncReaderImpl extends ReaderImpl implements AsyncReader {
     private final ReadEventHandler eventHandler;
 
     public AsyncReaderImpl(TopicRpc topicRpc, ReaderSettings settings, ReadEventHandlersSettings handlersSettings) {
-        super(topicRpc, settings);
+        this(null, topicRpc, settings, handlersSettings);
+    }
+
+    public AsyncReaderImpl(
+            String id,
+            TopicRpc topicRpc,
+            ReaderSettings settings,
+            ReadEventHandlersSettings handlersSettings
+    ) {
+        super(id, topicRpc, settings);
         this.eventHandler = handlersSettings.getEventHandler();
 
-        if (handlersSettings.getExecutor() != null) {
+        if (handlersSettings.getExecutor()!=null) {
             logger.debug("Using handler executor provided by user");
             this.defaultHandlerExecutorService = null;
             this.handlerExecutor = handlersSettings.getExecutor();

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

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

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

@@ -45,6 +45,10 @@ public SyncReaderImpl(TopicRpc topicRpc, ReaderSettings settings) {
         super(topicRpc, settings);
     }
 
+    public SyncReaderImpl(String id, TopicRpc topicRpc, ReaderSettings settings) {
+        super(id, topicRpc, settings);
+    }
+
     private static class MessageBatchWrapper {
         private final List<Message> messages;
         private final CompletableFuture<Void> future;

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

@@ -17,11 +17,14 @@
  * @author Nikolay Perfilov
  */
 public class AsyncWriterImpl extends WriterImpl implements AsyncWriter {
-
     public AsyncWriterImpl(TopicRpc topicRpc, WriterSettings settings, Executor compressionExecutor) {
         super(topicRpc, settings, compressionExecutor);
     }
 
+    public AsyncWriterImpl(String id, TopicRpc topicRpc, WriterSettings settings, Executor compressionExecutor) {
+        super(id, topicRpc, settings, compressionExecutor);
+    }
+
     @Override
     public CompletableFuture<InitResult> init() {
         return initImpl();

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

@@ -22,6 +22,10 @@ public SyncWriterImpl(TopicRpc topicRpc, WriterSettings settings, Executor compr
         super(topicRpc, settings, compressionExecutor);
     }
 
+    public SyncWriterImpl(String id, TopicRpc topicRpc, WriterSettings settings, Executor compressionExecutor) {
+        super(topicRpc, settings, compressionExecutor);
+    }
+
     @Override
     public void init() {
         initImpl();

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

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