Supported topic autopartitioning for the transfer (#19043)

Author: nshestakov

ydb/core/persqueue/pq_impl.cpp

@@ -1802,8 +1802,6 @@ void TPersQueue::Handle(TEvPersQueue::TEvStatus::TPtr& ev, const TActorContext&
 {
     PQ_LOG_T("Handle TEvPersQueue::TEvStatus");
 
-    ReadBalancerActorId = ev->Sender;
-
     if (!ConfigInited || !AllOriginalPartitionsInited()) {
         PQ_LOG_D("Postpone the request." <<
                  " ConfigInited " << static_cast<int>(ConfigInited) <<

ydb/core/persqueue/read_balancer.cpp

@@ -534,6 +534,8 @@ void TPersQueueReadBalancer::RequestTabletIfNeeded(const ui64 tabletId, const TA
     } else {
         TActorId pipeClient = GetPipeClient(tabletId, ctx);
 
+        NTabletPipe::SendData(ctx, pipeClient, new TEvPQ::TEvSubDomainStatus(SubDomainOutOfSpace));
+
         auto it = AggregatedStats.Cookies.find(tabletId);
         if (!pipeReconnected || it != AggregatedStats.Cookies.end()) {
             ui64 cookie;
@@ -548,8 +550,6 @@ void TPersQueueReadBalancer::RequestTabletIfNeeded(const ui64 tabletId, const TA
                 TStringBuilder() << "Send TEvPersQueue::TEvStatus TabletId: " << tabletId << " Cookie: " << cookie);
             NTabletPipe::SendData(ctx, pipeClient, new TEvPersQueue::TEvStatus("", true), cookie);
         }
-
-        NTabletPipe::SendData(ctx, pipeClient, new TEvPQ::TEvSubDomainStatus(SubDomainOutOfSpace));
     }
 }
 

ydb/core/transfer/transfer_writer.cpp

@@ -818,6 +818,10 @@ class TTransferWriter
             Send(Worker, new TEvWorker::TEvPoll());
         }
 
+        if (LastWriteTime) {
+            LastWriteTime = TInstant::Now();
+        }
+
         return StartWork();
     }
 

ydb/core/transfer/ut/common/utils.h

@@ -168,7 +168,7 @@ struct MainTestCase {
         if (user) {
             config.SetAuthToken(TStringBuilder() << user.value() << "@builtin");
         }
-        // config.SetLog(std::unique_ptr<TLogBackend>(CreateLogBackend("cerr", ELogPriority::TLOG_INFO).Release()))
+        // config.SetLog(std::unique_ptr<TLogBackend>(CreateLogBackend("cerr", ELogPriority::TLOG_DEBUG).Release()));
         return config;
     }
 
@@ -184,7 +184,6 @@ struct MainTestCase {
         , TransferName(TStringBuilder() << "Transfer_" << Id)
         , Driver(CreateDriverConfig(ConnectionString, user))
         , TableClient(Driver)
-        , Session(TableClient.GetSession().GetValueSync().GetSession())
         , TopicClient(Driver)
     {
     }
@@ -193,9 +192,13 @@ struct MainTestCase {
         Driver.Stop(true);
     }
 
+    auto Session() {
+        return TableClient.GetSession().GetValueSync().GetSession();
+    }
+
     void ExecuteDDL(const std::string& ddl, bool checkResult = true, const std::optional<std::string> expectedMessage = std::nullopt) {
         Cerr << "DDL: " << ddl << Endl << Flush;
-        auto res = Session.ExecuteQuery(ddl, TTxControl::NoTx()).GetValueSync();
+        auto res = Session().ExecuteQuery(ddl, TTxControl::NoTx()).GetValueSync();
         if (checkResult) {
             if (expectedMessage) {
                 UNIT_ASSERT(!res.IsSuccess());
@@ -211,7 +214,7 @@ struct MainTestCase {
     auto ExecuteQuery(const std::string& query, bool retry = true) {
         for (size_t i = 10; i--;) {
             Cerr << ">>>>> Query: " << query << Endl << Flush;
-            auto res = Session.ExecuteQuery(query, TTxControl::NoTx()).GetValueSync();
+            auto res = Session().ExecuteQuery(query, TTxControl::NoTx()).GetValueSync();
             if (!res.IsSuccess()) {
                 Cerr << ">>>>> Query error: " << res.GetIssues().ToString() << Endl << Flush;
             }
@@ -282,13 +285,41 @@ struct MainTestCase {
         )", SourceTableName.data(), ChangefeedName.data()));
     }
 
+    struct CreatTopicSettings {
+        size_t MinPartitionCount = 1;
+        size_t MaxPartitionCount = 100;
+        bool AutoPartitioningEnabled = false;
+    };
+
     void CreateTopic(size_t partitionCount = 10) {
-        ExecuteDDL(Sprintf(R"(
-            CREATE TOPIC `%s`
-            WITH (
-                min_active_partitions = %d
-            );
-        )", TopicName.data(), partitionCount));
+        CreateTopic({
+            .MinPartitionCount = partitionCount
+        });
+    }
+
+    void CreateTopic(const CreatTopicSettings& settings) {
+        if (settings.AutoPartitioningEnabled) {
+            ExecuteDDL(Sprintf(R"(
+                CREATE TOPIC `%s`
+                WITH (
+                    MIN_ACTIVE_PARTITIONS = %d,
+                    MAX_ACTIVE_PARTITIONS = %d,
+                    AUTO_PARTITIONING_STRATEGY = 'UP',
+                    auto_partitioning_down_utilization_percent = 1,
+                    auto_partitioning_up_utilization_percent=2,
+                    auto_partitioning_stabilization_window = Interval('PT1S'),
+                    partition_write_speed_bytes_per_second = 3
+                );
+            )", TopicName.data(), settings.MinPartitionCount, settings.MaxPartitionCount));
+        } else {
+            ExecuteDDL(Sprintf(R"(
+                CREATE TOPIC `%s`
+                WITH (
+                    MIN_ACTIVE_PARTITIONS = %d
+
+                );
+            )", TopicName.data(), settings.MinPartitionCount));
+        }
     }
 
     void DropTopic() {
@@ -427,7 +458,7 @@ struct MainTestCase {
             setOptions = TStringBuilder() << "SET (" << sb << " )";
         }
 
-        auto res = Session.ExecuteQuery(Sprintf(R"(
+        auto res = Session().ExecuteQuery(Sprintf(R"(
             %s;
 
             ALTER TRANSFER `%s`
@@ -474,7 +505,7 @@ struct MainTestCase {
         settings.IncludeStats(true);
 
         auto c = TopicClient.DescribeConsumer(TopicName, consumerName, settings).GetValueSync();
-        UNIT_ASSERT(c.IsSuccess());
+        UNIT_ASSERT_C(c.IsSuccess(), c.GetIssues().ToOneLineString());
         return c;
     }
 
@@ -554,7 +585,9 @@ struct MainTestCase {
         settings.IncludeLocation(true);
         settings.IncludeStats(true);
 
-        return TopicClient.DescribeTopic(TopicName, settings).ExtractValueSync();
+        auto result = TopicClient.DescribeTopic(TopicName, settings).ExtractValueSync();
+        UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToOneLineString());
+        return result;
     }
 
     void CreateUser(const std::string& username) {
@@ -704,7 +737,6 @@ struct MainTestCase {
 
     TDriver Driver;
     TQueryClient TableClient;
-    TSession Session;
     TTopicClient TopicClient;
 };