Merge 1c87540 into 03d0bea

Author: nshestakov

ydb/core/persqueue/pqrb/mirror_describer.cpp

@@ -4,6 +4,8 @@
 
 #include <google/protobuf/util/message_differencer.h>
 
+#define PQ_ENSURE(condition) AFL_ENSURE(condition)("topic", TopicName)
+
 using namespace NPersQueue;
 
 namespace NKikimr {
@@ -80,7 +82,7 @@ void TMirrorDescriber::DescribeTopic(const TActorContext& ctx) {
     }
 
     auto factory = AppData(ctx)->PersQueueMirrorReaderFactory;
-    Y_ABORT_UNLESS(factory);
+    PQ_ENSURE(factory);
     auto future = factory->GetTopicDescription(Config, CredentialsProvider);
     future.Subscribe(
         [
@@ -114,7 +116,7 @@ void TMirrorDescriber::HandleInitCredentials(TEvPQ::TEvInitCredentials::TPtr& /*
     CredentialsProvider = nullptr;
 
     auto factory = AppData(ctx)->PersQueueMirrorReaderFactory;
-    Y_ABORT_UNLESS(factory);
+    PQ_ENSURE(factory);
     auto future = factory->GetCredentialsProvider(Config.GetCredentials());
     future.Subscribe(
         [
@@ -174,6 +176,24 @@ TString TMirrorDescriber::GetCurrentState() const {
     return "UNKNOWN";
 }
 
+ bool TMirrorDescriber::OnUnhandledException(const std::exception& exc) {
+    LOG_CRIT_S(*TlsActivationContext, NKikimrServices::PQ_MIRROR_DESCRIBER,
+        LogDescription() << "unhandled exception " << TypeName(exc) << ": " << exc.what() << Endl
+            << TBackTrace::FromCurrentException().PrintToString());
+
+    Send(ReadBalancerActorId, new TEvents::TEvPoison());
+    PassAway();
+    return true;
+}
+
+NActors::IActor* CreateMirrorDescriber(
+    const NActors::TActorId& readBalancerActorId,
+    const TString& topicName,
+    const NKikimrPQ::TMirrorPartitionConfig& config
+) {
+    return new TMirrorDescriber(readBalancerActorId, topicName, config);
+}
+
 
 }// NPQ
 }// NKikimr

ydb/core/persqueue/pqrb/mirror_describer.h

@@ -14,7 +14,8 @@
 namespace NKikimr::NPQ {
 
 
-class TMirrorDescriber : public TActorBootstrapped<TMirrorDescriber> {
+class TMirrorDescriber : public TActorBootstrapped<TMirrorDescriber>
+                       , public IActorExceptionHandler {
 private:
     static constexpr TDuration INIT_INTERVAL_MAX = TDuration::Seconds(240);
     static constexpr TDuration INIT_INTERVAL_START = TDuration::Seconds(1);
@@ -76,7 +77,7 @@ class TMirrorDescriber : public TActorBootstrapped<TMirrorDescriber> {
     void HandleCredentialsCreated(TEvPQ::TEvCredentialsCreated::TPtr& ev, const TActorContext& ctx);
     void HandleWakeup(const TActorContext& ctx);
     void HandleDescriptionResult(TEvPQ::TEvMirrorTopicDescription::TPtr& ev, const TActorContext& ctx);
-
+    bool OnUnhandledException(const std::exception&) override;
 private:
     TActorId ReadBalancerActorId;
     TString TopicName;

ydb/core/persqueue/pqrb/mirror_describer_factory.h

@@ -0,0 +1,14 @@
+#pragma once
+
+#include <ydb/core/persqueue/public/config.h>
+#include <ydb/library/actors/core/actorsystem_fwd.h>
+
+namespace NKikimr::NPQ {
+
+NActors::IActor* CreateMirrorDescriber(
+    const NActors::TActorId& readBalancerActorId,
+    const TString& topicName,
+    const NKikimrPQ::TMirrorPartitionConfig& config
+);
+
+}

ydb/core/persqueue/pqrb/read_balancer.cpp

@@ -3,7 +3,7 @@
 #include "read_balancer__txpreinit.h"
 #include "read_balancer__txwrite.h"
 #include "read_balancer_log.h"
-#include "mirror_describer.h"
+#include "mirror_describer_factory.h"
 
 #include <ydb/core/persqueue/events/internal.h>
 #include <ydb/core/protos/counters_pq.pb.h>
@@ -13,6 +13,8 @@
 #include <library/cpp/string_utils/base64/base64.h>
 #include <library/cpp/random_provider/random_provider.h>
 
+#define PQ_ENSURE(condition) AFL_ENSURE(condition)("tablet_id", TabletID())("path", Path)("topic", Topic)
+
 namespace NKikimr {
 namespace NPQ {
 
@@ -289,7 +291,7 @@ void TPersQueueReadBalancer::Handle(TEvPersQueue::TEvUpdateBalancerConfig::TPtr
         if (MirrorTopicDescriberActorId) {
             ctx.Send(MirrorTopicDescriberActorId, new TEvPQ::TEvChangePartitionConfig(nullptr, TabletConfig));
         } else {
-            MirrorTopicDescriberActorId = ctx.Register(new TMirrorDescriber(SelfId(), Topic, TabletConfig.GetPartitionConfig().GetMirrorFrom()));
+            MirrorTopicDescriberActorId = ctx.Register(CreateMirrorDescriber(SelfId(), Topic, TabletConfig.GetPartitionConfig().GetMirrorFrom()));
         }
     } else {
         if (MirrorTopicDescriberActorId) {
@@ -320,7 +322,7 @@ void TPersQueueReadBalancer::Handle(TEvPersQueue::TEvUpdateBalancerConfig::TPtr
     for (auto& p : record.GetPartitions()) {
         auto it = PartitionsInfo.find(p.GetPartition());
         if (it == PartitionsInfo.end()) {
-            Y_ABORT_UNLESS(p.GetPartition() >= prevNextPartitionId && p.GetPartition() < NextPartitionId || NextPartitionId == 0);
+            PQ_ENSURE(p.GetPartition() >= prevNextPartitionId && p.GetPartition() < NextPartitionId || NextPartitionId == 0);
 
             partitionsInfo[p.GetPartition()] = {p.GetTabletId()};
 
@@ -422,7 +424,7 @@ TActorId TPersQueueReadBalancer::GetPipeClient(const ui64 tabletId, const TActor
         pipeClient = ctx.RegisterWithSameMailbox(NTabletPipe::CreateClient(ctx.SelfID, tabletId, clientConfig));
         TabletPipes[tabletId].PipeActor = pipeClient;
         auto res = PipesRequested.insert(tabletId);
-        Y_ABORT_UNLESS(res.second);
+        PQ_ENSURE(res.second);
     } else {
         pipeClient = it->second.PipeActor;
     }

ydb/core/persqueue/pqrb/read_balancer.h

@@ -54,7 +54,8 @@ class TMetricsTimeKeeper {
 };
 
 
-class TPersQueueReadBalancer : public TActor<TPersQueueReadBalancer>, public TTabletExecutedFlat {
+class TPersQueueReadBalancer : public TActor<TPersQueueReadBalancer>,
+                               public TTabletExecutedFlat {
     struct TTxPreInit;
     struct TTxInit;
     struct TTxWrite;

ydb/core/persqueue/pqrb/read_balancer__txinit.h

@@ -33,7 +33,7 @@ struct TPersQueueReadBalancer::TTxInit : public ITransaction {
                 return false;
 
             while (!dataRowset.EndOfSet()) { //found out topic info
-                Y_ABORT_UNLESS(!Self->Inited);
+                AFL_ENSURE(!Self->Inited)("tablet_id", Self->TabletID());
                 Self->PathId  = dataRowset.GetValue<Schema::Data::PathId>();
                 Self->Topic   = dataRowset.GetValue<Schema::Data::Topic>();
                 Self->Path    = dataRowset.GetValue<Schema::Data::Path>();
@@ -50,7 +50,7 @@ struct TPersQueueReadBalancer::TTxInit : public ITransaction {
                 TString config = dataRowset.GetValueOrDefault<Schema::Data::Config>("");
                 if (!config.empty()) {
                     bool res = Self->TabletConfig.ParseFromString(config);
-                    Y_ABORT_UNLESS(res);
+                    AFL_ENSURE(res)("tablet_id", Self->TabletID())("path", Self->Path)("topic", Self->Topic);
 
                     Migrate(Self->TabletConfig);
                     Self->Consumers.clear();

ydb/core/persqueue/pqrb/read_balancer__txwrite.h

@@ -48,7 +48,7 @@ struct TPersQueueReadBalancer::TTxWrite : public ITransaction {
         NIceDb::TNiceDb db(txc.DB);
         TString config;
         bool res = Self->TabletConfig.SerializeToString(&config);
-        Y_ABORT_UNLESS(res);
+        AFL_ENSURE(res)("tablet_id", Self->TabletID())("path", Self->Path)("topic", Self->Topic);
         db.Table<Schema::Data>().Key(1).Update(
             NIceDb::TUpdate<Schema::Data::PathId>(Self->PathId),
             NIceDb::TUpdate<Schema::Data::Topic>(Self->Topic),

ydb/core/persqueue/pqtablet/cache/cache_eviction.h

@@ -4,6 +4,7 @@
 #include "pq_l2_service.h"
 
 #include <ydb/core/base/appdata.h>
+#include <ydb/core/keyvalue/keyvalue_events.h>
 #include <ydb/core/persqueue/events/internal.h>
 #include <ydb/core/persqueue/pqtablet/blob/blob.h>
 

ydb/core/persqueue/pqtablet/cache/read.h

@@ -1,6 +1,5 @@
 #pragma once
 
-#include <ydb/core/persqueue/pqtablet/partition/partition.h>
 #include "pq_l2_service.h"
 #include "cache_eviction.h"
 

ydb/core/persqueue/pqtablet/partition/account_read_quoter.cpp

@@ -12,6 +12,8 @@
 #include <util/string/join.h>
 #include <util/string/vector.h>
 
+#define PQ_ENSURE(condition) AFL_ENSURE(condition)("tablet_id", TabletId)("partition_id", Partition)
+
 namespace NKikimr {
 namespace NPQ {
 
@@ -96,7 +98,7 @@ void TBasicAccountQuoter::HandleQuotaConsumed(NAccountQuoterEvents::TEvConsumed:
         << ", consumed in credit " << ConsumedBytesInCredit << "/" << CreditBytes
     );
     auto it = InProcessQuotaRequestCookies.find(ev->Get()->RequestCookie);
-    Y_ABORT_UNLESS(it != InProcessQuotaRequestCookies.end());
+    PQ_ENSURE(it != InProcessQuotaRequestCookies.end());
     InProcessQuotaRequestCookies.erase(it);
 
     if (!QuotaRequestInFlight) {
@@ -125,14 +127,14 @@ void TBasicAccountQuoter::HandleClearance(TEvQuota::TEvClearance::TPtr& ev, cons
         LimiterDescription() << "Got quota from Kesus:" << ev->Get()->Result << ". Cookie: " << cookie
     );
 
-    Y_ABORT_UNLESS(CurrentQuotaRequestCookie == cookie);
+    PQ_ENSURE(CurrentQuotaRequestCookie == cookie);
     if (!Queue.empty()) {
         ApproveQuota(Queue.front().Request, Queue.front().StartWait, ctx);
         Queue.pop_front();
     }
 
     if (Y_UNLIKELY(ev->Get()->Result != TEvQuota::TEvClearance::EResult::Success)) {
-        Y_ABORT_UNLESS(ev->Get()->Result != TEvQuota::TEvClearance::EResult::Deadline); // We set deadline == inf in quota request.
+        PQ_ENSURE(ev->Get()->Result != TEvQuota::TEvClearance::EResult::Deadline); // We set deadline == inf in quota request.
         if (ctx.Now() - LastReportedErrorTime > TDuration::Minutes(1)) {
             LOG_ERROR_S(ctx, NKikimrServices::PQ_RATE_LIMITER, LimiterDescription() << "Got quota request error: " << ev->Get()->Result);
             LastReportedErrorTime = ctx.Now();
@@ -240,7 +242,7 @@ TQuoterParams TAccountWriteQuoter::CreateQuoterParams(
 ) {
     TQuoterParams params;
     const auto& quotingConfig = pqConfig.GetQuotingConfig();
-    Y_ABORT_UNLESS(quotingConfig.GetTopicWriteQuotaEntityToLimit() != NKikimrPQ::TPQConfig::TQuotingConfig::UNSPECIFIED);
+    AFL_ENSURE(quotingConfig.GetTopicWriteQuotaEntityToLimit() != NKikimrPQ::TPQConfig::TQuotingConfig::UNSPECIFIED);
     auto topicPath = topicConverter->GetFederationPath();
 
     auto topicParts = SplitPath(topicPath); // account/folder/topic // account is first element
@@ -282,5 +284,15 @@ constexpr NKikimrServices::TActivity::EType TAccountWriteQuoter::ActorActivityTy
     return NKikimrServices::TActivity::PERSQUEUE_ACCOUNT_WRITE_QUOTER;
 }
 
+bool TBasicAccountQuoter::OnUnhandledException(const std::exception& exc) {
+    LOG_CRIT_S(*TlsActivationContext, NKikimrServices::PERSQUEUE,
+        LimiterDescription() << " unhandled exception " << TypeName(exc) << ": " << exc.what() << Endl
+            << TBackTrace::FromCurrentException().PrintToString());
+
+    Send(TabletActor, new TEvents::TEvPoison());
+    PassAway();
+    return true;
+}
+
 }// NPQ
 }// NKikimr