From fbdf9a520a638534d4cd9e01a184eb1c23da53ca Mon Sep 17 00:00:00 2001 From: KirillKurdyukov Date: Thu, 15 Oct 2026 23:02:10 +0300 Subject: [PATCH 1/6] Add Topic Reader commit and buffer metrics --- .../tech/ydb/topic/read/impl/BatchMeta.java | 14 ++++++ .../tech/ydb/topic/read/impl/MessageImpl.java | 8 ++++ .../tech/ydb/topic/read/impl/ReadConfig.java | 16 +++++++ .../ydb/topic/read/impl/ReadPartition.java | 16 ++++++- .../read/impl/ReadPartitionCommitter.java | 27 ++++++++++- .../topic/read/impl/ReadPartitionQueue.java | 14 +++++- .../tech/ydb/topic/read/impl/ReadSession.java | 12 ++++- .../ydb/topic/read/impl/ReaderMetrics.java | 48 ++++++++++++++++++- .../ydb/topic/read/impl/SyncReaderImpl.java | 8 +++- 9 files changed, 154 insertions(+), 9 deletions(-) diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/BatchMeta.java b/topic/src/main/java/tech/ydb/topic/read/impl/BatchMeta.java index ecc8a3a11..c348d2edc 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/BatchMeta.java +++ b/topic/src/main/java/tech/ydb/topic/read/impl/BatchMeta.java @@ -14,12 +14,26 @@ public class BatchMeta { private final Map 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() { diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/MessageImpl.java b/topic/src/main/java/tech/ydb/topic/read/impl/MessageImpl.java index 0fb94e9d6..be3ef2d3b 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/MessageImpl.java +++ b/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; diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/ReadConfig.java b/topic/src/main/java/tech/ydb/topic/read/impl/ReadConfig.java index b01a84fbe..528387cf6 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/ReadConfig.java +++ b/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() { diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartition.java b/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartition.java index 88909462c..047cfdb6d 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartition.java +++ b/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 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 batchList) { + public boolean addBatches(List batchList, long receivedAt) { if (!isActive()) { return false; } @@ -130,7 +142,7 @@ public boolean addBatches(List ba metrics.reportReceivedMessages(messagesCount, partition.getPath()); if (messagesCount > 0) { - queue.addBatches(batchList); + queue.addBatches(batchList, receivedAt); sendDataToReaders(); } diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionCommitter.java b/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionCommitter.java index 7993ce524..d2b616963 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionCommitter.java +++ b/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 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 commit(OffsetsRange range) { @Override public void commitRanges(List 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() { diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionQueue.java b/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionQueue.java index 50ca6b33d..a0c1cd775 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionQueue.java +++ b/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionQueue.java @@ -36,14 +36,24 @@ void updateLastReadOffset(long offset) { lastReadOffset = offset; } - void addBatches(List 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 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 messages = new ArrayList<>(batch.getMessageDataCount()); for (YdbTopic.StreamReadMessage.ReadResponse.MessageData msg : batch.getMessageDataList()) { if (lastReadOffset > msg.getOffset()) { diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/ReadSession.java b/topic/src/main/java/tech/ydb/topic/read/impl/ReadSession.java index 2f2ec0038..67f0e4ee2 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/ReadSession.java +++ b/topic/src/main/java/tech/ydb/topic/read/impl/ReadSession.java @@ -128,6 +128,7 @@ public boolean commitOffsets(PartitionSession session, List 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 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,6 +278,7 @@ 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()); @@ -278,7 +286,7 @@ public void onRead(YdbTopic.StreamReadMessage.ReadResponse response) { 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); 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 index c61a780fe..59f021bd9 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/ReaderMetrics.java +++ b/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,12 +45,17 @@ 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.", @@ -50,13 +63,32 @@ void register(LongSupplier partitionCount, LongSupplier bufferBudget) { 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, String failureReason) { sessionErrors.add(1, attributes); } + void reportCommitQueued(List 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)); 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 0efd13af4..cd0aeedff 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 @@ -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(); From 3e015b387e3ff23898d6abcb928c17d75ec299fb Mon Sep 17 00:00:00 2001 From: KirillKurdyukov Date: Mon, 19 Oct 2026 15:22:14 +0300 Subject: [PATCH 2/6] Simplify Topic Reader metric timestamp collection --- .../tech/ydb/topic/read/impl/BatchMeta.java | 11 +------ .../tech/ydb/topic/read/impl/MessageImpl.java | 4 --- .../tech/ydb/topic/read/impl/ReadConfig.java | 7 ++-- .../ydb/topic/read/impl/ReadPartition.java | 8 ++--- .../topic/read/impl/ReadPartitionQueue.java | 10 +++--- .../tech/ydb/topic/read/impl/ReadSession.java | 32 ++++++++++++++----- .../ydb/topic/read/impl/ReaderMetrics.java | 13 ++++---- .../ydb/topic/read/impl/SyncReaderImpl.java | 12 +++---- 8 files changed, 48 insertions(+), 49 deletions(-) diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/BatchMeta.java b/topic/src/main/java/tech/ydb/topic/read/impl/BatchMeta.java index c348d2edc..0a37315d4 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/BatchMeta.java +++ b/topic/src/main/java/tech/ydb/topic/read/impl/BatchMeta.java @@ -14,22 +14,13 @@ public class BatchMeta { private final Map writeSessionMeta; private final int codec; private final Instant writtenAt; - private final long receivedAt; + private final long receivedAt = System.nanoTime(); 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() { diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/MessageImpl.java b/topic/src/main/java/tech/ydb/topic/read/impl/MessageImpl.java index be3ef2d3b..d0b827092 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/MessageImpl.java +++ b/topic/src/main/java/tech/ydb/topic/read/impl/MessageImpl.java @@ -48,10 +48,6 @@ public MessageImpl(PartitionSession session, MessageCommitter committer, BatchMe public abstract boolean isReady(); - double getLocalBufferMessageAge(long now) { - return batchMeta.age(now); - } - long getReceivedAt() { return batchMeta.getReceivedAt(); } diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/ReadConfig.java b/topic/src/main/java/tech/ydb/topic/read/impl/ReadConfig.java index 528387cf6..ebcca3f65 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/ReadConfig.java +++ b/topic/src/main/java/tech/ydb/topic/read/impl/ReadConfig.java @@ -1,7 +1,6 @@ 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; @@ -21,7 +20,7 @@ public class ReadConfig { private final int maxBatchSize; private final ReaderMetrics metrics; final LongSupplier readyMessages; - final DoubleSupplier readyMessageAgeMax; + final LongSupplier readyOldestTimestamp; public ReadConfig(CodecRegistry codecRegistry, Executor control, Executor processor, Executor decompressor, ReaderSettings settings) { @@ -34,7 +33,7 @@ public ReadConfig(CodecRegistry codecRegistry, Executor control, Executor proces } public ReadConfig(CodecRegistry codecRegistry, Executor control, Executor processor, Executor decompressor, - ReaderSettings settings, LongSupplier readyMessages, DoubleSupplier readyMessageAgeMax) { + ReaderSettings settings, LongSupplier readyMessages, LongSupplier readyOldestTimestamp) { this.codecRegistry = codecRegistry; this.control = control; this.processor = processor; @@ -44,7 +43,7 @@ public ReadConfig(CodecRegistry codecRegistry, Executor control, Executor proces this.maxBatchSize = settings.getMaxBatchSize(); this.metrics = new ReaderMetrics(settings.getMeter(), settings.getConsumerName(), settings.getReaderName()); this.readyMessages = readyMessages; - this.readyMessageAgeMax = readyMessageAgeMax; + this.readyOldestTimestamp = readyOldestTimestamp; } public CodecRegistry getCodecRegistry() { diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartition.java b/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartition.java index 047cfdb6d..f390d8943 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartition.java +++ b/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartition.java @@ -108,8 +108,8 @@ public void confirmPartitionEnded(List childs) { sendDataToReaders(); } - double getLocalBufferMessageAgeMax() { - return queue.getLocalBufferMessageAgeMax(); + long getOldestReceivedTimestamp() { + return queue.getOldestReceivedTimestamp(); } long getLocalBufferMessages() { @@ -131,7 +131,7 @@ public void unpause() { sendDataToReaders(); } - public boolean addBatches(List batchList, long receivedAt) { + public boolean addBatches(List batchList) { if (!isActive()) { return false; } @@ -142,7 +142,7 @@ public boolean addBatches(List ba metrics.reportReceivedMessages(messagesCount, partition.getPath()); if (messagesCount > 0) { - queue.addBatches(batchList, receivedAt); + queue.addBatches(batchList); sendDataToReaders(); } diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionQueue.java b/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionQueue.java index a0c1cd775..66ad1f34d 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionQueue.java +++ b/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionQueue.java @@ -40,20 +40,18 @@ 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; + long getOldestReceivedTimestamp() { + return readingQueue.stream().mapToLong(MessageImpl::getReceivedAt).min().orElse(0); } - void addBatches(List batchList, long receivedAt) { + void addBatches(List batchList) { 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, receivedAt); + BatchMeta meta = new BatchMeta(batch); List messages = new ArrayList<>(batch.getMessageDataCount()); for (YdbTopic.StreamReadMessage.ReadResponse.MessageData msg : batch.getMessageDataList()) { if (lastReadOffset > msg.getOffset()) { diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/ReadSession.java b/topic/src/main/java/tech/ydb/topic/read/impl/ReadSession.java index d5b87fcfb..6b0a213ec 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/ReadSession.java +++ b/topic/src/main/java/tech/ydb/topic/read/impl/ReadSession.java @@ -191,12 +191,29 @@ public void removeLocks(PartitionSession ps, List childs) { public void onInit(YdbTopic.StreamReadMessage.InitResponse response) { bufferManager.init(response.getSessionId()); 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))); + this::getCommitOffsetLagMax, this::getLocalBufferMessages, this::getOldestReceivedTimestamp); + } + + private long getCommitOffsetLagMax() { + return partitions.values().stream().mapToLong(ReadPartition::getCommitOffsetLag).max().orElse(0); + } + + private long getLocalBufferMessages() { + return config.readyMessages.getAsLong() + partitions.values().stream() + .filter(ReadPartition::isActive).mapToLong(ReadPartition::getLocalBufferMessages).sum(); + } + + private long getOldestReceivedTimestamp() { + long oldestTimestamp = config.readyOldestTimestamp.getAsLong(); + for (ReadPartition partition : partitions.values()) { + if (partition.isActive()) { + long timestamp = partition.getOldestReceivedTimestamp(); + if (timestamp != 0 && (oldestTimestamp == 0 || timestamp < oldestTimestamp)) { + oldestTimestamp = timestamp; + } + } + } + return oldestTimestamp; } public void onStartPartition(YdbTopic.StreamReadMessage.StartPartitionSessionRequest req) { @@ -278,7 +295,6 @@ 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()); @@ -286,7 +302,7 @@ public void onRead(YdbTopic.StreamReadMessage.ReadResponse response) { for (YdbTopic.StreamReadMessage.ReadResponse.PartitionData data: response.getPartitionDataList()) { long psid = data.getPartitionSessionId(); ReadPartition partition = partitions.get(psid); - if (partition == null || !partition.addBatches(data.getBatchesList(), receivedAt)) { + if (partition == null || !partition.addBatches(data.getBatchesList())) { logger.warn("[{}] Received PartitionData for unknown(most likely already closed) PartitionSessionId={}", debugId, psid); bufferManager.releasePartition(psid); 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 index 7f2e179f2..f922cc13c 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/ReaderMetrics.java +++ b/topic/src/main/java/tech/ydb/topic/read/impl/ReaderMetrics.java @@ -2,7 +2,6 @@ import java.util.Arrays; import java.util.List; -import java.util.function.DoubleSupplier; import java.util.function.LongSupplier; import tech.ydb.core.Status; @@ -55,7 +54,7 @@ final class ReaderMetrics { } void register(LongSupplier partitionCount, LongSupplier bufferBudget, LongSupplier commitLag, - LongSupplier bufferMessages, DoubleSupplier messageAgeMax) { + LongSupplier bufferMessages, LongSupplier oldestTimestamp) { partitionsGauge = meter.registerLongGauge( "ydb.topic.reader.partition_session.count", "{session}", "The number of partition sessions currently in the reader session processing lifecycle.", @@ -71,7 +70,11 @@ void register(LongSupplier partitionCount, LongSupplier bufferBudget, LongSuppli 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)); + m -> { + long timestamp = oldestTimestamp.getAsLong(); + m.record(timestamp == 0 ? 0 : (System.nanoTime() - timestamp) / 1_000_000_000d, + commonAttributes); + }); } void unregister() { @@ -87,10 +90,6 @@ void unregister() { ageGauge = MetricRegistration.NOOP; } - long reportReadResponseStart() { - return enabled ? System.nanoTime() : 0; - } - void reportDelivered(long messages, String topic) { report(deliveredMessages, messages, topic); } 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 cd0aeedff..2d68515e1 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 @@ -66,12 +66,7 @@ public SyncReaderImpl(TopicRpc topicRpc, ReaderSettings settings, @Nonnull Codec this.decompressor = new LazyExecutor("reader[" + debugId + "]-decoder", settings.getDecompressionExecutor()); 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; - }); + queue::size, this::getOldestReceivedTimestamp); this.impl = new Impl(topicRpc, debugId, settings, config); String readerName = settings.getReaderName(); @@ -89,6 +84,11 @@ public String getSessionId() { return sessionId; } + private long getOldestReceivedTimestamp() { + return queue.stream().filter(MessageWrapper::isActive) + .mapToLong(message -> ((MessageImpl) message.msg).getReceivedAt()).min().orElse(0); + } + @Override public void init() { impl.start(); From 15553f52c08f70d041c40972fb07f7bf66b5fe2d Mon Sep 17 00:00:00 2001 From: KirillKurdyukov Date: Mon, 19 Oct 2026 16:55:03 +0300 Subject: [PATCH 3/6] Count Topic Reader commit acknowledgements in read session --- .../ydb/topic/read/impl/ReadPartition.java | 5 +++-- .../read/impl/ReadPartitionCommitter.java | 19 +++++++------------ .../tech/ydb/topic/read/impl/ReadSession.java | 3 ++- .../ydb/topic/read/impl/ReaderMetrics.java | 2 +- 4 files changed, 13 insertions(+), 16 deletions(-) diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartition.java b/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartition.java index f390d8943..075fbcf21 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartition.java +++ b/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartition.java @@ -116,11 +116,12 @@ long getLocalBufferMessages() { return queue.getLocalBufferMessages(); } - public void confirmCommittedOffset(long committedOffset) { - committer.updateCommittedOffset(committedOffset); + public long confirmCommittedOffset(long committedOffset) { + long acknowledged = committer.updateCommittedOffset(committedOffset); committer.completePendingCommits(); commitOffsetAck.set(new CommitOffsetAcknowledgementEventImpl(partition, committedOffset)); sendDataToReaders(); + return acknowledged; } public void unpause() { diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionCommitter.java b/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionCommitter.java index d2b616963..a43e228a6 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionCommitter.java +++ b/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionCommitter.java @@ -53,12 +53,14 @@ public long getCommitOffsetLag() { } } - public void updateCommittedOffset(long offset) { - long old = lastCommittedOffset.get(); - if (old != lastCommittedOffset.accumulateAndGet(offset, Math::max)) { + public long updateCommittedOffset(long offset) { + long old = lastCommittedOffset.getAndAccumulate(offset, Math::max); + if (offset > old) { logger.debug("[{}] Updated last committed offset: {}. Previous committed offset: {} " + "(diff is {} message(s)).", traceID, offset, old, offset - old); + return offset - old; } + return 0; } public void completePendingCommits() { @@ -94,9 +96,6 @@ public CompletableFuture 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(); } @@ -121,12 +120,8 @@ public CompletableFuture commit(OffsetsRange range) { public void commitRanges(List 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())); - }); + ranges.forEach(range -> + commitFutures.computeIfAbsent(range.getEnd(), offset -> new CompletableFuture<>())); } finally { commitFuturesLock.unlock(); } diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/ReadSession.java b/topic/src/main/java/tech/ydb/topic/read/impl/ReadSession.java index 6b0a213ec..9d96b2547 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/ReadSession.java +++ b/topic/src/main/java/tech/ydb/topic/read/impl/ReadSession.java @@ -323,7 +323,8 @@ public void onCommitOffset(YdbTopic.StreamReadMessage.CommitOffsetResponse respo continue; } - partition.confirmCommittedOffset(offset.getCommittedOffset()); + long acknowledged = partition.confirmCommittedOffset(offset.getCommittedOffset()); + config.getMetrics().reportCommitAcknowledged(acknowledged, partition.getPartition().getPath()); } } 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 index f922cc13c..0e59ea69a 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/ReaderMetrics.java +++ b/topic/src/main/java/tech/ydb/topic/read/impl/ReaderMetrics.java @@ -47,7 +47,7 @@ final class ReaderMetrics { 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."); + "The number of messages acknowledged by advances of the committed offset."); 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); From dca6bcd741d8a2f92814ceeb10e56ce3e800925f Mon Sep 17 00:00:00 2001 From: KirillKurdyukov Date: Mon, 19 Oct 2026 17:22:55 +0300 Subject: [PATCH 4/6] Remove metric-only Topic Reader bulk futures --- .../ydb/topic/read/impl/ReadPartition.java | 4 +++ .../read/impl/ReadPartitionCommitter.java | 27 ++++++++----------- .../tech/ydb/topic/read/impl/ReadSession.java | 5 +++- .../ydb/topic/read/impl/ReaderMetrics.java | 12 ++------- 4 files changed, 21 insertions(+), 27 deletions(-) diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartition.java b/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartition.java index 075fbcf21..632007f37 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartition.java +++ b/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartition.java @@ -94,6 +94,10 @@ public long getCommitOffsetLag() { return committer.getCommitOffsetLag(); } + void updateCommitTarget(long offset) { + committer.updateCommitTarget(offset); + } + @Override public void confirmRangeProcessed(OffsetsRange range) { decoder.releaseRange(range); diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionCommitter.java b/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionCommitter.java index a43e228a6..5133ed8ca 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionCommitter.java +++ b/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionCommitter.java @@ -32,12 +32,14 @@ class ReadPartitionCommitter implements MessageCommitter { private final ReentrantLock commitFuturesLock = new ReentrantLock(); private final AtomicLong lastCommittedOffset; + private final AtomicLong commitTargetOffset; ReadPartitionCommitter(String traceID, ReadSession session, PartitionSession partition, long lastCommittedOffset) { this.traceID = traceID; this.session = session; this.partition = partition; this.lastCommittedOffset = new AtomicLong(lastCommittedOffset); + this.commitTargetOffset = new AtomicLong(lastCommittedOffset); } private RuntimeException partitionIsClosedException() { @@ -45,15 +47,17 @@ private RuntimeException partitionIsClosedException() { } public long getCommitOffsetLag() { - commitFuturesLock.lock(); - try { - return commitFutures.isEmpty() ? 0 : commitFutures.lastKey() - lastCommittedOffset.get(); - } finally { - commitFuturesLock.unlock(); - } + long committed = lastCommittedOffset.get(); + return commitTargetOffset.get() - committed; + } + + void updateCommitTarget(long offset) { + commitTargetOffset.accumulateAndGet(offset, Math::max); } public long updateCommittedOffset(long offset) { + // Advance the target first so lag cannot observe an acknowledged offset ahead of it. + updateCommitTarget(offset); long old = lastCommittedOffset.getAndAccumulate(offset, Math::max); if (offset > old) { logger.debug("[{}] Updated last committed offset: {}. Previous committed offset: {} " @@ -118,16 +122,7 @@ public CompletableFuture commit(OffsetsRange range) { @Override public void commitRanges(List ranges) { - commitFuturesLock.lock(); - try { - ranges.forEach(range -> - commitFutures.computeIfAbsent(range.getEnd(), offset -> new CompletableFuture<>())); - } finally { - commitFuturesLock.unlock(); - } - if (!session.commitOffsets(partition, ranges)) { - close(); - } + session.commitOffsets(partition, ranges); } public void close() { diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/ReadSession.java b/topic/src/main/java/tech/ydb/topic/read/impl/ReadSession.java index 9d96b2547..0a26311f9 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/ReadSession.java +++ b/topic/src/main/java/tech/ydb/topic/read/impl/ReadSession.java @@ -128,7 +128,10 @@ public boolean commitOffsets(PartitionSession session, List ranges .build()) .build(); - config.getMetrics().reportCommitQueued(rangesToCommit, session.getPath()); + for (OffsetsRange range : rangesToCommit) { + partition.updateCommitTarget(range.getEnd()); + config.getMetrics().reportCommitQueued(range.getEnd() - range.getStart(), session.getPath()); + } send(FromClient.newBuilder().setCommitOffsetRequest(req).build()); return true; } 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 index 0e59ea69a..1ef203108 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/ReaderMetrics.java +++ b/topic/src/main/java/tech/ydb/topic/read/impl/ReaderMetrics.java @@ -1,7 +1,6 @@ package tech.ydb.topic.read.impl; import java.util.Arrays; -import java.util.List; import java.util.function.LongSupplier; import tech.ydb.core.Status; @@ -10,7 +9,6 @@ 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. @@ -116,14 +114,8 @@ void reportSessionError(Status status, boolean retry) { sessionErrors.add(1, attributes); } - void reportCommitQueued(List ranges, String topic) { - if (enabled) { - long messages = 0; - for (OffsetsRange range : ranges) { - messages += range.getEnd() - range.getStart(); - } - report(commitQueued, messages, topic); - } + void reportCommitQueued(long messages, String topic) { + report(commitQueued, messages, topic); } void reportCommitAcknowledged(long messages, String topic) { From 9e755ada16467f7be592001fc653139c6accfc56 Mon Sep 17 00:00:00 2001 From: KirillKurdyukov Date: Mon, 19 Oct 2026 17:49:42 +0300 Subject: [PATCH 5/6] Use FIFO head for Reader age and simplify commit lag --- .../tech/ydb/topic/read/impl/ReadPartitionCommitter.java | 5 +---- .../java/tech/ydb/topic/read/impl/ReadPartitionQueue.java | 3 ++- 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionCommitter.java b/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionCommitter.java index 5133ed8ca..9f5a6afa9 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionCommitter.java +++ b/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionCommitter.java @@ -47,8 +47,7 @@ private RuntimeException partitionIsClosedException() { } public long getCommitOffsetLag() { - long committed = lastCommittedOffset.get(); - return commitTargetOffset.get() - committed; + return Math.max(0, commitTargetOffset.get() - lastCommittedOffset.get()); } void updateCommitTarget(long offset) { @@ -56,8 +55,6 @@ void updateCommitTarget(long offset) { } public long updateCommittedOffset(long offset) { - // Advance the target first so lag cannot observe an acknowledged offset ahead of it. - updateCommitTarget(offset); long old = lastCommittedOffset.getAndAccumulate(offset, Math::max); if (offset > old) { logger.debug("[{}] Updated last committed offset: {}. Previous committed offset: {} " diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionQueue.java b/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionQueue.java index 66ad1f34d..f2854da05 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionQueue.java +++ b/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionQueue.java @@ -41,7 +41,8 @@ long getLocalBufferMessages() { } long getOldestReceivedTimestamp() { - return readingQueue.stream().mapToLong(MessageImpl::getReceivedAt).min().orElse(0); + MessageImpl first = readingQueue.peek(); + return first == null ? 0 : first.getReceivedAt(); } void addBatches(List batchList) { From 7dfb60ee2986c2fe654e095865192f6c57d59c7a Mon Sep 17 00:00:00 2001 From: KirillKurdyukov Date: Mon, 19 Oct 2026 18:18:35 +0300 Subject: [PATCH 6/6] Encapsulate Topic Reader ready queue suppliers --- .../tech/ydb/topic/read/impl/ReadConfig.java | 19 +++++++++++-------- .../tech/ydb/topic/read/impl/ReadSession.java | 4 ++-- 2 files changed, 13 insertions(+), 10 deletions(-) diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/ReadConfig.java b/topic/src/main/java/tech/ydb/topic/read/impl/ReadConfig.java index ebcca3f65..f3720622f 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/ReadConfig.java +++ b/topic/src/main/java/tech/ydb/topic/read/impl/ReadConfig.java @@ -19,17 +19,12 @@ public class ReadConfig { private final long maxMemoryUsageBytes; private final int maxBatchSize; private final ReaderMetrics metrics; - final LongSupplier readyMessages; - final LongSupplier readyOldestTimestamp; + private final LongSupplier readyMessages; + private final LongSupplier readyOldestTimestamp; 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); + this(codecRegistry, control, processor, decompressor, settings, () -> 0, () -> 0); } public ReadConfig(CodecRegistry codecRegistry, Executor control, Executor processor, Executor decompressor, @@ -74,6 +69,14 @@ public int getMaxBatchSize() { return maxBatchSize; } + public long getReadyMessages() { + return readyMessages.getAsLong(); + } + + public long getReadyOldestTimestamp() { + return readyOldestTimestamp.getAsLong(); + } + ReaderMetrics getMetrics() { return metrics; } diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/ReadSession.java b/topic/src/main/java/tech/ydb/topic/read/impl/ReadSession.java index 0a26311f9..b3d878bd1 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/ReadSession.java +++ b/topic/src/main/java/tech/ydb/topic/read/impl/ReadSession.java @@ -202,12 +202,12 @@ private long getCommitOffsetLagMax() { } private long getLocalBufferMessages() { - return config.readyMessages.getAsLong() + partitions.values().stream() + return config.getReadyMessages() + partitions.values().stream() .filter(ReadPartition::isActive).mapToLong(ReadPartition::getLocalBufferMessages).sum(); } private long getOldestReceivedTimestamp() { - long oldestTimestamp = config.readyOldestTimestamp.getAsLong(); + long oldestTimestamp = config.getReadyOldestTimestamp(); for (ReadPartition partition : partitions.values()) { if (partition.isActive()) { long timestamp = partition.getOldestReceivedTimestamp();