An empty list of consumers for the background partition (#15889)

Author: Alek5andr-Kotov

Diff for: ydb/core/persqueue/partition_init.cpp

@@ -400,6 +400,7 @@ void TInitInfoRangeStep::Handle(TEvKeyValue::TEvResponse::TPtr &ev, const TActor
         case NKikimrProto::OVERRUN: {
             auto& sourceIdStorage = Partition()->SourceIdStorage;
             auto& usersInfoStorage = Partition()->UsersInfoStorage;
+            const bool isSupportive = Partition()->IsSupportive();
 
             for (ui32 i = 0; i < range.PairSize(); ++i) {
                 const auto& pair = range.GetPair(i);
@@ -421,9 +422,9 @@ void TInitInfoRangeStep::Handle(TEvKeyValue::TEvResponse::TPtr &ev, const TActor
                     sourceIdStorage.LoadSourceIdInfo(*key, pair.GetValue(), now);
                 } else if (type == TKeyPrefix::MarkProtoSourceId) {
                     sourceIdStorage.LoadSourceIdInfo(*key, pair.GetValue(), now);
-                } else if (type == TKeyPrefix::MarkUser) {
+                } else if ((type == TKeyPrefix::MarkUser) && !isSupportive) {
                     usersInfoStorage->Parse(*key, pair.GetValue(), ctx);
-                } else if (type == TKeyPrefix::MarkUserDeprecated) {
+                } else if ((type == TKeyPrefix::MarkUserDeprecated) && !isSupportive) {
                     usersInfoStorage->ParseDeprecated(*key, pair.GetValue(), ctx);
                 }
             }

Diff for: ydb/core/persqueue/partition_read.cpp

@@ -51,6 +51,10 @@ void TPartition::SendReadingFinished(const TString& consumer) {
 }
 
 void TPartition::FillReadFromTimestamps(const TActorContext& ctx) {
+    if (IsSupportive()) {
+        return;
+    }
+
     TSet<TString> hasReadRule;
 
     for (auto& [consumer, userInfo] : UsersInfoStorage->GetAll()) {

Diff for: ydb/public/sdk/cpp/src/client/topic/ut/resources/topic_A_partition_0_v24-4-2.dat

@@ -16,6 +16,7 @@
 
 #include <library/cpp/logger/stream.h>
 #include <library/cpp/testing/unittest/registar.h>
+#include <library/cpp/streams/bzip2/bzip2.h>
 
 namespace NYdb::NTopic::NTests {
 
@@ -150,6 +151,9 @@ class TFixture : public NUnitTest::TBaseFixture {
     void RestartLongTxService();
     void RestartPQTablet(const TString& topicPath, ui32 partition);
     void DumpPQTabletKeys(const TString& topicName, ui32 partition);
+    void PQTabletPrepareFromResource(const TString& topicPath,
+                                     ui32 partitionId,
+                                     const TString& resourceName);
 
     void DeleteSupportivePartition(const TString& topicName,
                                    ui32 partition);
@@ -1857,6 +1861,51 @@ void TFixture::DumpPQTabletKeys(const TString& topicName)
     }
 }
 
+void TFixture::PQTabletPrepareFromResource(const TString& topicPath,
+                                           ui32 partitionId,
+                                           const TString& resourceName)
+{
+    auto& runtime = Setup->GetRuntime();
+    TActorId edge = runtime.AllocateEdgeActor();
+    ui64 tabletId = GetTopicTabletId(edge, "/Root/" + topicPath, partitionId);
+
+    auto request = MakeHolder<TEvKeyValue::TEvRequest>();
+    size_t count = 0;
+
+    for (TStringStream stream(NResource::Find(resourceName)); true; ++count) {
+        TString key, encoded;
+
+        if (!stream.ReadTo(key, ' ')) {
+            break;
+        }
+        encoded = stream.ReadLine();
+
+        auto decoded = Base64Decode(encoded);
+        TStringInput decodedStream(decoded);
+        TBZipDecompress decompressor(&decodedStream);
+
+        auto* cmd = request->Record.AddCmdWrite();
+        cmd->SetKey(key);
+        cmd->SetValue(decompressor.ReadAll());
+    }
+
+    runtime.SendToPipe(tabletId, edge, request.Release(), 0, GetPipeConfigWithRetries());
+
+    TAutoPtr<IEventHandle> handle;
+    auto* response = runtime.GrabEdgeEvent<TEvKeyValue::TEvResponse>(handle);
+    UNIT_ASSERT(response);
+    UNIT_ASSERT(response->Record.HasStatus());
+    UNIT_ASSERT_EQUAL(response->Record.GetStatus(), NMsgBusProxy::MSTATUS_OK);
+
+    UNIT_ASSERT_VALUES_EQUAL(response->Record.WriteResultSize(), count);
+
+    for (size_t i = 0; i < response->Record.WriteResultSize(); ++i) {
+        const auto &result = response->Record.GetWriteResult(i);
+        UNIT_ASSERT(result.HasStatus());
+        UNIT_ASSERT_EQUAL(result.GetStatus(), NKikimrProto::OK);
+    }
+}
+
 void TFixture::TestTheCompletionOfATransaction(const TTransactionCompletionTestDescription& d)
 {
     for (auto& topic : d.Topics) {
@@ -3207,6 +3256,20 @@ Y_UNIT_TEST_F(Transactions_Conflict_On_SeqNo, TFixture)
     UNIT_ASSERT_VALUES_UNEQUAL(successCount, TXS_COUNT);
 }
 
+Y_UNIT_TEST_F(The_Transaction_Starts_On_One_Version_And_Ends_On_The_Other, TFixture)
+{
+    // In the test, we check the compatibility between versions `24-4-2` and `24-4-*/25-1-*`. To do this, the data
+    // obtained on the `24-4-2` version is loaded into the PQ tablets.
+
+    CreateTopic("topic_A", TEST_CONSUMER, 2);
+
+    PQTabletPrepareFromResource("topic_A", 0, "topic_A_partition_0_v24-4-2.dat");
+    PQTabletPrepareFromResource("topic_A", 1, "topic_A_partition_1_v24-4-2.dat");
+
+    RestartPQTablet("topic_A", 0);
+    RestartPQTablet("topic_A", 1);
+}
+
 }
 
 }

Diff for: ydb/public/sdk/cpp/src/client/topic/ut/resources/topic_A_partition_1_v24-4-2.dat

@@ -40,4 +40,9 @@ SRCS(
     trace_ut.cpp
 )
 
+RESOURCE(
+    ydb/public/sdk/cpp/src/client/topic/ut/resources/topic_A_partition_0_v24-4-2.dat topic_A_partition_0_v24-4-2.dat
+    ydb/public/sdk/cpp/src/client/topic/ut/resources/topic_A_partition_1_v24-4-2.dat topic_A_partition_1_v24-4-2.dat
+)
+
 END()