Fixed errors of the distributed commit offset to the partition (#17423)

Author: nshestakov

src/client/topic/ut/ut_utils/topic_sdk_test_setup.cpp

@@ -98,14 +98,22 @@ void TTopicSdkTestSetup::Write(const std::string& message, ui32 partitionId, con
     session->Close(TDuration::Seconds(5));
 }
 
-TTopicSdkTestSetup::ReadResult TTopicSdkTestSetup::Read(const std::string& topic, const std::string& consumer, std::function<bool (NYdb::NTopic::TReadSessionEvent::TDataReceivedEvent&)> handler, const TDuration timeout) {
+TTopicSdkTestSetup::ReadResult TTopicSdkTestSetup::Read(const std::string& topic, const std::string& consumer,
+    std::function<bool (NYdb::NTopic::TReadSessionEvent::TDataReceivedEvent&)> handler,
+    std::optional<size_t> partition, const TDuration timeout) {
     TTopicClient client(MakeDriver());
 
-    auto reader = client.CreateReadSession(
-        TReadSessionSettings()
-            .AutoPartitioningSupport(true)
-            .AppendTopics(TTopicReadSettings(topic))
-            .ConsumerName(consumer));
+    auto topicSettings = TTopicReadSettings(topic);
+    if (partition) {
+        topicSettings.AppendPartitionIds(partition.value());
+    }
+
+    auto settins = TReadSessionSettings()
+        .AutoPartitioningSupport(true)
+        .AppendTopics(topicSettings)
+        .ConsumerName(consumer);
+
+    auto reader = client.CreateReadSession(settins);
 
     TInstant deadlineTime = TInstant::Now() + timeout;
 

src/client/topic/ut/ut_utils/topic_sdk_test_setup.h

@@ -32,7 +32,9 @@ class TTopicSdkTestSetup {
 
         std::vector<NYdb::NTopic::TReadSessionEvent::TStartPartitionSessionEvent> StartPartitionSessionEvents;
     };
-    ReadResult Read(const std::string& topic, const std::string& consumer, std::function<bool (NYdb::NTopic::TReadSessionEvent::TDataReceivedEvent&)> handler, const TDuration timeout = TDuration::Seconds(5));
+    ReadResult Read(const std::string& topic, const std::string& consumer,
+        std::function<bool (NYdb::NTopic::TReadSessionEvent::TDataReceivedEvent&)> handler,
+        std::optional<size_t> partition = std::nullopt, const TDuration timeout = TDuration::Seconds(5));
     TStatus Commit(const std::string& path, const std::string& consumerName, size_t partitionId, size_t offset, std::optional<std::string> sessionId = std::nullopt);
 
     TString GetEndpoint() const;