Return topic desctuctors

Author: vgvoleg

tests/topics/test_topic_reader.py

@@ -17,8 +17,8 @@ async def test_read_batch(self, driver, topic_with_messages, topic_consumer):
         await reader.close()
 
     async def test_link_to_client(self, driver, topic_path, topic_consumer):
-        reader = driver.topic_client.reader(topic_path, topic_consumer)
-        assert reader._parent is driver.topic_client
+        async with driver.topic_client.reader(topic_path, topic_consumer) as reader:
+            assert reader._parent is driver.topic_client
 
     async def test_read_message(self, driver, topic_with_messages, topic_consumer):
         reader = driver.topic_client.reader(topic_with_messages, topic_consumer)
@@ -138,8 +138,8 @@ def test_read_batch(self, driver_sync, topic_with_messages, topic_consumer):
         reader.close()
 
     def test_link_to_client(self, driver_sync, topic_path, topic_consumer):
-        reader = driver_sync.topic_client.reader(topic_path, topic_consumer)
-        assert reader._parent is driver_sync.topic_client
+        with driver_sync.topic_client.reader(topic_path, topic_consumer) as reader:
+            assert reader._parent is driver_sync.topic_client
 
     def test_read_message(self, driver_sync, topic_with_messages, topic_consumer):
         reader = driver_sync.topic_client.reader(topic_with_messages, topic_consumer)

tests/topics/test_topic_writer.py

@@ -37,8 +37,8 @@ async def test_wait_last_seqno(self, driver: ydb.aio.Driver, topic_path):
             assert init_info.last_seqno == 5
 
     async def test_link_to_client(self, driver, topic_path, topic_consumer):
-        writer = driver.topic_client.writer(topic_path)
-        assert writer._parent is driver.topic_client
+        async with driver.topic_client.writer(topic_path) as writer:
+            assert writer._parent is driver.topic_client
 
     async def test_random_producer_id(self, driver: ydb.aio.Driver, topic_path, topic_reader: ydb.TopicReaderAsyncIO):
         async with driver.topic_client.writer(topic_path) as writer:
@@ -180,8 +180,8 @@ def test_auto_flush_on_close(self, driver_sync: ydb.Driver, topic_path):
             assert init_info.last_seqno == last_seqno
 
     def test_link_to_client(self, driver_sync, topic_path, topic_consumer):
-        writer = driver_sync.topic_client.writer(topic_path)
-        assert writer._parent is driver_sync.topic_client
+        with driver_sync.topic_client.writer(topic_path) as writer:
+            assert writer._parent is driver_sync.topic_client
 
     def test_random_producer_id(
         self,

ydb/_grpc/grpcwrapper/common_utils.py

@@ -161,6 +161,9 @@ def __init__(self, convert_server_grpc_to_wrapper):
         self._stream_call = None
         self._wait_executor = None
 
+    def __del__(self):
+        self._clean_executor(wait=False)
+
     async def start(self, driver: SupportedDriverType, stub, method):
         if asyncio.iscoroutinefunction(driver.__call__):
             await self._start_asyncio_driver(driver, stub, method)

ydb/_topic_reader/topic_reader_asyncio.py

@@ -96,6 +96,8 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
     def __del__(self):
         if not self._closed:
             logger.warning("Topic reader was not closed properly. Consider using method close().")
+            task = self._loop.create_task(self.close(flush=False))
+            topic_common.wrap_set_name_for_asyncio_task(task, task_name="close reader")
 
     async def wait_message(self):
         """

ydb/_topic_reader/topic_reader_sync.py

@@ -60,6 +60,7 @@ async def create_reader():
     def __del__(self):
         if not self._closed:
             logger.warning("Topic reader was not closed properly. Consider using method close().")
+            self.close(flush=False)
 
     def __enter__(self):
         return self

ydb/_topic_writer/topic_writer_asyncio.py

@@ -1,6 +1,7 @@
 import asyncio
 import concurrent.futures
 import datetime
+import functools
 import gzip
 import typing
 from collections import deque
@@ -79,8 +80,11 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
                 raise
 
     def __del__(self):
-        if not self._closed:
-            logger.warning("Topic writer was not closed properly. Consider using method close().")
+        if self._closed or self._loop.is_closed():
+            return
+
+        logger.warning("Topic writer was not closed properly. Consider using method close().")
+        self._loop.call_soon(functools.partial(self.close, flush=False))
 
     async def close(self, *, flush: bool = True):
         if self._closed:

ydb/_topic_writer/topic_writer_sync.py

@@ -76,6 +76,7 @@ def __exit__(self, exc_type, exc_val, exc_tb):
     def __del__(self):
         if not self._closed:
             logger.warning("Topic writer was not closed properly. Consider using method close().")
+            self.close(flush=False)
 
     def close(self, *, flush: bool = True, timeout: TimeoutType = None):
         if self._closed:

ydb/topic.py

@@ -117,6 +117,7 @@ def __init__(self, driver: aio.Driver, settings: Optional[TopicClientSettings] =
     def __del__(self):
         if not self._closed:
             logger.warning("Topic client was not closed properly. Consider using method close().")
+            self.close()
 
     async def create_topic(
         self,
@@ -349,6 +350,7 @@ def __init__(self, driver: driver.Driver, settings: Optional[TopicClientSettings
     def __del__(self):
         if not self._closed:
             logger.warning("Topic client was not closed properly. Consider using method close().")
+            self.close()
 
     def create_topic(
         self,