Fixed the deletion of blobs by retention in the supportive partition of the topic (#19916)

Author: nshestakov

ydb/core/persqueue/partition.cpp

@@ -403,6 +403,10 @@ void TPartition::AddMetaKey(TEvKeyValue::TEvRequest* request) {
 }
 
 bool TPartition::CleanUp(TEvKeyValue::TEvRequest* request, const TActorContext& ctx) {
+    if (IsSupportive()) {
+        return false;
+    }
+
     bool haveChanges = CleanUpBlobs(request, ctx);
 
     PQ_LOG_T("Have " << request->Record.CmdDeleteRangeSize() << " items to delete old stuff");

ydb/public/sdk/cpp/src/client/topic/ut/topic_to_table_ut.cpp

@@ -84,7 +84,9 @@ class TFixture : public NUnitTest::TBaseFixture {
     void CreateTopic(const TString& path = TString{TEST_TOPIC},
                      const TString& consumer = TEST_CONSUMER,
                      size_t partitionCount = 1,
-                     std::optional<size_t> maxPartitionCount = std::nullopt);
+                     std::optional<size_t> maxPartitionCount = std::nullopt,
+                     const TDuration retention = TDuration::Hours(1),
+                     bool important = false);
     TTopicDescription DescribeTopic(const TString& path);
 
     void AddConsumer(const TString& topicPath, const TVector<TString>& consumers);
@@ -474,10 +476,12 @@ void TFixture::WriteMessages(const TVector<TString>& messages,
 void TFixture::CreateTopic(const TString& path,
                            const TString& consumer,
                            size_t partitionCount,
-                           std::optional<size_t> maxPartitionCount)
+                           std::optional<size_t> maxPartitionCount,
+                           const TDuration retention,
+                           bool important)
 
 {
-    Setup->CreateTopic(path, consumer, partitionCount, maxPartitionCount);
+    Setup->CreateTopic(path, consumer, partitionCount, maxPartitionCount, retention, important);
 }
 
 void TFixture::AddConsumer(const TString& topicPath,
@@ -3286,6 +3290,56 @@ Y_UNIT_TEST_F(The_Transaction_Starts_On_One_Version_And_Ends_On_The_Other, TFixt
     RestartPQTablet("topic_A", 1);
 }
 
+Y_UNIT_TEST_F(TestRetentionOnLongTxAndBigMessages, TFixture)
+{
+
+    auto bigMessage = []() {
+        TStringBuilder sb;
+        sb.reserve(10_MB);
+        for (size_t i = 0; i < sb.capacity(); ++i) {
+            sb << RandomNumber<char>();
+        }
+
+        return sb;
+    };
+
+    TString msg = bigMessage();
+
+    CreateTopic("topic_A", TEST_CONSUMER, 1, 1, TDuration::Seconds(1), true);
+
+    auto session = CreateTableSession();
+    auto tx0 = BeginTx(session);
+    auto tx1 = BeginTx(session);
+
+    WriteToTopic("topic_A", "grp-0", msg, &tx0);
+    WriteToTopic("topic_A", "grp-1", msg, &tx1);
+
+    Sleep(TDuration::Seconds(3));
+
+    WriteToTopic("topic_A", "grp-0", "short-message", &tx0);
+    WriteToTopic("topic_A", "grp-1", "short-message", &tx1);
+
+    WriteToTopic("topic_A", "grp-0", msg, &tx0);
+    WriteToTopic("topic_A", "grp-1", msg, &tx1);
+
+    Sleep(TDuration::Seconds(3));
+
+    WriteToTopic("topic_A", "grp-0", msg, &tx0);
+    WriteToTopic("topic_A", "grp-1", msg, &tx1);
+
+    Sleep(TDuration::Seconds(3));
+
+    CommitTx(tx0);
+    CommitTx(tx1);
+
+    //RestartPQTablet("topic_A", 0);
+
+    auto read = ReadFromTopic("topic_A", TEST_CONSUMER, TDuration::Seconds(2));
+    UNIT_ASSERT(read.size() > 0);
+    UNIT_ASSERT_EQUAL(msg, read[0]);
+}
+
+
 }
 
 }

ydb/public/sdk/cpp/src/client/topic/ut/ut_utils/topic_sdk_test_setup.cpp

@@ -26,12 +26,13 @@ void TTopicSdkTestSetup::CreateTopicWithAutoscale(const std::string& path, const
     CreateTopic(path, consumer, partitionCount, maxPartitionCount);
 }
 
-void TTopicSdkTestSetup::CreateTopic(const std::string& path, const std::string& consumer, size_t partitionCount, std::optional<size_t> maxPartitionCount)
+void TTopicSdkTestSetup::CreateTopic(const std::string& path, const std::string& consumer, size_t partitionCount, std::optional<size_t> maxPartitionCount, const TDuration retention, bool important)
 {
     TTopicClient client(MakeDriver());
 
     TCreateTopicSettings topics;
     topics
+        .RetentionPeriod(retention)
         .BeginConfigurePartitioningSettings()
         .MinActivePartitions(partitionCount)
         .MaxActivePartitions(maxPartitionCount.value_or(partitionCount));
@@ -44,6 +45,7 @@ void TTopicSdkTestSetup::CreateTopic(const std::string& path, const std::string&
     }
 
     TConsumerSettings<TCreateTopicSettings> consumers(topics, consumer);
+    consumers.Important(important);
     topics.AppendConsumers(consumers);
 
     auto status = client.CreateTopic(path, topics).GetValueSync();

ydb/public/sdk/cpp/src/client/topic/ut/ut_utils/topic_sdk_test_setup.h

@@ -17,7 +17,7 @@ class TTopicSdkTestSetup {
     TTopicSdkTestSetup(const TString& testCaseName, const NKikimr::Tests::TServerSettings& settings = MakeServerSettings(), bool createTopic = true);
 
     void CreateTopic(const std::string& path = TEST_TOPIC, const std::string& consumer = TEST_CONSUMER, size_t partitionCount = 1,
-                     std::optional<size_t> maxPartitionCount = std::nullopt);
+                     std::optional<size_t> maxPartitionCount = std::nullopt, const TDuration retention = TDuration::Hours(1), bool important = false);
     void CreateTopicWithAutoscale(const std::string& path = TEST_TOPIC, const std::string& consumer = TEST_CONSUMER, size_t partitionCount = 1,
                      size_t maxPartitionCount = 100);