Add PAUSED strategy for autopartitioning of the topic (#7002)

nshestakov · web-flow · commit 184bf806a7d2 · 2024-07-24T10:45:03.000+05:00
diff --git a/ydb/core/persqueue/ut/ut_with_sdk/autoscaling_ut.cpp b/ydb/core/persqueue/ut/ut_with_sdk/autoscaling_ut.cpp
@@ -741,6 +741,57 @@ Y_UNIT_TEST_SUITE(TopicAutoscaling) {
             auto v = f.GetValueSync();
             UNIT_ASSERT_C(v.IsSuccess(),  "Error: " << v);
         }
+
+        {
+            auto describeAfterAlter = client.DescribeTopic(topicName).GetValueSync();
+
+            UNIT_ASSERT_VALUES_EQUAL(describeAfterAlter.GetTopicDescription().GetPartitioningSettings().GetAutoPartitioningSettings().GetStrategy(), EAutoPartitioningStrategy::Disabled);
+        }
+    }
+
+    Y_UNIT_TEST(ControlPlane_PauseAutoPartitioning) {
+        auto topicName = "autoscalit-topic";
+
+        TTopicSdkTestSetup setup = CreateSetup();
+        TTopicClient client = setup.MakeClient();
+
+        {
+            TCreateTopicSettings createSettings;
+            createSettings
+                .BeginConfigurePartitioningSettings()
+                    .MinActivePartitions(1)
+                    .MaxActivePartitions(100)
+                    .BeginConfigureAutoPartitioningSettings()
+                        .Strategy(EAutoPartitioningStrategy::ScaleUp)
+                    .EndConfigureAutoPartitioningSettings()
+                .EndConfigurePartitioningSettings();
+            client.CreateTopic(topicName, createSettings).Wait();
+        }
+
+        {
+            TAlterTopicSettings alterSettings;
+            alterSettings
+                .BeginAlterPartitioningSettings()
+                    .MinActivePartitions(3)
+                    .MaxActivePartitions(107)
+                    .BeginAlterAutoPartitioningSettings()
+                        .Strategy(EAutoPartitioningStrategy::Paused)
+                    .EndAlterAutoPartitioningSettings()
+                .EndAlterTopicPartitioningSettings();
+            auto f = client.AlterTopic(topicName, alterSettings);
+            f.Wait();
+
+            auto v = f.GetValueSync();
+            UNIT_ASSERT_C(v.IsSuccess(),  "Error: " << v);
+        }
+
+        {
+            auto describeAfterAlter = client.DescribeTopic(topicName).GetValueSync();
+
+            UNIT_ASSERT_VALUES_EQUAL(describeAfterAlter.GetTopicDescription().GetPartitioningSettings().GetMinActivePartitions(), 3);
+            UNIT_ASSERT_VALUES_EQUAL(describeAfterAlter.GetTopicDescription().GetPartitioningSettings().GetMaxActivePartitions(), 107);
+            UNIT_ASSERT_VALUES_EQUAL(describeAfterAlter.GetTopicDescription().GetPartitioningSettings().GetAutoPartitioningSettings().GetStrategy(), EAutoPartitioningStrategy::Paused);
+        }
     }
 
     Y_UNIT_TEST(ControlPlane_AutoscalingWithStorageSizeRetention) {
diff --git a/ydb/core/protos/pqconfig.proto b/ydb/core/protos/pqconfig.proto
@@ -404,6 +404,7 @@ message TPQTabletConfig {
         CAN_SPLIT = 1;
         // The autoscaling algorithm will both increase and decrease partitions count depending on the load characteristics.
         CAN_SPLIT_AND_MERGE = 2;
+        PAUSED = 3;
     }
 
     // Strategy for automatically changing the number of topic partitions depending on the load
diff --git a/ydb/public/api/protos/draft/datastreams.proto b/ydb/public/api/protos/draft/datastreams.proto
@@ -289,6 +289,8 @@ enum AutoPartitioningStrategy {
     AUTO_PARTITIONING_STRATEGY_SCALE_UP = 2;
     // The auto partitioning algorithm will both increase and decrease partitions count depending on the load characteristics.
     AUTO_PARTITIONING_STRATEGY_SCALE_UP_AND_DOWN = 3;
+    // The auto partitioning is paused.
+    AUTO_PARTITIONING_STRATEGY_PAUSED = 4;
 }
 
 // Partitioning settings for stream.
diff --git a/ydb/public/api/protos/ydb_persqueue_v1.proto b/ydb/public/api/protos/ydb_persqueue_v1.proto
@@ -1092,6 +1092,8 @@ enum AutoPartitioningStrategy {
     AUTO_PARTITIONING_STRATEGY_SCALE_UP = 2;
     // The auto partitioning algorithm will both increase and decrease partitions count depending on the load characteristics.
     AUTO_PARTITIONING_STRATEGY_SCALE_UP_AND_DOWN = 3;
+    // The auto partitioning is paused.
+    AUTO_PARTITIONING_STRATEGY_PAUSED = 4;
 }
 
 /**
diff --git a/ydb/public/api/protos/ydb_topic.proto b/ydb/public/api/protos/ydb_topic.proto
@@ -842,6 +842,8 @@ enum AutoPartitioningStrategy {
     AUTO_PARTITIONING_STRATEGY_SCALE_UP = 2;
     // The auto partitioning algorithm will both increase and decrease partitions count depending on the load characteristics.
     AUTO_PARTITIONING_STRATEGY_SCALE_UP_AND_DOWN = 3;
+    // The auto partitioning is paused.
+    AUTO_PARTITIONING_STRATEGY_PAUSED = 4;
 }
 
 // Partitioning settings for topic.
diff --git a/ydb/public/sdk/cpp/client/ydb_datastreams/datastreams.cpp b/ydb/public/sdk/cpp/client/ydb_datastreams/datastreams.cpp
@@ -35,6 +35,9 @@ namespace NYdb::NDataStreams::V1 {
             case EAutoPartitioningStrategy::ScaleUpAndDown:
                 strategy = ::Ydb::DataStreams::V1::AutoPartitioningStrategy::AUTO_PARTITIONING_STRATEGY_SCALE_UP_AND_DOWN;
                 break;
+            case EAutoPartitioningStrategy::Paused:
+                strategy = ::Ydb::DataStreams::V1::AutoPartitioningStrategy::AUTO_PARTITIONING_STRATEGY_PAUSED;
+                break;
         }
 
         pt->mutable_auto_partitioning_settings()->set_strategy(strategy);
diff --git a/ydb/public/sdk/cpp/client/ydb_datastreams/datastreams.h b/ydb/public/sdk/cpp/client/ydb_datastreams/datastreams.h
@@ -106,6 +106,7 @@ namespace NYdb::NDataStreams::V1 {
         Disabled = 1,
         ScaleUp = 2,
         ScaleUpAndDown = 3,
+        Paused = 4
     };
 
     struct TCreateStreamSettings;
diff --git a/ydb/public/sdk/cpp/client/ydb_topic/include/control_plane.h b/ydb/public/sdk/cpp/client/ydb_topic/include/control_plane.h
@@ -35,6 +35,7 @@ enum class EAutoPartitioningStrategy: ui32 {
     Disabled = 1,
     ScaleUp = 2,
     ScaleUpAndDown = 3,
+    Paused = 4
 };
 
 class TConsumer {
diff --git a/ydb/services/datastreams/datastreams_proxy.cpp b/ydb/services/datastreams/datastreams_proxy.cpp
@@ -66,7 +66,8 @@ namespace NKikimr::NDataStreams::V1 {
 
         TString ValidatePartitioningSettings(const ::Ydb::DataStreams::V1::PartitioningSettings& s) {
             if (s.auto_partitioning_settings().strategy() == ::Ydb::DataStreams::V1::AutoPartitioningStrategy::AUTO_PARTITIONING_STRATEGY_SCALE_UP
-                || s.auto_partitioning_settings().strategy() == ::Ydb::DataStreams::V1::AutoPartitioningStrategy::AUTO_PARTITIONING_STRATEGY_SCALE_UP_AND_DOWN) {
+                || s.auto_partitioning_settings().strategy() == ::Ydb::DataStreams::V1::AutoPartitioningStrategy::AUTO_PARTITIONING_STRATEGY_SCALE_UP_AND_DOWN
+                || s.auto_partitioning_settings().strategy() == ::Ydb::DataStreams::V1::AutoPartitioningStrategy::AUTO_PARTITIONING_STRATEGY_PAUSED) {
 
                 if (s.min_active_partitions() < 0) {
                     return "min_active_partitions must be great than 0";
@@ -502,6 +503,10 @@ namespace NKikimr::NDataStreams::V1 {
                 case  Ydb::DataStreams::V1::AutoPartitioningStrategy::AUTO_PARTITIONING_STRATEGY_SCALE_UP_AND_DOWN:
                     t->SetPartitionStrategyType(NKikimrPQ::TPQTabletConfig_TPartitionStrategyType::TPQTabletConfig_TPartitionStrategyType_CAN_SPLIT_AND_MERGE);
                     break;
+
+                case  Ydb::DataStreams::V1::AutoPartitioningStrategy::AUTO_PARTITIONING_STRATEGY_PAUSED:
+                    t->SetPartitionStrategyType(NKikimrPQ::TPQTabletConfig_TPartitionStrategyType::TPQTabletConfig_TPartitionStrategyType_PAUSED);
+                    break;
             }
 
             auto& ws = as.partition_write_speed();
@@ -785,10 +790,13 @@ namespace NKikimr::NDataStreams::V1 {
         if (ps.GetPartitionStrategyType() != NKikimrPQ::TPQTabletConfig::TPartitionStrategyType::TPQTabletConfig_TPartitionStrategyType_DISABLED) {
             pt->set_min_active_partitions(ps.GetMinPartitionCount());
             pt->set_max_active_partitions(ps.GetMaxPartitionCount());
+
             if (ps.GetPartitionStrategyType() == NKikimrPQ::TPQTabletConfig::TPartitionStrategyType::TPQTabletConfig_TPartitionStrategyType_CAN_SPLIT) {
                 pt->mutable_auto_partitioning_settings()->set_strategy(::Ydb::DataStreams::V1::AutoPartitioningStrategy::AUTO_PARTITIONING_STRATEGY_SCALE_UP);
-            } else {
+            } else if (ps.GetPartitionStrategyType() == NKikimrPQ::TPQTabletConfig::TPartitionStrategyType::TPQTabletConfig_TPartitionStrategyType_CAN_SPLIT_AND_MERGE) {
                 pt->mutable_auto_partitioning_settings()->set_strategy(::Ydb::DataStreams::V1::AutoPartitioningStrategy::AUTO_PARTITIONING_STRATEGY_SCALE_UP_AND_DOWN);
+            } else {
+                pt->mutable_auto_partitioning_settings()->set_strategy(::Ydb::DataStreams::V1::AutoPartitioningStrategy::AUTO_PARTITIONING_STRATEGY_PAUSED);
             }
 
             pt->mutable_auto_partitioning_settings()->mutable_partition_write_speed()->mutable_stabilization_window()->set_seconds(ps.GetScaleThresholdSeconds());
diff --git a/ydb/services/lib/actors/pq_schema_actor.cpp b/ydb/services/lib/actors/pq_schema_actor.cpp
@@ -785,6 +785,9 @@ namespace NKikimr::NGRpcProxy::V1 {
                     case ::Ydb::PersQueue::V1::AutoPartitioningStrategy::AUTO_PARTITIONING_STRATEGY_SCALE_UP_AND_DOWN:
                         pqTabletConfigPartStrategy->SetPartitionStrategyType(::NKikimrPQ::TPQTabletConfig_TPartitionStrategyType::TPQTabletConfig_TPartitionStrategyType_CAN_SPLIT_AND_MERGE);
                         break;
+                    case ::Ydb::PersQueue::V1::AutoPartitioningStrategy::AUTO_PARTITIONING_STRATEGY_PAUSED:
+                        pqTabletConfigPartStrategy->SetPartitionStrategyType(::NKikimrPQ::TPQTabletConfig_TPartitionStrategyType::TPQTabletConfig_TPartitionStrategyType_PAUSED);
+                        break;
                     default:
                         pqTabletConfigPartStrategy->SetPartitionStrategyType(::NKikimrPQ::TPQTabletConfig_TPartitionStrategyType::TPQTabletConfig_TPartitionStrategyType_DISABLED);
                         break;
@@ -1110,6 +1113,9 @@ namespace NKikimr::NGRpcProxy::V1 {
                     case ::Ydb::Topic::AutoPartitioningStrategy::AUTO_PARTITIONING_STRATEGY_SCALE_UP_AND_DOWN:
                         pqTabletConfigPartStrategy->SetPartitionStrategyType(::NKikimrPQ::TPQTabletConfig_TPartitionStrategyType::TPQTabletConfig_TPartitionStrategyType_CAN_SPLIT_AND_MERGE);
                         break;
+                    case ::Ydb::Topic::AutoPartitioningStrategy::AUTO_PARTITIONING_STRATEGY_PAUSED:
+                        pqTabletConfigPartStrategy->SetPartitionStrategyType(::NKikimrPQ::TPQTabletConfig_TPartitionStrategyType::TPQTabletConfig_TPartitionStrategyType_PAUSED);
+                        break;
                     default:
                         pqTabletConfigPartStrategy->SetPartitionStrategyType(::NKikimrPQ::TPQTabletConfig_TPartitionStrategyType::TPQTabletConfig_TPartitionStrategyType_DISABLED);
                         break;
@@ -1290,6 +1296,9 @@ namespace NKikimr::NGRpcProxy::V1 {
                             case ::Ydb::Topic::AutoPartitioningStrategy::AUTO_PARTITIONING_STRATEGY_SCALE_UP_AND_DOWN:
                                 pqTabletConfig->MutablePartitionStrategy()->SetPartitionStrategyType(::NKikimrPQ::TPQTabletConfig_TPartitionStrategyType::TPQTabletConfig_TPartitionStrategyType_CAN_SPLIT_AND_MERGE);
                                 break;
+                            case ::Ydb::Topic::AutoPartitioningStrategy::AUTO_PARTITIONING_STRATEGY_PAUSED:
+                                pqTabletConfig->MutablePartitionStrategy()->SetPartitionStrategyType(::NKikimrPQ::TPQTabletConfig_TPartitionStrategyType::TPQTabletConfig_TPartitionStrategyType_PAUSED);
+                                break;
                             default:
                                 pqTabletConfig->MutablePartitionStrategy()->SetPartitionStrategyType(::NKikimrPQ::TPQTabletConfig_TPartitionStrategyType::TPQTabletConfig_TPartitionStrategyType_DISABLED);
                                 break;
diff --git a/ydb/services/persqueue_v1/actors/schema_actors.cpp b/ydb/services/persqueue_v1/actors/schema_actors.cpp
@@ -1091,6 +1091,9 @@ void TDescribeTopicActor::HandleCacheNavigateResponse(TEvTxProxySchemeCache::TEv
             case ::NKikimrPQ::TPQTabletConfig_TPartitionStrategyType::TPQTabletConfig_TPartitionStrategyType_CAN_SPLIT_AND_MERGE:
                 Result.mutable_partitioning_settings()->mutable_auto_partitioning_settings()->set_strategy(Ydb::Topic::AutoPartitioningStrategy::AUTO_PARTITIONING_STRATEGY_SCALE_UP_AND_DOWN);
                 break;
+            case ::NKikimrPQ::TPQTabletConfig_TPartitionStrategyType::TPQTabletConfig_TPartitionStrategyType_PAUSED:
+                Result.mutable_partitioning_settings()->mutable_auto_partitioning_settings()->set_strategy(Ydb::Topic::AutoPartitioningStrategy::AUTO_PARTITIONING_STRATEGY_PAUSED);
+                break;
             default:
                 Result.mutable_partitioning_settings()->mutable_auto_partitioning_settings()->set_strategy(Ydb::Topic::AutoPartitioningStrategy::AUTO_PARTITIONING_STRATEGY_DISABLED);
                 break;

Original file line number	Diff line number	Diff line change
`@@ -404,6 +404,7 @@ message TPQTabletConfig {`
`404`	`404`	`CAN_SPLIT = 1;`
`405`	`405`	`// The autoscaling algorithm will both increase and decrease partitions count depending on the load characteristics.`
`406`	`406`	`CAN_SPLIT_AND_MERGE = 2;`
	`407`	`+ PAUSED = 3;`
`407`	`408`	`}`
`408`	`409`
`409`	`410`	`// Strategy for automatically changing the number of topic partitions depending on the load`
Original file line number	Diff line number	Diff line change
`@@ -289,6 +289,8 @@ enum AutoPartitioningStrategy {`
`289`	`289`	`AUTO_PARTITIONING_STRATEGY_SCALE_UP = 2;`
`290`	`290`	`// The auto partitioning algorithm will both increase and decrease partitions count depending on the load characteristics.`
`291`	`291`	`AUTO_PARTITIONING_STRATEGY_SCALE_UP_AND_DOWN = 3;`
	`292`	`+ // The auto partitioning is paused.`
	`293`	`+ AUTO_PARTITIONING_STRATEGY_PAUSED = 4;`
`292`	`294`	`}`
`293`	`295`
`294`	`296`	`// Partitioning settings for stream.`
Original file line number	Diff line number	Diff line change
`@@ -1092,6 +1092,8 @@ enum AutoPartitioningStrategy {`
`1092`	`1092`	`AUTO_PARTITIONING_STRATEGY_SCALE_UP = 2;`
`1093`	`1093`	`// The auto partitioning algorithm will both increase and decrease partitions count depending on the load characteristics.`
`1094`	`1094`	`AUTO_PARTITIONING_STRATEGY_SCALE_UP_AND_DOWN = 3;`
	`1095`	`+ // The auto partitioning is paused.`
	`1096`	`+ AUTO_PARTITIONING_STRATEGY_PAUSED = 4;`
`1095`	`1097`	`}`
`1096`	`1098`
`1097`	`1099`	`/**`
Original file line number	Diff line number	Diff line change
`@@ -842,6 +842,8 @@ enum AutoPartitioningStrategy {`
`842`	`842`	`AUTO_PARTITIONING_STRATEGY_SCALE_UP = 2;`
`843`	`843`	`// The auto partitioning algorithm will both increase and decrease partitions count depending on the load characteristics.`
`844`	`844`	`AUTO_PARTITIONING_STRATEGY_SCALE_UP_AND_DOWN = 3;`
	`845`	`+ // The auto partitioning is paused.`
	`846`	`+ AUTO_PARTITIONING_STRATEGY_PAUSED = 4;`
`845`	`847`	`}`
`846`	`848`
`847`	`849`	`// Partitioning settings for topic.`
Original file line number	Diff line number	Diff line change
`@@ -35,6 +35,9 @@ namespace NYdb::NDataStreams::V1 {`
`35`	`35`	`case EAutoPartitioningStrategy::ScaleUpAndDown:`
`36`	`36`	`strategy = ::Ydb::DataStreams::V1::AutoPartitioningStrategy::AUTO_PARTITIONING_STRATEGY_SCALE_UP_AND_DOWN;`
`37`	`37`	`break;`
	`38`	`+ case EAutoPartitioningStrategy::Paused:`
	`39`	`+ strategy = ::Ydb::DataStreams::V1::AutoPartitioningStrategy::AUTO_PARTITIONING_STRATEGY_PAUSED;`
	`40`	`+ break;`
`38`	`41`	`}`
`39`	`42`
`40`	`43`	`pt->mutable_auto_partitioning_settings()->set_strategy(strategy);`