support statistics reporting for topic writes

Author: zinal

topic/src/main/java/tech/ydb/topic/write/WriteAck.java

@@ -1,17 +1,23 @@
 package tech.ydb.topic.write;
 
+import java.time.Duration;
+
+import tech.ydb.proto.topic.YdbTopic;
+
 /**
  * @author Nikolay Perfilov
  */
 public class WriteAck {
     private final long seqNo;
     private final State state;
     private final Details details;
+    private final Statistics statistics;
 
-    public WriteAck(long seqNo, State state, Details details) {
+    public WriteAck(long seqNo, State state, Details details, Statistics statistics) {
         this.seqNo = seqNo;
         this.state = state;
         this.details = details;
+        this.statistics = statistics;
     }
 
     public enum State {
@@ -20,18 +26,6 @@ public enum State {
         WRITTEN_IN_TX
     }
 
-    public static class Details {
-        private final long offset;
-
-        public Details(long offset) {
-            this.offset = offset;
-        }
-
-        public long getOffset() {
-            return offset;
-        }
-    }
-
     public long getSeqNo() {
         return seqNo;
     }
@@ -47,4 +41,68 @@ public State getState() {
     public Details getDetails() {
         return details;
     }
+
+    /**
+     * Obtain message write statistics
+     * @return {@link Statistics} with timings if statistics are available or null otherwise
+     */
+    public Statistics getStatistics() {
+        return statistics;
+    }
+
+    private static Duration convert(com.google.protobuf.Duration d) {
+        if (d == null) {
+            return Duration.ZERO;
+        }
+        return Duration.ofSeconds(d.getSeconds(), d.getNanos());
+    }
+
+    public static class Details {
+        private final long offset;
+
+        public Details(long offset) {
+            this.offset = offset;
+        }
+
+        public long getOffset() {
+            return offset;
+        }
+    }
+
+    public static class Statistics {
+        private final Duration persistingTime;
+        private final Duration partitionQuotaWaitTime;
+        private final Duration topicQuotaWaitTime;
+        private final Duration maxQueueWaitTime;
+        private final Duration minQueueWaitTime;
+
+        public Statistics(YdbTopic.StreamWriteMessage.WriteResponse.WriteStatistics src) {
+            this.persistingTime          = convert(src.getPersistingTime());
+            this.partitionQuotaWaitTime  = convert(src.getPartitionQuotaWaitTime());
+            this.topicQuotaWaitTime      = convert(src.getTopicQuotaWaitTime());
+            this.maxQueueWaitTime        = convert(src.getMaxQueueWaitTime());
+            this.minQueueWaitTime        = convert(src.getMinQueueWaitTime());
+        }
+
+        public Duration getPersistingTime() {
+            return persistingTime;
+        }
+
+        public Duration getPartitionQuotaWaitTime() {
+            return partitionQuotaWaitTime;
+        }
+
+        public Duration getTopicQuotaWaitTime() {
+            return topicQuotaWaitTime;
+        }
+
+        public Duration getMaxQueueWaitTime() {
+            return maxQueueWaitTime;
+        }
+
+        public Duration getMinQueueWaitTime() {
+            return minQueueWaitTime;
+        }
+    }
+
 }

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

@@ -421,6 +421,10 @@ private void onInitResponse(YdbTopic.StreamWriteMessage.InitResponse response) {
         private void onWriteResponse(YdbTopic.StreamWriteMessage.WriteResponse response) {
             List<YdbTopic.StreamWriteMessage.WriteResponse.WriteAck> acks = response.getAcksList();
             logger.debug("[{}] Received WriteResponse with {} WriteAcks", streamId, acks.size());
+            WriteAck.Statistics statistics = null;
+            if (response.getWriteStatistics() != null) {
+                statistics = new WriteAck.Statistics(response.getWriteStatistics());
+            }
             int inFlightFreed = 0;
             long bytesFreed = 0;
             for (YdbTopic.StreamWriteMessage.WriteResponse.WriteAck ack : acks) {
@@ -433,7 +437,7 @@ private void onWriteResponse(YdbTopic.StreamWriteMessage.WriteResponse response)
                         inFlightFreed++;
                         bytesFreed += sentMessage.getSize();
                         sentMessages.remove();
-                        processWriteAck(sentMessage, ack);
+                        processWriteAck(sentMessage, statistics, ack);
                         break;
                     }
                     if (sentMessage.getSeqNo() < ack.getSeqNo()) {
@@ -474,20 +478,20 @@ private void processMessage(YdbTopic.StreamWriteMessage.FromServer message) {
             }
         }
 
-        private void processWriteAck(EnqueuedMessage message,
+        private void processWriteAck(EnqueuedMessage message, WriteAck.Statistics statistics,
                                      YdbTopic.StreamWriteMessage.WriteResponse.WriteAck ack) {
             logger.debug("[{}] Received WriteAck with seqNo {} and status {}", streamId, ack.getSeqNo(),
                     ack.getMessageWriteStatusCase());
             WriteAck resultAck;
             switch (ack.getMessageWriteStatusCase()) {
                 case WRITTEN:
                     WriteAck.Details details = new WriteAck.Details(ack.getWritten().getOffset());
-                    resultAck = new WriteAck(ack.getSeqNo(), WriteAck.State.WRITTEN, details);
+                    resultAck = new WriteAck(ack.getSeqNo(), WriteAck.State.WRITTEN, details, statistics);
                     break;
                 case SKIPPED:
                     switch (ack.getSkipped().getReason()) {
                         case REASON_ALREADY_WRITTEN:
-                            resultAck = new WriteAck(ack.getSeqNo(), WriteAck.State.ALREADY_WRITTEN, null);
+                            resultAck = new WriteAck(ack.getSeqNo(), WriteAck.State.ALREADY_WRITTEN, null, statistics);
                             break;
                         case REASON_UNSPECIFIED:
                         default:
@@ -497,7 +501,7 @@ private void processWriteAck(EnqueuedMessage message,
                     }
                     break;
                 case WRITTEN_IN_TX:
-                    resultAck = new WriteAck(ack.getSeqNo(), WriteAck.State.WRITTEN_IN_TX, null);
+                    resultAck = new WriteAck(ack.getSeqNo(), WriteAck.State.WRITTEN_IN_TX, null, statistics);
                     break;
                 default:
                     message.getFuture().completeExceptionally(
@@ -519,5 +523,6 @@ private void closeDueToError(Status status, Throwable th) {
         protected void onStop() {
             logger.debug("[{}] Session {} onStop called", streamId, sessionId);
         }
+
     }
 }