diff --git a/CHANGELOG.md b/CHANGELOG.md
index c53465784..e370b4186 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,4 @@
+* Add optional topic reader names for identifying logical readers in server diagnostics and SDK logs
 * Deprecated the table client scan query methods — `TableClient.scan_query`, `TableClient.async_scan_query` and the async `ydb.aio.TableClient.scan_query`: they now emit a `DeprecationWarning` and keep working as before, use QueryService (`ydb.QuerySessionPool` / `ydb.aio.QuerySessionPool`) instead
 * Mark the package as typed so type checkers use the SDK's inline annotations
 
diff --git a/docs/topic.rst b/docs/topic.rst
index bd5455691..999d8733c 100644
--- a/docs/topic.rst
+++ b/docs/topic.rst
@@ -378,8 +378,13 @@ Reader Parameters
         consumer="my-consumer",
         buffer_size_bytes=50 * 1024 * 1024,  # client-side buffer (default: 50 MB)
         buffer_release_threshold=0.5,        # see below (default: 0.5)
+        reader_name="payments-worker",       # optional diagnostic name
     )
 
+``reader_name`` identifies a logical reader in server-side diagnostics and SDK logs. If it is
+omitted or empty, the SDK generates a process-local name in the form ``reader-N``. The name
+remains unchanged when the reader reconnects. Explicit names do not have to be unique.
+
 ``buffer_size_bytes`` controls how many bytes the server is allowed to send before the client
 signals that it is ready for more. The server will not exceed this limit.
 
diff --git a/examples/topic/reader_async_example.py b/examples/topic/reader_async_example.py
index 71e774555..464293c1a 100644
--- a/examples/topic/reader_async_example.py
+++ b/examples/topic/reader_async_example.py
@@ -9,7 +9,11 @@ async def connect():
         connection_string="grpc://localhost:2135?database=/local",
         credentials=ydb.credentials.AnonymousCredentials(),
     )
-    reader = db.topic_client.reader("/local/topic", consumer="consumer")
+    reader = db.topic_client.reader(
+        "/local/topic",
+        consumer="consumer",
+        reader_name="payments-worker",
+    )
     return reader
 
 
diff --git a/examples/topic/reader_example.py b/examples/topic/reader_example.py
index ab8157a9b..53caab4bd 100644
--- a/examples/topic/reader_example.py
+++ b/examples/topic/reader_example.py
@@ -8,7 +8,11 @@ def connect():
         connection_string="grpc://localhost:2135?database=/local",
         credentials=ydb.credentials.AnonymousCredentials(),
     )
-    reader = db.topic_client.reader("/local/topic", consumer="consumer")
+    reader = db.topic_client.reader(
+        "/local/topic",
+        consumer="consumer",
+        reader_name="payments-worker",
+    )
     return reader
 
 
diff --git a/ydb/_grpc/grpcwrapper/ydb_topic.py b/ydb/_grpc/grpcwrapper/ydb_topic.py
index 19b38bf01..0de3a7495 100644
--- a/ydb/_grpc/grpcwrapper/ydb_topic.py
+++ b/ydb/_grpc/grpcwrapper/ydb_topic.py
@@ -486,11 +486,14 @@ class InitRequest(IToProto):
         topics_read_settings: List["StreamReadMessage.InitRequest.TopicReadSettings"]
         consumer: Optional[str]
         auto_partitioning_support: bool
+        reader_name: Optional[str] = None
 
         def to_proto(self) -> ydb_topic_pb2.StreamReadMessage.InitRequest:
             res = ydb_topic_pb2.StreamReadMessage.InitRequest()
             if self.consumer is not None:
                 res.consumer = self.consumer
+            if self.reader_name is not None:
+                res.reader_name = self.reader_name
             for settings in self.topics_read_settings:
                 res.topics_read_settings.append(settings.to_proto())
             res.auto_partitioning_support = self.auto_partitioning_support
