Merge 61db074 into 6df1995

Author: stanislav-shchetinin

ydb/core/tx/schemeshard/schemeshard_import__create.cpp

@@ -541,6 +541,21 @@ struct TSchemeShard::TImport::TTxProgress: public TSchemeShard::TXxport::TTxBase
  Send(Self->SelfId(), CreateChangefeedPropose(Self, txId, item));
  }
 
+ void CreateConsumers(TImportInfo::TPtr importInfo, ui32 itemIdx, TTxId txId) {
+ Y_ABORT_UNLESS(itemIdx < importInfo->Items.size());
+ auto& item = importInfo->Items.at(itemIdx);
+ item.SubState = ESubState::Proposed;
+
+ LOG_I("TImport::TTxProgress: CreateConsumers propose"
+ << ": info# " << importInfo->ToString()
+ << ", item# " << item.ToString(itemIdx)
+ << ", txId# " << txId);
+
+ Y_ABORT_UNLESS(item.WaitTxId == InvalidTxId);
+
+ Send(Self->SelfId(), CreateConsumersPropose(Self, txId, item));
+ }
+
  void AllocateTxId(TImportInfo::TPtr importInfo, ui32 itemIdx) {
  Y_ABORT_UNLESS(itemIdx < importInfo->Items.size());
  auto& item = importInfo->Items.at(itemIdx);
@@ -1046,6 +1061,7 @@ struct TSchemeShard::TImport::TTxProgress: public TSchemeShard::TXxport::TTxBase
 
  case EState::CreateChangefeed:
  CreateChangefeed(importInfo, i, txId);
+ CreateConsumers(importInfo, i, txId);
  itemIdx = i;
  break;
 

ydb/core/tx/schemeshard/schemeshard_import_flow_proposals.cpp

@@ -291,5 +291,33 @@ THolder<TEvSchemeShard::TEvModifySchemeTransaction> CreateChangefeedPropose(
  return propose;
 }
 
+THolder<TEvSchemeShard::TEvModifySchemeTransaction> CreateConsumersPropose(
+ TSchemeShard* ss,
+ TTxId txId,
+ const TImportInfo::TItem& item
+) {
+ Y_ABORT_UNLESS(item.NextChangefeedIdx < item.Changefeeds.GetChangefeeds().size());
+
+ const auto& importChangefeedTopic = item.Changefeeds.GetChangefeeds()[item.NextChangefeedIdx];
+ const auto& topic = importChangefeedTopic.GetTopic();
+
+ auto propose = MakeHolder<TEvSchemeShard::TEvModifySchemeTransaction>(ui64(txId), ss->TabletID());
+ auto& record = propose->Record;
+ auto& modifyScheme = *record.AddTransaction();
+ modifyScheme.SetOperationType(NKikimrSchemeOp::EOperationType::ESchemeOpAlterPersQueueGroup);
+ auto& pqGroup = *modifyScheme.MutableAlterPersQueueGroup();
+
+ const TPath dstPath = TPath::Init(item.DstPathId, ss);
+ modifyScheme.SetWorkingDir(dstPath.PathString());
+ pqGroup.SetName(importChangefeedTopic.GetChangefeed().name());
+
+ for (const auto& consumer : topic.consumers()) {
+ auto pqConsumer = *pqGroup.MutablePQTabletConfig()->AddConsumers();
+ *pqConsumer.MutableName() = consumer.name();
+ }
+ 
+ return propose;
+}
+
 } // NSchemeShard
 } // NKikimr

ydb/core/tx/schemeshard/schemeshard_import_flow_proposals.h

@@ -52,5 +52,11 @@ THolder<TEvSchemeShard::TEvModifySchemeTransaction> CreateChangefeedPropose(
  const TImportInfo::TItem& item
 );
 
+THolder<TEvSchemeShard::TEvModifySchemeTransaction> CreateConsumersPropose(
+ TSchemeShard* ss,
+ TTxId txId,
+ const TImportInfo::TItem& item
+);
+
 } // NSchemeShard
 } // NKikimr

ydb/core/tx/schemeshard/ut_helpers/ls_checks.cpp

@@ -968,6 +968,12 @@ TCheckFunc RetentionPeriod(const TDuration& value) {
  };
 }
 
+TCheckFunc ConsumerExist(const TString& name) {
+ return [=] (const NKikimrScheme::TEvDescribeSchemeResult& record) {
+ UNIT_ASSERT_VALUES_EQUAL(record.GetPathDescription().GetPersQueueGroup().GetPQTabletConfig().GetConsumers(0).GetName(), name);
+ };
+}
+
 void NoChildren(const NKikimrScheme::TEvDescribeSchemeResult& record) {
  ChildrenCount(0)(record);
 }

ydb/core/tx/schemeshard/ut_helpers/ls_checks.h

@@ -169,6 +169,7 @@ namespace NLs {
  TCheckFunc StreamAwsRegion(const TString& value);
  TCheckFunc StreamInitialScanProgress(ui32 total, ui32 completed);
  TCheckFunc RetentionPeriod(const TDuration& value);
+ TCheckFunc ConsumerExist(const TString& name);
 
  TCheckFunc HasBackupInFly(ui64 txId);
  void NoBackupInFly(const NKikimrScheme::TEvDescribeSchemeResult& record);

ydb/core/tx/schemeshard/ut_restore/ut_restore.cpp

@@ -4897,7 +4897,8 @@ Y_UNIT_TEST_SUITE(TImportTests) {
  {changefeedPath, GenerateTestData({EPathTypeCdcStream, changefeedDesc, std::move(attr)})},
  [changefeedPath = TString(changefeedPath)](TTestBasicRuntime& runtime) {
  TestDescribeResult(DescribePath(runtime, "/MyRoot/Table" + changefeedPath, false, false, true), {
- NLs::PathExist
+ NLs::PathExist,
+ NLs::ConsumerExist("my_consumer")
  });
  }
  };