Skip to content

Commit eaef93f

Browse files
nshestakovnshestakov
committed
Revert "Revert "Choose partition for topic split/merge" (ydb-platform#1243)"
This reverts commit 4295d15.
1 parent 78daf8b commit eaef93f

34 files changed

+2130
-760
lines changed

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

+17-1
Original file line numberDiff line numberDiff line change
@@ -169,6 +169,8 @@ struct TEvPQ {
169169
EvCacheProxyForgetRead,
170170
EvGetFullDirectReadData,
171171
EvProvideDirectReadInfo,
172+
EvCheckPartitionStatusRequest,
173+
EvCheckPartitionStatusResponse,
172174
EvEnd
173175
};
174176

@@ -491,19 +493,21 @@ struct TEvPQ {
491493
};
492494

493495
struct TEvChangeOwner : public TEventLocal<TEvChangeOwner, EvChangeOwner> {
494-
explicit TEvChangeOwner(const ui64 cookie, const TString& owner, const TActorId& pipeClient, const TActorId& sender, const bool force)
496+
explicit TEvChangeOwner(const ui64 cookie, const TString& owner, const TActorId& pipeClient, const TActorId& sender, const bool force, const bool registerIfNotExists)
495497
: Cookie(cookie)
496498
, Owner(owner)
497499
, PipeClient(pipeClient)
498500
, Sender(sender)
499501
, Force(force)
502+
, RegisterIfNotExists(registerIfNotExists)
500503
{}
501504

502505
ui64 Cookie;
503506
TString Owner;
504507
TActorId PipeClient;
505508
TActorId Sender;
506509
bool Force;
510+
bool RegisterIfNotExists;
507511
};
508512

509513
struct TEvPipeDisconnected : public TEventLocal<TEvPipeDisconnected, EvPipeDisconnected> {
@@ -989,6 +993,18 @@ struct TEvPQ {
989993
struct TEvProvideDirectReadInfo : public TEventLocal<TEvProvideDirectReadInfo, EvProvideDirectReadInfo> {
990994
};
991995

996+
struct TEvCheckPartitionStatusRequest : public TEventPB<TEvCheckPartitionStatusRequest, NKikimrPQ::TEvCheckPartitionStatusRequest, EvCheckPartitionStatusRequest> {
997+
TEvCheckPartitionStatusRequest() = default;
998+
999+
TEvCheckPartitionStatusRequest(ui32 partitionId) {
1000+
Record.SetPartition(partitionId);
1001+
}
1002+
};
1003+
1004+
struct TEvCheckPartitionStatusResponse : public TEventPB<TEvCheckPartitionStatusResponse, NKikimrPQ::TEvCheckPartitionStatusResponse, EvCheckPartitionStatusResponse> {
1005+
};
1006+
1007+
9921008
};
9931009

9941010
} //NKikimr

‎ydb/core/persqueue/partition.cpp

+20-1
Original file line numberDiff line numberDiff line change
@@ -1796,7 +1796,7 @@ void TPartition::EndChangePartitionConfig(const NKikimrPQ::TPQTabletConfig& conf
17961796
{
17971797
Config = config;
17981798
PartitionConfig = GetPartitionConfig(Config, Partition);
1799-
PartitionGraph.Rebuild(Config);
1799+
PartitionGraph = MakePartitionGraph(Config);
18001800
TopicConverter = topicConverter;
18011801
NewPartition = false;
18021802

@@ -2616,4 +2616,23 @@ void TPartition::Handle(TEvPQ::TEvSourceIdRequest::TPtr& ev, const TActorContext
26162616
Send(ev->Sender, response.Release());
26172617
}
26182618

2619+
void TPartition::Handle(TEvPQ::TEvCheckPartitionStatusRequest::TPtr& ev, const TActorContext& ctx) {
2620+
auto& record = ev->Get()->Record;
2621+
2622+
if (Partition != record.GetPartition()) {
2623+
LOG_INFO_S(
2624+
ctx, NKikimrServices::PERSQUEUE,
2625+
"TEvCheckPartitionStatusRequest for wrong partition " << record.GetPartition() << "." <<
2626+
" Topic: \"" << TopicName() << "\"." <<
2627+
" Partition: " << Partition << "."
2628+
);
2629+
return;
2630+
}
2631+
2632+
auto response = MakeHolder<TEvPQ::TEvCheckPartitionStatusResponse>();
2633+
response->Record.SetStatus(PartitionConfig ? PartitionConfig->GetStatus() : NKikimrPQ::ETopicPartitionStatus::Active);
2634+
2635+
Send(ev->Sender, response.Release());
2636+
}
2637+
26192638
} // namespace NKikimr::NPQ

‎ydb/core/persqueue/partition.h

+4-1
Original file line numberDiff line numberDiff line change
@@ -100,7 +100,7 @@ class TPartition : public TActorBootstrapped<TPartition> {
100100

101101
void ReplyGetClientOffsetOk(const TActorContext& ctx, const ui64 dst, const i64 offset, const TInstant writeTimestamp, const TInstant createTimestamp);
102102
void ReplyOk(const TActorContext& ctx, const ui64 dst);
103-
void ReplyOwnerOk(const TActorContext& ctx, const ui64 dst, const TString& ownerCookie);
103+
void ReplyOwnerOk(const TActorContext& ctx, const ui64 dst, const TString& ownerCookie, ui64 seqNo);
104104

105105
void ReplyWrite(const TActorContext& ctx, ui64 dst, const TString& sourceId, ui64 seqNo, ui16 partNo, ui16 totalParts, ui64 offset, TInstant writeTimestamp, bool already, ui64 maxSeqNo, TDuration partitionQuotedTime, TDuration topicQuotedTime, TDuration queueTime, TDuration writeTime);
106106

@@ -344,6 +344,7 @@ class TPartition : public TActorBootstrapped<TPartition> {
344344
// void DestroyReadSession(const TReadSessionKey& key);
345345

346346
void Handle(TEvPQ::TEvSourceIdRequest::TPtr& ev, const TActorContext& ctx);
347+
void Handle(TEvPQ::TEvCheckPartitionStatusRequest::TPtr& ev, const TActorContext& ctx);
347348

348349
TString LogPrefix() const;
349350

@@ -481,6 +482,7 @@ class TPartition : public TActorBootstrapped<TPartition> {
481482
HFuncTraced(TEvPQ::TEvTxRollback, Handle);
482483
HFuncTraced(TEvPQ::TEvSubDomainStatus, Handle);
483484
HFuncTraced(TEvPQ::TEvSourceIdRequest, Handle);
485+
HFuncTraced(TEvPQ::TEvCheckPartitionStatusRequest, Handle);
484486
HFuncTraced(TEvPQ::TEvSourceIdResponse, SourceManager.Handle);
485487
HFuncTraced(NReadQuoterEvents::TEvQuotaUpdated, Handle);
486488
HFuncTraced(NReadQuoterEvents::TEvAccountQuotaCountersUpdated, Handle);
@@ -538,6 +540,7 @@ class TPartition : public TActorBootstrapped<TPartition> {
538540
HFuncTraced(TEvPQ::TEvTxRollback, Handle);
539541
HFuncTraced(TEvPQ::TEvSubDomainStatus, Handle);
540542
HFuncTraced(TEvPQ::TEvSourceIdRequest, Handle);
543+
HFuncTraced(TEvPQ::TEvCheckPartitionStatusRequest, Handle);
541544
HFuncTraced(TEvPQ::TEvSourceIdResponse, SourceManager.Handle);
542545
HFuncTraced(NReadQuoterEvents::TEvQuotaUpdated, Handle);
543546
HFuncTraced(NReadQuoterEvents::TEvAccountQuotaCountersUpdated, Handle);

‎ydb/core/persqueue/partition_init.cpp

+1-1
Original file line numberDiff line numberDiff line change
@@ -180,7 +180,7 @@ void TInitConfigStep::Handle(TEvKeyValue::TEvResponse::TPtr& ev, const TActorCon
180180
case NKikimrProto::NODATA:
181181
Partition()->Config = Partition()->TabletConfig;
182182
Partition()->PartitionConfig = GetPartitionConfig(Partition()->Config, Partition()->Partition);
183-
Partition()->PartitionGraph.Rebuild(Partition()->Config);
183+
Partition()->PartitionGraph = MakePartitionGraph(Partition()->Config);
184184
break;
185185

186186
case NKikimrProto::ERROR:

‎ydb/core/persqueue/partition_sourcemanager.cpp

+6-6
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88
namespace NKikimr::NPQ {
99

1010
IActor* CreateRequester(TActorId parent, TPartitionSourceManager::TPartitionId partition, ui64 tabletId);
11-
bool IsResearchRequires(std::optional<const TPartitionGraph::Node*> node);
11+
bool IsResearchRequires(const TPartitionGraph::Node* node);
1212

1313
//
1414
// TPartitionSourceManager
@@ -37,7 +37,7 @@ void TPartitionSourceManager::ScheduleBatch() {
3737

3838
PendingSourceIds = std::move(UnknownSourceIds);
3939

40-
for(const auto* parent : node.value()->HierarhicalParents) {
40+
for(const auto* parent : node->HierarhicalParents) {
4141
PendingCookies.insert(++Cookie);
4242

4343
TActorId actorId = PartitionRequester(parent->Id, parent->TabletId);
@@ -141,7 +141,7 @@ void TPartitionSourceManager::Handle(TEvPQ::TEvSourceIdResponse::TPtr& ev, const
141141
}
142142
}
143143

144-
TPartitionSourceManager::TPartitionNode TPartitionSourceManager::GetPartitionNode() const {
144+
const TPartitionSourceManager::TPartitionNode* TPartitionSourceManager::GetPartitionNode() const {
145145
return Partition.PartitionGraph.GetPartition(Partition.Partition);
146146
}
147147

@@ -185,7 +185,7 @@ TSourceIdStorage& TPartitionSourceManager::GetSourceIdStorage() const {
185185

186186
bool TPartitionSourceManager::HasParents() const {
187187
auto node = Partition.PartitionGraph.GetPartition(Partition.Partition);
188-
return node && !node.value()->Parents.empty();
188+
return node && !node->Parents.empty();
189189
}
190190

191191
TActorId TPartitionSourceManager::PartitionRequester(TPartitionId id, ui64 tabletId) {
@@ -484,8 +484,8 @@ IActor* CreateRequester(TActorId parent, TPartitionSourceManager::TPartitionId p
484484
return new TSourceIdRequester(parent, partition, tabletId);
485485
}
486486

487-
bool IsResearchRequires(std::optional<const TPartitionGraph::Node*> node) {
488-
return node && !node.value()->Parents.empty();
487+
bool IsResearchRequires(const TPartitionGraph::Node* node) {
488+
return node && !node->Parents.empty();
489489
}
490490

491491
NKikimrPQ::TEvSourceIdResponse::EState Convert(TSourceIdInfo::EState value) {

‎ydb/core/persqueue/partition_sourcemanager.h

+3-3
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ class TPartition;
1212

1313
class TPartitionSourceManager {
1414
private:
15-
using TPartitionNode = std::optional<const TPartitionGraph::Node *>;
15+
using TPartitionNode = TPartitionGraph::Node;
1616

1717
public:
1818
using TPartitionId = ui32;
@@ -96,7 +96,7 @@ class TPartitionSourceManager {
9696
private:
9797
TPartitionSourceManager& Manager;
9898

99-
TPartitionNode Node;
99+
const TPartitionNode* Node;
100100
TSourceIdWriter SourceIdWriter;
101101
THeartbeatEmitter HeartbeatEmitter;
102102
};
@@ -125,7 +125,7 @@ class TPartitionSourceManager {
125125
void FinishBatch(const TActorContext& ctx);
126126
bool RequireEnqueue(const TString& sourceId);
127127

128-
TPartitionNode GetPartitionNode() const;
128+
const TPartitionNode* GetPartitionNode() const;
129129
TSourceIdStorage& GetSourceIdStorage() const;
130130
bool HasParents() const;
131131

‎ydb/core/persqueue/partition_write.cpp

+17-6
Original file line numberDiff line numberDiff line change
@@ -31,14 +31,18 @@ static const ui32 MAX_INLINE_SIZE = 1000;
3131

3232
static constexpr NPersQueue::NErrorCode::EErrorCode InactivePartitionErrorCode = NPersQueue::NErrorCode::WRITE_ERROR_PARTITION_IS_FULL;
3333

34-
void TPartition::ReplyOwnerOk(const TActorContext& ctx, const ui64 dst, const TString& cookie) {
34+
void TPartition::ReplyOwnerOk(const TActorContext& ctx, const ui64 dst, const TString& cookie, ui64 seqNo) {
3535
LOG_DEBUG_S(ctx, NKikimrServices::PERSQUEUE, "TPartition::ReplyOwnerOk. Partition: " << Partition);
3636

3737
THolder<TEvPQ::TEvProxyResponse> response = MakeHolder<TEvPQ::TEvProxyResponse>(dst);
3838
NKikimrClient::TResponse& resp = *response->Response;
3939
resp.SetStatus(NMsgBusProxy::MSTATUS_OK);
4040
resp.SetErrorCode(NPersQueue::NErrorCode::OK);
41-
resp.MutablePartitionResponse()->MutableCmdGetOwnershipResult()->SetOwnerCookie(cookie);
41+
auto* r = resp.MutablePartitionResponse()->MutableCmdGetOwnershipResult();
42+
r->SetOwnerCookie(cookie);
43+
r->SetStatus(PartitionConfig ? PartitionConfig->GetStatus() : NKikimrPQ::ETopicPartitionStatus::Active);
44+
r->SetSeqNo(seqNo);
45+
4246
ctx.Send(Tablet, response.Release());
4347
}
4448

@@ -146,8 +150,12 @@ void TPartition::ProcessChangeOwnerRequest(TAutoPtr<TEvPQ::TEvChangeOwner> ev, c
146150
auto &owner = ev->Owner;
147151
auto it = Owners.find(owner);
148152
if (it == Owners.end()) {
149-
Owners[owner];
150-
it = Owners.find(owner);
153+
if (ev->RegisterIfNotExists) {
154+
Owners[owner];
155+
it = Owners.find(owner);
156+
} else {
157+
return ReplyError(ctx, ev->Cookie, NPersQueue::NErrorCode::SOURCEID_DELETED, "SourceId isn't registered");
158+
}
151159
}
152160
if (it->second.NeedResetOwner || ev->Force) { //change owner
153161
Y_ABORT_UNLESS(ReservedSize >= it->second.ReservedSize);
@@ -346,10 +354,13 @@ void TPartition::AnswerCurrentWrites(const TActorContext& ctx) {
346354
if (!already && partNo + 1 == totalParts && !writeResponse.Msg.HeartbeatVersion)
347355
++offset;
348356
} else if (response.IsOwnership()) {
349-
const TString& ownerCookie = response.GetOwnership().OwnerCookie;
357+
const auto& r = response.GetOwnership();
358+
const TString& ownerCookie = r.OwnerCookie;
350359
auto it = Owners.find(TOwnerInfo::GetOwnerFromOwnerCookie(ownerCookie));
351360
if (it != Owners.end() && it->second.OwnerCookie == ownerCookie) {
352-
ReplyOwnerOk(ctx, response.GetCookie(), ownerCookie);
361+
auto sit = SourceIdStorage.GetInMemorySourceIds().find(NSourceIdEncoding::EncodeSimple(it->first));
362+
auto seqNo = sit == SourceIdStorage.GetInMemorySourceIds().end() ? 0 : sit->second.SeqNo;
363+
ReplyOwnerOk(ctx, response.GetCookie(), ownerCookie, seqNo);
353364
} else {
354365
ReplyError(ctx, response.GetCookie(), NPersQueue::NErrorCode::WRONG_COOKIE, "new GetOwnership request is dropped already");
355366
}

‎ydb/core/persqueue/pq_impl.cpp

+34-1
Original file line numberDiff line numberDiff line change
@@ -1230,6 +1230,7 @@ void TPersQueue::Handle(TEvPQ::TEvInitComplete::TPtr& ev, const TActorContext& c
12301230
}
12311231

12321232
ProcessSourceIdRequests(partitionId);
1233+
ProcessCheckPartitionStatusRequests(partitionId);
12331234
if (allInitialized) {
12341235
SourceIdRequests.clear();
12351236
}
@@ -2048,7 +2049,8 @@ void TPersQueue::HandleGetOwnershipRequest(const ui64 responseCookie, const TAct
20482049
it->second = TPipeInfo::ForOwner(partActor, owner, it->second.ServerActors);
20492050

20502051
InitResponseBuilder(responseCookie, 1, COUNTER_LATENCY_PQ_GET_OWNERSHIP);
2051-
THolder<TEvPQ::TEvChangeOwner> event = MakeHolder<TEvPQ::TEvChangeOwner>(responseCookie, owner, pipeClient, sender, req.GetCmdGetOwnership().GetForce());
2052+
THolder<TEvPQ::TEvChangeOwner> event = MakeHolder<TEvPQ::TEvChangeOwner>(responseCookie, owner, pipeClient, sender,
2053+
req.GetCmdGetOwnership().GetForce(), req.GetCmdGetOwnership().GetRegisterIfNotExists());
20522054
ctx.Send(partActor, event.Release());
20532055
}
20542056

@@ -3915,6 +3917,37 @@ void TPersQueue::ProcessSourceIdRequests(ui32 partitionId) {
39153917
}
39163918
}
39173919

3920+
void TPersQueue::Handle(TEvPQ::TEvCheckPartitionStatusRequest::TPtr& ev, const TActorContext& ctx) {
3921+
auto& record = ev->Get()->Record;
3922+
auto it = Partitions.find(record.GetPartition());
3923+
if (it == Partitions.end()) {
3924+
LOG_INFO_S(ctx, NKikimrServices::PERSQUEUE, "Unknown partition " << record.GetPartition());
3925+
3926+
auto response = THolder<TEvPQ::TEvCheckPartitionStatusResponse>();
3927+
response->Record.SetStatus(NKikimrPQ::ETopicPartitionStatus::Deleted);
3928+
Send(ev->Sender, response.Release());
3929+
3930+
return;
3931+
}
3932+
3933+
if (it->second.InitDone) {
3934+
Forward(ev, it->second.Actor);
3935+
} else {
3936+
CheckPartitionStatusRequests[record.GetPartition()].push_back(ev);
3937+
}
3938+
}
3939+
3940+
void TPersQueue::ProcessCheckPartitionStatusRequests(ui32 partitionId) {
3941+
auto sit = CheckPartitionStatusRequests.find(partitionId);
3942+
if (sit != CheckPartitionStatusRequests.end()) {
3943+
auto it = Partitions.find(partitionId);
3944+
for (auto& r : sit->second) {
3945+
Forward(r, it->second.Actor);
3946+
}
3947+
CheckPartitionStatusRequests.erase(partitionId);
3948+
}
3949+
}
3950+
39183951
TString TPersQueue::LogPrefix() const {
39193952
return TStringBuilder() << SelfId() << " ";
39203953
}

‎ydb/core/persqueue/pq_impl.h

+4
Original file line numberDiff line numberDiff line change
@@ -168,6 +168,9 @@ class TPersQueue : public NKeyValue::TKeyValueFlat {
168168
void Handle(TEvPQ::TEvSourceIdRequest::TPtr& ev, const TActorContext& ctx);
169169
void ProcessSourceIdRequests(ui32 partitionId);
170170

171+
void Handle(TEvPQ::TEvCheckPartitionStatusRequest::TPtr& ev, const TActorContext& ctx);
172+
void ProcessCheckPartitionStatusRequests(ui32 partitionId);
173+
171174
TString LogPrefix() const;
172175

173176
static constexpr const char * KeyConfig() { return "_config"; }
@@ -405,6 +408,7 @@ class TPersQueue : public NKeyValue::TKeyValueFlat {
405408
bool UseMediatorTimeCast = true;
406409

407410
THashMap<ui32, TVector<TEvPQ::TEvSourceIdRequest::TPtr>> SourceIdRequests;
411+
THashMap<ui32, TVector<TEvPQ::TEvCheckPartitionStatusRequest::TPtr>> CheckPartitionStatusRequests;
408412
TMaybe<ui64> TabletGeneration;
409413
};
410414

‎ydb/core/persqueue/transaction.cpp

+4-5
Original file line numberDiff line numberDiff line change
@@ -147,24 +147,23 @@ void TDistributedTransaction::OnProposeTransaction(const NKikimrPQ::TConfigTrans
147147
TabletConfig = txBody.GetTabletConfig();
148148
BootstrapConfig = txBody.GetBootstrapConfig();
149149

150-
TPartitionGraph graph;
151-
graph.Rebuild(TabletConfig);
150+
TPartitionGraph graph = MakePartitionGraph(TabletConfig);
152151

153152
for (const auto& p : TabletConfig.GetPartitions()) {
154153
auto node = graph.GetPartition(p.GetPartitionId());
155154
if (!node) {
156155
// Old configuration format without AllPartitions. Split/Merge is not supported.
157156
continue;
158157
}
159-
if (node.value()->Children.empty()) {
160-
for (const auto* r : node.value()->Parents) {
158+
if (node->Children.empty()) {
159+
for (const auto* r : node->Parents) {
161160
if (extractTabletId != r->TabletId) {
162161
Senders.insert(r->TabletId);
163162
}
164163
}
165164
}
166165

167-
for (const auto* r : node.value()->Children) {
166+
for (const auto* r : node->Children) {
168167
if (r->Children.empty()) {
169168
if (extractTabletId != r->TabletId) {
170169
Receivers.insert(r->TabletId);

0 commit comments

Comments
 (0)