diff --git a/ydb/_grpc/grpcwrapper/ydb_topic_test.py b/ydb/_grpc/grpcwrapper/ydb_topic_test.py
index b9e306034..5e6ff1c78 100644
--- a/ydb/_grpc/grpcwrapper/ydb_topic_test.py
+++ b/ydb/_grpc/grpcwrapper/ydb_topic_test.py
@@ -2,7 +2,7 @@
 
 from google.protobuf.json_format import MessageToDict
 
-from ydb._grpc.grpcwrapper.ydb_topic import OffsetsRange
+from ydb._grpc.grpcwrapper.ydb_topic import OffsetsRange, StreamReadMessage
 from .ydb_topic import AlterTopicRequest
 from .ydb_topic_public_types import (
     AlterTopicRequestParams,
@@ -96,3 +96,25 @@ def test_alter_topic_request_from_public_to_proto():
     }
 
     assert msg_dict == expected_dict
+
+
+def test_stream_read_init_request_serializes_reader_name():
+    request = StreamReadMessage.InitRequest(
+        topics_read_settings=[],
+        consumer="analytics",
+        auto_partitioning_support=True,
+        reader_name="payments-worker",
+    )
+
+    assert request.to_proto().reader_name == "payments-worker"
+
+
+def test_stream_read_init_request_omits_reader_name_by_default():
+    request = StreamReadMessage.InitRequest(
+        topics_read_settings=[],
+        consumer="analytics",
+        auto_partitioning_support=True,
+    )
+
+    assert request.reader_name is None
+    assert request.to_proto().reader_name == ""
diff --git a/ydb/_topic_reader/topic_reader.py b/ydb/_topic_reader/topic_reader.py
index e4545f3a6..59f280d59 100644
--- a/ydb/_topic_reader/topic_reader.py
+++ b/ydb/_topic_reader/topic_reader.py
@@ -60,13 +60,18 @@ class PublicReaderSettings:
     buffer_release_threshold: float = 0.5
     """Min fraction of buffer_size_bytes to accumulate before sending a new ReadRequest (0.0 = immediately after every batch)."""
 
+    reader_name: Optional[str] = None
+    """Optional reader name used to identify this logical reader in diagnostics."""
+
     def __post_init__(self):
+        if self.reader_name is not None and not isinstance(self.reader_name, str):
+            raise TypeError("Unsupported type for reader_name field: '%s'" % type(self.reader_name))
         if not (0.0 <= self.buffer_release_threshold <= 1.0):
             raise ValueError("buffer_release_threshold must be in [0.0, 1.0], got %s" % self.buffer_release_threshold)
         # check possible create init message
         _ = self._init_message()
 
-    def _init_message(self) -> StreamReadMessage.InitRequest:
+    def _init_message(self, *, reader_name: Optional[str] = None) -> StreamReadMessage.InitRequest:
         if self.consumer is not None and not isinstance(self.consumer, str):
             raise TypeError("Unsupported type for customer field: '%s'" % type(self.consumer))
 
@@ -87,6 +92,7 @@ def _init_message(self) -> StreamReadMessage.InitRequest:
             topics_read_settings=list(map(PublicTopicSelector._to_topic_read_settings, selectors)),  # type: ignore
             consumer=self.consumer,
             auto_partitioning_support=self.auto_partitioning_support,
+            reader_name=self.reader_name if reader_name is None else reader_name,
         )
 
     def _retry_settings(self) -> RetrySettings:
diff --git a/ydb/_topic_reader/topic_reader_asyncio.py b/ydb/_topic_reader/topic_reader_asyncio.py
index e5c9f3a59..11c56ef85 100644
--- a/ydb/_topic_reader/topic_reader_asyncio.py
+++ b/ydb/_topic_reader/topic_reader_asyncio.py
@@ -87,6 +87,7 @@ def __init__(self):
 class PublicAsyncIOReader:
     _loop: asyncio.AbstractEventLoop
     _closed: bool
+    _log_prefix: str
     _settings: topic_reader.PublicReaderSettings
     _reconnector: ReaderReconnector
     _parent: typing.Any  # need for prevent close parent client by GC
@@ -100,8 +101,10 @@ def __init__(
     ):
         self._loop = asyncio.get_running_loop()
         self._closed = False
