fix

Author: nshestakov

ydb/core/persqueue/partition.cpp

@@ -2361,6 +2361,9 @@ std::pair<TString, bool> TPartition::ValidatePartitionOperation(const NKikimrPQ:
     TUserInfoBase& userInfo = GetOrCreatePendingUser(consumer);
 
     if (!operation.GetReadSessionId().empty() && operation.GetReadSessionId() != userInfo.Session) {
+        if (!IsActive() && operation.GetCommitOffsetsEnd() >= EndOffset && userInfo.Offset == i64(EndOffset)) {
+            return {"", false};
+        }
         PQ_LOG_D("Partition " << Partition <<
             " Consumer '" << consumer << "'" <<
             " Bad request (session already dead) " <<

ydb/core/persqueue/ut/ut_with_sdk/commitoffset_ut.cpp

@@ -245,18 +245,11 @@ Y_UNIT_TEST_SUITE(CommitOffset) {
         PrepareAutopartitionedTopic(setup);
         TTopicClient client(setup.MakeDriver());
 
-        auto createReadSession = [&](size_t partitionId) {
-            return client.CreateReadSession(
-                TReadSessionSettings()
-                    .AutoPartitioningSupport(true)
-                    .AppendTopics(TTopicReadSettings(TEST_TOPIC).AppendPartitionIds(partitionId))
-                    .ConsumerName(TEST_CONSUMER));
-        };
-
-        auto session0 = createReadSession(0);
-        auto session1 = createReadSession(1);
+        auto r0 = setup.Read(TEST_TOPIC, TEST_CONSUMER, [](auto&) { return false; }, 0);
+        auto r1 = setup.Read(TEST_TOPIC, TEST_CONSUMER, [](auto&) { return false; }, 1);
 
-        auto result = setup.Commit(TEST_TOPIC, TEST_CONSUMER, 1, 1, session1->GetSessionId());
+        auto result = setup.Commit(TEST_TOPIC, TEST_CONSUMER, 1, 1,
+            r1.StartPartitionSessionEvents.back().GetPartitionSession()->GetReadSessionId());
         UNIT_ASSERT(!result.IsSuccess());
 
         UNIT_ASSERT_VALUES_EQUAL(0, GetCommittedOffset(setup, 0));
@@ -266,6 +259,27 @@ Y_UNIT_TEST_SUITE(CommitOffset) {
         UNIT_ASSERT_VALUES_EQUAL(0, GetCommittedOffset(setup, 4));
     }
 
+    Y_UNIT_TEST(Commit_WithSession_ParentNotFinished_OtherSession_ParentCommittedToEnd) {
+        TTopicSdkTestSetup setup = CreateSetup();
+        PrepareAutopartitionedTopic(setup);
+        TTopicClient client(setup.MakeDriver());
+
+        setup.Commit(TEST_TOPIC, TEST_CONSUMER, 0, 3);
+
+        auto r0 = setup.Read(TEST_TOPIC, TEST_CONSUMER, [](auto&) { return false; }, 0);
+        auto r1 = setup.Read(TEST_TOPIC, TEST_CONSUMER, [](auto&) { return false; }, 1);
+
+        auto result = setup.Commit(TEST_TOPIC, TEST_CONSUMER, 1, 1,
+            r1.StartPartitionSessionEvents.back().GetPartitionSession()->GetReadSessionId());
+        UNIT_ASSERT(result.IsSuccess());
+
+        UNIT_ASSERT_VALUES_EQUAL(3, GetCommittedOffset(setup, 0));
+        UNIT_ASSERT_VALUES_EQUAL(1, GetCommittedOffset(setup, 1));
+        UNIT_ASSERT_VALUES_EQUAL(0, GetCommittedOffset(setup, 2));
+        UNIT_ASSERT_VALUES_EQUAL(0, GetCommittedOffset(setup, 3));
+        UNIT_ASSERT_VALUES_EQUAL(0, GetCommittedOffset(setup, 4));
+    }
+
     Y_UNIT_TEST(Commit_WithSession_ToPastParentPartition) {
         TTopicSdkTestSetup setup = CreateSetup();
         PrepareAutopartitionedTopic(setup);

ydb/public/sdk/cpp/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;
 

ydb/public/sdk/cpp/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;