Skip to content

Commit cbc199b

Browse files
nshestakovnshestakov
authored
Merge 2443d0c into 2764b01 
2 parents 2764b01 + 2443d0c commit cbc199b

File tree

7 files changed

+42
-11
lines changed

7 files changed

+42
-11
lines changed

‎ydb/core/persqueue/events/internal.h‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -576,13 +576,15 @@ struct TEvPQ {
576576
};
577577

578578
struct TEvChangePartitionConfig : public TEventLocal<TEvChangePartitionConfig, EvChangePartitionConfig> {
579-
TEvChangePartitionConfig(const NPersQueue::TTopicConverterPtr& topicConverter, const NKikimrPQ::TPQTabletConfig& config)
579+
TEvChangePartitionConfig(const NPersQueue::TTopicConverterPtr& topicConverter, const NKikimrPQ::TPQTabletConfig& config, const NKikimrPQ::TBootstrapConfig& bootstrapConfig)
580580
: TopicConverter(topicConverter)
581581
, Config(config)
582+
, BootstrapConfig(bootstrapConfig)
582583
{}
583584

584585
NPersQueue::TTopicConverterPtr TopicConverter;
585586
NKikimrPQ::TPQTabletConfig Config;
587+
NKikimrPQ::TBootstrapConfig BootstrapConfig;
586588
};
587589

588590
struct TEvPartitionConfigChanged : public TEventLocal<TEvPartitionConfigChanged, EvPartitionConfigChanged> {
@@ -845,6 +847,7 @@ struct TEvPQ {
845847
ui64 TxId;
846848
NPersQueue::TTopicConverterPtr TopicConverter;
847849
NKikimrPQ::TPQTabletConfig Config;
850+
NKikimrPQ::TBootstrapConfig BootstrapConfig;
848851
};
849852

850853
struct TEvProposePartitionConfigResult : public TEventLocal<TEvProposePartitionConfigResult, EvProposePartitionConfigResult> {

‎ydb/core/persqueue/partition.cpp‎

Lines changed: 21 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -2037,7 +2037,8 @@ bool TPartition::ExecUserActionOrTransaction(TSimpleSharedPtr<TTransaction>& t,
20372037
} else if (t->ProposeConfig) {
20382038
Y_ABORT_UNLESS(ChangingConfig);
20392039
ChangeConfig = MakeSimpleShared<TEvPQ::TEvChangePartitionConfig>(TopicConverter,
2040-
t->ProposeConfig->Config);
2040+
t->ProposeConfig->Config,
2041+
t->ProposeConfig->BootstrapConfig);
20412042
PendingPartitionConfig = GetPartitionConfig(ChangeConfig->Config);
20422043
SendChangeConfigReply = false;
20432044
}
@@ -2123,7 +2124,8 @@ bool TPartition::BeginTransaction(const TEvPQ::TEvProposePartitionConfig& event)
21232124
{
21242125
ChangeConfig =
21252126
MakeSimpleShared<TEvPQ::TEvChangePartitionConfig>(TopicConverter,
2126-
event.Config);
2127+
event.Config,
2128+
event.BootstrapConfig);
21272129
PendingPartitionConfig = GetPartitionConfig(ChangeConfig->Config);
21282130

21292131
SendChangeConfigReply = false;
@@ -2360,6 +2362,7 @@ void TPartition::OnProcessTxsAndUserActsWriteComplete(const TActorContext& ctx)
23602362

23612363
if (ChangeConfig) {
23622364
EndChangePartitionConfig(std::move(ChangeConfig->Config),
2365+
std::move(ChangeConfig->BootstrapConfig),
23632366
ChangeConfig->TopicConverter,
23642367
ctx);
23652368
}
@@ -2426,12 +2429,24 @@ void TPartition::OnProcessTxsAndUserActsWriteComplete(const TActorContext& ctx)
24262429
}
24272430

24282431
void TPartition::EndChangePartitionConfig(NKikimrPQ::TPQTabletConfig&& config,
2432+
NKikimrPQ::TBootstrapConfig&& bootstrapConfig,
24292433
NPersQueue::TTopicConverterPtr topicConverter,
24302434
const TActorContext& ctx)
24312435
{
24322436
Config = std::move(config);
24332437
PartitionConfig = GetPartitionConfig(Config);
24342438
PartitionGraph = MakePartitionGraph(Config);
2439+
2440+
for (const auto& mg : bootstrapConfig.GetExplicitMessageGroups()) {
2441+
TMaybe<TPartitionKeyRange> keyRange;
2442+
if (mg.HasKeyRange()) {
2443+
keyRange = TPartitionKeyRange::Parse(mg.GetKeyRange());
2444+
}
2445+
2446+
TSourceIdInfo sourceId(0, 0, ctx.Now(), std::move(keyRange), false);
2447+
SourceIdStorage.RegisterSourceIdInfo(mg.GetId(), std::move(sourceId), true);
2448+
}
2449+
24352450
TopicConverter = topicConverter;
24362451
NewPartition = false;
24372452

@@ -2441,14 +2456,15 @@ void TPartition::EndChangePartitionConfig(NKikimrPQ::TPQTabletConfig&& config,
24412456
InitSplitMergeSlidingWindow();
24422457
}
24432458

2444-
Send(ReadQuotaTrackerActor, new TEvPQ::TEvChangePartitionConfig(TopicConverter, Config));
2445-
Send(WriteQuotaTrackerActor, new TEvPQ::TEvChangePartitionConfig(TopicConverter, Config));
2459+
Send(ReadQuotaTrackerActor, new TEvPQ::TEvChangePartitionConfig(TopicConverter, Config, bootstrapConfig));
2460+
Send(WriteQuotaTrackerActor, new TEvPQ::TEvChangePartitionConfig(TopicConverter, Config, bootstrapConfig));
24462461
TotalPartitionWriteSpeed = config.GetPartitionConfig().GetWriteSpeedInBytesPerSecond();
24472462

24482463
if (Config.GetPartitionConfig().HasMirrorFrom()) {
24492464
if (Mirrorer) {
24502465
ctx.Send(Mirrorer->Actor, new TEvPQ::TEvChangePartitionConfig(TopicConverter,
2451-
Config));
2466+
Config,
2467+
bootstrapConfig));
24522468
} else {
24532469
CreateMirrorerActor();
24542470
}

