Merge pull request #359 from ydb-platform/logging-improvements

Logging improvements for topic reader & writer

Author: pnv1

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

@@ -1,6 +1,6 @@
 package tech.ydb.topic.impl;
 
-import java.util.UUID;
+import java.util.Random;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ScheduledExecutorService;
@@ -22,6 +22,9 @@ public abstract class GrpcStreamRetrier {
     private static final int EXP_BACKOFF_BASE_MS = 256;
     private static final int EXP_BACKOFF_CEILING_MS = 40000; // 40 sec (max delays would be 40-80 sec)
     private static final int EXP_BACKOFF_MAX_POWER = 7;
+    private static final int ID_LENGTH = 6;
+    private static final char[] ID_ALPHABET = "abcdefghijklmnopqrstuvwxyzABSDEFGHIJKLMNOPQRSTUVWXYZ1234567890"
+            .toCharArray();
 
     protected final String id;
     protected final AtomicBoolean isReconnecting = new AtomicBoolean(false);
@@ -31,14 +34,22 @@ public abstract class GrpcStreamRetrier {
 
     protected GrpcStreamRetrier(ScheduledExecutorService scheduler) {
         this.scheduler = scheduler;
-        this.id = UUID.randomUUID().toString();
+        this.id = generateRandomId(ID_LENGTH);
     }
 
     protected abstract Logger getLogger();
     protected abstract String getStreamName();
     protected abstract void onStreamReconnect();
     protected abstract void onShutdown(String reason);
 
+    protected static String generateRandomId(int length) {
+        return new Random().ints(0, ID_ALPHABET.length)
+                .limit(length)
+                .map(charId -> ID_ALPHABET[charId])
+                .collect(StringBuilder::new, StringBuilder::appendCodePoint, StringBuilder::append)
+                .toString();
+    }
+
     private void tryScheduleReconnect() {
         int currentReconnectCounter = reconnectCounter.get() + 1;
         if (MAX_RECONNECT_COUNT > 0 && currentReconnectCounter > MAX_RECONNECT_COUNT) {
@@ -49,8 +60,8 @@ private void tryScheduleReconnect() {
                 shutdownImpl(errorMessage);
                 return;
             } else {
-                getLogger().debug("[{}] Maximum retry count ({}}) exceeded. But {} is already shut down.", id,
-                        MAX_RECONNECT_COUNT, getStreamName());
+                getLogger().info("[{}] Maximum retry count ({}}) exceeded. Need to shutdown {} but it's already " +
+                                "shut down.", id, MAX_RECONNECT_COUNT, getStreamName());
             }
         }
         if (isReconnecting.compareAndSet(false, true)) {

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

@@ -29,10 +29,8 @@ public CompletableFuture<Void> commit() {
 
     public CompletableFuture<Void> commitImpl(boolean fromCommitter) {
         if (logger.isDebugEnabled()) {
-            logger.debug("[{}] partition session {} (partition {}): committing {} message(s), offsets" +
-                            " [{},{})" + (fromCommitter ? " from Committer" : ""), partitionSession.getPath(),
-                    partitionSession.getId(), partitionSession.getPartitionId(), messageCount,
-                    offsetsToCommit.getStart(), offsetsToCommit.getEnd());
+            logger.debug("[{}] Committing {} message(s), offsets [{},{})" + (fromCommitter ? " from Committer" : ""),
+                    partitionSession.getFullId(), messageCount, offsetsToCommit.getStart(), offsetsToCommit.getEnd());
         }
         return partitionSession.commitOffsetRange(offsetsToCommit);
     }

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

@@ -41,9 +41,9 @@ private void add(OffsetsRange offsetRange) {
                     rangesLock.unlock();
                 }
             } catch (RuntimeException exception) {
-                String errorMessage = "Error adding new offset range to DeferredCommitter for partition session " +
-                        partitionSession.getId() + " (partition " + partitionSession.getPartitionId() + "): " +
-                        exception.getMessage();
+                String errorMessage = "[" + partitionSession.getFullId() + "] Error adding new offset range to " +
+                        "DeferredCommitter for partition session " + partitionSession.getId() + " (partition " +
+                        partitionSession.getPartitionId() + "): " + exception.getMessage();
                 logger.error(errorMessage);
                 throw new RuntimeException(errorMessage, exception);
             }