From 7bd0abc1a5f405f59600a22b2b65a08e820e880c Mon Sep 17 00:00:00 2001 From: KirillKurdyukov Date: Thu, 15 Oct 2026 22:50:44 +0300 Subject: [PATCH 1/2] Add Topic Writer duration and age metrics --- .../ydb/topic/write/impl/BufferManager.java | 49 +++++++++++++++++-- .../ydb/topic/write/impl/EnqueuedMessage.java | 10 ++++ .../ydb/topic/write/impl/SentMessage.java | 16 ++++++ .../ydb/topic/write/impl/WriterMetrics.java | 41 ++++++++++++++++ .../ydb/topic/write/impl/WriterQueue.java | 12 +++-- 5 files changed, 120 insertions(+), 8 deletions(-) diff --git a/topic/src/main/java/tech/ydb/topic/write/impl/BufferManager.java b/topic/src/main/java/tech/ydb/topic/write/impl/BufferManager.java index d8a4a21a4..4227f6b2d 100644 --- a/topic/src/main/java/tech/ydb/topic/write/impl/BufferManager.java +++ b/topic/src/main/java/tech/ydb/topic/write/impl/BufferManager.java @@ -60,11 +60,21 @@ public void close(Status status) { } public void acquire(long messageSize) throws InterruptedException, QueueOverflowException { + acquire(messageSize, null); + } + + void acquire(long messageSize, WriterMetrics metrics) throws InterruptedException, QueueOverflowException { if (closed != null) { throw new IllegalStateException("Writer was closed with status " + closed); } - countAvailable.acquire(); + long waitStarted = 0; + if (!countAvailable.tryAcquire(0, TimeUnit.NANOSECONDS)) { + if (metrics != null) { + waitStarted = metrics.reportBufferWaitStart(); + } + countAvailable.acquire(); + } if (closed != null) { countAvailable.release(); @@ -74,7 +84,12 @@ public void acquire(long messageSize) throws InterruptedException, QueueOverflow int messageBlocks = calculateBlocksCount(messageSize, blockBitsCount); try { - blocksAvailable.acquire(messageBlocks); + if (!blocksAvailable.tryAcquire(messageBlocks, 0, TimeUnit.NANOSECONDS)) { + if (waitStarted == 0 && metrics != null) { + waitStarted = metrics.reportBufferWaitStart(); + } + blocksAvailable.acquire(messageBlocks); + } } catch (InterruptedException ex) { countAvailable.release(); throw ex; @@ -85,6 +100,9 @@ public void acquire(long messageSize) throws InterruptedException, QueueOverflow countAvailable.release(); throw new IllegalStateException("Writer was closed with status " + closed); } + if (metrics != null) { + metrics.reportBufferWaitDuration(waitStarted); + } } public void tryAcquire(long messageSize) throws QueueOverflowException { @@ -125,12 +143,25 @@ public void tryAcquire(long messageSize) throws QueueOverflowException { public void tryAcquire(long messageSize, long timeout, TimeUnit unit) throws InterruptedException, QueueOverflowException, TimeoutException { + tryAcquire(messageSize, timeout, unit, null); + } + + void tryAcquire(long messageSize, long timeout, TimeUnit unit, WriterMetrics metrics) throws InterruptedException, + QueueOverflowException, TimeoutException { if (closed != null) { throw new IllegalStateException("Writer was closed with status " + closed); } long expireAt = System.nanoTime() + unit.toNanos(timeout); - if (!countAvailable.tryAcquire(timeout, unit)) { + long waitStarted = 0; + boolean countAcquired = countAvailable.tryAcquire(0, TimeUnit.NANOSECONDS); + if (!countAcquired) { + if (metrics != null) { + waitStarted = metrics.reportBufferWaitStart(); + } + countAcquired = countAvailable.tryAcquire(timeout, unit); + } + if (!countAcquired) { String errorMsg = "[" + debugId + "] Rejecting a message due to reaching message queue in-flight limit of " + maxCount; logger.warn(errorMsg); @@ -147,7 +178,14 @@ public void tryAcquire(long messageSize, long timeout, TimeUnit unit) throws Int try { // negative timeout is allowed for tryAcquire long timeout2 = expireAt - System.nanoTime(); - if (!blocksAvailable.tryAcquire(messageBlocks, timeout2, TimeUnit.NANOSECONDS)) { + boolean blocksAcquired = blocksAvailable.tryAcquire(messageBlocks, 0, TimeUnit.NANOSECONDS); + if (!blocksAcquired) { + if (waitStarted == 0 && metrics != null) { + waitStarted = metrics.reportBufferWaitStart(); + } + blocksAcquired = blocksAvailable.tryAcquire(messageBlocks, timeout2, TimeUnit.NANOSECONDS); + } + if (!blocksAcquired) { countAvailable.release(); int count = maxCount - countAvailable.availablePermits(); long size = ((long) blocksAvailable.availablePermits()) << blockBitsCount; @@ -167,6 +205,9 @@ public void tryAcquire(long messageSize, long timeout, TimeUnit unit) throws Int countAvailable.release(); throw new IllegalStateException("Writer was closed with status " + closed); } + if (metrics != null) { + metrics.reportBufferWaitDuration(waitStarted); + } } public void releaseMessage(long messageSize) { diff --git a/topic/src/main/java/tech/ydb/topic/write/impl/EnqueuedMessage.java b/topic/src/main/java/tech/ydb/topic/write/impl/EnqueuedMessage.java index 309fa7a33..38a75652b 100644 --- a/topic/src/main/java/tech/ydb/topic/write/impl/EnqueuedMessage.java +++ b/topic/src/main/java/tech/ydb/topic/write/impl/EnqueuedMessage.java @@ -12,6 +12,7 @@ */ public class EnqueuedMessage { private final MessageMeta meta; + private final long acceptedTimestamp; private final CompletableFuture ackFuture = new CompletableFuture<>(); private volatile ByteString data = null; @@ -20,7 +21,12 @@ public class EnqueuedMessage { private volatile boolean isReady = false; public EnqueuedMessage(MessageMeta meta, long bufferSize) { + this(meta, bufferSize, 0); + } + + EnqueuedMessage(MessageMeta meta, long bufferSize, long acceptedTimestamp) { this.meta = meta; + this.acceptedTimestamp = acceptedTimestamp; this.data = null; this.bufferSize = bufferSize; } @@ -29,6 +35,10 @@ public MessageMeta getMeta() { return meta; } + long getAcceptedTimestamp() { + return acceptedTimestamp; + } + public ByteString getData() { return data; } diff --git a/topic/src/main/java/tech/ydb/topic/write/impl/SentMessage.java b/topic/src/main/java/tech/ydb/topic/write/impl/SentMessage.java index 53b840062..88c73699f 100644 --- a/topic/src/main/java/tech/ydb/topic/write/impl/SentMessage.java +++ b/topic/src/main/java/tech/ydb/topic/write/impl/SentMessage.java @@ -17,13 +17,21 @@ public class SentMessage implements WriterQueue.EncodedMsg { private final long seqNo; private final long bufferSize; + private final long sendTimestamp; + private final long acceptedTimestamp; private final YdbTransaction tx; private final YdbTopic.StreamWriteMessage.WriteRequest.MessageData pb; private final CompletableFuture ackFuture; public SentMessage(EnqueuedMessage msg, long seqNo) { + this(msg, seqNo, 0); + } + + SentMessage(EnqueuedMessage msg, long seqNo, long sendTimestamp) { this.seqNo = seqNo; this.bufferSize = msg.getBufferSize(); + this.sendTimestamp = sendTimestamp; + this.acceptedTimestamp = msg.getAcceptedTimestamp(); this.tx = msg.getMeta().getTransaction(); this.pb = buildPb(seqNo, msg); this.ackFuture = msg.getAckFuture(); @@ -42,6 +50,14 @@ public long getBufferSize() { return bufferSize; } + long getSendTimestamp() { + return sendTimestamp; + } + + long getAcceptedTimestamp() { + return acceptedTimestamp; + } + @Override public void confirm(WriteAck ack) { this.ackFuture.complete(ack); diff --git a/topic/src/main/java/tech/ydb/topic/write/impl/WriterMetrics.java b/topic/src/main/java/tech/ydb/topic/write/impl/WriterMetrics.java index 3fc70f6da..51882cf5d 100644 --- a/topic/src/main/java/tech/ydb/topic/write/impl/WriterMetrics.java +++ b/topic/src/main/java/tech/ydb/topic/write/impl/WriterMetrics.java @@ -6,6 +6,7 @@ import tech.ydb.core.Status; import tech.ydb.core.StatusCode; import tech.ydb.core.metrics.Attr; +import tech.ydb.core.metrics.DoubleHistogram; import tech.ydb.core.metrics.LongCounter; import tech.ydb.core.metrics.Meter; import tech.ydb.core.metrics.MetricRegistration; @@ -20,13 +21,18 @@ final class WriterMetrics { private final LongCounter sendingBytes; private final LongCounter writtenMessages; private final LongCounter sessionErrors; + private final DoubleHistogram messageAckDuration; + private final DoubleHistogram bufferWaitDuration; private final Attr[] commonAttributes; private final Meter meter; + private final boolean enabled; private MetricRegistration bufferUsedGauge = MetricRegistration.NOOP; private MetricRegistration bufferLimitGauge = MetricRegistration.NOOP; + private MetricRegistration oldestAgeGauge = MetricRegistration.NOOP; WriterMetrics(Meter meter, String topic, String writerName) { this.meter = meter; + this.enabled = meter != Meter.NOOP; this.sendingMessages = meter.createCounter("ydb.topic.writer.sending.messages", MESSAGE_UNIT, "The number of messages accepted by the SDK for sending."); this.sendingBytes = meter.createCounter("ydb.topic.writer.sending.bytes", "By", @@ -35,6 +41,10 @@ final class WriterMetrics { "The number of messages confirmed written by the server, including already written messages."); this.sessionErrors = meter.createCounter("ydb.topic.writer.session.errors", "{error}", "The number of writer stream session errors by retry decision."); + this.messageAckDuration = meter.createHistogram("ydb.topic.writer.message.ack.duration", "s", + "Time from first sending a message to the server to its acknowledgement."); + this.bufferWaitDuration = meter.createHistogram("ydb.topic.writer.buffer.wait.duration", "s", + "Time waiting for buffer capacity before a message is accepted."); this.commonAttributes = writerName == null ? new Attr[0] : new Attr[]{Attr.of("topic", topic), Attr.of("writer.name", writerName)}; @@ -49,11 +59,42 @@ void register(LongSupplier bufferUsed, LongSupplier bufferLimit) { m -> m.record(bufferLimit.getAsLong(), commonAttributes)); } + void register(LongSupplier oldestTimestamp) { + oldestAgeGauge = meter.registerDoubleGauge("ydb.topic.writer.sending.oldest_age", "s", + "The age of the oldest message in the writer's in-flight buffer.", m -> { + long timestamp = oldestTimestamp.getAsLong(); + m.record(timestamp == 0 ? 0 : (System.nanoTime() - timestamp) / 1_000_000_000d, + commonAttributes); + }); + } + void unregister() { bufferUsedGauge.close(); bufferLimitGauge.close(); + oldestAgeGauge.close(); bufferUsedGauge = MetricRegistration.NOOP; bufferLimitGauge = MetricRegistration.NOOP; + oldestAgeGauge = MetricRegistration.NOOP; + } + + long reportMessageSendStart() { + return enabled ? System.nanoTime() : 0; + } + + void reportMessageAckDuration(long timestamp) { + if (timestamp != 0) { + messageAckDuration.record((System.nanoTime() - timestamp) / 1_000_000_000d, commonAttributes); + } + } + + long reportBufferWaitStart() { + return enabled ? System.nanoTime() : 0; + } + + void reportBufferWaitDuration(long timestamp) { + if (timestamp != 0) { + bufferWaitDuration.record((System.nanoTime() - timestamp) / 1_000_000_000d, commonAttributes); + } } void reportSending(long bytes) { diff --git a/topic/src/main/java/tech/ydb/topic/write/impl/WriterQueue.java b/topic/src/main/java/tech/ydb/topic/write/impl/WriterQueue.java index f08a91034..ab5307e74 100644 --- a/topic/src/main/java/tech/ydb/topic/write/impl/WriterQueue.java +++ b/topic/src/main/java/tech/ydb/topic/write/impl/WriterQueue.java @@ -81,6 +81,8 @@ public WriterQueue(String debugId, WriterSettings settings, WriterMetrics metric this.readyNotify = readyNotify; this.metrics = metrics; this.metrics.register(buffer::getUsedSize, buffer::getMaxSize); + this.metrics.register(() -> sent.stream().map(EncodedMsg::getSentMessage) + .filter(msg -> msg != null).mapToLong(SentMessage::getAcceptedTimestamp).findFirst().orElse(0)); } CompletableFuture flush() { @@ -121,7 +123,7 @@ SentMessage nextMessageToSend() { } lastSeqNo = actualSeqNo; - SentMessage sentMsg = new SentMessage(next, actualSeqNo); + SentMessage sentMsg = new SentMessage(next, actualSeqNo, metrics.reportMessageSendStart()); logger.trace("[{}] prepare sent message with seqNo {}", debugId, actualSeqNo); sent.offer(sentMsg); return sentMsg; @@ -159,6 +161,7 @@ void confirmAck(WriteAck ack) { sentIt.remove(); buffer.releaseMessage(sentMsg.getBufferSize()); metrics.reportWritten(); + metrics.reportMessageAckDuration(sentMsg.getSendTimestamp()); msg.confirm(ack); } } @@ -206,7 +209,7 @@ List updateSeqNo(long newSeqNo) { CompletableFuture enqueue(Message message, YdbTransaction tx) throws QueueOverflowException, InterruptedException { long reservedSizeBytes = reservationSizeBytes(message.getData().length); - buffer.acquire(reservedSizeBytes); + buffer.acquire(reservedSizeBytes, metrics); return accept(message, tx, reservedSizeBytes); } @@ -219,7 +222,7 @@ CompletableFuture tryEnqueue(Message message, YdbTransaction tx) throw CompletableFuture tryEnqueue(Message message, YdbTransaction tx, long timeout, TimeUnit unit) throws QueueOverflowException, InterruptedException, TimeoutException { long reservedSizeBytes = reservationSizeBytes(message.getData().length); - buffer.tryAcquire(reservedSizeBytes, timeout, unit); + buffer.tryAcquire(reservedSizeBytes, timeout, unit, metrics); return accept(message, tx, reservedSizeBytes); } @@ -232,7 +235,8 @@ private long reservationSizeBytes(int inputSizeBytes) { } private CompletableFuture accept(Message message, YdbTransaction tx, long reservedSizeBytes) { - EnqueuedMessage msg = new EnqueuedMessage(new MessageMeta(message, tx), reservedSizeBytes); + EnqueuedMessage msg = new EnqueuedMessage(new MessageMeta(message, tx), reservedSizeBytes, + metrics.reportMessageSendStart()); lastAcceptedAckFuture = msg.getAckFuture(); queue.add(msg); metrics.reportSending(message.getData().length); From 96d73143bc6637474d703d301893b4898a25c681 Mon Sep 17 00:00:00 2001 From: KirillKurdyukov Date: Mon, 19 Oct 2026 13:39:39 +0300 Subject: [PATCH 2/2] Simplify Topic Writer metric timing and registration --- .../ydb/topic/write/impl/BufferManager.java | 49 ++----------------- .../ydb/topic/write/impl/EnqueuedMessage.java | 7 +-- .../ydb/topic/write/impl/SentMessage.java | 7 +-- .../ydb/topic/write/impl/WriterMetrics.java | 17 ++----- .../ydb/topic/write/impl/WriterQueue.java | 27 +++++++--- 5 files changed, 28 insertions(+), 79 deletions(-) diff --git a/topic/src/main/java/tech/ydb/topic/write/impl/BufferManager.java b/topic/src/main/java/tech/ydb/topic/write/impl/BufferManager.java index 4227f6b2d..d8a4a21a4 100644 --- a/topic/src/main/java/tech/ydb/topic/write/impl/BufferManager.java +++ b/topic/src/main/java/tech/ydb/topic/write/impl/BufferManager.java @@ -60,21 +60,11 @@ public void close(Status status) { } public void acquire(long messageSize) throws InterruptedException, QueueOverflowException { - acquire(messageSize, null); - } - - void acquire(long messageSize, WriterMetrics metrics) throws InterruptedException, QueueOverflowException { if (closed != null) { throw new IllegalStateException("Writer was closed with status " + closed); } - long waitStarted = 0; - if (!countAvailable.tryAcquire(0, TimeUnit.NANOSECONDS)) { - if (metrics != null) { - waitStarted = metrics.reportBufferWaitStart(); - } - countAvailable.acquire(); - } + countAvailable.acquire(); if (closed != null) { countAvailable.release(); @@ -84,12 +74,7 @@ void acquire(long messageSize, WriterMetrics metrics) throws InterruptedExceptio int messageBlocks = calculateBlocksCount(messageSize, blockBitsCount); try { - if (!blocksAvailable.tryAcquire(messageBlocks, 0, TimeUnit.NANOSECONDS)) { - if (waitStarted == 0 && metrics != null) { - waitStarted = metrics.reportBufferWaitStart(); - } - blocksAvailable.acquire(messageBlocks); - } + blocksAvailable.acquire(messageBlocks); } catch (InterruptedException ex) { countAvailable.release(); throw ex; @@ -100,9 +85,6 @@ void acquire(long messageSize, WriterMetrics metrics) throws InterruptedExceptio countAvailable.release(); throw new IllegalStateException("Writer was closed with status " + closed); } - if (metrics != null) { - metrics.reportBufferWaitDuration(waitStarted); - } } public void tryAcquire(long messageSize) throws QueueOverflowException { @@ -143,25 +125,12 @@ public void tryAcquire(long messageSize) throws QueueOverflowException { public void tryAcquire(long messageSize, long timeout, TimeUnit unit) throws InterruptedException, QueueOverflowException, TimeoutException { - tryAcquire(messageSize, timeout, unit, null); - } - - void tryAcquire(long messageSize, long timeout, TimeUnit unit, WriterMetrics metrics) throws InterruptedException, - QueueOverflowException, TimeoutException { if (closed != null) { throw new IllegalStateException("Writer was closed with status " + closed); } long expireAt = System.nanoTime() + unit.toNanos(timeout); - long waitStarted = 0; - boolean countAcquired = countAvailable.tryAcquire(0, TimeUnit.NANOSECONDS); - if (!countAcquired) { - if (metrics != null) { - waitStarted = metrics.reportBufferWaitStart(); - } - countAcquired = countAvailable.tryAcquire(timeout, unit); - } - if (!countAcquired) { + if (!countAvailable.tryAcquire(timeout, unit)) { String errorMsg = "[" + debugId + "] Rejecting a message due to reaching message queue in-flight limit of " + maxCount; logger.warn(errorMsg); @@ -178,14 +147,7 @@ void tryAcquire(long messageSize, long timeout, TimeUnit unit, WriterMetrics met try { // negative timeout is allowed for tryAcquire long timeout2 = expireAt - System.nanoTime(); - boolean blocksAcquired = blocksAvailable.tryAcquire(messageBlocks, 0, TimeUnit.NANOSECONDS); - if (!blocksAcquired) { - if (waitStarted == 0 && metrics != null) { - waitStarted = metrics.reportBufferWaitStart(); - } - blocksAcquired = blocksAvailable.tryAcquire(messageBlocks, timeout2, TimeUnit.NANOSECONDS); - } - if (!blocksAcquired) { + if (!blocksAvailable.tryAcquire(messageBlocks, timeout2, TimeUnit.NANOSECONDS)) { countAvailable.release(); int count = maxCount - countAvailable.availablePermits(); long size = ((long) blocksAvailable.availablePermits()) << blockBitsCount; @@ -205,9 +167,6 @@ void tryAcquire(long messageSize, long timeout, TimeUnit unit, WriterMetrics met countAvailable.release(); throw new IllegalStateException("Writer was closed with status " + closed); } - if (metrics != null) { - metrics.reportBufferWaitDuration(waitStarted); - } } public void releaseMessage(long messageSize) { diff --git a/topic/src/main/java/tech/ydb/topic/write/impl/EnqueuedMessage.java b/topic/src/main/java/tech/ydb/topic/write/impl/EnqueuedMessage.java index 38a75652b..c7edfc95d 100644 --- a/topic/src/main/java/tech/ydb/topic/write/impl/EnqueuedMessage.java +++ b/topic/src/main/java/tech/ydb/topic/write/impl/EnqueuedMessage.java @@ -12,7 +12,7 @@ */ public class EnqueuedMessage { private final MessageMeta meta; - private final long acceptedTimestamp; + private final long acceptedTimestamp = System.nanoTime(); private final CompletableFuture ackFuture = new CompletableFuture<>(); private volatile ByteString data = null; @@ -21,12 +21,7 @@ public class EnqueuedMessage { private volatile boolean isReady = false; public EnqueuedMessage(MessageMeta meta, long bufferSize) { - this(meta, bufferSize, 0); - } - - EnqueuedMessage(MessageMeta meta, long bufferSize, long acceptedTimestamp) { this.meta = meta; - this.acceptedTimestamp = acceptedTimestamp; this.data = null; this.bufferSize = bufferSize; } diff --git a/topic/src/main/java/tech/ydb/topic/write/impl/SentMessage.java b/topic/src/main/java/tech/ydb/topic/write/impl/SentMessage.java index 88c73699f..c96bba406 100644 --- a/topic/src/main/java/tech/ydb/topic/write/impl/SentMessage.java +++ b/topic/src/main/java/tech/ydb/topic/write/impl/SentMessage.java @@ -17,20 +17,15 @@ public class SentMessage implements WriterQueue.EncodedMsg { private final long seqNo; private final long bufferSize; - private final long sendTimestamp; + private final long sendTimestamp = System.nanoTime(); private final long acceptedTimestamp; private final YdbTransaction tx; private final YdbTopic.StreamWriteMessage.WriteRequest.MessageData pb; private final CompletableFuture ackFuture; public SentMessage(EnqueuedMessage msg, long seqNo) { - this(msg, seqNo, 0); - } - - SentMessage(EnqueuedMessage msg, long seqNo, long sendTimestamp) { this.seqNo = seqNo; this.bufferSize = msg.getBufferSize(); - this.sendTimestamp = sendTimestamp; this.acceptedTimestamp = msg.getAcceptedTimestamp(); this.tx = msg.getMeta().getTransaction(); this.pb = buildPb(seqNo, msg); diff --git a/topic/src/main/java/tech/ydb/topic/write/impl/WriterMetrics.java b/topic/src/main/java/tech/ydb/topic/write/impl/WriterMetrics.java index 51882cf5d..29cfbc8a1 100644 --- a/topic/src/main/java/tech/ydb/topic/write/impl/WriterMetrics.java +++ b/topic/src/main/java/tech/ydb/topic/write/impl/WriterMetrics.java @@ -50,16 +50,13 @@ final class WriterMetrics { : new Attr[]{Attr.of("topic", topic), Attr.of("writer.name", writerName)}; } - void register(LongSupplier bufferUsed, LongSupplier bufferLimit) { + void register(LongSupplier bufferUsed, LongSupplier bufferLimit, LongSupplier oldestTimestamp) { bufferUsedGauge = meter.registerLongGauge("ydb.topic.writer.buffer.used.bytes", "By", "The occupied budget of the writer buffer limiter.", m -> m.record(bufferUsed.getAsLong(), commonAttributes)); bufferLimitGauge = meter.registerLongGauge("ydb.topic.writer.buffer.limit.bytes", "By", "The configured limit of the writer buffer limiter.", m -> m.record(bufferLimit.getAsLong(), commonAttributes)); - } - - void register(LongSupplier oldestTimestamp) { oldestAgeGauge = meter.registerDoubleGauge("ydb.topic.writer.sending.oldest_age", "s", "The age of the oldest message in the writer's in-flight buffer.", m -> { long timestamp = oldestTimestamp.getAsLong(); @@ -77,22 +74,14 @@ void unregister() { oldestAgeGauge = MetricRegistration.NOOP; } - long reportMessageSendStart() { - return enabled ? System.nanoTime() : 0; - } - void reportMessageAckDuration(long timestamp) { - if (timestamp != 0) { + if (enabled) { messageAckDuration.record((System.nanoTime() - timestamp) / 1_000_000_000d, commonAttributes); } } - long reportBufferWaitStart() { - return enabled ? System.nanoTime() : 0; - } - void reportBufferWaitDuration(long timestamp) { - if (timestamp != 0) { + if (enabled) { bufferWaitDuration.record((System.nanoTime() - timestamp) / 1_000_000_000d, commonAttributes); } } diff --git a/topic/src/main/java/tech/ydb/topic/write/impl/WriterQueue.java b/topic/src/main/java/tech/ydb/topic/write/impl/WriterQueue.java index ab5307e74..e0c918e5c 100644 --- a/topic/src/main/java/tech/ydb/topic/write/impl/WriterQueue.java +++ b/topic/src/main/java/tech/ydb/topic/write/impl/WriterQueue.java @@ -80,9 +80,17 @@ public WriterQueue(String debugId, WriterSettings settings, WriterMetrics metric this.compressionExecutor = compressionExecutor; this.readyNotify = readyNotify; this.metrics = metrics; - this.metrics.register(buffer::getUsedSize, buffer::getMaxSize); - this.metrics.register(() -> sent.stream().map(EncodedMsg::getSentMessage) - .filter(msg -> msg != null).mapToLong(SentMessage::getAcceptedTimestamp).findFirst().orElse(0)); + this.metrics.register(buffer::getUsedSize, buffer::getMaxSize, this::getOlderAcceptedTimestamp); + } + + private long getOlderAcceptedTimestamp() { + for (EncodedMsg msg : sent) { + SentMessage sentMsg = msg.getSentMessage(); + if (sentMsg != null) { + return sentMsg.getAcceptedTimestamp(); + } + } + return 0; } CompletableFuture flush() { @@ -123,7 +131,7 @@ SentMessage nextMessageToSend() { } lastSeqNo = actualSeqNo; - SentMessage sentMsg = new SentMessage(next, actualSeqNo, metrics.reportMessageSendStart()); + SentMessage sentMsg = new SentMessage(next, actualSeqNo); logger.trace("[{}] prepare sent message with seqNo {}", debugId, actualSeqNo); sent.offer(sentMsg); return sentMsg; @@ -209,7 +217,9 @@ List updateSeqNo(long newSeqNo) { CompletableFuture enqueue(Message message, YdbTransaction tx) throws QueueOverflowException, InterruptedException { long reservedSizeBytes = reservationSizeBytes(message.getData().length); - buffer.acquire(reservedSizeBytes, metrics); + long waitStarted = System.nanoTime(); + buffer.acquire(reservedSizeBytes); + metrics.reportBufferWaitDuration(waitStarted); return accept(message, tx, reservedSizeBytes); } @@ -222,7 +232,9 @@ CompletableFuture tryEnqueue(Message message, YdbTransaction tx) throw CompletableFuture tryEnqueue(Message message, YdbTransaction tx, long timeout, TimeUnit unit) throws QueueOverflowException, InterruptedException, TimeoutException { long reservedSizeBytes = reservationSizeBytes(message.getData().length); - buffer.tryAcquire(reservedSizeBytes, timeout, unit, metrics); + long waitStarted = System.nanoTime(); + buffer.tryAcquire(reservedSizeBytes, timeout, unit); + metrics.reportBufferWaitDuration(waitStarted); return accept(message, tx, reservedSizeBytes); } @@ -235,8 +247,7 @@ private long reservationSizeBytes(int inputSizeBytes) { } private CompletableFuture accept(Message message, YdbTransaction tx, long reservedSizeBytes) { - EnqueuedMessage msg = new EnqueuedMessage(new MessageMeta(message, tx), reservedSizeBytes, - metrics.reportMessageSendStart()); + EnqueuedMessage msg = new EnqueuedMessage(new MessageMeta(message, tx), reservedSizeBytes); lastAcceptedAckFuture = msg.getAckFuture(); queue.add(msg); metrics.reportSending(message.getData().length);