From 8c05e0e7d6ddeb6b5dadbd6f74d124a98e315ec4 Mon Sep 17 00:00:00 2001 From: Daniil Zulin Date: Thu, 26 Sep 2024 22:54:44 +0700 Subject: [PATCH 1/2] Remove incomingMessage from the incomingQueue ony once --- topic/src/main/java/tech/ydb/topic/write/impl/WriterImpl.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/topic/src/main/java/tech/ydb/topic/write/impl/WriterImpl.java b/topic/src/main/java/tech/ydb/topic/write/impl/WriterImpl.java index 1c01d0f8..94218e11 100644 --- a/topic/src/main/java/tech/ydb/topic/write/impl/WriterImpl.java +++ b/topic/src/main/java/tech/ydb/topic/write/impl/WriterImpl.java @@ -291,8 +291,8 @@ private void free(int messageCount, long sizeBytes) { logger.trace("[{}] Putting a message into send buffer after freeing some space", id); if (incomingMessage.future.complete(null)) { acceptMessageIntoSendingQueue(incomingMessage.message); + incomingQueue.remove(); } - incomingQueue.remove(); } logger.trace("[{}] All messages from incomingQueue are accepted into send buffer", id); } From 2f92d0a3499388bc71889625f946e0023a7e14ec Mon Sep 17 00:00:00 2001 From: Daniil Zulin Date: Fri, 27 Sep 2024 16:18:25 +0700 Subject: [PATCH 2/2] Fix PR issue --- topic/src/main/java/tech/ydb/topic/write/impl/WriterImpl.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/topic/src/main/java/tech/ydb/topic/write/impl/WriterImpl.java b/topic/src/main/java/tech/ydb/topic/write/impl/WriterImpl.java index 94218e11..9674de49 100644 --- a/topic/src/main/java/tech/ydb/topic/write/impl/WriterImpl.java +++ b/topic/src/main/java/tech/ydb/topic/write/impl/WriterImpl.java @@ -289,9 +289,9 @@ private void free(int messageCount, long sizeBytes) { return; } logger.trace("[{}] Putting a message into send buffer after freeing some space", id); + incomingQueue.remove(); if (incomingMessage.future.complete(null)) { acceptMessageIntoSendingQueue(incomingMessage.message); - incomingQueue.remove(); } } logger.trace("[{}] All messages from incomingQueue are accepted into send buffer", id);