‎ydb/core/persqueue/partition.h‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -390,6 +390,7 @@ class TPartition : public TActorBootstrapped<TPartition> {
390390
void OnProcessTxsAndUserActsWriteComplete(const TActorContext& ctx);
391391

392392
void EndChangePartitionConfig(NKikimrPQ::TPQTabletConfig&& config,
393+
NKikimrPQ::TBootstrapConfig&& bootstrapConfig,
393394
NPersQueue::TTopicConverterPtr topicConverter,
394395
const TActorContext& ctx);
395396
TString GetKeyConfig() const;

‎ydb/core/persqueue/partition_init.cpp‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -169,7 +169,8 @@ void TInitConfigStep::Handle(TEvKeyValue::TEvResponse::TPtr& ev, const TActorCon
169169

170170
if (Partition()->Config.GetVersion() < Partition()->TabletConfig.GetVersion()) {
171171
auto event = MakeHolder<TEvPQ::TEvChangePartitionConfig>(Partition()->TopicConverter,
172-
Partition()->TabletConfig);
172+
Partition()->TabletConfig,
173+
NKikimrPQ::TBootstrapConfig());
173174
Partition()->PushFrontDistrTx(event.Release());
174175
}
175176
break;

‎ydb/core/persqueue/pq_impl.cpp‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -691,7 +691,7 @@ void TPersQueue::ApplyNewConfigAndReply(const TActorContext& ctx)
691691
ClearNewConfig();
692692

693693
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()));
695695
}
696696
ChangePartitionConfigInflight += Partitions.size();
697697

@@ -1807,7 +1807,7 @@ void TPersQueue::AddCmdWriteConfig(TEvKeyValue::TEvRequest* request,
18071807
keyRange = TPartitionKeyRange::Parse(mg.GetKeyRange());
18081808
}
18091809

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);
18111811
}
18121812

18131813
for (const auto& partition : cfg.GetPartitions()) {
@@ -4545,6 +4545,7 @@ void TPersQueue::SendEvProposePartitionConfig(const TActorContext& ctx,
45454545

45464546
event->TopicConverter = tx.TopicConverter;
45474547
event->Config = tx.TabletConfig;
4548+
event->BootstrapConfig = tx.BootstrapConfig;
45484549

45494550
ctx.Send(partition.Actor, std::move(event));
45504551
}

‎ydb/core/persqueue/sourceid.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,7 @@ class TSourceIdStorage: private THeartbeatProcessor {
5151
void RegisterSourceId(const TString& sourceId, Args&&... args) {
5252
RegisterSourceIdInfo(sourceId, TSourceIdInfo(std::forward<Args>(args)...), false);
5353
}
54+
void RegisterSourceIdInfo(const TString& sourceId, TSourceIdInfo&& sourceIdInfo, bool load);
5455

5556
void DeregisterSourceId(const TString& sourceId);
5657

@@ -65,7 +66,6 @@ class TSourceIdStorage: private THeartbeatProcessor {
6566
private:
6667
void LoadRawSourceIdInfo(const TString& key, const TString& data, TInstant now);
6768
void LoadProtoSourceIdInfo(const TString& key, const TString& data);
68-
void RegisterSourceIdInfo(const TString& sourceId, TSourceIdInfo&& sourceIdInfo, bool load);
6969

7070
private:
7171
TSourceIdMap InMemorySourceIds;

‎ydb/core/tx/datashard/datashard_ut_change_exchange.cpp‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2654,12 +2654,13 @@ Y_UNIT_TEST_SUITE(Cdc) {
26542654
}
26552655
}
26562656

2657-
Y_UNIT_TEST(InitialScan) {
2657+
void InitialScanTest(bool withTopicSchemeTx) {
26582658
TPortManager portManager;
26592659
TServer::TPtr server = new TServer(TServerSettings(portManager.GetPort(2134), {}, DefaultPQConfig())
26602660
.SetUseRealThreads(false)
26612661
.SetDomainName("Root")
26622662
.SetEnableChangefeedInitialScan(true)
2663+
.SetEnablePQConfigTransactionsAtSchemeShard(withTopicSchemeTx)
26632664
);
26642665

26652666
auto& runtime = *server->GetRuntime();
@@ -2702,6 +2703,14 @@ Y_UNIT_TEST_SUITE(Cdc) {
27022703
});
27032704
}
27042705

2706+
Y_UNIT_TEST(InitialScan) {
2707+
InitialScanTest(false);
2708+
}
2709+
2710+
Y_UNIT_TEST(InitialScan_WithTopicSchemeTx) {
2711+
InitialScanTest(true);
2712+
}
2713+
27052714
Y_UNIT_TEST(InitialScanDebezium) {
27062715
TTestTopicEnv env(SimpleTable(), KeysOnly(NKikimrSchemeOp::ECdcStreamFormatDebeziumJson, "UnusedStream"));
27072716
auto& client = env.GetClient();

0 commit comments

Comments
 (0)