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

Author: nshestakov

.github/last_commit.txt

@@ -1 +1 @@
-6065f1e3f7655f65b808df9cee77081380d85c3b
+693a9330006deec1f2ced3ef39abf3db14e13fc4

src/client/topic/ut/topic_to_table_ut.cpp

@@ -111,7 +111,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);
@@ -862,10 +864,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,
@@ -4429,6 +4433,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, TFixtureQuery)
+{
+    // TODO uncomment
+    return;
+
+    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 = CreateSession();
+    auto tx0 = session->BeginTx();
+    auto tx1 = session->BeginTx();
+
+    WriteToTopic("topic_A", "grp-0", msg, tx0.get());
+    WriteToTopic("topic_A", "grp-1", msg, tx1.get());
+
+    Sleep(TDuration::Seconds(3));
+
+    WriteToTopic("topic_A", "grp-0", "short-msg", tx0.get());
+    WriteToTopic("topic_A", "grp-1", "short-msg", tx1.get());
+
+    WriteToTopic("topic_A", "grp-0", msg, tx0.get());
+    WriteToTopic("topic_A", "grp-1", msg, tx1.get());
+
+    Sleep(TDuration::Seconds(3));
+
+    WriteToTopic("topic_A", "grp-0", msg, tx0.get());
+    WriteToTopic("topic_A", "grp-1", msg, tx1.get());
+
+    Sleep(TDuration::Seconds(3));
+
+    session->CommitTx(*tx0, EStatus::SUCCESS);
+    session->CommitTx(*tx1, EStatus::SUCCESS);
+
+    //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]);
+}
+
 }
 
 }

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();

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);