From 80fdba0feb66cd1c40750e4949fd04beeedc0441 Mon Sep 17 00:00:00 2001 From: KirillKurdyukov Date: Fri, 4 Sep 2026 16:37:10 +0300 Subject: [PATCH] Add Topic Reader delivered metric --- .../main/java/tech/ydb/topic/TopicClient.java | 9 +++ .../topic/impl/TopicClientBuilderImpl.java | 11 +++ .../tech/ydb/topic/impl/TopicClientImpl.java | 9 ++- .../ydb/topic/read/impl/AsyncReaderImpl.java | 12 +++- .../tech/ydb/topic/read/impl/ReaderImpl.java | 12 ++++ .../ydb/topic/read/impl/ReaderMetrics.java | 48 +++++++++++++ .../ydb/topic/read/impl/SyncReaderImpl.java | 9 ++- .../topic/read/impl/ReaderMetricsTest.java | 71 +++++++++++++++++++ 8 files changed, 177 insertions(+), 4 deletions(-) create mode 100644 topic/src/main/java/tech/ydb/topic/read/impl/ReaderMetrics.java create mode 100644 topic/src/test/java/tech/ydb/topic/read/impl/ReaderMetricsTest.java diff --git a/topic/src/main/java/tech/ydb/topic/TopicClient.java b/topic/src/main/java/tech/ydb/topic/TopicClient.java index 785f55b39..ebbcb12bd 100644 --- a/topic/src/main/java/tech/ydb/topic/TopicClient.java +++ b/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 diff --git a/topic/src/main/java/tech/ydb/topic/impl/TopicClientBuilderImpl.java b/topic/src/main/java/tech/ydb/topic/impl/TopicClientBuilderImpl.java index 2d2157d05..d0fab3240 100644 --- a/topic/src/main/java/tech/ydb/topic/impl/TopicClientBuilderImpl.java +++ b/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 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) { diff --git a/topic/src/main/java/tech/ydb/topic/impl/TopicClientImpl.java b/topic/src/main/java/tech/ydb/topic/impl/TopicClientImpl.java index d76dd4912..bd3c09ed0 100644 --- a/topic/src/main/java/tech/ydb/topic/impl/TopicClientImpl.java +++ b/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 diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/AsyncReaderImpl.java b/topic/src/main/java/tech/ydb/topic/read/impl/AsyncReaderImpl.java index bf5d4a10a..6a0aab2cd 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/AsyncReaderImpl.java +++ b/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); diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/ReaderImpl.java b/topic/src/main/java/tech/ydb/topic/read/impl/ReaderImpl.java index f27236991..8cf8d3f98 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/ReaderImpl.java +++ b/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 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; diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/ReaderMetrics.java b/topic/src/main/java/tech/ydb/topic/read/impl/ReaderMetrics.java new file mode 100644 index 000000000..dd8f9f871 --- /dev/null +++ b/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)}; + } +} diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/SyncReaderImpl.java b/topic/src/main/java/tech/ydb/topic/read/impl/SyncReaderImpl.java index 3914e8db6..66d77d93f 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/SyncReaderImpl.java +++ b/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; } } diff --git a/topic/src/test/java/tech/ydb/topic/read/impl/ReaderMetricsTest.java b/topic/src/test/java/tech/ydb/topic/read/impl/ReaderMetricsTest.java new file mode 100644 index 000000000..e1f225bcf --- /dev/null +++ b/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 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(); + } + } +}