Return topic desctuctors

Author: vgvoleg

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: