workaround specific partition bug LOGBROKER-8319

Author: rekby

tests/topics/test_topic_reader.py

@@ -167,44 +167,41 @@ def decode(b: bytes):
 
 @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
+            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")
+            await writer.write_with_ack("00")
 
         async with driver.topic_client.writer(topic, partition_id=1) as writer:
-            await writer.write_with_ack("1")
+            await writer.write_with_ack("01")
 
         # 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,
-        )
+        reader1 = driver.topic_client.reader(topic, consumer=topic_consumer)
 
-        await asyncio.sleep(0.1)
+        # receive uncommited message
+        await reader1.receive_message()
+
+        # write one message for every partition
+        async with driver.topic_client.writer(topic, partition_id=0) as writer:
+            await writer.write_with_ack("10")
+        async with driver.topic_client.writer(topic, partition_id=0) as writer:
+            await writer.write_with_ack("11")
 
-        # receive uncommited message from partition 1
-        msg = await wait(reader1.receive_message())
-        assert msg.data.decode() == "1"
+        msg0 = await wait(reader0.receive_message())
+        msg1 = await wait(reader1.receive_message())
 
-        # 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")
+        datas = [msg0.data.decode(), msg1.data.decode]
+        datas.sort()
 
-        msg = await wait(reader0.receive_message())
-        assert msg.data.decode() == "02"
+        assert datas == ["10", "11"]