No Consumer Reader

Author: vgvoleg

tests/topics/test_topic_reader.py

@@ -251,3 +251,40 @@ async def wait(fut):
 
         await reader0.close()
         await reader1.close()
+
+
+@pytest.mark.asyncio
+class TestTopicNoConsumerReaderAsyncIO:
+    async def test_read_message(self, driver, topic_with_messages):
+        def get_offset_lambda(partition_id: int) -> int:
+            return 0
+
+        reader = driver.topic_client.no_consumer_reader(topic_with_messages, [0], get_offset_lambda)
+        await reader.wait_message()
+        msg = await reader.receive_message()
+
+        assert msg is not None
+        assert msg.seqno
+
+        await reader.close()
+
+    async def test_read_after_reconnect(self, driver, topic_with_messages):
+        reader = driver.topic_client.no_consumer_reader(topic_with_messages, [0], lambda x: None)
+        await reader.wait_message()
+        msg1 = await reader.receive_message()
+
+        assert msg1 is not None
+        assert msg1.seqno
+
+        await reader.close()
+
+        reader = driver.topic_client.no_consumer_reader(topic_with_messages, [0], lambda x: None)
+        await reader.wait_message()
+        msg2 = await reader.receive_message()
+
+        assert msg2 is not None
+        assert msg2.seqno
+
+        await reader.close()
+
+        assert msg1.seqno == msg2.seqno

ydb/_grpc/grpcwrapper/ydb_topic.py

@@ -439,12 +439,13 @@ def from_proto(
     @dataclass
     class InitRequest(IToProto):
         topics_read_settings: List["StreamReadMessage.InitRequest.TopicReadSettings"]
-        consumer: str
+        consumer: Optional[str]
         auto_partitioning_support: bool
 
         def to_proto(self) -> ydb_topic_pb2.StreamReadMessage.InitRequest:
             res = ydb_topic_pb2.StreamReadMessage.InitRequest()
-            res.consumer = self.consumer
+            if self.consumer is not None:
+                res.consumer = self.consumer
             for settings in self.topics_read_settings:
                 res.topics_read_settings.append(settings.to_proto())
             res.auto_partitioning_support = self.auto_partitioning_support

ydb/_topic_reader/topic_reader.py

@@ -42,7 +42,7 @@ def _to_topic_read_settings(self) -> StreamReadMessage.InitRequest.TopicReadSett
 
 @dataclass
 class PublicReaderSettings:
-    consumer: str
+    consumer: Optional[str]
     topic: TopicSelectorTypes
     buffer_size_bytes: int = 50 * 1024 * 1024
     auto_partitioning_support: bool = True
@@ -54,12 +54,15 @@ class PublicReaderSettings:
     decoder_executor: Optional[concurrent.futures.Executor] = None
     update_token_interval: Union[int, float] = 3600
 
+    partition_ids: Optional[List[int]] = None
+    get_start_offset_lambda: Optional[Callable[[int], int]] = None
+
     def __post_init__(self):
         # check possible create init message
         _ = self._init_message()
 
     def _init_message(self) -> StreamReadMessage.InitRequest:
-        if not isinstance(self.consumer, str):
+        if self.consumer is not None and not isinstance(self.consumer, str):
             raise TypeError("Unsupported type for customer field: '%s'" % type(self.consumer))
 
         if isinstance(self.topic, list):
@@ -69,7 +72,7 @@ def _init_message(self) -> StreamReadMessage.InitRequest:
 
         for index, selector in enumerate(selectors):
             if isinstance(selector, str):
-                selectors[index] = PublicTopicSelector(path=selector)
+                selectors[index] = PublicTopicSelector(path=selector, partitions=self.partition_ids)
             elif isinstance(selector, PublicTopicSelector):
                 pass
             else:

ydb/_topic_reader/topic_reader_asyncio.py

@@ -182,6 +182,14 @@ async def close(self, flush: bool = True):
         await self._reconnector.close(flush)
 
 
+class PublicAsyncIONoConsumerReader(PublicAsyncIOReader):
+    def commit(self, batch: typing.Union[datatypes.PublicMessage, datatypes.PublicBatch]):
+        raise issues.Error("Commit operations are not supported for topic reader without consumer.")
+
+    async def commit_with_ack(self, batch: typing.Union[datatypes.PublicMessage, datatypes.PublicBatch]):
+        raise issues.Error("Commit operations are not supported for topic reader without consumer.")
+
+
 class ReaderReconnector:
     _static_reader_reconnector_counter = AtomicCounter()
 

ydb/topic.py

@@ -14,6 +14,7 @@
     "TopicError",
     "TopicMeteringMode",
     "TopicReader",
+    "TopicNoConsumerReaderAsyncIO",
     "TopicReaderAsyncIO",
     "TopicReaderBatch",
     "TopicReaderMessage",
@@ -56,6 +57,7 @@
 
 from ._topic_reader.topic_reader_asyncio import (
     PublicAsyncIOReader as TopicReaderAsyncIO,
+    PublicAsyncIONoConsumerReader as TopicNoConsumerReaderAsyncIO,
     PublicTopicReaderPartitionExpiredError as TopicReaderPartitionExpiredError,
     PublicTopicReaderUnexpectedCodecError as TopicReaderUnexpectedCodecError,
 )
@@ -261,6 +263,31 @@ def reader(
 
         return TopicReaderAsyncIO(self._driver, settings, _parent=self)
 
+    def no_consumer_reader(
+        self,
+        topic: Union[str, TopicReaderSelector, List[Union[str, TopicReaderSelector]]],
+        partition_ids: List[int],
+        get_start_offset_lambda: Callable[[int], int],
+        buffer_size_bytes: int = 50 * 1024 * 1024,
+        # decoders: map[codec_code] func(encoded_bytes)->decoded_bytes
+        # the func will be called from multiply threads in parallel
+        decoders: Union[Mapping[int, Callable[[bytes], bytes]], None] = None,
+        # custom decoder executor for call builtin and custom decoders. If None - use shared executor pool.
+        # if max_worker in the executor is 1 - then decoders will be called from the thread without parallel
+        decoder_executor: Optional[concurrent.futures.Executor] = None,
+        auto_partitioning_support: Optional[bool] = True,  # Auto partitioning feature flag. Default - True.
+    ) -> TopicNoConsumerReaderAsyncIO:
+        if not decoder_executor:
+            decoder_executor = self._executor
+
+        args = locals().copy()
+        del args["self"]
+        args['consumer'] = None
+
+        settings = TopicReaderSettings(**args)
+
+        return TopicNoConsumerReaderAsyncIO(self._driver, settings, _parent=self)
+
     def writer(
         self,
         topic,