GrpcStreamRetrier's ID was made configurable

Author: Igor Melnichenko

.editorconfig

@@ -17,5 +17,20 @@ ij_java_insert_override_annotation = true
 ij_java_lambda_brace_style = end_of_line
 ij_java_method_brace_style = end_of_line
 ij_java_names_count_to_use_import_on_demand = 9000
+ij_java_space_after_colon = true
+ij_java_space_before_colon = true
+ij_java_spaces_around_additive_operators = true
+ij_java_spaces_around_annotation_eq = true
+ij_java_spaces_around_assignment_operators = true
+ij_java_spaces_around_bitwise_operators = true
+ij_java_spaces_around_equality_operators = true
+ij_java_spaces_around_lambda_arrow = true
+ij_java_spaces_around_logical_operators = true
+ij_java_spaces_around_method_ref_dbl_colon = false
+ij_java_spaces_around_multiplicative_operators = true
+ij_java_spaces_around_relational_operators = true
+ij_java_spaces_around_shift_operators = true
+ij_java_spaces_around_type_bounds_in_type_parameters = true
+ij_java_spaces_around_unary_operator = false
 ij_java_use_single_class_imports = true
 ij_java_while_brace_force = always

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/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(), settings.getErrorsHandler());
+        super(settings.getDebugId(), topicRpc.getScheduler(), settings.getErrorsHandler());
         this.topicRpc = topicRpc;
         this.settings = settings;
         this.session = new ReadSessionImpl();

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

@@ -17,6 +17,7 @@
 public class ReaderSettings {
     private static final long MAX_MEMORY_USAGE_BYTES_DEFAULT = 100 * 1024 * 1024; // 100 MB
 
+    private final String debugId;
     private final String consumerName;
     private final String readerName;
     private final List<TopicReadSettings> topics;
@@ -25,6 +26,7 @@ public class ReaderSettings {
     private final BiConsumer<Status, Throwable> errorsHandler;
 
     private ReaderSettings(Builder builder) {
+        this.debugId = builder.debugId;
         this.consumerName = builder.consumerName;
         this.readerName = builder.readerName;
         this.topics = ImmutableList.copyOf(builder.topics);
@@ -33,6 +35,10 @@ private ReaderSettings(Builder builder) {
         this.errorsHandler = builder.errorsHandler;
     }
 
+    public String getDebugId() {
+        return debugId;
+    }
+
     public String getConsumerName() {
         return consumerName;
     }
@@ -66,6 +72,7 @@ public static Builder newBuilder() {
      * BUILDER
      */
     public static class Builder {
+        private String debugId = null;
         private String consumerName = null;
         private boolean readWithoutConsumer = false;
         private String readerName = null;
@@ -74,6 +81,19 @@ public static class Builder {
         private Executor decompressionExecutor = null;
         private BiConsumer<Status, Throwable> errorsHandler = null;
 
+        /**
+         * Sets an ID of a reader to be used for logging, tracing, etc. If is {@code null}, an autogenerated value will
+         * be used.
+         *
+         * @param debugId the ID
+         *
+         * @return this builder
+         */
+        public Builder setDebugId(String debugId) {
+            this.debugId = debugId;
+            return this;
+        }
+
         public Builder setConsumerName(String consumerName) {
             this.consumerName = consumerName;
             return this;

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

@@ -12,6 +12,7 @@ public class WriterSettings {
     private static final long MAX_MEMORY_USAGE_BYTES_DEFAULT = 20 * 1024 * 1024; // 20 MB
     private static final int MAX_IN_FLIGHT_COUNT_DEFAULT = 100000;
 
+    private final String debugId;
     private final String topicPath;
     private final String producerId;
     private final String messageGroupId;
@@ -22,6 +23,7 @@ public class WriterSettings {
     private final BiConsumer<Status, Throwable> errorsHandler;
 
     private WriterSettings(Builder builder) {
+        this.debugId = builder.debugId;
         this.topicPath = builder.topicPath;
         this.producerId = builder.producerId;
         this.messageGroupId = builder.messageGroupId;
@@ -36,6 +38,10 @@ public static Builder newBuilder() {
         return new Builder();
     }
 
+    public String getDebugId() {
+        return debugId;
+    }
+
     public String getTopicPath() {
         return topicPath;
     }
@@ -72,6 +78,7 @@ public int getMaxSendBufferMessagesCount() {
      * BUILDER
      */
     public static class Builder {
+        private String debugId = null;
         private String topicPath = null;
         private String producerId = null;
         private String messageGroupId = null;
@@ -81,6 +88,19 @@ public static class Builder {
         private int maxSendBufferMessagesCount = MAX_IN_FLIGHT_COUNT_DEFAULT;
         private BiConsumer<Status, Throwable> errorsHandler = null;
 
+        /**
+         * Sets an ID of a writer to be used for logging, tracing, etc. If is {@code null}, an autogenerated value will
+         * be used.
+         *
+         * @param debugId the ID
+         *
+         * @return this builder
+         */
+        public Builder setDebugId(String debugId) {
+            this.debugId = debugId;
+            return this;
+        }
+
         /**
          * Set path to a topic to write to
          * @param topicPath  path to a topic

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(), settings.getErrorsHandler());
+        super(settings.getDebugId(), topicRpc.getScheduler(), settings.getErrorsHandler());
         this.topicRpc = topicRpc;
         this.settings = settings;
         this.session = new WriteSessionImpl();