Change topics __del__ to warn if not closed

Author: vgvoleg

tests/topics/test_topic_reader.py

@@ -174,12 +174,13 @@ def test_read_and_commit_with_close_reader(self, driver_sync, topic_with_message
             assert message != message2
 
     def test_read_and_commit_with_ack(self, driver_sync, topic_with_messages, topic_consumer):
-        reader = driver_sync.topic_client.reader(topic_with_messages, topic_consumer)
-        message = reader.receive_message()
-        reader.commit_with_ack(message)
+        with driver_sync.topic_client.reader(topic_with_messages, topic_consumer) as reader:
+            message = reader.receive_message()
+            reader.commit_with_ack(message)
+
+        with driver_sync.topic_client.reader(topic_with_messages, topic_consumer) as reader:
+            batch = reader.receive_batch()
 
-        reader = driver_sync.topic_client.reader(topic_with_messages, topic_consumer)
-        batch = reader.receive_batch()
         assert message != batch.messages[0]
 
     def test_read_compressed_messages(self, driver_sync, topic_path, topic_consumer):
@@ -247,3 +248,6 @@ async def wait(fut):
         datas.sort()
 
         assert datas == ["10", "11"]
+
+        await reader0.close()
+        await reader1.close()

ydb/_grpc/grpcwrapper/common_utils.py

@@ -161,9 +161,6 @@ 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

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

ydb/_topic_reader/topic_reader_sync.py

@@ -1,5 +1,6 @@
 import asyncio
 import concurrent.futures
+import logging
 import typing
 from typing import List, Union, Optional
 
@@ -23,6 +24,8 @@
 if typing.TYPE_CHECKING:
     from ..query.transaction import BaseQueryTxContext
 
+logger = logging.getLogger(__name__)
+
 
 class TopicReaderSync:
     _caller: CallFromSyncToAsync
@@ -55,7 +58,8 @@ async def create_reader():
         self._parent = _parent
 
     def __del__(self):
-        self.close(flush=False)
+        if not self._closed:
+            logger.warning("Topic reader was not closed properly. Consider using method close().")
 
     def __enter__(self):
         return self

ydb/_topic_writer/topic_writer_asyncio.py

@@ -80,10 +80,8 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
                 raise
 
     def __del__(self):
-        if self._closed or self._loop.is_closed():
-            return
-
-        self._loop.call_soon(functools.partial(self.close, flush=False))
+        if not self._closed:
+            logger.warning("Topic writer was not closed properly. Consider using method close().")
 
     async def close(self, *, flush: bool = True):
         if self._closed:

ydb/_topic_writer/topic_writer_sync.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import logging
 import typing
 from concurrent.futures import Future
 from typing import Union, List, Optional
@@ -29,6 +30,8 @@
 if typing.TYPE_CHECKING:
     from ..query.transaction import BaseQueryTxContext
 
+logger = logging.getLogger(__name__)
+
 
 class WriterSync:
     _caller: CallFromSyncToAsync
@@ -71,7 +74,8 @@ def __exit__(self, exc_type, exc_val, exc_tb):
                 raise
 
     def __del__(self):
-        self.close(flush=False)
+        if not self._closed:
+            logger.warning("Topic writer was not closed properly. Consider using method close().")
 
     def close(self, *, flush: bool = True, timeout: TimeoutType = None):
         if self._closed:

ydb/aio/driver.py

@@ -62,4 +62,5 @@ def __init__(
 
     async def stop(self, timeout=10):
         await self.table_client._stop_pool_if_needed(timeout=timeout)
+        self.topic_client.close()
         await super().stop(timeout=timeout)

ydb/driver.py

@@ -288,4 +288,5 @@ def __init__(
 
     def stop(self, timeout=10):
         self.table_client._stop_pool_if_needed(timeout=timeout)
+        self.topic_client.close()
         super().stop(timeout=timeout)

ydb/topic.py

@@ -35,6 +35,7 @@
 import concurrent.futures
 import datetime
 from dataclasses import dataclass
+import logging
 from typing import List, Union, Mapping, Optional, Dict, Callable
 
 from . import aio, Credentials, _apis, issues
@@ -92,6 +93,8 @@
     PublicAlterAutoPartitioningSettings as TopicAlterAutoPartitioningSettings,
 )
 
+logger = logging.getLogger(__name__)
+
 
 class TopicClientAsyncIO:
     _closed: bool
@@ -112,7 +115,8 @@ def __init__(self, driver: aio.Driver, settings: Optional[TopicClientSettings] =
         )
 
     def __del__(self):
-        self.close()
+        if not self._closed:
+            logger.warning("Topic client was not closed properly. Consider using method close().")
 
     async def create_topic(
         self,
@@ -320,7 +324,7 @@ def _check_closed(self):
         if not self._closed:
             return
 
-        raise RuntimeError("Topic client closed")
+        raise issues.Error("Topic client closed")
 
 
 class TopicClient:
@@ -343,7 +347,8 @@ def __init__(self, driver: driver.Driver, settings: Optional[TopicClientSettings
         )
 
     def __del__(self):
-        self.close()
+        if not self._closed:
+            logger.warning("Topic client was not closed properly. Consider using method close().")
 
     def create_topic(
         self,
@@ -561,7 +566,7 @@ def _check_closed(self):
         if not self._closed:
             return
 
-        raise RuntimeError("Topic client closed")
+        raise issues.Error("Topic client closed")
 
 
 @dataclass