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..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,6 +14,7 @@ public class BatchMeta { private final Map writeSessionMeta; private final int codec; private final Instant writtenAt; + private final long receivedAt = System.nanoTime(); public BatchMeta(YdbTopic.StreamReadMessage.ReadResponse.Batch batch) { this.producerId = batch.getProducerId(); @@ -22,6 +23,10 @@ public BatchMeta(YdbTopic.StreamReadMessage.ReadResponse.Batch batch) { this.writtenAt = ProtobufUtils.protoToInstant(batch.getWrittenAt()); } + long getReceivedAt() { + return receivedAt; + } + public String getProducerId() { return producerId; } 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..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,6 +48,10 @@ public MessageImpl(PartitionSession session, MessageCommitter committer, BatchMe public abstract boolean isReady(); + 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..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 @@ -1,6 +1,7 @@ package tech.ydb.topic.read.impl; import java.util.concurrent.Executor; +import java.util.function.LongSupplier; import tech.ydb.topic.description.CodecRegistry; import tech.ydb.topic.settings.ReaderSettings; @@ -18,9 +19,16 @@ public class ReadConfig { private final long maxMemoryUsageBytes; private final int maxBatchSize; private final ReaderMetrics metrics; + 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, () -> 0); + } + + public ReadConfig(CodecRegistry codecRegistry, Executor control, Executor processor, Executor decompressor, + ReaderSettings settings, LongSupplier readyMessages, LongSupplier readyOldestTimestamp) { this.codecRegistry = codecRegistry; this.control = control; this.processor = processor; @@ -29,6 +37,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.readyOldestTimestamp = readyOldestTimestamp; } public CodecRegistry getCodecRegistry() { @@ -59,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/ReadPartition.java b/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartition.java index 88909462c..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 @@ -90,6 +90,14 @@ public PartitionSession getPartition() { return partition; } + public long getCommitOffsetLag() { + return committer.getCommitOffsetLag(); + } + + void updateCommitTarget(long offset) { + committer.updateCommitTarget(offset); + } + @Override public void confirmRangeProcessed(OffsetsRange range) { decoder.releaseRange(range); @@ -104,11 +112,20 @@ public void confirmPartitionEnded(List childs) { sendDataToReaders(); } - public void confirmCommittedOffset(long committedOffset) { - committer.updateCommittedOffset(committedOffset); + long getOldestReceivedTimestamp() { + return queue.getOldestReceivedTimestamp(); + } + + long getLocalBufferMessages() { + return queue.getLocalBufferMessages(); + } + + 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 7993ce524..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 @@ -32,24 +32,36 @@ 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() { return new RuntimeException("" + partition + " is already stopped"); } - public void updateCommittedOffset(long offset) { - long old = lastCommittedOffset.get(); - if (old != lastCommittedOffset.accumulateAndGet(offset, Math::max)) { + public long getCommitOffsetLag() { + return Math.max(0, commitTargetOffset.get() - lastCommittedOffset.get()); + } + + void updateCommitTarget(long offset) { + commitTargetOffset.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() { 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..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 @@ -36,6 +36,15 @@ void updateLastReadOffset(long offset) { lastReadOffset = offset; } + long getLocalBufferMessages() { + return readingQueue.size(); + } + + long getOldestReceivedTimestamp() { + MessageImpl first = readingQueue.peek(); + return first == null ? 0 : first.getReceivedAt(); + } + void addBatches(List batchList) { for (YdbTopic.StreamReadMessage.ReadResponse.Batch batch : batchList) { if (batch.getMessageDataCount() == 0) { 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..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 @@ -128,6 +128,10 @@ public boolean commitOffsets(PartitionSession session, List ranges .build()) .build(); + 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; } @@ -189,7 +193,30 @@ 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, + this::getCommitOffsetLagMax, this::getLocalBufferMessages, this::getOldestReceivedTimestamp); + } + + private long getCommitOffsetLagMax() { + return partitions.values().stream().mapToLong(ReadPartition::getCommitOffsetLag).max().orElse(0); + } + + private long getLocalBufferMessages() { + return config.getReadyMessages() + partitions.values().stream() + .filter(ReadPartition::isActive).mapToLong(ReadPartition::getLocalBufferMessages).sum(); + } + + private long getOldestReceivedTimestamp() { + long oldestTimestamp = config.getReadyOldestTimestamp(); + 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) { @@ -299,7 +326,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 3ccd4d520..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 @@ -19,12 +19,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 +42,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 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); } - void register(LongSupplier partitionCount, LongSupplier bufferBudget) { + void register(LongSupplier partitionCount, LongSupplier bufferBudget, LongSupplier commitLag, + 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.", @@ -50,13 +60,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 -> { + long timestamp = oldestTimestamp.getAsLong(); + m.record(timestamp == 0 ? 0 : (System.nanoTime() - timestamp) / 1_000_000_000d, + 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; } void reportDelivered(long messages, String topic) { @@ -85,6 +114,14 @@ void reportSessionError(Status status, boolean retry) { sessionErrors.add(1, attributes); } + void reportCommitQueued(long messages, String topic) { + 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..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 @@ -65,7 +65,8 @@ 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, this::getOldestReceivedTimestamp); this.impl = new Impl(topicRpc, debugId, settings, config); String readerName = settings.getReaderName(); @@ -83,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();