bug: When trying to write to the Logbroker topic using TopicAPI from YDB Python SDK versions 3.3.1 and 3.3.2 with a stopped YDB driver, the program hangs completely if timeout is not specified, or throws an empty TimeoutError exception if it is specified

[pltnk]

Bug Report

YDB Python SDK version:

3.3.1 and 3.3.2

Environment

Python 3.9.6

Current behavior:

If you create a YDB driver, create a writer from it, then stop the driver, and then try to write something to the topic with the writer, in versions 3.3.1 and 3.3.2 of the YDB Python SDK, when using write_with_ack or async_write_with_ack methods to write, program execution hangs forever unless a timeout is specified.

If the timeout is specified, program execution stops with an empty TimeoutError exception, from which it is impossible to get information about the reason for stopping in order to use it for troubleshooting.

Traceback:

Traceback (most recent call last):
  File "/Users/pltnkv/code/ydb-python-sdk/venv/lib/python3.9/site-packages/ydb/_topic_common/common.py", line 83, in unsafe_call_with_result
    return f.result(timeout)
  File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/concurrent/futures/_base.py", line 447, in result
    raise TimeoutError()
concurrent.futures._base.TimeoutError
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
  File "/Applications/PyCharm.app/Contents/plugins/python/helpers/pydev/pydevconsole.py", line 364, in runcode
    coro = func()
  File "<input>", line 21, in <module>
  File "/Users/pltnkv/code/ydb-python-sdk/venv/lib/python3.9/site-packages/ydb/_topic_writer/topic_writer_sync.py", line 120, in write_with_ack
    return self._caller.unsafe_call_with_result(self._async_writer.write_with_ack(messages), timeout=timeout)
  File "/Users/pltnkv/code/ydb-python-sdk/venv/lib/python3.9/site-packages/ydb/_topic_common/common.py", line 85, in unsafe_call_with_result
    raise TimeoutError()
TimeoutError

Expected behavior:

In version 3.2.2 of the YDB Python SDK, an explicit RuntimeError("Can not write on closed stream.") exception was thrown immediately in this case, without having to wait. It is expected to have this behavior instead of the current one.

Traceback:

Traceback (most recent call last):
  File "/Applications/PyCharm.app/Contents/plugins/python/helpers/pydev/pydevconsole.py", line 364, in runcode
    coro = func()
  File "<input>", line 21, in <module>
  File "/Users/pltnkv/.pyenv/versions/billing/lib/python3.9/site-packages/ydb/_topic_writer/topic_writer_sync.py", line 120, in write_with_ack
    return self._caller.unsafe_call_with_result(self._async_writer.write_with_ack(messages), timeout=timeout)
  File "/Users/pltnkv/.pyenv/versions/billing/lib/python3.9/site-packages/ydb/_topic_common/common.py", line 83, in unsafe_call_with_result
    return f.result(timeout)
  File "/Users/pltnkv/.pyenv/versions/3.9.10/lib/python3.9/concurrent/futures/_base.py", line 446, in result
    return self.__get_result()
  File "/Users/pltnkv/.pyenv/versions/3.9.10/lib/python3.9/concurrent/futures/_base.py", line 391, in __get_result
    raise self._exception
  File "/Users/pltnkv/.pyenv/versions/billing/lib/python3.9/site-packages/ydb/_topic_writer/topic_writer_asyncio.py", line 101, in write_with_ack
    results = [f.result() for f in futures]
  File "/Users/pltnkv/.pyenv/versions/billing/lib/python3.9/site-packages/ydb/_topic_writer/topic_writer_asyncio.py", line 101, in <listcomp>
    results = [f.result() for f in futures]
  File "/Users/pltnkv/.pyenv/versions/billing/lib/python3.9/site-packages/ydb/_topic_writer/topic_writer_asyncio.py", line 535, in _send_loop
    writer.write([m])
  File "/Users/pltnkv/.pyenv/versions/billing/lib/python3.9/site-packages/ydb/_topic_writer/topic_writer_asyncio.py", line 657, in write
    raise RuntimeError("Can not write on closed stream.")
RuntimeError: Can not write on closed stream.

Steps to reproduce:

Run the snippet below.

Related code:

import ydb

TOPIC = "/local/test-topic"
TIMEOUT = 10

db = ydb.Driver(
    connection_string="grpc://localhost:2135?database=/local",
    credentials=ydb.credentials.AnonymousCredentials(),
)
db.wait(timeout=TIMEOUT)

db.topic_client.create_topic(path=TOPIC)

writer = db.topic_client.writer(topic=TOPIC, producer_id="test-producer")
init_info = writer.wait_init(timeout=TIMEOUT)

db.stop(timeout=TIMEOUT)

msg = ydb.TopicWriterMessage(data="test")

writer.write_with_ack(messages=msg, timeout=TIMEOUT)  # hangs forever if timeout is not specified

# future = writer.async_write_with_ack(messages=msg)
# future.result(timeout=TIMEOUT)  # hangs forever if timeout is not specified