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 6b26f3aa1..d5b87fcfb 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 3ccd4d520..7f2e179f2 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) { 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();