review fixes

Author: vgvoleg

ydb/_topic_reader/datatypes.py

@@ -182,6 +182,10 @@ def _pop(self) -> Tuple[List[PublicMessage], bool]:
 
     def _pop_batch(self, size: int) -> PublicBatch:
         initial_length = len(self.messages)
+
+        if size >= initial_length:
+            raise ValueError("Pop batch with size >= actual size is not supported.")
+
         one_message_size = self._bytes_size // initial_length
 
         new_batch = PublicBatch(
@@ -192,6 +196,6 @@ def _pop_batch(self, size: int) -> PublicBatch:
         )
 
         self.messages = self.messages[size:]
-        self._bytes_size = one_message_size * (initial_length - size)
+        self._bytes_size = self._bytes_size - new_batch._bytes_size
 
         return new_batch

ydb/_topic_reader/topic_reader_asyncio.py

@@ -388,26 +388,6 @@ def _get_first_batch(self) -> typing.Tuple[int, datatypes.PublicBatch]:
         partition_session_id, batch = self._message_batches.popitem(last=False)
         return partition_session_id, batch
 
-    def _cut_batch_by_max_messages(
-        self,
-        batch: datatypes.PublicBatch,
-        max_messages: int,
-    ) -> typing.Tuple[datatypes.PublicBatch, datatypes.PublicBatch]:
-        initial_length = len(batch.messages)
-        one_message_size = batch._bytes_size // initial_length
-
-        new_batch = datatypes.PublicBatch(
-            messages=batch.messages[:max_messages],
-            _partition_session=batch._partition_session,
-            _bytes_size=one_message_size * max_messages,
-            _codec=batch._codec,
-        )
-
-        batch.messages = batch.messages[max_messages:]
-        batch._bytes_size = one_message_size * (initial_length - max_messages)
-
-        return new_batch, batch
-
     def receive_batch_nowait(self, max_messages: Optional[int] = None):
         if self._get_first_error():
             raise self._get_first_error()