Add topic and consumer names for each write / read session start

Author: pnv1

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

@@ -39,7 +39,8 @@ public class PartitionSessionImpl {
 
     private final long id;
     private final String fullId;
-    private final String path;
+    private final String topicPath;
+    private final String consumerName;
     private final long partitionId;
     private final PartitionSession sessionInfo;
     private final Executor decompressionExecutor;
@@ -60,17 +61,18 @@ public class PartitionSessionImpl {
     private PartitionSessionImpl(Builder builder) {
         this.id = builder.id;
         this.fullId = builder.fullId;
-        this.path = builder.path;
+        this.topicPath = builder.topicPath;
+        this.consumerName = builder.consumerName;
         this.partitionId = builder.partitionId;
-        this.sessionInfo = new PartitionSession(id, partitionId, path);
+        this.sessionInfo = new PartitionSession(id, partitionId, topicPath);
         this.lastReadOffset = builder.committedOffset;
         this.lastCommittedOffset = builder.committedOffset;
         this.decompressionExecutor = builder.decompressionExecutor;
         this.dataEventCallback = builder.dataEventCallback;
         this.commitFunction = builder.commitFunction;
-        logger.info("[{}] Partition session for {} is started. CommittedOffset: {}. " +
-                "Partition offsets: {}-{}", fullId, path, lastReadOffset, builder.partitionOffsets.getStart(),
-                builder.partitionOffsets.getEnd());
+        logger.info("[{}] Partition session is started for Topic \"{}\" and Consumer \"{}\". CommittedOffset: {}. " +
+                "Partition offsets: {}-{}", fullId, topicPath, consumerName, lastReadOffset,
+                builder.partitionOffsets.getStart(), builder.partitionOffsets.getEnd());
     }
 
     public static Builder newBuilder() {
@@ -89,8 +91,8 @@ public long getPartitionId() {
         return partitionId;
     }
 
-    public String getPath() {
-        return path;
+    public String getTopicPath() {
+        return topicPath;
     }
 
     public PartitionSession getSessionInfo() {
@@ -216,7 +218,7 @@ public CompletableFuture<Void> commitOffsetRange(OffsetsRange rangeToCommit) {
                 logger.info("[{}] Offset range [{}, {}) is requested to be committed, but partition session " +
                         "is already closed", fullId, rangeToCommit.getStart(), rangeToCommit.getEnd());
                 resultFuture.completeExceptionally(new RuntimeException("Partition session " + id + " (partition " +
-                        partitionId + ") for " + path + " is already closed"));
+                        partitionId + ") for " + topicPath + " is already closed"));
                 return resultFuture;
             }
         } finally {
@@ -315,7 +317,7 @@ private void sendDataToReadersIfNeeded() {
             DataReceivedEvent event = new DataReceivedEventImpl(this, messagesToRead, offsetsToCommit);
             if (logger.isDebugEnabled()) {
                 logger.debug("[{}] DataReceivedEvent callback with {} message(s) (offsets {}-{}) is about " +
-                                "to be called...", fullId, messagesToRead.size(),messagesToRead.get(0).getOffset(),
+                                "to be called...", fullId, messagesToRead.size(), messagesToRead.get(0).getOffset(),
                         messagesToRead.get(messagesToRead.size() - 1).getOffset());
             }
             dataEventCallback.apply(event)
@@ -348,9 +350,9 @@ public void shutdown() {
         try {
             isWorking.set(false);
             logger.info("[{}] Partition session for {} is shutting down. Failing {} commit futures...", fullId,
-                    path, commitFutures.size());
+                    topicPath, commitFutures.size());
             commitFutures.values().forEach(f -> f.completeExceptionally(new RuntimeException("Partition session " + id +
-                    " (partition " + partitionId + ") for " + path + " is closed")));
+                    " (partition " + partitionId + ") for " + topicPath + " is closed")));
         } finally {
             commitFuturesLock.unlock();
         }
@@ -371,7 +373,8 @@ public void shutdown() {
     public static class Builder {
         private long id;
         private String fullId;
-        private String path;
+        private String topicPath;
+        private String consumerName;
         private long partitionId;
         private long committedOffset;
         private OffsetsRange partitionOffsets;
@@ -389,8 +392,13 @@ public Builder setFullId(String fullId) {
             return this;
         }
 
-        public Builder setPath(String path) {
-            this.path = path;
+        public Builder setTopicPath(String topicPath) {
+            this.topicPath = topicPath;
+            return this;
+        }
+
+        public Builder setConsumerName(String consumerName) {
+            this.consumerName = consumerName;
             return this;
         }
 

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

@@ -71,15 +71,15 @@ public ReaderImpl(TopicRpc topicRpc, ReaderSettings settings) {
             message.append(" \"").append(settings.getReaderName()).append("\"");
         }
         message.append(" (generated id ").append(id).append(")");
-        message.append(" created for topic(s): ");
+        message.append(" created for Topic(s) ");
         for (TopicReadSettings topic : settings.getTopics()) {
             if (topic != settings.getTopics().get(0)) {
                 message.append(", ");
             }
             message.append("\"").append(topic.getPath()).append("\"");
         }
         if (settings.getConsumerName() != null) {
-            message.append(" and Consumer: \"").append(settings.getConsumerName()).append("\"");
+            message.append(" and Consumer \"").append(settings.getConsumerName());
             consumerName = settings.getConsumerName();
         } else {
             message.append(" without a Consumer");
@@ -392,7 +392,7 @@ private void onStartPartitionSessionRequest(YdbTopic.StreamReadMessage.StartPart
             long partitionSessionId = request.getPartitionSession().getPartitionSessionId();
             long partitionId = request.getPartitionSession().getPartitionId();
             String partitionSessionFullId = streamId + '/' + partitionSessionId + "-p" + partitionId;
-            logger.info("[{}] Received StartPartitionSessionRequest for Topic {} and Consumer {}. " +
+            logger.info("[{}] Received StartPartitionSessionRequest for Topic \"{}\" and Consumer \"{}\". " +
                             "Partition session {} (partition {}) with committedOffset {} and partitionOffsets [{}-{})",
                     partitionSessionFullId, request.getPartitionSession().getPath(), consumerName,
                     partitionSessionId, partitionId, request.getCommittedOffset(),
@@ -401,7 +401,7 @@ private void onStartPartitionSessionRequest(YdbTopic.StreamReadMessage.StartPart
             PartitionSessionImpl partitionSession = PartitionSessionImpl.newBuilder()
                     .setId(partitionSessionId)
                     .setFullId(partitionSessionFullId)
-                    .setPath(request.getPartitionSession().getPath())
+                    .setTopicPath(request.getPartitionSession().getPath())
                     .setPartitionId(partitionId)
                     .setCommittedOffset(request.getCommittedOffset())
                     .setPartitionOffsets(new OffsetsRangeImpl(request.getPartitionOffsets().getStart(),

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

@@ -413,8 +413,9 @@ private void onInitResponse(YdbTopic.StreamWriteMessage.InitResponse response) {
             sessionId = response.getSessionId();
             long lastSeqNo = response.getLastSeqNo();
             long actualLastSeqNo = lastSeqNo;
-            logger.info("[{}] Session initialized with id {} (partition {}), lastSeqNo {}, actualLastSeqNo {}",
-                    streamId, sessionId, response.getPartitionId(), lastSeqNo, actualLastSeqNo);
+            logger.info("[{}] Session with id {} (partition {}) initialized for topic \"{}\", lastSeqNo {}," +
+                            " actualLastSeqNo {}", streamId, sessionId, response.getPartitionId(),
+                    settings.getTopicPath(), lastSeqNo, actualLastSeqNo);
             // If there are messages that were already sent before reconnect but haven't received acks,
             // their highest seqNo should also be taken in consideration when calculating next seqNo automatically
             if (!sentMessages.isEmpty()) {