From 0de33811b99cb50b03d688556cc68bc04d4efaf8 Mon Sep 17 00:00:00 2001
From: Igor Melnichenko <i-melnichenko@yandex-team.ru>
Date: Wed, 5 Aug 2026 01:43:26 +0300
Subject: [PATCH] Skip an empty read batch instead of dropping the rest of the
 response

addBatches returned as soon as it saw a batch with no messages, so every
batch after it in the same ReadResponse was silently discarded: the
messages never reached the user and the read futures of those batches
never completed, which also stopped ReadSession.onReadResponse from
requesting more data.

Skip the offending batch and keep processing the rest of the response.
---
 .../topic/read/impl/ReadPartitionSession.java |  2 +-
 .../read/impl/ReadPartitionSessionTest.java   | 72 +++++++++++++++++++
 2 files changed, 73 insertions(+), 1 deletion(-)
 create mode 100644 topic/src/test/java/tech/ydb/topic/read/impl/ReadPartitionSessionTest.java

diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionSession.java b/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionSession.java
index 0a055a0e3..3fd8099ff 100644
--- a/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionSession.java
+++ b/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionSession.java
@@ -92,7 +92,7 @@ public CompletableFuture<Void> addBatches(List<YdbTopic.StreamReadMessage.ReadRe
         for (YdbTopic.StreamReadMessage.ReadResponse.Batch batch: batchList) {
             if (batch.getMessageDataCount() == 0) {
                 logger.error("{} Received empty batch. This shouldn't happen", this);
-                return CompletableFuture.completedFuture(null);
+                continue;
             }
 
             BatchMeta meta = new BatchMeta(batch);
diff --git a/topic/src/test/java/tech/ydb/topic/read/impl/ReadPartitionSessionTest.java b/topic/src/test/java/tech/ydb/topic/read/impl/ReadPartitionSessionTest.java
new file mode 100644
index 000000000..fbab3da1d
--- /dev/null
+++ b/topic/src/test/java/tech/ydb/topic/read/impl/ReadPartitionSessionTest.java
@@ -0,0 +1,72 @@
+package tech.ydb.topic.read.impl;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+import com.google.protobuf.ByteString;
+import org.junit.Assert;
+import org.junit.Test;
+
+import tech.ydb.proto.topic.YdbTopic;
+import tech.ydb.topic.description.Codec;
+import tech.ydb.topic.description.CodecRegistry;
+import tech.ydb.topic.read.PartitionSession;
+import tech.ydb.topic.read.events.DataReceivedEvent;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class ReadPartitionSessionTest {
+    /**
+     * An empty batch is a server side anomaly, but it must not affect the batches that follow it in the
+     * same ReadResponse.
+     */
+    @Test(timeout = 30_000)
+    public void batchesAfterAnEmptyBatchAreStillRead() throws Exception {
+        ReadSession session = mock(ReadSession.class);
+        when(session.getMaxBatchSize()).thenReturn(0);
+        when(session.getMessageDecoder()).thenReturn(new MessageDecoder(1024, Runnable::run, new CodecRegistry()));
+
+        PartitionSession partition = new PartitionSession(1, 1, "/topic");
+        List<Long> delivered = Collections.synchronizedList(new ArrayList<>());
+
+        ReadPartitionSession reader = new ReadPartitionSession("test", session, partition, 0) {
+            @Override
+            CompletableFuture<Void> handleDataReceivedEvent(DataReceivedEvent event) {
+                event.getMessages().forEach(message -> delivered.add(message.getOffset()));
+                return CompletableFuture.completedFuture(null);
+            }
+        };
+
+        CompletableFuture<Void> batchesRead = reader.addBatches(
+                Arrays.asList(emptyProtoBatch(), rawProtoBatch(1), rawProtoBatch(2))
+        );
+
+        batchesRead.get();
+        Assert.assertEquals(Arrays.asList(1L, 2L), delivered);
+    }
+
+    private static YdbTopic.StreamReadMessage.ReadResponse.Batch emptyProtoBatch() {
+        return YdbTopic.StreamReadMessage.ReadResponse.Batch.newBuilder()
+                .setCodec(Codec.RAW)
+                .setProducerId("producer")
+                .build();
+    }
+
+    private static YdbTopic.StreamReadMessage.ReadResponse.Batch rawProtoBatch(long offset) {
+        return YdbTopic.StreamReadMessage.ReadResponse.Batch.newBuilder()
+                .setCodec(Codec.RAW)
+                .setProducerId("producer")
+                .addMessageData(
+                        YdbTopic.StreamReadMessage.ReadResponse.MessageData.newBuilder()
+                                .setOffset(offset)
+                                .setSeqNo(offset)
+                                .setData(ByteString.copyFromUtf8("data"))
+                                .build()
+                )
+                .build();
+    }
+}