Fix hang while write messages to closed driver.

Close #296

Author: rekby

tests/topics/test_topic_writer.py

@@ -1,4 +1,6 @@
 from __future__ import annotations
+
+import asyncio
 from typing import List
 
 import pytest
@@ -120,11 +122,12 @@ class TestException(Exception):
 
         with pytest.raises(TestException):
             async with driver.topic_client.writer(topic_path) as writer:
-                driver.stop()  # will raise exception on topic writer __exit__
-                try:
-                    writer.write("123")
-                except ydb.Error:
-                    pass
+                await writer.wait_init()
+                await driver.stop()  # will raise exception on topic writer __exit__
+
+                # ensure writer has exception internally
+                with pytest.raises((ydb.Error, asyncio.CancelledError)):
+                    await writer.write_with_ack("123")
 
                 raise TestException()
 
@@ -257,10 +260,11 @@ class TestException(Exception):
 
         with pytest.raises(TestException):
             with driver_sync.topic_client.writer(topic_path) as writer:
+                writer.wait_init()
                 driver_sync.stop()  # will raise exception on topic writer __exit__
-                try:
-                    writer.write("123")
-                except ydb.Error:
-                    pass
+
+                # ensure writer has exception internally
+                with pytest.raises(ydb.Error):
+                    writer.write_with_ack("123")
 
                 raise TestException()

ydb/_topic_writer/topic_writer_asyncio.py

@@ -65,7 +65,11 @@ async def __aenter__(self) -> "WriterAsyncIO":
         return self
 
     async def __aexit__(self, exc_type, exc_val, exc_tb):
-        await self.close()
+        try:
+            await self.close()
+        except BaseException as e:
+            if exc_val is None:
+                raise
 
     def __del__(self):
         if self._closed or self._loop.is_closed():

ydb/_topic_writer/topic_writer_sync.py

@@ -59,10 +59,9 @@ def __enter__(self):
     def __exit__(self, exc_type, exc_val, exc_tb):
         try:
             self.close()
-        except ydb.Error:
-            if exc_val:
-                raise exc_val
-            raise
+        except BaseException as e:
+            if exc_val is None:
+                raise
 
     def __del__(self):
         self.close(flush=False)

ydb/topic.py

@@ -337,7 +337,7 @@ def writer(
         encoders: Optional[Mapping[_ydb_topic_public_types.PublicCodec, Callable[[bytes], bytes]]] = None,
         encoder_executor: Optional[concurrent.futures.Executor] = None,  # default shared client executor pool
     ) -> TopicWriter:
-        args = locals().copy()
+        args = locals()
         del args["self"]
         self._check_closed()