Add Topic Reader delivered metric

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

@@ -8,6 +8,7 @@
 import tech.ydb.core.Result;
 import tech.ydb.core.Status;
 import tech.ydb.core.grpc.GrpcTransport;
+import tech.ydb.core.metrics.Meter;
 import tech.ydb.topic.description.Codec;
 import tech.ydb.topic.description.ConsumerDescription;
 import tech.ydb.topic.description.TopicDescription;
@@ -201,6 +202,14 @@ interface Builder {
          */
         Builder setCompressionPoolThreadCount(Integer compressionPoolThreadCount);
 
+        /**
+         * Enable client metrics using the supplied meter.
+         *
+         * @param meter meter used to create Topic client instruments
+         * @return settings builder
+         */
+        Builder withMeter(Meter meter);
+
         /**
          * Register a custom codec used to compress and decompress topic messages.
          * A codec is identified by its {@link Codec#getId()}; registering a codec with the id of an already

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

@@ -4,6 +4,7 @@
 import java.util.List;
 import java.util.concurrent.Executor;
 
+import tech.ydb.core.metrics.Meter;
 import tech.ydb.topic.TopicClient;
 import tech.ydb.topic.TopicRpc;
 import tech.ydb.topic.description.Codec;
@@ -18,6 +19,7 @@ public class TopicClientBuilderImpl implements TopicClient.Builder {
     protected final List<Codec> codecs = new ArrayList<>(StandardCodecs.getAvailableCodecs());
     protected Integer compressionExecutorThreadCount;
     protected Executor compressionExecutor;
+    protected Meter meter = Meter.NOOP;
 
     public TopicClientBuilderImpl(TopicRpc topicRpc) {
         this.topicRpc = topicRpc;
@@ -35,6 +37,15 @@ public TopicClientBuilderImpl setCompressionExecutor(Executor compressionExecuto
         return this;
     }
 
+    @Override
+    public TopicClientBuilderImpl withMeter(Meter meter) {
+        if (meter == null) {
+            throw new IllegalArgumentException("Meter must be not null");
+        }
+        this.meter = meter;
+        return this;
+    }
+
     @Override
     public TopicClientBuilderImpl registerCodec(Codec codec) {
         if (codec == null) {

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

@@ -16,6 +16,8 @@
 import tech.ydb.core.Result;
 import tech.ydb.core.Status;
 import tech.ydb.core.grpc.GrpcRequestSettings;
+import tech.ydb.core.impl.Observability;
+import tech.ydb.core.metrics.Meter;
 import tech.ydb.core.operation.Operation;
 import tech.ydb.core.utils.ProtobufUtils;
 import tech.ydb.proto.topic.YdbTopic;
@@ -66,10 +68,13 @@ public class TopicClientImpl implements TopicClient {
     private final Executor compressionExecutor;
     private final ExecutorService defaultCompressionExecutorService;
     private final CodecRegistry codecRegistry;
+    private final Meter meter;
 
     TopicClientImpl(TopicClientBuilderImpl builder) {
         this.topicRpc = builder.topicRpc;
         this.codecRegistry = new CodecRegistry(builder.codecs);
+        this.meter = builder.meter;
+        Observability.reportMetricsUsage(meter);
         if (builder.compressionExecutor != null) {
             this.defaultCompressionExecutorService = null;
             this.compressionExecutor = builder.compressionExecutor;
@@ -372,12 +377,12 @@ private TopicDescription mapDescribeTopic(YdbTopic.DescribeTopicResult result, b
 
     @Override
     public SyncReader createSyncReader(ReaderSettings settings) {
-        return new SyncReaderImpl(topicRpc, settings, codecRegistry);
+        return new SyncReaderImpl(topicRpc, settings, codecRegistry, meter);
     }
 
     @Override
     public AsyncReader createAsyncReader(ReaderSettings settings, ReadEventHandlersSettings handlersSettings) {
-        return new AsyncReaderImpl(topicRpc, settings, handlersSettings, codecRegistry);
+        return new AsyncReaderImpl(topicRpc, settings, handlersSettings, codecRegistry, meter);
     }
 
     @Override

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

@@ -14,6 +14,7 @@
 
 import tech.ydb.common.transaction.YdbTransaction;
 import tech.ydb.core.Status;
+import tech.ydb.core.metrics.Meter;
 import tech.ydb.topic.TopicRpc;
 import tech.ydb.topic.description.CodecRegistry;
 import tech.ydb.topic.impl.SerialExecutor;
@@ -48,7 +49,15 @@ public AsyncReaderImpl(TopicRpc topicRpc,
                            ReaderSettings settings,
                            ReadEventHandlersSettings handlersSettings,
                            @Nonnull CodecRegistry codecRegistry) {
-        super(topicRpc, settings, codecRegistry);
+        this(topicRpc, settings, handlersSettings, codecRegistry, Meter.NOOP);
+    }
+
+    public AsyncReaderImpl(TopicRpc topicRpc,
+                           ReaderSettings settings,
+                           ReadEventHandlersSettings handlersSettings,
+                           @Nonnull CodecRegistry codecRegistry,
+                           Meter meter) {
+        super(topicRpc, settings, codecRegistry, meter);
         this.eventHandler = handlersSettings.getEventHandler();
 
         if (handlersSettings.getExecutor() != null) {
@@ -100,6 +109,7 @@ protected void handleDataReceivedEvent(ReadPartitionSession session, DataReceive
             long offsetEnd = event.getMessages().get(event.getMessages().size() - 1).getOffset();
             logger.debug("{} DataReceivedEvent callback with {} message(s) (offsets {}-{}) is about "
                     + "to be called...", session, messagesCount, offsetStart, offsetEnd);
+            getMetrics().reportDelivered(messagesCount, session.getPartition().getPath());
             eventHandler.onMessages(event);
             logger.debug("{} DataReceivedEvent callback with {} message(s) (offsets {}-{}) "
                     + "successfully finished", session, messagesCount, offsetStart, offsetEnd);

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

@@ -16,6 +16,7 @@
 
 import tech.ydb.common.transaction.YdbTransaction;
 import tech.ydb.core.Status;
+import tech.ydb.core.metrics.Meter;
 import tech.ydb.topic.TopicRpc;
 import tech.ydb.topic.description.CodecRegistry;
 import tech.ydb.topic.impl.GrpcStreamRetrier;
@@ -36,13 +37,20 @@ public abstract class ReaderImpl extends GrpcStreamRetrier {
     private static final int DEFAULT_DECOMPRESSION_THREAD_COUNT = 4;
     private final ExecutorService defaultDecompressionExecutorService;
     private final ReadSessionFactory sessionFactory;
+    private final ReaderMetrics metrics;
 
     private final CompletableFuture<Void> sessionReady = new CompletableFuture<>();
     private volatile ReadSession session = null;
 
     public ReaderImpl(TopicRpc topicRpc, ReaderSettings settings, @Nonnull CodecRegistry codecRegistry) {
+        this(topicRpc, settings, codecRegistry, Meter.NOOP);
+    }
+
+    public ReaderImpl(TopicRpc topicRpc, ReaderSettings settings, @Nonnull CodecRegistry codecRegistry, Meter meter) {
         super(settings.getLogPrefix(), topicRpc.getScheduler(), settings.getErrorsHandler());
 
+        this.metrics = new ReaderMetrics(meter, settings.getConsumerName(), settings.getReaderName());
+
         Executor decompressionExecutor = settings.getDecompressionExecutor();
         if (decompressionExecutor != null) {
             this.defaultDecompressionExecutorService = null;
@@ -71,6 +79,10 @@ public ReaderImpl(TopicRpc topicRpc, ReaderSettings settings, @Nonnull CodecRegi
     protected abstract void handleStopPartitionSession(StopPartitionSessionEvent event);
     protected abstract void handleClosePartitionSession(PartitionSession partition);
 
+    final ReaderMetrics getMetrics() {
+        return metrics;
+    }
+
     @Override
     protected Logger getLogger() {
         return logger;

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

@@ -0,0 +1,48 @@
+package tech.ydb.topic.read.impl;
+
+import java.util.Arrays;
+
+import tech.ydb.core.metrics.Attr;
+import tech.ydb.core.metrics.LongCounter;
+import tech.ydb.core.metrics.Meter;
+
+/**
+ * Topic reader delivery metrics.
+ */
+final class ReaderMetrics {
+    private static final String MESSAGE_UNIT = "{message}";
+
+    private final LongCounter deliveredMessages;
+    private final Attr[] commonAttributes;
+
+    ReaderMetrics(Meter meter, String consumer, String readerName) {
+        this.deliveredMessages = meter.createCounter(
+                "ydb.topic.reader.delivered.messages",
+                MESSAGE_UNIT,
+                "The number of messages delivered by the SDK to application code.");
+        this.commonAttributes = createCommonAttributes(consumer, readerName);
+    }
+
+    void reportDelivered(long messages, String topic) {
+        record(deliveredMessages, messages, topic);
+    }
+
+    private void record(LongCounter counter, long value, String topic) {
+        if (value > 0) {
+            Attr[] attributes = Arrays.copyOf(commonAttributes, commonAttributes.length + 1);
+            attributes[commonAttributes.length] = Attr.of("topic", topic);
+            counter.add(value, attributes);
+        }
+    }
+
+    private static Attr[] createCommonAttributes(String consumer, String readerName) {
+        boolean hasReaderName = readerName != null && !readerName.isEmpty();
+        if (consumer == null) {
+            return hasReaderName ? new Attr[]{Attr.of("reader.name", readerName)} : new Attr[0];
+        }
+        if (!hasReaderName) {
+            return new Attr[]{Attr.of("consumer", consumer)};
+        }
+        return new Attr[]{Attr.of("consumer", consumer), Attr.of("reader.name", readerName)};
+    }
+}

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

@@ -19,6 +19,7 @@
 import org.slf4j.LoggerFactory;
 
 import tech.ydb.core.Status;
+import tech.ydb.core.metrics.Meter;
 import tech.ydb.topic.TopicRpc;
 import tech.ydb.topic.description.CodecRegistry;
 import tech.ydb.topic.description.OffsetsRange;
@@ -46,7 +47,12 @@ public class SyncReaderImpl extends ReaderImpl implements SyncReader {
     private volatile String sessionId = null;
 
     public SyncReaderImpl(TopicRpc topicRpc, ReaderSettings settings, @Nonnull CodecRegistry codecRegistry) {
-        super(topicRpc, settings, codecRegistry);
+        this(topicRpc, settings, codecRegistry, Meter.NOOP);
+    }
+
+    public SyncReaderImpl(TopicRpc topicRpc, ReaderSettings settings, @Nonnull CodecRegistry codecRegistry,
+            Meter meter) {
+        super(topicRpc, settings, codecRegistry, meter);
     }
 
     private static class MessageWrapper {
@@ -157,6 +163,7 @@ public Message receiveInternal(ReceiveSettings receiveSettings, long timeout, Ti
             }
 
             next.release();
+            getMetrics().reportDelivered(1, result.getPartitionSession().getPath());
             return result;
         }
     }

topic/src/test/java/tech/ydb/topic/read/impl/ReaderMetricsTest.java

@@ -0,0 +1,71 @@
+package tech.ydb.topic.read.impl;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import tech.ydb.core.metrics.LongCounter;
+import tech.ydb.core.metrics.Meter;
+import tech.ydb.topic.TopicClient;
+import tech.ydb.topic.TopicRpc;
+import tech.ydb.topic.description.Codec;
+import tech.ydb.topic.impl.TopicClientImpl;
+import tech.ydb.topic.read.SyncReader;
+import tech.ydb.topic.settings.ReaderSettings;
+import tech.ydb.topic.settings.TopicReadSettings;
+
+public class ReaderMetricsTest {
+    private static final String DELIVERED = "ydb.topic.reader.delivered.messages";
+
+    @Test
+    public void deliveredMessageIncrementsCounter() throws InterruptedException {
+        RecordingMeter meter = new RecordingMeter();
+        ReadStreamMock stream = new ReadStreamMock();
+        TopicRpc rpc = Mockito.mock(TopicRpc.class);
+        Mockito.when(rpc.getScheduler()).thenReturn(Mockito.mock(ScheduledExecutorService.class));
+        Mockito.when(rpc.readSession(Mockito.any(String.class))).thenReturn(stream);
+
+        TopicClient client = TopicClientImpl.newClient(rpc).withMeter(meter).build();
+        SyncReader reader = client.createSyncReader(ReaderSettings.newBuilder()
+                .addTopic(TopicReadSettings.newBuilder().setPath("/topic").build())
+                .setConsumerName("consumer")
+                .build());
+
+        try {
+            reader.init();
+            stream.responseInit("read-session");
+            stream.responseStartPartition("/topic", 42);
+            stream.responseData(25).partition(1, -1)
+                    .batch(Codec.RAW, new byte[] { 1 })
+                    .and().send();
+
+            Assert.assertEquals(0, meter.value(DELIVERED));
+
+            Assert.assertNotNull(reader.receive());
+            Assert.assertEquals(1, meter.value(DELIVERED));
+        } finally {
+            reader.shutdown();
+            client.close();
+        }
+    }
+
+    private static class RecordingMeter implements Meter {
+        private final Map<String, AtomicLong> counters = new ConcurrentHashMap<>();
+
+        @Override
+        public LongCounter createCounter(String name, String unit, String description) {
+            AtomicLong counter = counters.computeIfAbsent(name, key -> new AtomicLong());
+            return (value, attrs) -> counter.addAndGet(value);
+        }
+
+        long value(String name) {
+            AtomicLong counter = counters.get(name);
+            return counter == null ? 0 : counter.get();
+        }
+    }
+}