Merge branch 'test-297' into 297-stop-partition

Author: rekby

tests/conftest.py

@@ -176,6 +176,26 @@ async def topic2_path(driver, topic_consumer, database) -> str:
     return topic_path
 
 
+@pytest.fixture()
+@pytest.mark.asyncio()
+async def topic_with_two_partitions_path(driver, topic_consumer, database) -> str:
+    topic_path = database + "/test-topic-two-partitions"
+
+    try:
+        await driver.topic_client.drop_topic(topic_path)
+    except issues.SchemeError:
+        pass
+
+    await driver.topic_client.create_topic(
+        path=topic_path,
+        consumers=[topic_consumer],
+        min_active_partitions=2,
+        partition_count_limit=2,
+    )
+
+    return topic_path
+
+
 @pytest.fixture()
 @pytest.mark.asyncio()
 async def topic_with_messages(driver, topic_consumer, database):

tests/topics/test_topic_reader.py

@@ -1,3 +1,5 @@
+import asyncio
+
 import pytest
 
 import ydb
@@ -161,3 +163,44 @@ def decode(b: bytes):
         with driver_sync.topic_client.reader(topic_path, topic_consumer, decoders={codec: decode}) as reader:
             batch = reader.receive_batch()
             assert batch.messages[0].data.decode() == "123"
+
+
+@pytest.mark.asyncio
+class TestBugFixesAsync:
+    @pytest.mark.skip("LOGBROKER-8319")
+    async def test_issue_297_bad_handle_stop_partition(self, driver, topic_consumer, topic_with_two_partitions_path: str):
+
+        async def wait(fut):
+            return await asyncio.wait_for(fut, timeout=10)
+
+        topic = topic_with_two_partitions_path  # type: str
+
+        async with driver.topic_client.writer(topic, partition_id=0) as writer:
+            await writer.write_with_ack("01")
+
+        async with driver.topic_client.writer(topic, partition_id=1) as writer:
+            await writer.write_with_ack("1")
+
+        # Start first reader and receive messages from both partitions
+        reader0 = driver.topic_client.reader(topic, consumer=topic_consumer)
+        await wait(reader0.receive_message())
+        await wait(reader0.receive_message())
+
+        # Start second reader for same topic, same consumer, partition 1
+        reader1 = driver.topic_client.reader(ydb.TopicReaderSelector(
+            path=topic,
+            partitions=1,
+        ), consumer=topic_consumer)
+
+        await asyncio.sleep(0.1)
+
+        # receive uncommited message from partition 1
+        msg = await wait(reader1.receive_message())
+        assert msg.data.decode() == "1"
+
+        # write message to partition 0 - for reader 0
+        # async with driver.topic_client.writer(topic, partition_id=0) as writer:
+        #     await writer.write_with_ack("02")
+
+        msg = await wait(reader0.receive_message())
+        assert msg.data.decode() == "02"

ydb/_topic_reader/topic_reader_asyncio.py

@@ -451,7 +451,7 @@ async def _read_messages_loop(self):
                 self._state_changed.set()
         except Exception as e:
             self._set_first_error(e)
-            raise
+            return
 
     async def _update_token_loop(self):
         while True: