fixed CDC write to the autopartitioning topic (#20243)

Author: nshestakov

ydb/core/persqueue/ut/slow/autopartitioning_ut.cpp

@@ -24,9 +24,20 @@ Y_UNIT_TEST_SUITE(SlowTopicAutopartitioning) {
         UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), NYdb::EStatus::SUCCESS, result.GetIssues().ToString());
     }
 
-    void ExecuteDataQuery(NYdb::NTable::TSession& session, const TString& query ) {
-        const auto result = session.ExecuteDataQuery(query, NYdb::NTable::TTxControl::BeginTx().CommitTx()).GetValueSync();
-        UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), NYdb::EStatus::SUCCESS, result.GetIssues().ToString());
+    void ExecuteDataQuery(NYdb::NTable::TSession& session, const TString& query) {
+        TString error;
+        for (size_t i = 0; i < 20; ++i) {
+            const auto result = session.ExecuteDataQuery(query, NYdb::NTable::TTxControl::BeginTx().CommitTx()).GetValueSync();
+            if (NYdb::EStatus::SUCCESS != result.GetStatus()) {
+                error = result.GetIssues().ToString();
+                Sleep(TDuration::MilliSeconds(500));
+                continue;
+            }
+            UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), NYdb::EStatus::SUCCESS, result.GetIssues().ToString());
+            return;
+        }
+
+        UNIT_ASSERT_C(false, "ExecuteDataQuery error: " << error);
     }
 
     ui64 GetBalancerTabletId(TTopicSdkTestSetup& setup, const TString& topicPath) {
@@ -68,6 +79,8 @@ Y_UNIT_TEST_SUITE(SlowTopicAutopartitioning) {
 
         size_t count = 0;
 
+        std::set<ui32> partitions;
+
         auto reader = client.CreateReadSession(
             TReadSessionSettings()
                 .AutoPartitioningSupport(true)
@@ -78,6 +91,7 @@ Y_UNIT_TEST_SUITE(SlowTopicAutopartitioning) {
                 if (auto* x = std::get_if<NYdb::NTopic::TReadSessionEvent::TDataReceivedEvent>(&event)) {
                     count += x->GetMessages().size();
                 } else if (auto* x = std::get_if<NYdb::NTopic::TReadSessionEvent::TStartPartitionSessionEvent>(&event)) {
+                    partitions.insert(x->GetPartitionSession()->GetPartitionId());
                     x->Confirm();
                     Cerr << ">>>>> " << x->DebugString() << Endl << Flush;
                 } else if (auto* x = std::get_if<NYdb::NTopic::TReadSessionEvent::TCommitOffsetAcknowledgementEvent>(&event)) {
@@ -105,6 +119,8 @@ Y_UNIT_TEST_SUITE(SlowTopicAutopartitioning) {
         }
 
         UNIT_ASSERT_VALUES_EQUAL(expected, count);
+        UNIT_ASSERT_C(partitions.size() > 1, "Split must be happened");
+        Cerr << ">>>>> Partition count: " << partitions.size() << Endl << Flush;
     }
 
     Y_UNIT_TEST(CDC_Write) {
@@ -121,9 +137,12 @@ Y_UNIT_TEST_SUITE(SlowTopicAutopartitioning) {
                 value Text,
                 PRIMARY KEY (id, order)
             ) WITH (
-                AUTO_PARTITIONING_MIN_PARTITIONS_COUNT = 64,
+                AUTO_PARTITIONING_BY_SIZE = ENABLED,
+                AUTO_PARTITIONING_PARTITION_SIZE_MB = 5,
+                AUTO_PARTITIONING_BY_LOAD = ENABLED,
+                AUTO_PARTITIONING_MIN_PARTITIONS_COUNT = 1,
                 AUTO_PARTITIONING_MAX_PARTITIONS_COUNT = 64,
-                UNIFORM_PARTITIONS = 64
+                UNIFORM_PARTITIONS = 1
             );
         )");
 
@@ -134,7 +153,8 @@ Y_UNIT_TEST_SUITE(SlowTopicAutopartitioning) {
                     MODE = 'UPDATES',
                     FORMAT = 'JSON',
                     TOPIC_AUTO_PARTITIONING = 'ENABLED',
-                    TOPIC_MIN_ACTIVE_PARTITIONS = 2
+                    TOPIC_MIN_ACTIVE_PARTITIONS = 1,
+                    TOPIC_MAX_ACTIVE_PARTITIONS = 100
                 );
         )");
 
@@ -143,10 +163,10 @@ Y_UNIT_TEST_SUITE(SlowTopicAutopartitioning) {
             alterSettings
                 .BeginAlterPartitioningSettings()
                     .MinActivePartitions(1)
-                    .MaxActivePartitions(10000)
+                    .MaxActivePartitions(100)
                     .BeginAlterAutoPartitioningSettings()
                         .Strategy(EAutoPartitioningStrategy::ScaleUp)
-                        .StabilizationWindow(TDuration::Seconds(1))
+                        .StabilizationWindow(TDuration::Seconds(15))
                         .DownUtilizationPercent(1)
                         .UpUtilizationPercent(2)
                     .EndAlterAutoPartitioningSettings()
@@ -162,23 +182,25 @@ Y_UNIT_TEST_SUITE(SlowTopicAutopartitioning) {
         }
 
         Cerr << ">>>>> " << TInstant::Now() << " Start table insert" << Endl << Flush;
-        ExecuteDataQuery(session, R"(
-            --!syntax_v1
-            $sample = AsList(
-                AsStruct(ListFromRange(0, 150000) AS v)
-            );
+        for (size_t i = 0; i < 50; ++i) {
+            ExecuteDataQuery(session, R"(
+                --!syntax_v1
+                $sample = AsList(
+                    AsStruct(ListFromRange(0, 5000) AS v)
+                );
 
-            UPSERT INTO `/Root/origin` (id, order, value)
-            SELECT
-                RandomNumber(v) AS id,
-                v AS order,
-                CAST('0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF' AS Utf8?)  AS value
-            FROM as_table($sample)
-                FLATTEN BY (v);
-        )");
+                UPSERT INTO `/Root/origin` (id, order, value)
+                SELECT
+                    RandomNumber(v) AS id,
+                    CAST(v AS Uint64) AS order,
+                    CAST('0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF' AS Utf8?)  AS value
+                FROM as_table($sample)
+                    FLATTEN BY (v);
+            )");
+        }
 
         Cerr << ">>>>> " << TInstant::Now() << " Start read topic" << Endl << Flush;
-        AssertMessageCountInTopic(client, "/Root/origin/feed/streamImpl", 150000);
+        AssertMessageCountInTopic(client, "/Root/origin/feed/streamImpl", 250000);
         Cerr << ">>>>> " << TInstant::Now() << " End" << Endl << Flush;
     }
 }

ydb/core/tx/datashard/change_exchange_split.cpp

@@ -360,6 +360,10 @@ class TCdcWorker
             const auto partitionId = partition.GetPartitionId();
             const auto tabletId = partition.GetTabletId();
 
+            if (NKikimrPQ::ETopicPartitionStatus::Active != partition.GetStatus()) {
+                continue;
+            }
+
             auto it = Workers.find(partitionId);
             if (it != Workers.end()) {
                 workers.emplace(partitionId, it->second);