fix usage internal codec

Author: rekby

tests/conftest.py

@@ -142,11 +142,10 @@ async def topic_path(driver, topic_consumer, database) -> str:
 @pytest.fixture()
 @pytest.mark.asyncio()
 async def topic_with_messages(driver, topic_path):
-    pass
     writer = driver.topic_client.topic_writer(
         topic_path, producer_and_message_group_id="fixture-producer-id"
     )
-    await writer.write_with_ack(
+    res = await writer.write_with_ack(
         ydb.TopicWriterMessage(data="123".encode()),
         ydb.TopicWriterMessage(data="456".encode()),
     )

ydb/_topic_writer/topic_writer.py

@@ -303,7 +303,7 @@ def messages_to_proto_requests(
         req = StreamWriteMessage.FromClient(
             StreamWriteMessage.WriteRequest(
                 messages=[msg.to_message_data()],
-                codec=Codec.CODEC_RAW.value,
+                codec=Codec.CODEC_RAW,
             )
         )
         res.append(req)

ydb/_topic_writer/topic_writer_asyncio.py

@@ -363,6 +363,8 @@ async def _send_loop(self, writer: "WriterAsyncIOStream"):
                 m = await self._new_messages.get()  # type: InternalMessage
                 if m.seq_no > last_seq_no:
                     writer.write([m])
+        except Exception as e:
+            await self._stop(e)
         finally:
             pass