[+] test

Author: Alek5andr-Kotov

ydb/public/sdk/cpp/client/ydb_topic/ut/topic_to_table_ut.cpp

@@ -42,6 +42,10 @@ class TFixture : public NUnitTest::TBaseFixture {
         void WaitForEvent();
     };
 
+    struct TAlterTopicParameters {
+        TMaybe<ui32> PartitionsCount;
+    };
+
     void SetUp(NUnitTest::TTestContext&) override;
 
     NTable::TSession CreateTableSession();
@@ -67,6 +71,8 @@ class TFixture : public NUnitTest::TBaseFixture {
                      std::optional<size_t> maxPartitionCount = std::nullopt);
     void DescribeTopic(const TString& path);
 
+    void AddConsumer(const TString& topic, const TVector<TString>& consumers);
+
     void WriteToTopicWithInvalidTxId(bool invalidTxId);
 
     TTopicWriteSessionPtr CreateTopicWriteSession(const TString& topicPath,
@@ -198,6 +204,7 @@ void TFixture::SetUp(NUnitTest::TTestContext&)
 {
     NKikimr::Tests::TServerSettings settings = TTopicSdkTestSetup::MakeServerSettings();
     settings.SetEnableTopicServiceTx(true);
+    settings.SetEnablePQConfigTransactionsAtSchemeShard(true);
     Setup = std::make_unique<TTopicSdkTestSetup>(TEST_CASE_NAME, settings);
 
     Driver = std::make_unique<TDriver>(Setup->MakeDriver());
@@ -329,6 +336,20 @@ void TFixture::CreateTopic(const TString& path,
     Setup->CreateTopic(path, consumer, partitionCount, maxPartitionCount);
 }
 
+void TFixture::AddConsumer(const TString& path,
+                           const TVector<TString>& consumers)
+{
+    NTopic::TTopicClient client(GetDriver());
+    NTopic::TAlterTopicSettings settings;
+
+    for (const auto& consumer : consumers) {
+        settings.BeginAddConsumer(consumer);
+    }
+
+    auto result = client.AlterTopic(path, settings).GetValueSync();
+    UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToString());
+}
+
 void TFixture::DescribeTopic(const TString& path)
 {
     Setup->DescribeTopic(path);
@@ -871,6 +892,23 @@ void TFixture::RestartLongTxService()
     }
 }
 
+Y_UNIT_TEST_F(WriteToTopic_Demo_39, TFixture)
+{
+    CreateTopic("topic_A", TEST_CONSUMER);
+
+    NTable::TSession tableSession = CreateTableSession();
+    NTable::TTransaction tx = BeginTx(tableSession);
+
+    WriteToTopic("topic_A", TEST_MESSAGE_GROUP_ID, "message #1", &tx);
+    WriteToTopic("topic_A", TEST_MESSAGE_GROUP_ID, "message #2", &tx);
+
+    WaitForAcks("topic_A", TEST_MESSAGE_GROUP_ID);
+
+    AddConsumer("topic_A", {"consumer"});
+
+    CommitTx(tx, EStatus::SUCCESS);
+}
+
 Y_UNIT_TEST_F(WriteToTopic_Demo_1, TFixture)
 {
     CreateTopic("topic_A");