fix write to inactive partition (#5481)

3 files changed, 39 insertions, 7 deletions

diff --git a/ydb/core/persqueue/partition_write.cpp b/ydb/core/persqueue/partition_write.cpp
index 6c7bf5dcfe7..4140fd01da8 100644
--- a/ydb/core/persqueue/partition_write.cpp
+++ b/ydb/core/persqueue/partition_write.cpp
@@ -867,6 +867,11 @@ void TPartition::CancelOneWriteOnWrite(const TActorContext& ctx,
 }
 
 TPartition::EProcessResult TPartition::PreProcessRequest(TRegisterMessageGroupMsg& msg) {
+    if (!CanWrite()) {
+        ScheduleReplyError(msg.Cookie, InactivePartitionErrorCode,
+            TStringBuilder() << "Write to inactive partition");
+        return EProcessResult::ContinueDrop;
+    }
     if (DiskIsFull) {
         ScheduleReplyError(msg.Cookie,
                            NPersQueue::NErrorCode::WRITE_ERROR_DISK_IS_FULL,
@@ -894,6 +899,11 @@ void TPartition::ExecRequest(TRegisterMessageGroupMsg& msg, ProcessParameters& p
 }
 
 TPartition::EProcessResult TPartition::PreProcessRequest(TDeregisterMessageGroupMsg& msg) {
+    if (!CanWrite()) {
+        ScheduleReplyError(msg.Cookie, InactivePartitionErrorCode,
+            TStringBuilder() << "Write to inactive partition");
+        return EProcessResult::ContinueDrop;
+    }
     if (DiskIsFull) {
         ScheduleReplyError(msg.Cookie,
                            NPersQueue::NErrorCode::WRITE_ERROR_DISK_IS_FULL,
@@ -913,6 +923,11 @@ void TPartition::ExecRequest(TDeregisterMessageGroupMsg& msg, ProcessParameters&
 
 
 TPartition::EProcessResult TPartition::PreProcessRequest(TSplitMessageGroupMsg& msg) {
+    if (!CanWrite()) {
+        ScheduleReplyError(msg.Cookie, InactivePartitionErrorCode,
+            TStringBuilder() << "Write to inactive partition");
+        return EProcessResult::ContinueDrop;
+    }
     if (DiskIsFull) {
         ScheduleReplyError(msg.Cookie,
                            NPersQueue::NErrorCode::WRITE_ERROR_DISK_IS_FULL,
@@ -953,6 +968,11 @@ void TPartition::ExecRequest(TSplitMessageGroupMsg& msg, ProcessParameters& para
 }
 
 TPartition::EProcessResult TPartition::PreProcessRequest(TWriteMsg& p) {
+    if (!CanWrite()) {
+        ScheduleReplyError(p.Cookie, InactivePartitionErrorCode,
+            TStringBuilder() << "Write to inactive partition");
+        return EProcessResult::ContinueDrop;
+    }
     if (DiskIsFull) {
         ScheduleReplyError(p.Cookie,
                             NPersQueue::NErrorCode::WRITE_ERROR_DISK_IS_FULL,
@@ -967,6 +987,11 @@ TPartition::EProcessResult TPartition::PreProcessRequest(TWriteMsg& p) {
 }
 
 bool TPartition::ExecRequest(TWriteMsg& p, ProcessParameters& parameters, TEvKeyValue::TEvRequest* request) {
+    if (!CanWrite()) {
+        ScheduleReplyError(p.Cookie, InactivePartitionErrorCode,
+            TStringBuilder() << "Write to inactive partition");
+        return false;
+    }
     if (DiskIsFull) {
         ScheduleReplyError(p.Cookie,
                             NPersQueue::NErrorCode::WRITE_ERROR_DISK_IS_FULL,
diff --git a/ydb/core/persqueue/read_balancer__balancing.cpp b/ydb/core/persqueue/read_balancer__balancing.cpp
index 98f6ba637bd..4130dd83b74 100644
--- a/ydb/core/persqueue/read_balancer__balancing.cpp
+++ b/ydb/core/persqueue/read_balancer__balancing.cpp
@@ -680,6 +680,9 @@ bool TConsumer::BreakUpFamily(TPartitionFamily* family, ui32 partitionId, bool d
     std::vector<TPartitionFamily*> newFamilies;
 
     if (!family->IsLonely()) {
+        LOG_DEBUG_S(ctx, NKikimrServices::PERSQUEUE_READ_BALANCER,
+                GetPrefix() << "break up " << family->DebugStr() << " partition=" << partitionId);
+
         std::unordered_set<ui32> partitions;
         partitions.insert(family->Partitions.begin(), family->Partitions.end());
 
@@ -748,10 +751,10 @@ bool TConsumer::BreakUpFamily(TPartitionFamily* family, ui32 partitionId, bool d
                     }
                 }
             }
+        } else {
+            LOG_DEBUG_S(ctx, NKikimrServices::PERSQUEUE_READ_BALANCER,
+                    GetPrefix() << "can't break up " << family->DebugStr() << " because partition=" << partitionId << " is not root of family");
         }
-    } else {
-        LOG_DEBUG_S(ctx, NKikimrServices::PERSQUEUE_READ_BALANCER,
-                GetPrefix() << "can't break up " << family->DebugStr() << " because partition is not root of family " << family->DebugStr());
     }
 
     family->WantedPartitions.clear();
diff --git a/ydb/core/persqueue/ut/autoscaling_ut.cpp b/ydb/core/persqueue/ut/autoscaling_ut.cpp
index fa9113b4a1c..98fa43c53fe 100644
--- a/ydb/core/persqueue/ut/autoscaling_ut.cpp
+++ b/ydb/core/persqueue/ut/autoscaling_ut.cpp
@@ -77,7 +77,7 @@ Y_UNIT_TEST_SUITE(TopicAutoscaling) {
         SimpleTest(true);
     }
 
-    void ReadingAfterSplitTest(bool autoscaleAwareSDK) {
+    void ReadingAfterSplitTest(bool autoscaleAwareSDK, bool autoCommit) {
         TTopicSdkTestSetup setup = CreateSetup();
         setup.CreateTopic();
 
@@ -96,7 +96,7 @@ Y_UNIT_TEST_SUITE(TopicAutoscaling) {
 
         UNIT_ASSERT(writeSession->Write(Msg("message_3.1", 5)));
 
-        TTestReadSession readSession("Session-0", client, 3, !autoscaleAwareSDK, {}, autoscaleAwareSDK);
+        TTestReadSession readSession("Session-0", client, 3, autoCommit, {}, autoscaleAwareSDK);
         readSession.Run();
         readSession.WaitAllMessages();
 
@@ -120,11 +120,15 @@ Y_UNIT_TEST_SUITE(TopicAutoscaling) {
     }
 
     Y_UNIT_TEST(ReadingAfterSplitTest_BeforeAutoscaleAwareSDK) {
-        ReadingAfterSplitTest(false);
+        ReadingAfterSplitTest(false, true);
     }
 
     Y_UNIT_TEST(ReadingAfterSplitTest_AutoscaleAwareSDK) {
-        ReadingAfterSplitTest(true);
+        ReadingAfterSplitTest(true, false);
+    }
+
+    Y_UNIT_TEST(ReadingAfterSplitTest_AutoscaleAwareSDK_AutoCommit) {
+        ReadingAfterSplitTest(true, false);
     }
 
     void ReadingAfterSplitTest_PreferedPartition(bool autoscaleAwareSDK) {