Fix initialization of explicit messages groups (ydb-platform#8739)

nshestakov · spuchin · commit 6de66af86c09 · 2024-09-09T14:52:49.000Z
diff --git a/ydb/core/persqueue/events/internal.h b/ydb/core/persqueue/events/internal.h
@@ -576,13 +576,15 @@ struct TEvPQ {
     };
 
     struct TEvChangePartitionConfig : public TEventLocal<TEvChangePartitionConfig, EvChangePartitionConfig> {
-        TEvChangePartitionConfig(const NPersQueue::TTopicConverterPtr& topicConverter, const NKikimrPQ::TPQTabletConfig& config)
+        TEvChangePartitionConfig(const NPersQueue::TTopicConverterPtr& topicConverter, const NKikimrPQ::TPQTabletConfig& config, const NKikimrPQ::TBootstrapConfig& bootstrapConfig)
             : TopicConverter(topicConverter)
             , Config(config)
+            , BootstrapConfig(bootstrapConfig)
         {}
 
         NPersQueue::TTopicConverterPtr TopicConverter;
         NKikimrPQ::TPQTabletConfig Config;
+        NKikimrPQ::TBootstrapConfig BootstrapConfig;
     };
 
     struct TEvPartitionConfigChanged : public TEventLocal<TEvPartitionConfigChanged, EvPartitionConfigChanged> {
@@ -845,6 +847,7 @@ struct TEvPQ {
         ui64 TxId;
         NPersQueue::TTopicConverterPtr TopicConverter;
         NKikimrPQ::TPQTabletConfig Config;
+        NKikimrPQ::TBootstrapConfig BootstrapConfig;
     };
 
     struct TEvProposePartitionConfigResult : public TEventLocal<TEvProposePartitionConfigResult, EvProposePartitionConfigResult> {
diff --git a/ydb/core/persqueue/partition.cpp b/ydb/core/persqueue/partition.cpp
@@ -2037,7 +2037,8 @@ bool TPartition::ExecUserActionOrTransaction(TSimpleSharedPtr<TTransaction>& t,
     } else if (t->ProposeConfig) {
         Y_ABORT_UNLESS(ChangingConfig);
         ChangeConfig = MakeSimpleShared<TEvPQ::TEvChangePartitionConfig>(TopicConverter,
-                                                                         t->ProposeConfig->Config);
+                                                                         t->ProposeConfig->Config,
+                                                                         t->ProposeConfig->BootstrapConfig);
         PendingPartitionConfig = GetPartitionConfig(ChangeConfig->Config);
         SendChangeConfigReply = false;
     }
@@ -2123,7 +2124,8 @@ bool TPartition::BeginTransaction(const TEvPQ::TEvProposePartitionConfig& event)
 {
     ChangeConfig =
         MakeSimpleShared<TEvPQ::TEvChangePartitionConfig>(TopicConverter,
-                                                          event.Config);
+                                                          event.Config,
+                                                          event.BootstrapConfig);
     PendingPartitionConfig = GetPartitionConfig(ChangeConfig->Config);
 
     SendChangeConfigReply = false;
@@ -2360,6 +2362,7 @@ void TPartition::OnProcessTxsAndUserActsWriteComplete(const TActorContext& ctx)
 
     if (ChangeConfig) {
         EndChangePartitionConfig(std::move(ChangeConfig->Config),
+                                 std::move(ChangeConfig->BootstrapConfig),
                                  ChangeConfig->TopicConverter,
                                  ctx);
     }
@@ -2426,12 +2429,24 @@ void TPartition::OnProcessTxsAndUserActsWriteComplete(const TActorContext& ctx)
 }
 
 void TPartition::EndChangePartitionConfig(NKikimrPQ::TPQTabletConfig&& config,
+                                          NKikimrPQ::TBootstrapConfig&& bootstrapConfig,
                                           NPersQueue::TTopicConverterPtr topicConverter,
                                           const TActorContext& ctx)
 {
     Config = std::move(config);
     PartitionConfig = GetPartitionConfig(Config);
     PartitionGraph = MakePartitionGraph(Config);
+
+    for (const auto& mg : bootstrapConfig.GetExplicitMessageGroups()) {
+        TMaybe<TPartitionKeyRange> keyRange;
+        if (mg.HasKeyRange()) {
+            keyRange = TPartitionKeyRange::Parse(mg.GetKeyRange());
+        }
+
+        TSourceIdInfo sourceId(0, 0, ctx.Now(), std::move(keyRange), false);
+        SourceIdStorage.RegisterSourceIdInfo(mg.GetId(), std::move(sourceId), true);
+    }
+
     TopicConverter = topicConverter;
     NewPartition = false;
 
@@ -2441,14 +2456,15 @@ void TPartition::EndChangePartitionConfig(NKikimrPQ::TPQTabletConfig&& config,
         InitSplitMergeSlidingWindow();
     }
 
-    Send(ReadQuotaTrackerActor, new TEvPQ::TEvChangePartitionConfig(TopicConverter, Config));
-    Send(WriteQuotaTrackerActor, new TEvPQ::TEvChangePartitionConfig(TopicConverter, Config));
+    Send(ReadQuotaTrackerActor, new TEvPQ::TEvChangePartitionConfig(TopicConverter, Config, bootstrapConfig));
+    Send(WriteQuotaTrackerActor, new TEvPQ::TEvChangePartitionConfig(TopicConverter, Config, bootstrapConfig));
     TotalPartitionWriteSpeed = config.GetPartitionConfig().GetWriteSpeedInBytesPerSecond();
 
     if (Config.GetPartitionConfig().HasMirrorFrom()) {
         if (Mirrorer) {
             ctx.Send(Mirrorer->Actor, new TEvPQ::TEvChangePartitionConfig(TopicConverter,
-                                                                          Config));
+                                                                          Config,
+                                                                          bootstrapConfig));
         } else {
             CreateMirrorerActor();
         }
diff --git a/ydb/core/persqueue/partition.h b/ydb/core/persqueue/partition.h
@@ -390,6 +390,7 @@ class TPartition : public TActorBootstrapped<TPartition> {
     void OnProcessTxsAndUserActsWriteComplete(const TActorContext& ctx);
 
     void EndChangePartitionConfig(NKikimrPQ::TPQTabletConfig&& config,
+                                  NKikimrPQ::TBootstrapConfig&& bootstrapConfig,
                                   NPersQueue::TTopicConverterPtr topicConverter,
                                   const TActorContext& ctx);
     TString GetKeyConfig() const;
diff --git a/ydb/core/persqueue/partition_init.cpp b/ydb/core/persqueue/partition_init.cpp
@@ -169,7 +169,8 @@ void TInitConfigStep::Handle(TEvKeyValue::TEvResponse::TPtr& ev, const TActorCon
 
         if (Partition()->Config.GetVersion() < Partition()->TabletConfig.GetVersion()) {
             auto event = MakeHolder<TEvPQ::TEvChangePartitionConfig>(Partition()->TopicConverter,
-                                                                     Partition()->TabletConfig);
+                                                                     Partition()->TabletConfig,
+                                                                     NKikimrPQ::TBootstrapConfig());
             Partition()->PushFrontDistrTx(event.Release());
         }
         break;
diff --git a/ydb/core/persqueue/pq_impl.cpp b/ydb/core/persqueue/pq_impl.cpp
@@ -691,7 +691,7 @@ void TPersQueue::ApplyNewConfigAndReply(const TActorContext& ctx)
     ClearNewConfig();
 
     for (auto& p : Partitions) { //change config for already created partitions
-        ctx.Send(p.second.Actor, new TEvPQ::TEvChangePartitionConfig(TopicConverter, Config));
+        ctx.Send(p.second.Actor, new TEvPQ::TEvChangePartitionConfig(TopicConverter, Config, BootstrapConfigTx ? *BootstrapConfigTx : NKikimrPQ::TBootstrapConfig()));
     }
     ChangePartitionConfigInflight += Partitions.size();
 
@@ -1807,7 +1807,7 @@ void TPersQueue::AddCmdWriteConfig(TEvKeyValue::TEvRequest* request,
             keyRange = TPartitionKeyRange::Parse(mg.GetKeyRange());
         }
 
-        sourceIdWriter.RegisterSourceId(mg.GetId(), 0, 0, ctx.Now(), std::move(keyRange));
+        sourceIdWriter.RegisterSourceId(mg.GetId(), 0, 0, ctx.Now(), std::move(keyRange), false);
     }
 
     for (const auto& partition : cfg.GetPartitions()) {
@@ -4545,6 +4545,7 @@ void TPersQueue::SendEvProposePartitionConfig(const TActorContext& ctx,
 
         event->TopicConverter = tx.TopicConverter;
         event->Config = tx.TabletConfig;
+        event->BootstrapConfig = tx.BootstrapConfig;
 
         ctx.Send(partition.Actor, std::move(event));
     }
diff --git a/ydb/core/persqueue/sourceid.h b/ydb/core/persqueue/sourceid.h
@@ -51,6 +51,7 @@ class TSourceIdStorage: private THeartbeatProcessor {
     void RegisterSourceId(const TString& sourceId, Args&&... args) {
         RegisterSourceIdInfo(sourceId, TSourceIdInfo(std::forward<Args>(args)...), false);
     }
+    void RegisterSourceIdInfo(const TString& sourceId, TSourceIdInfo&& sourceIdInfo, bool load);
 
     void DeregisterSourceId(const TString& sourceId);
 
@@ -65,7 +66,6 @@ class TSourceIdStorage: private THeartbeatProcessor {
 private:
     void LoadRawSourceIdInfo(const TString& key, const TString& data, TInstant now);
     void LoadProtoSourceIdInfo(const TString& key, const TString& data);
-    void RegisterSourceIdInfo(const TString& sourceId, TSourceIdInfo&& sourceIdInfo, bool load);
 
 private:
     TSourceIdMap InMemorySourceIds;
diff --git a/ydb/core/persqueue/ut/partition_ut.cpp b/ydb/core/persqueue/ut/partition_ut.cpp
@@ -988,7 +988,8 @@ void TPartitionFixture::SendChangePartitionConfig(const TConfigParams& config)
     auto event = MakeHolder<TEvPQ::TEvChangePartitionConfig>(TopicConverter, MakeConfig(config.Version,
                                                                                         config.Consumers,
                                                                                         1,
-                                                                                        config.MeteringMode));
+                                                                                        config.MeteringMode),
+                                                                            NKikimrPQ::TBootstrapConfig());
     Ctx->Runtime->SingleSys()->Send(new IEventHandle(ActorId, Ctx->Edge, event.Release()));
 }
 
diff --git a/ydb/core/tx/datashard/datashard_ut_change_exchange.cpp b/ydb/core/tx/datashard/datashard_ut_change_exchange.cpp
@@ -2654,12 +2654,13 @@ Y_UNIT_TEST_SUITE(Cdc) {
         }
     }
 