+        self._log_prefix = "topic reader"
         self._settings = settings
         self._reconnector = ReaderReconnector(driver, settings, self._loop)
+        self._log_prefix = self._reconnector._log_prefix
         self._parent = _parent
 
     async def __aenter__(self):
@@ -111,13 +114,19 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
         await self.close()
 
     def __del__(self):
-        if not self._closed:
-            try:
-                logger.debug("Topic reader was not closed properly. Consider using method close().")
-                task = self._loop.create_task(self.close(flush=False))
-                task.set_name("close reader")
-            except BaseException:
-                logger.warning("Something went wrong during reader close in __del__")
+        if getattr(self, "_closed", True):
+            return
+
+        loop = getattr(self, "_loop", None)
+        if getattr(self, "_reconnector", None) is None or loop is None or loop.is_closed() or not loop.is_running():
+            return
+
+        try:
+            logger.debug("%s was not closed properly. Consider using method close().", self._log_prefix)
+            task = loop.create_task(self.close(flush=False))
+            task.set_name("close reader")
+        except BaseException:
+            logger.warning("%s failed to close in __del__", self._log_prefix)
 
     async def wait_message(self):
         """
@@ -140,7 +149,7 @@ async def receive_batch(
 
         use asyncio.wait_for for wait with timeout.
         """
