Merge pull request #357 from alex268/master

Unite trace logs between transport and topics

Author: alex268

core/src/main/java/tech/ydb/core/impl/call/ReadWriteStreamCall.java

@@ -88,6 +88,10 @@ public CompletableFuture<Status> start(Observer<R> observer) {
     public void sendNext(W message) {
         synchronized (call) {
             if (flush()) {
+                if (logger.isTraceEnabled()) {
+                    String msg = TextFormat.shortDebugString((Message) message);
+                    logger.trace("ReadWriteStreamCall[{}] --> {}", traceId, msg);
+                }
                 call.sendMessage(message);
             } else {
                 messagesQueue.add(message);
@@ -103,7 +107,8 @@ private boolean flush() {
             }
 
             if (logger.isTraceEnabled()) {
-                logger.trace("ReadWriteStreamCall[{}] --> {}", traceId, TextFormat.shortDebugString((Message) next));
+                String msg = TextFormat.shortDebugString((Message) next);
+                logger.trace("ReadWriteStreamCall[{}] --> {}", traceId, msg);
             }
             call.sendMessage(next);
         }

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

@@ -75,9 +75,13 @@ CompletableFuture<Result<YdbTopic.DescribeConsumerResult>> describeConsumer(
     CompletableFuture<Status> updateOffsetsInTransaction(YdbTopic.UpdateOffsetsInTransactionRequest request,
                                                          GrpcRequestSettings settings);
 
-    GrpcReadWriteStream<YdbTopic.StreamWriteMessage.FromServer, YdbTopic.StreamWriteMessage.FromClient> writeSession();
+    GrpcReadWriteStream<YdbTopic.StreamWriteMessage.FromServer, YdbTopic.StreamWriteMessage.FromClient> writeSession(
+            String traceId
+    );
 
-    GrpcReadWriteStream<YdbTopic.StreamReadMessage.FromServer, YdbTopic.StreamReadMessage.FromClient> readSession();
+    GrpcReadWriteStream<YdbTopic.StreamReadMessage.FromServer, YdbTopic.StreamReadMessage.FromClient> readSession(
+            String traceId
+    );
 
     ScheduledExecutorService getScheduler();
 }

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

@@ -91,21 +91,21 @@ public CompletableFuture<Status> updateOffsetsInTransaction(YdbTopic.UpdateOffse
     }
 
     @Override
-    public GrpcReadWriteStream<
-        YdbTopic.StreamWriteMessage.FromServer,
-        YdbTopic.StreamWriteMessage.FromClient
-        > writeSession() {
-        return transport.readWriteStreamCall(TopicServiceGrpc.getStreamWriteMethod(),
-                GrpcRequestSettings.newBuilder().build());
+    public GrpcReadWriteStream<YdbTopic.StreamWriteMessage.FromServer, YdbTopic.StreamWriteMessage.FromClient>
+            writeSession(String streamId) {
+        return transport.readWriteStreamCall(
+                TopicServiceGrpc.getStreamWriteMethod(),
+                GrpcRequestSettings.newBuilder().withTraceId(streamId).build()
+        );
     }
 
     @Override
-    public GrpcReadWriteStream<
-            YdbTopic.StreamReadMessage.FromServer,
-            YdbTopic.StreamReadMessage.FromClient
-            > readSession() {
-        return transport.readWriteStreamCall(TopicServiceGrpc.getStreamReadMethod(),
-                GrpcRequestSettings.newBuilder().build());
+    public GrpcReadWriteStream<YdbTopic.StreamReadMessage.FromServer, YdbTopic.StreamReadMessage.FromClient>
+            readSession(String streamId) {
+        return transport.readWriteStreamCall(
+                TopicServiceGrpc.getStreamReadMethod(),
+                GrpcRequestSettings.newBuilder().withTraceId(streamId).build()
+        );
     }
 
     @Override

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

@@ -15,8 +15,11 @@
 public abstract class ReadSession extends SessionBase<FromServer, FromClient> {
     private static final Logger logger = LoggerFactory.getLogger(ReadSession.class);
 
-    public ReadSession(TopicRpc rpc) {
-        super(rpc.readSession());
+    protected final String streamId;
+
+    public ReadSession(TopicRpc rpc, String streamId) {
+        super(rpc.readSession(streamId));
+        this.streamId = streamId;
     }
 
     @Override