-    Y_UNIT_TEST(InitialScan) {
+    void InitialScanTest(bool withTopicSchemeTx) {
         TPortManager portManager;
         TServer::TPtr server = new TServer(TServerSettings(portManager.GetPort(2134), {}, DefaultPQConfig())
             .SetUseRealThreads(false)
             .SetDomainName("Root")
             .SetEnableChangefeedInitialScan(true)
+            .SetEnablePQConfigTransactionsAtSchemeShard(withTopicSchemeTx)
         );
 
         auto& runtime = *server->GetRuntime();
@@ -2702,6 +2703,14 @@ Y_UNIT_TEST_SUITE(Cdc) {
         });
     }
 
+    Y_UNIT_TEST(InitialScan) {
+        InitialScanTest(false);
+    }
+
+    Y_UNIT_TEST(InitialScan_WithTopicSchemeTx) {
+        InitialScanTest(true);
+    }
+
     Y_UNIT_TEST(InitialScanDebezium) {
         TTestTopicEnv env(SimpleTable(), KeysOnly(NKikimrSchemeOp::ECdcStreamFormatDebeziumJson, "UnusedStream"));
         auto& client = env.GetClient();

Original file line number	Diff line number	Diff line change
`@@ -169,7 +169,8 @@ void TInitConfigStep::Handle(TEvKeyValue::TEvResponse::TPtr& ev, const TActorCon`
`169`	`169`
`170`	`170`	`if (Partition()->Config.GetVersion() < Partition()->TabletConfig.GetVersion()) {`
`171`	`171`	`auto event = MakeHolder<TEvPQ::TEvChangePartitionConfig>(Partition()->TopicConverter,`
`172`		`- Partition()->TabletConfig);`
	`172`	`+ Partition()->TabletConfig,`
	`173`	`+ NKikimrPQ::TBootstrapConfig());`
`173`	`174`	`Partition()->PushFrontDistrTx(event.Release());`
`174`	`175`	`}`
`175`	`176`	`break;`
Original file line number	Diff line number	Diff line change
`@@ -691,7 +691,7 @@ void TPersQueue::ApplyNewConfigAndReply(const TActorContext& ctx)`
`691`	`691`	`ClearNewConfig();`
`692`	`692`
`693`	`693`	`for (auto& p : Partitions) { //change config for already created partitions`
`694`		`- ctx.Send(p.second.Actor, new TEvPQ::TEvChangePartitionConfig(TopicConverter, Config));`
	`694`	`+ ctx.Send(p.second.Actor, new TEvPQ::TEvChangePartitionConfig(TopicConverter, Config, BootstrapConfigTx ? *BootstrapConfigTx : NKikimrPQ::TBootstrapConfig()));`
`695`	`695`	`}`
`696`	`696`	`ChangePartitionConfigInflight += Partitions.size();`
`697`	`697`
`@@ -1807,7 +1807,7 @@ void TPersQueue::AddCmdWriteConfig(TEvKeyValue::TEvRequest* request,`
`1807`	`1807`	`keyRange = TPartitionKeyRange::Parse(mg.GetKeyRange());`
`1808`	`1808`	`}`
`1809`	`1809`
`1810`		`- sourceIdWriter.RegisterSourceId(mg.GetId(), 0, 0, ctx.Now(), std::move(keyRange));`
	`1810`	`+ sourceIdWriter.RegisterSourceId(mg.GetId(), 0, 0, ctx.Now(), std::move(keyRange), false);`
`1811`	`1811`	`}`
`1812`	`1812`
`1813`	`1813`	`for (const auto& partition : cfg.GetPartitions()) {`
`@@ -4545,6 +4545,7 @@ void TPersQueue::SendEvProposePartitionConfig(const TActorContext& ctx,`
`4545`	`4545`
`4546`	`4546`	`event->TopicConverter = tx.TopicConverter;`
`4547`	`4547`	`event->Config = tx.TabletConfig;`
	`4548`	`+ event->BootstrapConfig = tx.BootstrapConfig;`
`4548`	`4549`
`4549`	`4550`	`ctx.Send(partition.Actor, std::move(event));`
`4550`	`4551`	`}`
Original file line number	Diff line number	Diff line change
`@@ -988,7 +988,8 @@ void TPartitionFixture::SendChangePartitionConfig(const TConfigParams& config)`
`988`	`988`	`auto event = MakeHolder<TEvPQ::TEvChangePartitionConfig>(TopicConverter, MakeConfig(config.Version,`
`989`	`989`	`config.Consumers,`
`990`	`990`	`1,`
`991`		`- config.MeteringMode));`
	`991`	`+ config.MeteringMode),`
	`992`	`+ NKikimrPQ::TBootstrapConfig());`
`992`	`993`	`Ctx->Runtime->SingleSys()->Send(new IEventHandle(ActorId, Ctx->Edge, event.Release()));`
`993`	`994`	`}`
`994`	`995`