Add Topic Reader metrics

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

@@ -14,12 +14,26 @@ public class BatchMeta {
     private final Map<String, String> writeSessionMeta;
     private final int codec;
     private final Instant writtenAt;
+    private final long receivedAt;
 
     public BatchMeta(YdbTopic.StreamReadMessage.ReadResponse.Batch batch) {
+        this(batch, 0);
+    }
+
+    BatchMeta(YdbTopic.StreamReadMessage.ReadResponse.Batch batch, long receivedAt) {
         this.producerId = batch.getProducerId();
         this.writeSessionMeta = batch.getWriteSessionMetaMap();
         this.codec = batch.getCodec();
         this.writtenAt = ProtobufUtils.protoToInstant(batch.getWrittenAt());
+        this.receivedAt = receivedAt;
+    }
+
+    double age(long now) {
+        return receivedAt == 0 ? 0 : Math.max(0, now - receivedAt) / 1_000_000_000.0;
+    }
+
+    long getReceivedAt() {
+        return receivedAt;
     }
 
     public String getProducerId() {

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

@@ -48,6 +48,14 @@ public MessageImpl(PartitionSession session, MessageCommitter committer, BatchMe
 
     public abstract boolean isReady();
 
+    double getLocalBufferMessageAge(long now) {
+        return batchMeta.age(now);
+    }
+
+    long getReceivedAt() {
+        return batchMeta.getReceivedAt();
+    }
+
     @Override
     public long getOffset() {
         return offset;

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

@@ -1,6 +1,8 @@
 package tech.ydb.topic.read.impl;
 
 import java.util.concurrent.Executor;
+import java.util.function.DoubleSupplier;
+import java.util.function.LongSupplier;
 
 import tech.ydb.topic.description.CodecRegistry;
 import tech.ydb.topic.settings.ReaderSettings;
@@ -18,9 +20,21 @@ public class ReadConfig {
     private final long maxMemoryUsageBytes;
     private final int maxBatchSize;
     private final ReaderMetrics metrics;
+    final LongSupplier readyMessages;
+    final DoubleSupplier readyMessageAgeMax;
 
     public ReadConfig(CodecRegistry codecRegistry, Executor control, Executor processor, Executor decompressor,
             ReaderSettings settings) {
+        this(codecRegistry, control, processor, decompressor, settings, () -> 0);
+    }
+
+    ReadConfig(CodecRegistry codecRegistry, Executor control, Executor processor, Executor decompressor,
+            ReaderSettings settings, LongSupplier readyMessages) {
+        this(codecRegistry, control, processor, decompressor, settings, readyMessages, () -> 0);
+    }
+
+    public ReadConfig(CodecRegistry codecRegistry, Executor control, Executor processor, Executor decompressor,
+            ReaderSettings settings, LongSupplier readyMessages, DoubleSupplier readyMessageAgeMax) {
         this.codecRegistry = codecRegistry;
         this.control = control;
         this.processor = processor;
@@ -29,6 +43,8 @@ public ReadConfig(CodecRegistry codecRegistry, Executor control, Executor proces
         this.maxMemoryUsageBytes = settings.getMaxMemoryUsageBytes();
         this.maxBatchSize = settings.getMaxBatchSize();
         this.metrics = new ReaderMetrics(settings.getMeter(), settings.getConsumerName(), settings.getReaderName());
+        this.readyMessages = readyMessages;
+        this.readyMessageAgeMax = readyMessageAgeMax;
     }
 
     public CodecRegistry getCodecRegistry() {

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

@@ -90,6 +90,10 @@ public PartitionSession getPartition() {
         return partition;
     }
 
+    public long getCommitOffsetLag() {
+        return committer.getCommitOffsetLag();
+    }
+
     @Override
     public void confirmRangeProcessed(OffsetsRange range) {
         decoder.releaseRange(range);
@@ -104,6 +108,14 @@ public void confirmPartitionEnded(List<TopicPartition> childs) {
         sendDataToReaders();
     }
 
+    double getLocalBufferMessageAgeMax() {
+        return queue.getLocalBufferMessageAgeMax();
+    }
+
+    long getLocalBufferMessages() {
+        return queue.getLocalBufferMessages();
+    }
+
     public void confirmCommittedOffset(long committedOffset) {
         committer.updateCommittedOffset(committedOffset);
         committer.completePendingCommits();
@@ -119,7 +131,7 @@ public void unpause() {
         sendDataToReaders();
     }
 
-    public boolean addBatches(List<YdbTopic.StreamReadMessage.ReadResponse.Batch> batchList) {
+    public boolean addBatches(List<YdbTopic.StreamReadMessage.ReadResponse.Batch> batchList, long receivedAt) {
         if (!isActive()) {
             return false;
         }
@@ -130,7 +142,7 @@ public boolean addBatches(List<YdbTopic.StreamReadMessage.ReadResponse.Batch> ba
         metrics.reportReceivedMessages(messagesCount, partition.getPath());
 
         if (messagesCount > 0) {
-            queue.addBatches(batchList);
+            queue.addBatches(batchList, receivedAt);
             sendDataToReaders();
         }
 

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

@@ -44,6 +44,15 @@ private RuntimeException partitionIsClosedException() {
         return new RuntimeException("" + partition + " is already stopped");
     }
 
+    public long getCommitOffsetLag() {
+        commitFuturesLock.lock();
+        try {
+            return commitFutures.isEmpty() ? 0 : commitFutures.lastKey() - lastCommittedOffset.get();
+        } finally {
+            commitFuturesLock.unlock();
+        }
+    }
+
     public void updateCommittedOffset(long offset) {
         long old = lastCommittedOffset.get();
         if (old != lastCommittedOffset.accumulateAndGet(offset, Math::max)) {
@@ -85,6 +94,9 @@ public CompletableFuture<Void> commit(OffsetsRange range) {
                 future = new CompletableFuture<>();
                 commitFutures.put(range.getEnd(), future);
             }
+            long messages = range.getEnd() - range.getStart();
+            future.thenRun(() -> session.getConfig().getMetrics()
+                    .reportCommitAcknowledged(messages, partition.getPath()));
         } finally {
             commitFuturesLock.unlock();
         }
@@ -107,7 +119,20 @@ public CompletableFuture<Void> commit(OffsetsRange range) {
 
     @Override
     public void commitRanges(List<OffsetsRange> ranges) {
-        session.commitOffsets(partition, ranges);
+        commitFuturesLock.lock();
+        try {
+            ranges.forEach(range -> {
+                long messages = range.getEnd() - range.getStart();
+                commitFutures.computeIfAbsent(range.getEnd(), offset -> new CompletableFuture<>())
+                        .thenRun(() -> session.getConfig().getMetrics()
+                                .reportCommitAcknowledged(messages, partition.getPath()));
+            });
+        } finally {
+            commitFuturesLock.unlock();
+        }
+        if (!session.commitOffsets(partition, ranges)) {
+            close();
+        }
     }
 
     public void close() {

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

@@ -36,14 +36,24 @@ void updateLastReadOffset(long offset) {
         lastReadOffset = offset;
     }
 
-    void addBatches(List<YdbTopic.StreamReadMessage.ReadResponse.Batch> batchList) {
+    long getLocalBufferMessages() {
+        return readingQueue.size();
+    }
+
+    double getLocalBufferMessageAgeMax() {
+        long receivedAt = readingQueue.stream().mapToLong(MessageImpl::getReceivedAt)
+                .filter(timestamp -> timestamp != 0).min().orElse(0);
+        return receivedAt == 0 ? 0 : Math.max(0, System.nanoTime() - receivedAt) / 1_000_000_000.0;
+    }
+
+    void addBatches(List<YdbTopic.StreamReadMessage.ReadResponse.Batch> batchList, long receivedAt) {
         for (YdbTopic.StreamReadMessage.ReadResponse.Batch batch : batchList) {
             if (batch.getMessageDataCount() == 0) {
                 logger.error("[{}] Received empty batch. This shouldn't happen", traceID);
                 continue;
             }
 
-            BatchMeta meta = new BatchMeta(batch);
+            BatchMeta meta = new BatchMeta(batch, receivedAt);
             List<MessageImpl> messages = new ArrayList<>(batch.getMessageDataCount());
             for (YdbTopic.StreamReadMessage.ReadResponse.MessageData msg : batch.getMessageDataList()) {
                 if (lastReadOffset > msg.getOffset()) {

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

@@ -128,6 +128,7 @@ public boolean commitOffsets(PartitionSession session, List<OffsetsRange> ranges
                         .build())
                 .build();
 
+        config.getMetrics().reportCommitQueued(rangesToCommit, session.getPath());
         send(FromClient.newBuilder().setCommitOffsetRequest(req).build());
         return true;
     }
@@ -189,7 +190,13 @@ public void removeLocks(PartitionSession ps, List<TopicPartition> childs) {
 
     public void onInit(YdbTopic.StreamReadMessage.InitResponse response) {
         bufferManager.init(response.getSessionId());
-        config.getMetrics().register(partitions::size, bufferManager::getCreditBalanceBytes);
+        config.getMetrics().register(partitions::size, bufferManager::getCreditBalanceBytes,
+                () -> partitions.values().stream().mapToLong(ReadPartition::getCommitOffsetLag).max().orElse(0),
+                () -> config.readyMessages.getAsLong() + partitions.values().stream()
+                        .filter(ReadPartition::isActive).mapToLong(ReadPartition::getLocalBufferMessages).sum(),
+                () -> Math.max(config.readyMessageAgeMax.getAsDouble(), partitions.values().stream()
+                        .filter(ReadPartition::isActive)
+                        .mapToDouble(ReadPartition::getLocalBufferMessageAgeMax).max().orElse(0)));
     }
 
     public void onStartPartition(YdbTopic.StreamReadMessage.StartPartitionSessionRequest req) {
@@ -271,14 +278,15 @@ public void onStopPartition(YdbTopic.StreamReadMessage.StopPartitionSessionReque
     }
 
     public void onRead(YdbTopic.StreamReadMessage.ReadResponse response) {
+        long receivedAt = config.getMetrics().reportReadResponseStart();
         logger.debug("[{}] Received ReadResponse of {} bytes", debugId, response.getBytesSize());
         config.getMetrics().reportReceivedBytes(response.getBytesSize());
         bufferManager.allocate(response.getBytesSize(), response.getPartitionDataList());
 
         for (YdbTopic.StreamReadMessage.ReadResponse.PartitionData data: response.getPartitionDataList()) {
             long psid = data.getPartitionSessionId();
             ReadPartition partition = partitions.get(psid);
-            if (partition == null || !partition.addBatches(data.getBatchesList())) {
+            if (partition == null || !partition.addBatches(data.getBatchesList(), receivedAt)) {
                 logger.warn("[{}] Received PartitionData for unknown(most likely already closed) PartitionSessionId={}",
                         debugId, psid);
                 bufferManager.releasePartition(psid);

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

@@ -1,6 +1,8 @@
 package tech.ydb.topic.read.impl;
 
 import java.util.Arrays;
+import java.util.List;
+import java.util.function.DoubleSupplier;
 import java.util.function.LongSupplier;
 
 import tech.ydb.core.Status;
@@ -9,6 +11,7 @@
 import tech.ydb.core.metrics.LongCounter;
 import tech.ydb.core.metrics.Meter;
 import tech.ydb.core.metrics.MetricRegistration;
+import tech.ydb.topic.description.OffsetsRange;
 
 /**
  * Topic reader metrics.
@@ -19,12 +22,17 @@ final class ReaderMetrics {
     private final LongCounter deliveredMessages;
     private final LongCounter receivedMessages;
     private final LongCounter receivedBytes;
+    private final LongCounter commitQueued;
+    private final LongCounter commitAcknowledged;
     private final LongCounter sessionErrors;
     private final Attr[] commonAttributes;
     private final boolean enabled;
     private final Meter meter;
     private MetricRegistration partitionsGauge = MetricRegistration.NOOP;
     private MetricRegistration creditGauge = MetricRegistration.NOOP;
+    private MetricRegistration commitLagGauge = MetricRegistration.NOOP;
+    private MetricRegistration bufferMessagesGauge = MetricRegistration.NOOP;
+    private MetricRegistration ageGauge = MetricRegistration.NOOP;
 
     ReaderMetrics(Meter meter, String consumer, String readerName) {
         this.meter = meter;
@@ -37,26 +45,50 @@ final class ReaderMetrics {
                 "The number of messages accepted by the SDK for an active partition session.");
         this.receivedBytes = meter.createCounter("ydb.topic.reader.received.bytes", "By",
                 "The protocol bytes_size received in read responses.");
+        this.commitQueued = meter.createCounter("ydb.topic.reader.commit.queued", MESSAGE_UNIT,
+                "The number of messages in commit ranges accepted by the SDK.");
+        this.commitAcknowledged = meter.createCounter("ydb.topic.reader.commit.acknowledged", MESSAGE_UNIT,
+                "The number of messages in commit ranges completed by successful acknowledgements.");
         this.sessionErrors = meter.createCounter("ydb.topic.reader.session.errors", "{error}",
                 "The number of reader stream session errors by retry decision.");
         this.commonAttributes = createCommonAttributes(consumer, readerName);
     }
 
-    void register(LongSupplier partitionCount, LongSupplier bufferBudget) {
+    void register(LongSupplier partitionCount, LongSupplier bufferBudget, LongSupplier commitLag,
+            LongSupplier bufferMessages, DoubleSupplier messageAgeMax) {
         partitionsGauge = meter.registerLongGauge(
                 "ydb.topic.reader.partition_session.count", "{session}",
                 "The number of partition sessions currently in the reader session processing lifecycle.",
                 m -> m.record(partitionCount.getAsLong(), commonAttributes));
         creditGauge = meter.registerLongGauge("ydb.topic.reader.credit_balance_bytes", "By",
                 "The protocol credit granted to the server and not yet consumed by read responses.",
                 m -> m.record(bufferBudget.getAsLong(), commonAttributes));
+        commitLagGauge = meter.registerLongGauge("ydb.topic.reader.commit_offset.lag.max", MESSAGE_UNIT,
+                "The maximum gap between requested and acknowledged commit offsets.",
+                m -> m.record(commitLag.getAsLong(), commonAttributes));
+        bufferMessagesGauge = meter.registerLongGauge("ydb.topic.reader.local_buffer.messages", MESSAGE_UNIT,
+                "The number of messages currently buffered by the reader.",
+                m -> m.record(bufferMessages.getAsLong(), commonAttributes));
+        ageGauge = meter.registerDoubleGauge("ydb.topic.reader.local_buffer.message_age.max", "s",
+                "The age of the oldest batch retained in the reader's local buffer.",
+                m -> m.record(messageAgeMax.getAsDouble(), commonAttributes));
     }
 
     void unregister() {
         partitionsGauge.close();
         creditGauge.close();
+        commitLagGauge.close();
+        bufferMessagesGauge.close();
+        ageGauge.close();
         partitionsGauge = MetricRegistration.NOOP;
         creditGauge = MetricRegistration.NOOP;
+        commitLagGauge = MetricRegistration.NOOP;
+        bufferMessagesGauge = MetricRegistration.NOOP;
+        ageGauge = MetricRegistration.NOOP;
+    }
+
+    long reportReadResponseStart() {
+        return enabled ? System.nanoTime() : 0;
     }
 
     void reportDelivered(long messages, String topic) {
@@ -85,6 +117,20 @@ void reportSessionError(Status status, boolean retry) {
         sessionErrors.add(1, attributes);
     }
 
+    void reportCommitQueued(List<OffsetsRange> ranges, String topic) {
+        if (enabled) {
+            long messages = 0;
+            for (OffsetsRange range : ranges) {
+                messages += range.getEnd() - range.getStart();
+            }
+            report(commitQueued, messages, topic);
+        }
+    }
+
+    void reportCommitAcknowledged(long messages, String topic) {
+        report(commitAcknowledged, messages, topic);
+    }
+
     private void report(LongCounter counter, long messages, String topic) {
         if (enabled && messages > 0) {
             record(counter, messages, Attr.of("topic", topic));

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

@@ -65,7 +65,13 @@ public SyncReaderImpl(TopicRpc topicRpc, ReaderSettings settings, @Nonnull Codec
         this.debugId = DebugTools.createDebugId(settings.getLogPrefix());
         this.decompressor = new LazyExecutor("reader[" + debugId + "]-decoder", settings.getDecompressionExecutor());
 
-        this.config = new ReadConfig(codecRegistry, Runnable::run, Runnable::run, decompressor, settings);
+        this.config = new ReadConfig(codecRegistry, Runnable::run, Runnable::run, decompressor, settings,
+                queue::size, () -> {
+                    long receivedAt = queue.stream().filter(MessageWrapper::isActive)
+                            .mapToLong(message -> ((MessageImpl) message.msg).getReceivedAt())
+                            .filter(timestamp -> timestamp != 0).min().orElse(0);
+                    return receivedAt == 0 ? 0 : Math.max(0, System.nanoTime() - receivedAt) / 1_000_000_000.0;
+                });
         this.impl = new Impl(topicRpc, debugId, settings, config);
 
         String readerName = settings.getReaderName();