-        logger.debug("receive_batch max_messages=%s max_bytes=%s", max_messages, max_bytes)
+        logger.debug("%s receive_batch max_messages=%s max_bytes=%s", self._log_prefix, max_messages, max_bytes)
         await self._reconnector.wait_message()
         return self._reconnector.receive_batch_nowait(
             max_messages=max_messages,
@@ -163,7 +172,13 @@ async def receive_batch_with_tx(
 
         use asyncio.wait_for for wait with timeout.
         """
-        logger.debug("receive_batch_with_tx tx=%s max_messages=%s max_bytes=%s", tx, max_messages, max_bytes)
+        logger.debug(
+            "%s receive_batch_with_tx tx=%s max_messages=%s max_bytes=%s",
+            self._log_prefix,
+            tx,
+            max_messages,
+            max_bytes,
+        )
         await self._reconnector.wait_message()
         return self._reconnector.receive_batch_with_tx_nowait(
             tx=tx,
@@ -177,7 +192,7 @@ async def receive_message(self) -> typing.Optional[datatypes.PublicMessage]:
 
         use asyncio.wait_for for wait with timeout.
         """
-        logger.debug("receive_message")
+        logger.debug("%s receive_message", self._log_prefix)
         await self._reconnector.wait_message()
         return self._reconnector.receive_message_nowait()
 
@@ -188,7 +203,7 @@ def commit(self, batch: typing.Union[datatypes.PublicMessage, datatypes.PublicBa
         For the method no way check the commit result
         (for example if lost connection - commits will not re-send and committed messages will receive again).
         """
-        logger.debug("commit message or batch")
+        logger.debug("%s commit message or batch", self._log_prefix)
         if self._settings.consumer is None:
             raise issues.Error("Commit operations are not supported for topic reader without consumer.")
 
@@ -207,7 +222,7 @@ async def commit_with_ack(self, batch: typing.Union[datatypes.PublicMessage, dat
         before receive commit ack. Message may be acked or not (if not - it will send in other read session,
         to this or other reader).
         """
-        logger.debug("commit_with_ack message or batch")
+        logger.debug("%s commit_with_ack message or batch", self._log_prefix)
         if self._settings.consumer is None:
             raise issues.Error("Commit operations are not supported for topic reader without consumer.")
 
@@ -218,10 +233,10 @@ async def close(self, flush: bool = True):
         if self._closed:
             raise TopicReaderClosedError()
 
-        logger.debug("Close topic reader")
+        logger.debug("%s close", self._log_prefix)
         self._closed = True
         await self._reconnector.close(flush)
-        logger.debug("Topic reader was closed")
+        logger.debug("%s was closed", self._log_prefix)
 
     @property
     def read_session_id(self) -> Optional[str]:
@@ -235,6 +250,8 @@ class ReaderReconnector:
     _settings: topic_reader.PublicReaderSettings
     _driver: Driver
     _background_tasks: Set[Task]
+    _reader_name: str
+    _log_prefix: str
 
     _state_changed: asyncio.Event
     _stream_reader: Optional["ReaderStream"]
@@ -251,9 +268,11 @@ def __init__(
         self._id = ReaderReconnector._static_reader_reconnector_counter.inc_and_get()
         self._settings = settings
         self._driver = driver
+        self._reader_name = settings.reader_name or "reader-%d" % self._id
+        self._log_prefix = "topic reader reader_name=%r reader_id=%s" % (self._reader_name, self._id)
         self._loop = loop if loop is not None else asyncio.get_running_loop()
         self._background_tasks = set()
-        logger.debug("init reader reconnector id=%s", self._id)
+        logger.debug("%s initialize reconnector", self._log_prefix)
 
         self._state_changed = asyncio.Event()
         self._stream_reader = None
@@ -269,21 +288,30 @@ async def _connection_loop(self):
             if self._closed:
                 return
             try:
-                logger.debug("reader %s connect attempt %s", self._id, attempt)
-                self._stream_reader = await ReaderStream.create(self._id, self._driver, self._settings)
-                logger.debug("reader %s connected stream %s", self._id, self._stream_reader._id)
+                logger.debug("%s connect attempt=%s", self._log_prefix, attempt)
+                self._stream_reader = await ReaderStream.create(
+                    self._id,
+                    self._driver,
+                    self._settings,
+                    reader_name=self._reader_name,
+                )
+                logger.debug("%s connected stream_id=%s", self._log_prefix, self._stream_reader._id)
                 attempt = 0
                 self._state_changed.set()
                 await self._stream_reader.wait_error()
             except BaseException as err:
-                logger.debug("reader %s, attempt %s connection loop error %s", self._id, attempt, err)
+                logger.debug("%s connection attempt=%s failed: %s", self._log_prefix, attempt, err)
                 retry_info = check_retriable_error(err, self._settings._retry_settings(), attempt)
                 if not retry_info.is_retriable:
-                    logger.debug("reader %s stop connection loop due to %s", self._id, err)
+                    logger.debug("%s stop connection loop: %s", self._log_prefix, err)
                     self._set_first_error(err)
                     return
 
-                logger.debug("sleep before retry for %s seconds", retry_info.sleep_timeout_seconds)
+                logger.debug(
+                    "%s sleep before retry for %s seconds",
+                    self._log_prefix,
+                    retry_info.sleep_timeout_seconds,
+                )
 
                 await asyncio.sleep(retry_info.sleep_timeout_seconds)
 
@@ -465,7 +493,7 @@ def commit(self, batch: datatypes.ICommittable) -> datatypes.PartitionSession.Co
         return self._stream_reader.commit(batch)
 
     async def close(self, flush: bool):
-        logger.debug("reader reconnector %s close", self._id)
+        logger.debug("%s close reconnector", self._log_prefix)
         # Mark closed so the connection loop won't start a new stream, then close the
         # current stream with the requested flush before cancelling the loop. On a normal
         # close this flushes pending commits; cancelling the loop first would let it close
@@ -506,6 +534,7 @@ class ReaderStream:
     _loop: asyncio.AbstractEventLoop
     _id: int
     _reader_reconnector_id: int
+    _reader_name: str
     _session_id: str
     _stream: Optional[IGrpcWrapperAsyncIO]
     _started: bool
@@ -537,12 +566,16 @@ def __init__(
         reader_reconnector_id: int,
         settings: topic_reader.PublicReaderSettings,
         get_token_function: Optional[Callable[[], str]] = None,
+        *,
+        reader_name: str,
     ):
         self._loop = asyncio.get_running_loop()
         self._id = ReaderStream._static_id_counter.inc_and_get()
         self._reader_reconnector_id = reader_reconnector_id
+        self._reader_name = reader_name
         self._session_id = "not initialized"
-        self._log_prefix = "reader %s stream %s session=%s" % (
+        self._log_prefix = "topic reader reader_name=%r reader_id=%s stream_id=%s session_id=%s" % (
+            self._reader_name,
             self._reader_reconnector_id,
             self._id,
             self._session_id,
@@ -572,13 +605,15 @@ def __init__(
 
         self._settings = settings
 
-        logger.debug("created ReaderStream id=%s reconnector=%s", self._id, self._reader_reconnector_id)
+        logger.debug("%s created", self._log_prefix)
 
     @staticmethod
     async def create(
         reader_reconnector_id: int,
         driver: SupportedDriverType,
         settings: topic_reader.PublicReaderSettings,
+        *,
+        reader_name: str,
     ) -> "ReaderStream":
         stream = GrpcWrapperAsyncIO(StreamReadMessage.FromServer.from_proto)
         reader = None
@@ -590,8 +625,9 @@ async def create(
                 reader_reconnector_id,
                 settings,
                 get_token_function=creds.get_auth_token if creds else None,
+                reader_name=reader_name,
             )
-            await reader._start(stream, settings._init_message())
+            await reader._start(stream, settings._init_message(reader_name=reader_name))
         except BaseException:
             # If create() is interrupted (e.g. reader.close() cancels the connection loop
             # mid-reconnect) the in-flight stream is not yet assigned to the reconnector, so
@@ -623,7 +659,8 @@ async def _start(self, stream: IGrpcWrapperAsyncIO, init_message: StreamReadMess
 
         if isinstance(init_response.server_message, StreamReadMessage.InitResponse):
             self._session_id = init_response.server_message.session_id
-            self._log_prefix = "reader %s stream %s session=%s" % (
+            self._log_prefix = "topic reader reader_name=%r reader_id=%s stream_id=%s session_id=%s" % (
+                self._reader_name,
                 self._reader_reconnector_id,
                 self._id,
                 self._session_id,
@@ -820,7 +857,7 @@ async def _read_messages_loop(self):
                             "Unexpected message in _read_messages_loop: %s" % type(message.server_message)
                         )
                 except issues.UnexpectedGrpcMessage as e:
-                    logger.exception("unexpected message in stream reader: %s" % e)
+                    logger.exception("%s unexpected message in stream reader: %s", self._log_prefix, e)
 
                 self._state_changed.set()
         except asyncio.CancelledError as e:
diff --git a/ydb/_topic_reader/topic_reader_asyncio_test.py b/ydb/_topic_reader/topic_reader_asyncio_test.py
index c018604f2..989d13181 100644
--- a/ydb/_topic_reader/topic_reader_asyncio_test.py
+++ b/ydb/_topic_reader/topic_reader_asyncio_test.py
@@ -88,6 +88,20 @@ def stub_message(id: int):
     )
 
 
+def test_partially_initialized_public_reader_does_not_schedule_close():
+    reader = object.__new__(topic_reader_asyncio.PublicAsyncIOReader)
+    reader._closed = False
+    reader._log_prefix = "topic reader"
+    reader._loop = mock.Mock()
+    reader._loop.is_closed.return_value = False
+    reader._loop.is_running.return_value = True
+
+    reader.__del__()
+
+    reader._loop.create_task.assert_not_called()
+    reader._closed = True
+
+
 @pytest.fixture()
 def default_reader_settings(default_executor):
     return PublicReaderSettings(
@@ -155,6 +169,7 @@ def second_partition_session(self, default_reader_settings, stream_reader_starte
         return stream_reader_started._partition_sessions[partition_session.id]
 
     async def get_started_reader(self, stream, *args, **kwargs) -> ReaderStream:
+        kwargs.setdefault("reader_name", "reader-%d" % self.default_reader_reconnector_id)
         reader = ReaderStream(self.default_reader_reconnector_id, *args, **kwargs)
         init_message = object()
 
@@ -182,6 +197,20 @@ async def get_started_reader(self, stream, *args, **kwargs) -> ReaderStream:
 
         return reader
 
+    async def test_log_prefix_identifies_reader_stream_and_session(self, stream, default_reader_settings):
+        reader = await self.get_started_reader(
+            stream,
+            default_reader_settings,
+            reader_name="payments worker\nprimary",
+        )
+
+        assert reader._log_prefix == (
+            "topic reader reader_name='payments worker\\nprimary' "
+            "reader_id=%s stream_id=%s session_id=test-session" % (self.default_reader_reconnector_id, reader._id)
+        )
+
+        await reader.close(False)
+
     @pytest.fixture()
     async def stream_reader_started(self, stream, default_reader_settings) -> ReaderStream:
         return await self.get_started_reader(stream, default_reader_settings)
@@ -601,7 +630,11 @@ async def test_error_from_status_code(self, stream, stream_reader_finish_with_er
             stream_reader_finish_with_error.receive_batch_nowait()
 
     async def test_init_reader(self, stream, default_reader_settings):
-        reader = ReaderStream(self.default_reader_reconnector_id, default_reader_settings)
+        reader = ReaderStream(
+            self.default_reader_reconnector_id,
+            default_reader_settings,
+            reader_name="reader-%d" % self.default_reader_reconnector_id,
+        )
         init_message = StreamReadMessage.InitRequest(
             consumer="test-consumer",
             topics_read_settings=[
@@ -1549,7 +1582,11 @@ def logged():
 
     async def test_init_timeout_parameter(self, stream, default_reader_settings):
         """Test that ReaderStream._start calls stream.receive with timeout=10"""
-        reader = ReaderStream(self.default_reader_reconnector_id, default_reader_settings)
+        reader = ReaderStream(
+            self.default_reader_reconnector_id,
+            default_reader_settings,
+            reader_name="reader-%d" % self.default_reader_reconnector_id,
+        )
         init_message = default_reader_settings._init_message()
 
         # Mock stream.receive to check if timeout is passed
@@ -1568,7 +1605,11 @@ async def test_init_timeout_parameter(self, stream, default_reader_settings):
 
     async def test_init_timeout_behavior(self, stream, default_reader_settings):
         """Test that ReaderStream._start raises TopicReaderError when receive times out"""
-        reader = ReaderStream(self.default_reader_reconnector_id, default_reader_settings)
+        reader = ReaderStream(
+            self.default_reader_reconnector_id,
+            default_reader_settings,
+            reader_name="reader-%d" % self.default_reader_reconnector_id,
+        )
         init_message = default_reader_settings._init_message()
 
         # Mock stream.receive to directly raise TimeoutError when called with timeout
@@ -1590,7 +1631,8 @@ async def timeout_receive(timeout=None):
 
 @pytest.mark.asyncio
 class TestReaderReconnector:
-    async def test_reconnect_on_repeatable_error(self, monkeypatch):
+    @pytest.mark.parametrize("configured_reader_name", [None, ""])
+    async def test_reconnect_on_repeatable_error(self, monkeypatch, configured_reader_name):
         test_error = issues.Overloaded("test error")
 
         async def wait_error():
@@ -1620,8 +1662,11 @@ async def stream_create(
             reader_reconnector_id: int,
             driver: SupportedDriverType,
             settings: PublicReaderSettings,
+            *,
+            reader_name: str,
         ):
             nonlocal stream_index
+            reader_names.append(reader_name)
             stream_index += 1
             if stream_index == 1:
                 return reader_stream_mock_with_error
@@ -1630,12 +1675,19 @@ async def stream_create(
             else:
                 raise Exception("unexpected create stream")
 
+        reader_names = []
         with mock.patch.object(ReaderStream, "create", stream_create):
-            reconnector = ReaderReconnector(mock.Mock(), PublicReaderSettings("", ""))
+            reconnector = ReaderReconnector(
+                mock.Mock(),
+                PublicReaderSettings("", "", reader_name=configured_reader_name),
+            )
             await wait_for_fast(reconnector.wait_message())
 
+        assert reader_names == [reconnector._reader_name, reconnector._reader_name]
+        assert reconnector._reader_name == "reader-%d" % reconnector._id
         reader_stream_mock_with_error.wait_error.assert_any_await()
         reader_stream_mock_with_error.wait_messages.assert_any_await()
+        await reconnector.close(flush=False)
 
     async def test_close_during_reconnect_does_not_hang(self):
         # The connection loop must stop on reader.close() even while it is closing the old
@@ -1668,7 +1720,7 @@ async def wait_forever():
 
         create_calls = 0
 
-        async def stream_create(reader_reconnector_id, driver, settings):
+        async def stream_create(reader_reconnector_id, driver, settings, *, reader_name):
             nonlocal create_calls
             create_calls += 1
             return stream1 if create_calls == 1 else stream2
@@ -1702,7 +1754,14 @@ async def start(self, driver, stub, method):
         with mock.patch.object(topic_reader_asyncio, "GrpcWrapperAsyncIO", FakeStream):
             # Real create(); no InitResponse is sent, so it parks inside _start() on
             # `await stream.receive()` (the only reachable cancellation point in create()).
-            create_task = asyncio.create_task(ReaderStream.create(7, driver, default_reader_settings))
+            create_task = asyncio.create_task(
+                ReaderStream.create(
+                    7,
+                    driver,
+                    default_reader_settings,
+                    reader_name="reader-7",
+                )
+            )
             await wait_condition(lambda: bool(built) and not built[0].from_client.empty())
             assert not create_task.done()
 
@@ -1732,7 +1791,7 @@ async def receive(timeout=None):
         stream.write = mock.Mock()
         stream.close = mock.Mock()
 
-        reader = ReaderStream(4, default_reader_settings)
+        reader = ReaderStream(4, default_reader_settings, reader_name="reader-4")
         await reader._start(stream, default_reader_settings._init_message())
 
         # Bug: wait_error() hangs forever because _first_error is never set.
@@ -1766,7 +1825,11 @@ async def _get_started_reader(self, stream, threshold, buffer_size_bytes, defaul
             buffer_release_threshold=threshold,
             decoder_executor=default_executor,
         )
-        reader = ReaderStream(self.default_reader_reconnector_id, settings)
+        reader = ReaderStream(
+            self.default_reader_reconnector_id,
+            settings,
+            reader_name="reader-%d" % self.default_reader_reconnector_id,
+        )
         init_message = object()
         start = asyncio.create_task(reader._start(stream, init_message))
 
@@ -1930,3 +1993,78 @@ async def test_threshold_one_flushes_when_bytes_match_buffer_size(self, stream,
         assert msg.client_message.bytes_size == 1000
 
         await reader.close(False)
+
+
+def test_reader_settings_forward_reader_name():
+    settings = PublicReaderSettings(
+        consumer="analytics",
+        topic="/Root/events",
+        reader_name="payments-worker",
+    )
+
+    assert settings._init_message().reader_name == "payments-worker"
+    assert settings._init_message(reader_name="reader-42").to_proto().reader_name == "reader-42"
+
+
+def test_reader_settings_reject_non_string_reader_name():
+    with pytest.raises(TypeError, match="reader_name"):
+        PublicReaderSettings(
+            consumer="analytics",
+            topic="/Root/events",
+            reader_name=42,
+        )
+
+
+@pytest.mark.asyncio
+async def test_reader_reconnector_uses_configured_or_process_local_reader_name():
+    async def wait_forever(self):
+        await asyncio.Future()
+
+    with mock.patch.object(ReaderReconnector, "_connection_loop", wait_forever):
+        generated_from_none = ReaderReconnector(
+            mock.Mock(),
+            PublicReaderSettings("analytics", "/Root/events"),
+        )
+        generated_from_empty = ReaderReconnector(
+            mock.Mock(),
+            PublicReaderSettings("analytics", "/Root/events", reader_name=""),
+        )
+        configured = ReaderReconnector(
+            mock.Mock(),
+            PublicReaderSettings("analytics", "/Root/events", reader_name="payments-worker"),
+        )
+
+        assert generated_from_none._reader_name == "reader-%d" % generated_from_none._id
+        assert generated_from_empty._reader_name == "reader-%d" % generated_from_empty._id
+        assert generated_from_none._reader_name != generated_from_empty._reader_name
+        assert configured._reader_name == "payments-worker"
+        assert configured._log_prefix == "topic reader reader_name='payments-worker' reader_id=%s" % configured._id
+
+        await asyncio.gather(
+            generated_from_none.close(flush=False),
+            generated_from_empty.close(flush=False),
+            configured.close(flush=False),
+        )
+
+
+@pytest.mark.parametrize(
+    "client_class, reader_class_name", [("TopicClient", "TopicReader"), ("TopicClientAsyncIO", "TopicReaderAsyncIO")]
+)
+def test_topic_clients_forward_reader_name(client_class, reader_class_name):
+    from .. import topic as topic_module
+
+    client = object.__new__(getattr(topic_module, client_class))
+    client._closed = False
+    client._driver = mock.Mock()
+    client._executor = mock.Mock()
+
+    with mock.patch.object(topic_module, reader_class_name, return_value=mock.sentinel.reader) as reader_class:
+        result = client.reader(
+            "/Root/events",
+            consumer="analytics",
+            reader_name="payments-worker",
+        )
+
+    assert result is mock.sentinel.reader
+    assert reader_class.call_args.args[1].reader_name == "payments-worker"
+    client._closed = True
diff --git a/ydb/_topic_reader/topic_reader_sync.py b/ydb/_topic_reader/topic_reader_sync.py
index a2a24fef6..4a09ebdf3 100644
--- a/ydb/_topic_reader/topic_reader_sync.py
+++ b/ydb/_topic_reader/topic_reader_sync.py
@@ -32,6 +32,7 @@ class TopicReaderSync:
     _caller: CallFromSyncToAsync
     _async_reader: PublicAsyncIOReader
     _closed: bool
+    _log_prefix: str
     _settings: PublicReaderSettings
     _parent: typing.Any  # need for prevent stop the client by GC
 
@@ -44,6 +45,7 @@ def __init__(
         _parent=None,  # need for prevent stop the client by GC
     ):
         self._closed = False
+        self._log_prefix = "topic reader"
 
         if eventloop:
             loop = eventloop
@@ -56,6 +58,7 @@ async def create_reader():
             return PublicAsyncIOReader(driver, settings)
 
         self._async_reader = asyncio.run_coroutine_threadsafe(create_reader(), loop).result()
+        self._log_prefix = self._async_reader._log_prefix
 
         self._settings = settings
 
@@ -64,10 +67,10 @@ async def create_reader():
     def __del__(self):
         if not self._closed:
             try:
-                logger.debug("Topic reader was not closed properly. Consider using method close().")
+                logger.debug("%s was not closed properly. Consider using method close().", self._log_prefix)
                 self.close(flush=False)
             except BaseException:
-                logger.warning("Something went wrong during reader close in __del__")
+                logger.warning("%s failed to close in __del__", self._log_prefix)
 
     def __enter__(self):
         return self
diff --git a/ydb/topic.py b/ydb/topic.py
index e76956abc..2a588c25c 100644
--- a/ydb/topic.py
+++ b/ydb/topic.py
@@ -295,6 +295,7 @@ def reader(
         auto_partitioning_support: Optional[bool] = True,  # Auto partitioning feature flag. Default - True.
         event_handler: Optional[TopicReaderEvents.EventHandler] = None,
         buffer_release_threshold: float = 0.5,
+        reader_name: Optional[str] = None,
     ) -> TopicReaderAsyncIO:
 
         logger.debug("Create reader for topic=%s consumer=%s", topic, consumer)
@@ -631,6 +632,7 @@ def reader(
         auto_partitioning_support: Optional[bool] = True,  # Auto partitioning feature flag. Default - True.
         event_handler: Optional[TopicReaderEvents.EventHandler] = None,
         buffer_release_threshold: float = 0.5,
+        reader_name: Optional[str] = None,
     ) -> TopicReader:
         logger.debug("Create reader for topic=%s consumer=%s", topic, consumer)
         if not decoder_executor: