Skip to content

Commit a35c9c1

Browse files
FloatingCrowbarFloatingCrowbar
authored
Get rid of table with all topics list (#26335)
1 parent 5a4ab93 commit a35c9c1

10 files changed

+196
-501
lines changed

‎ydb/core/client/server/msgbus_server_persqueue.cpp‎

Lines changed: 75 additions & 82 deletions
Original file line numberDiff line numberDiff line change
@@ -32,28 +32,6 @@ const TString& TopicPrefix(const TActorContext& ctx) {
3232
return topicPrefix;
3333
}
3434

35-
TProcessingResult ProcessMetaCacheAllTopicsResponse(TEvPqNewMetaCache::TEvDescribeAllTopicsResponse::TPtr& ev) {
36-
auto& res = ev->Get()->Result;
37-
const TString& path = ev->Get()->Path;
38-
TProcessingResult result;
39-
if (!ev->Get()->Success) {
40-
return TProcessingResult{
41-
MSTATUS_ERROR,
42-
NPersQueue::NErrorCode::UNKNOWN_TOPIC,
43-
Sprintf("path '%s' has invalid/unknown root prefix, Marker# PQ14", path.c_str()),
44-
true
45-
};
46-
}
47-
if (!res) {
48-
return TProcessingResult{
49-
MSTATUS_ERROR,
50-
NPersQueue::NErrorCode::ERROR,
51-
Sprintf("path '%s' describe error, Status# no status, reason: no reason, Marker# PQ1", path.c_str()),
52-
true
53-
};
54-
}
55-
return {};
56-
}
5735

5836
TProcessingResult ProcessMetaCacheSingleTopicsResponse(
5937
const TSchemeCacheNavigate::TEntry& entry
@@ -165,7 +143,9 @@ struct TTabletInfo {
165143
TVector<NKikimrPQ::TStatusResponse::TPartResult> StatusResponses;
166144
};
167145

168-
TPersQueueBaseRequestProcessor::TPersQueueBaseRequestProcessor(const NKikimrClient::TPersQueueRequest& request, const TActorId& pqMetaCacheId, bool listNodes)
146+
TPersQueueBaseRequestProcessor::TPersQueueBaseRequestProcessor(
147+
const NKikimrClient::TPersQueueRequest& request, const TActorId& pqMetaCacheId, bool listNodes
148+
)
169149
: RequestProto(new NKikimrClient::TPersQueueRequest(request))
170150
, RequestId(RequestProto->HasRequestId() ? RequestProto->GetRequestId() : "<none>")
171151
, PqMetaCache(pqMetaCacheId)
@@ -192,9 +172,17 @@ void TPersQueueBaseRequestProcessor::AnswerAndDie(const TActorContext& ctx) {
192172
void TPersQueueBaseRequestProcessor::Bootstrap(const TActorContext& ctx) {
193173
StartTimestamp = ctx.Now();
194174

195-
LOG_TRACE_S(ctx, NKikimrServices::PERSQUEUE, "Send to PqMetaCache TEvDescribeAllTopicsRequest");
196-
bool ret = ctx.Send(PqMetaCache, new NPqMetaCacheV2::TEvPqNewMetaCache::TEvDescribeAllTopicsRequest());
197-
LOG_TRACE_S(ctx, NKikimrServices::PERSQUEUE, "Send to PqMetaCache TEvDescribeAllTopicsRequest Result:" << ret);
175+
if (TopicsToRequest.empty()) {
176+
throw std::runtime_error("No topics in request");
177+
}
178+
LOG_TRACE_S(ctx, NKikimrServices::PERSQUEUE, "Send to PqMetaCache TEvDescribeTopicsRequest");
179+
TVector<TString> topicsToRequest;
180+
topicsToRequest.reserve(TopicsToRequest.size());
181+
for (const auto& topic : TopicsToRequest) {
182+
topicsToRequest.push_back(topic);
183+
}
184+
bool ret = ctx.Send(PqMetaCache, new NPqMetaCacheV2::TEvPqNewMetaCache::TEvDescribeTopicsByNameRequest(topicsToRequest));
185+
LOG_TRACE_S(ctx, NKikimrServices::PERSQUEUE, "Send to PqMetaCache TEvDescribeTopicsRequest Result:" << ret);
198186

199187
if (ListNodes) {
200188
const TActorId nameserviceId = GetNameserviceActorId();
@@ -216,7 +204,6 @@ STFUNC(TPersQueueBaseRequestProcessor::StateFunc) {
216204
switch (ev->GetTypeRewrite()) {
217205
HFunc(TEvInterconnect::TEvNodesInfo, Handle);
218206
HFunc(NPqMetaCacheV2::TEvPqNewMetaCache::TEvDescribeTopicsResponse, Handle);
219-
HFunc(NPqMetaCacheV2::TEvPqNewMetaCache::TEvDescribeAllTopicsResponse, Handle);
220207
HFunc(NPqMetaCacheV2::TEvPqNewMetaCache::TEvGetNodesMappingResponse, Handle);
221208
HFunc(TEvPersQueue::TEvResponse, Handle);
222209
CFunc(TEvents::TSystem::Wakeup, HandleTimeout);
@@ -274,56 +261,54 @@ void TPersQueueBaseRequestProcessor::HandleTimeout(const TActorContext& ctx) {
274261
}
275262

276263

277-
void TPersQueueBaseRequestProcessor::GetTopicsListOrThrow(
264+
THashSet<TString> GetTopicsListOrThrow(
278265
const ::google::protobuf::RepeatedPtrField<::NKikimrClient::TPersQueueMetaRequest::TTopicRequest>& requests,
279-
THashMap<TString, std::shared_ptr<THashSet<ui64>>>& partitionsToRequest
266+
THashMap<TString, std::shared_ptr<THashSet<ui64>>>* partitionsToRequest
280267
) {
268+
THashSet<TString> ret;
281269
for (const auto& topicRequest : requests) {
282270
if (topicRequest.GetTopic().empty()) {
283271
throw std::runtime_error("TopicRequest must have Topic field.");
284272
}
285-
std::shared_ptr<THashSet<ui64>> partitionsToRequestOnTopic(new THashSet<ui64>()); // nonconst
286-
partitionsToRequest[topicRequest.GetTopic()] = partitionsToRequestOnTopic;
287-
for (ui32 partition : topicRequest.GetPartition()) {
288-
const bool inserted = partitionsToRequestOnTopic->insert(partition).second;
289-
if (!inserted) {
290-
TStringBuilder desc;
291-
desc << "multiple partition " << partition
292-
<< " in TopicRequest for topic '" << topicRequest.GetTopic() << "'";
293-
throw std::runtime_error(desc);
273+
if (partitionsToRequest != nullptr) {
274+
std::shared_ptr<THashSet<ui64>> partitionsToRequestOnTopic(new THashSet<ui64>()); // nonconst
275+
(*partitionsToRequest)[topicRequest.GetTopic()] = partitionsToRequestOnTopic;
276+
for (ui32 partition : topicRequest.GetPartition()) {
277+
const bool inserted = partitionsToRequestOnTopic->insert(partition).second;
278+
if (!inserted) {
279+
TStringBuilder desc;
280+
desc << "multiple partition " << partition
281+
<< " in TopicRequest for topic '" << topicRequest.GetTopic() << "'";
282+
throw std::runtime_error(desc);
283+
}
294284
}
295285
}
296-
297-
const bool res = TopicsToRequest.insert(topicRequest.GetTopic()).second;
286+
const bool res = ret.insert(topicRequest.GetTopic()).second;
298287
if (!res) {
299288
TStringBuilder desc;
300289
desc << "multiple TopicRequest for topic '" << topicRequest.GetTopic() << "'";
301290
throw std::runtime_error(desc);
302291
}
303292
}
304-
293+
return ret;
305294
}
306295

307296
void TPersQueueBaseRequestProcessor::Handle(
308-
NPqMetaCacheV2::TEvPqNewMetaCache::TEvDescribeTopicsResponse::TPtr&, const TActorContext&
297+
NPqMetaCacheV2::TEvPqNewMetaCache::TEvDescribeTopicsResponse::TPtr& ev, const TActorContext& ctx
309298
) {
310-
Y_ABORT();
311-
}
312-
313-
void TPersQueueBaseRequestProcessor::Handle(
314-
NPqMetaCacheV2::TEvPqNewMetaCache::TEvDescribeAllTopicsResponse::TPtr& ev, const TActorContext& ctx
315-
) {
316-
LOG_TRACE_S(ctx, NKikimrServices::PERSQUEUE, "TPersQueueBaseRequestProcessor::Handle");
317-
318-
auto& path = ev->Get()->Path;
319-
if (!ev->Get()->Success) {
320-
return SendErrorReplyAndDie(ctx, MSTATUS_ERROR, NPersQueue::NErrorCode::UNKNOWN_TOPIC,
321-
TStringBuilder() << "no path '" << path << "', Marker# PQ17");
299+
TopicsConverters.reserve(ev->Get()->TopicsRequested.size());
300+
Y_ABORT_UNLESS(ev->Get()->Result->ResultSet.size() == ev->Get()->TopicsRequested.size());
301+
for (ui32 i = 0; i < ev->Get()->TopicsRequested.size(); ++i) {
302+
if (ev->Get()->Result->ResultSet[i].PQGroupInfo) {
303+
const auto& pqTabletConfig = ev->Get()->Result->ResultSet[i].PQGroupInfo->Description.GetPQTabletConfig();
304+
TopicsConverters.push_back(ev->Get()->TopicsRequested[i]->UpgradeToFullConverter(
305+
pqTabletConfig,
306+
AppData(ctx)->PQConfig.GetTestDatabaseRoot()));
307+
} else {
308+
TopicsConverters.push_back(nullptr);
309+
}
322310
}
323-
324311
TopicsDescription = std::move(ev->Get()->Result);
325-
TopicsConverters = std::move(ev->Get()->Topics);
326-
Y_ABORT_UNLESS(TopicsDescription->ResultSet.size() == TopicsConverters.size());
327312
if (ReadyToCreateChildren()) {
328313
if (CreateChildren(ctx)) {
329314
return;
@@ -501,8 +486,6 @@ STFUNC(TTopicInfoBasedActor::StateFunc) {
501486

502487

503488
class TMessageBusServerPersQueueImpl : public TActorBootstrapped<TMessageBusServerPersQueueImpl> {
504-
using TEvDescribeAllTopicsRequest = NMsgBusProxy::NPqMetaCacheV2::TEvPqNewMetaCache::TEvDescribeAllTopicsRequest;
505-
using TEvDescribeAllTopicsResponse = NMsgBusProxy::NPqMetaCacheV2::TEvPqNewMetaCache::TEvDescribeAllTopicsResponse;
506489

507490
protected:
508491
NKikimrClient::TPersQueueRequest RequestProto;
@@ -589,6 +572,11 @@ class TMessageBusServerPersQueueImpl : public TActorBootstrapped<TMessageBusServ
589572
}
590573
TopicInfo[d.GetTopic(i)];
591574
}
575+
if (TopicInfo.empty()) {
576+
ErrorReason = "no topics in GetTopicMetadata request";
577+
return;
578+
}
579+
592580
} else if (meta.HasCmdGetPartitionStatus()) {
593581
if (!GetTopicsList(meta.GetCmdGetPartitionStatus().topicrequest()))
594582
return;
@@ -601,6 +589,10 @@ class TMessageBusServerPersQueueImpl : public TActorBootstrapped<TMessageBusServ
601589
}
602590
TopicInfo[d.GetTopic(i)];
603591
}
592+
if (TopicInfo.empty()) {
593+
ErrorReason = "no topics in GetReadSessionsInfo request";
594+
return;
595+
}
604596
}
605597
else
606598
ErrorReason = "Not implemented yet";
@@ -941,31 +933,22 @@ class TMessageBusServerPersQueueImpl : public TActorBootstrapped<TMessageBusServ
941933
}
942934

943935

944-
void Handle(TEvDescribeAllTopicsResponse::TPtr& ev, const TActorContext& ctx) {
936+
void Handle(TEvPqNewMetaCache::TEvDescribeTopicsResponse::TPtr& ev, const TActorContext& ctx) {
945937
--DescribeRequests;
946-
auto& res = ev->Get()->Result->ResultSet;
947-
auto& topics = ev->Get()->Topics;
948-
auto processResult = ProcessMetaCacheAllTopicsResponse(ev);
949-
if (processResult.IsFatal) {
950-
ErrorReason = processResult.Reason;
951-
return SendReplyAndDie(CreateErrorReply(processResult.Status, processResult.ErrorCode, ctx), ctx);
952-
}
953-
954-
NoTopicsAtStart = TopicInfo.empty();
955-
bool hasTopics = !NoTopicsAtStart;
938+
auto& resultSet = ev->Get()->Result->ResultSet;
956939

957-
Y_ABORT_UNLESS(topics.size() == res.size());
958-
auto factory = NPersQueue::TTopicNamesConverterFactory(AppData(ctx)->PQConfig, {});
959-
for (auto i = 0u; i != res.size(); i++) {
960-
auto& entry = res[i];
961-
auto& converter = ev->Get()->Topics[i];
940+
Y_ABORT_UNLESS(TopicInfo.size() == resultSet.size());
941+
for (auto i = 0u; i != resultSet.size(); i++) {
942+
auto& entry = resultSet[i];
943+
auto& converter = ev->Get()->TopicsRequested[i];
962944
if (entry.Kind == TSchemeCacheNavigate::EKind::KindTopic && entry.PQGroupInfo && converter) {
963945
auto& description = entry.PQGroupInfo->Description;
964-
if (!hasTopics || TopicInfo.find(converter->GetClientsideName()) != TopicInfo.end()) {
965-
auto& topicInfo = TopicInfo[converter->GetClientsideName()];
966-
topicInfo.BalancerTabletId = description.GetBalancerTabletID();
967-
topicInfo.PQInfo = entry.PQGroupInfo;
968-
}
946+
auto converter = ev->Get()->TopicsRequested[i]->UpgradeToFullConverter(description.GetPQTabletConfig(),
947+
AppData(ctx)->PQConfig.GetTestDatabaseRoot());
948+
Y_ABORT_UNLESS(TopicInfo.contains(converter->GetClientsideName()));
949+
auto& topicInfo = TopicInfo[converter->GetClientsideName()];
950+
topicInfo.BalancerTabletId = description.GetBalancerTabletID();
951+
topicInfo.PQInfo = entry.PQGroupInfo;
969952
}
970953
}
971954

@@ -1389,8 +1372,14 @@ class TMessageBusServerPersQueueImpl : public TActorBootstrapped<TMessageBusServ
13891372
LOG_DEBUG_S(ctx, NKikimrServices::PERSQUEUE, "scheduling HasDataInfoResponse in " << RequestProto.GetFetchRequest().GetWaitMs());
13901373
ctx.Schedule(TDuration::MilliSeconds(Min<ui32>(RequestProto.GetFetchRequest().GetWaitMs(), 30000)), new TEvPersQueue::TEvHasDataInfoResponse);
13911374
}
1375+
Y_ABORT_UNLESS(!TopicInfo.empty());
13921376

1393-
auto* request = new TEvDescribeAllTopicsRequest();
1377+
TVector<TString> topics;
1378+
topics.reserve(TopicInfo.size());
1379+
for (const auto& [topic, _] : TopicInfo) {
1380+
topics.push_back(topic);
1381+
}
1382+
auto* request = new TEvPqNewMetaCache::TEvDescribeTopicsByNameRequest(topics);
13941383
ctx.Send(SchemeCache, request);
13951384
++DescribeRequests;
13961385

@@ -1406,7 +1395,7 @@ class TMessageBusServerPersQueueImpl : public TActorBootstrapped<TMessageBusServ
14061395

14071396
STRICT_STFUNC(StateFunc,
14081397
HFunc(TEvInterconnect::TEvNodesInfo, Handle);
1409-
HFunc(TEvDescribeAllTopicsResponse, Handle);
1398+
HFunc(TEvPqNewMetaCache::TEvDescribeTopicsResponse, Handle);
14101399
HFunc(TEvTabletPipe::TEvClientDestroyed, Handle);
14111400
HFunc(TEvTabletPipe::TEvClientConnected, Handle);
14121401
HFunc(TEvPersQueue::TEvResponse, Handle);
@@ -1443,6 +1432,10 @@ class TMessageBusServerPersQueueImpl : public TActorBootstrapped<TMessageBusServ
14431432
TopicInfo[topic] = std::move(topicInfo);
14441433
}
14451434
}
1435+
if (TopicInfo.empty()) {
1436+
ErrorReason = "No topics in request";
1437+
return false;
1438+
}
14461439
return true;
14471440
}
14481441
};

‎ydb/core/client/server/msgbus_server_persqueue.h‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,6 @@ struct TProcessingResult {
2525
bool IsFatal = false;
2626
};
2727

28-
TProcessingResult ProcessMetaCacheAllTopicsResponse(NPqMetaCacheV2::TEvPqNewMetaCache::TEvDescribeAllTopicsResponse::TPtr& response);
2928
TProcessingResult ProcessMetaCacheSingleTopicsResponse(const NSchemeCache::TSchemeCacheNavigate::TEntry& entry);
3029

3130
// Worker actor creation
@@ -52,6 +51,9 @@ inline ui64 GetTabletId<TEvTabletPipe::TEvClientConnected>(const TEvTabletPipe::
5251
return ev->TabletId;
5352
}
5453

54+
THashSet<TString> GetTopicsListOrThrow(const ::google::protobuf::RepeatedPtrField<::NKikimrClient::TPersQueueMetaRequest::TTopicRequest>& requests,
55+
THashMap<TString, std::shared_ptr<THashSet<ui64>>>* partitionsToRequest);
56+
5557
// Base class for PQ requests. It requests EvGetNode and creates worker actors for concrete topics.
5658
// Than it starts merge over children responses.
5759
// To use actor you need to:
@@ -112,6 +114,9 @@ class TPersQueueBaseRequestProcessor : public TActorBootstrapped<TPersQueueBaseR
112114
}
113115

114116
protected:
117+
using TopicRequestsExctratorFunc = std::function<::google::protobuf::RepeatedPtrField<::NKikimrClient::TPersQueueMetaRequest::TTopicRequest>(
118+
std::shared_ptr<const NKikimrClient::TPersQueueRequest>&)>;
119+
115120
TPersQueueBaseRequestProcessor(const NKikimrClient::TPersQueueRequest& request, const TActorId& pqMetaCacheId, bool listNodes);
116121

117122
~TPersQueueBaseRequestProcessor();
@@ -135,14 +140,12 @@ class TPersQueueBaseRequestProcessor : public TActorBootstrapped<TPersQueueBaseR
135140

136141
virtual bool ReadyForAnswer(const TActorContext& ctx);
137142
void AnswerAndDie(const TActorContext& ctx);
138-
void GetTopicsListOrThrow(const ::google::protobuf::RepeatedPtrField<::NKikimrClient::TPersQueueMetaRequest::TTopicRequest>& requests, THashMap<TString, std::shared_ptr<THashSet<ui64>>>& partitionsToRequest);
139143

140144
virtual STFUNC(StateFunc);
141145

142146
void Handle(TEvInterconnect::TEvNodesInfo::TPtr& ev, const TActorContext& ctx);
143147
void Handle(NPqMetaCacheV2::TEvPqNewMetaCache::TEvGetNodesMappingResponse::TPtr& ev, const TActorContext& ctx);
144148
void Handle(NPqMetaCacheV2::TEvPqNewMetaCache::TEvDescribeTopicsResponse::TPtr& ev, const TActorContext& ctx);
145-
void Handle(NPqMetaCacheV2::TEvPqNewMetaCache::TEvDescribeAllTopicsResponse::TPtr& ev, const TActorContext& ctx);
146149
void Handle(TEvPersQueue::TEvResponse::TPtr& ev, const TActorContext& ctx);
147150
void HandleTimeout(const TActorContext& ctx);
148151

0 commit comments

Comments
 (0)