Fix TStartPartitionSessionEvent::Confirm (#16734)

Author: qyryq

src/client/topic/impl/read_session_impl.ipp

@@ -73,6 +73,9 @@ void TPartitionStreamImpl<UseMigrationProtocol>::RequestStatus() {
 template<bool UseMigrationProtocol>
 void TPartitionStreamImpl<UseMigrationProtocol>::ConfirmCreate(std::optional<ui64> readOffset, std::optional<ui64> commitOffset) {
     if (auto sessionShared = CbContext->LockShared()) {
+        if (commitOffset.has_value()) {
+            SetFirstNotReadOffset(commitOffset.value());
+        }
         sessionShared->ConfirmPartitionStreamCreate(this, readOffset, commitOffset);
     }
 }

src/client/topic/ut/basic_usage_ut.cpp

@@ -3,7 +3,7 @@
 #include <src/client/persqueue_public/ut/ut_utils/ut_utils.h>
 
 #include <ydb-cpp-sdk/client/topic/client.h>
- 
+
 #include <src/client/persqueue_public/persqueue.h>
 
 #include <src/client/topic/impl/common.h>
@@ -661,6 +661,64 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
         Sleep(TDuration::Seconds(5));
     }
 
+    Y_UNIT_TEST(ConfirmPartitionSessionWithCommitOffset) {
+        // TStartPartitionSessionEvent::Confirm(readOffset, commitOffset) should work,
+        // if commitOffset passed to Confirm is greater than the offset committed previously by the consumer.
+        // https://st.yandex-team.ru/KIKIMR-23015
+
+        auto setup = TTopicSdkTestSetup(TEST_CASE_NAME);
+
+        {
+            // Write 2 messages:
+            auto settings = NTopic::TWriteSessionSettings()
+                .Path(setup.GetTopicPath())
+                .MessageGroupId(TEST_MESSAGE_GROUP_ID)
+                .ProducerId(TEST_MESSAGE_GROUP_ID);
+            auto client = setup.MakeClient();
+            auto writer = client.CreateSimpleBlockingWriteSession(settings);
+            writer->Write("message");
+            writer->Write("message");
+            writer->Close();
+        }
+
+        {
+            // Read messages:
+            auto settings = NTopic::TReadSessionSettings()
+                .ConsumerName(TEST_CONSUMER)
+                .AppendTopics(std::string(setup.GetTopicPath()));
+
+            auto client = setup.MakeClient();
+            auto reader = client.CreateReadSession(settings);
+
+            {
+                // Start partition session and request to read from offset 1 and commit offset 1:
+                auto event = reader->GetEvent(true);
+                UNIT_ASSERT(event.has_value());
+                UNIT_ASSERT(std::holds_alternative<TReadSessionEvent::TStartPartitionSessionEvent>(*event));
+                auto& startPartitionSession = std::get<TReadSessionEvent::TStartPartitionSessionEvent>(*event);
+                startPartitionSession.Confirm(/*readOffset=*/ 1, /*commitOffset=*/ 1);
+            }
+
+            {
+                // Receive a message with offset 1 and commit it:
+                auto event = reader->GetEvent(true);
+                UNIT_ASSERT(event.has_value());
+                UNIT_ASSERT(std::holds_alternative<TReadSessionEvent::TDataReceivedEvent>(*event));
+                auto& dataReceived = std::get<TReadSessionEvent::TDataReceivedEvent>(*event);
+
+                // Here we should commit range [1, 2), not [0, 2):
+                dataReceived.Commit();
+            }
+
+            {
+                // And then get a TCommitOffsetAcknowledgementEvent:
+                auto event = reader->GetEvent(true);
+                UNIT_ASSERT(event.has_value());
+                UNIT_ASSERT(std::holds_alternative<TReadSessionEvent::TCommitOffsetAcknowledgementEvent>(*event));
+            }
+        }
+    }
+
     Y_UNIT_TEST(ConflictingWrites) {
 
         TTopicSdkTestSetup setup(TEST_CASE_NAME);