Use enshure in core/persqueue (#24858)

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/blob/blob.cpp

@@ -4,6 +4,7 @@
 #include <util/string/builder.h>
 #include <util/string/escape.h>
 #include <util/system/unaligned_mem.h>
+#include <ydb/library/actors/core/log.h>
 
 namespace NKikimr {
 namespace NPQ {
@@ -123,7 +124,7 @@ TBatch::TBatch(const NKikimrPQ::TBatchHeader &header, const char* data)
 }
 
 TBatch TBatch::FromBlobs(const ui64 offset, std::deque<TClientBlob>&& blobs) {
-    Y_ABORT_UNLESS(!blobs.empty());
+    AFL_ENSURE(!blobs.empty());
     TBatch batch(offset, blobs.front().GetPartNo());
     for (auto& b : blobs) {
         batch.AddBlob(b);
@@ -183,12 +184,12 @@ TInstant TBatch::GetEndWriteTimestamp() const {
 }
 
 ui32 TBatch::GetPackedSize() const {
-    Y_ABORT_UNLESS(Packed);
+    AFL_ENSURE(Packed);
     return sizeof(ui16) + PackedData.size() + Header.ByteSize();
 }
 
 ui32 TBatch::FindPos(const ui64 offset, const ui16 partNo) const {
-    Y_ABORT_UNLESS(!Packed);
+    AFL_ENSURE(!Packed);
     if (offset < GetOffset() || offset == GetOffset() && partNo < GetPartNo())
         return Max<ui32>();
     if (offset == GetOffset()) {
@@ -257,20 +258,20 @@ const TBatch& THead::GetBatch(ui32 idx) const {
 }
 
 const TBatch& THead::GetLastBatch() const {
-    Y_ABORT_UNLESS(!Batches.empty());
+    AFL_ENSURE(!Batches.empty());
     return Batches.back();
 }
 
 TBatch THead::ExtractFirstBatch() {
-    Y_ABORT_UNLESS(!Batches.empty());
+    AFL_ENSURE(!Batches.empty());
     auto batch = std::move(Batches.front());
     InternalPartsCount -= batch.GetInternalPartsCount();
     Batches.pop_front();
     return batch;
 }
 
 void THead::AddBlob(const TClientBlob& blob) {
-    Y_ABORT_UNLESS(!Batches.empty());
+    AFL_ENSURE(!Batches.empty());
     auto& batch = Batches.back();
     InternalPartsCount -= batch.GetInternalPartsCount();
     batch.AddBlob(blob);
@@ -299,9 +300,7 @@ ui32 THead::GetCount() const
         return 0;
 
     //how much offsets before last batch and how much offsets in last batch
-    Y_ABORT_UNLESS(Batches.front().GetOffset() == Offset,
-                   "front.Offset=%" PRIu64 ", offset=%" PRIu64,
-                   Batches.front().GetOffset(), Offset);
+    AFL_ENSURE(Batches.front().GetOffset() == Offset)("front.Offset", Batches.front().GetOffset())("offset", Offset);
 
     return Batches.back().GetOffset() - Offset + Batches.back().GetCount();
 }
@@ -312,12 +311,12 @@ ui32 THead::GetCount() const
 //
 
 THead::TBatchAccessor THead::MutableBatch(ui32 idx) {
-    Y_ABORT_UNLESS(idx < Batches.size());
+    AFL_ENSURE(idx < Batches.size());
     return TBatchAccessor(Batches[idx]);
 }
 
 THead::TBatchAccessor THead::MutableLastBatch() {
-    Y_ABORT_UNLESS(!Batches.empty());
+    AFL_ENSURE(!Batches.empty());
     return TBatchAccessor(Batches.back());
 }
 
@@ -437,7 +436,7 @@ TPartitionedBlob::TPartitionedBlob(const TPartitionId& partition, const ui64 off
     , MaxBlobSize(maxBlobSize)
     , FastWrite(fastWrite)
 {
-    Y_ABORT_UNLESS(NewHead.Offset == Head.GetNextOffset() && NewHead.PartNo == 0 || headCleared || needCompactHead || Head.PackedSize == 0); // if head not cleared, then NewHead is going after Head
+    AFL_ENSURE(NewHead.Offset == Head.GetNextOffset() && NewHead.PartNo == 0 || headCleared || needCompactHead || Head.PackedSize == 0); // if head not cleared, then NewHead is going after Head
     if (!headCleared) {
         HeadSize = Head.PackedSize + NewHead.PackedSize;
         InternalPartsCount = Head.GetInternalPartsCount() + NewHead.GetInternalPartsCount();
@@ -452,7 +451,7 @@ TPartitionedBlob::TPartitionedBlob(const TPartitionId& partition, const ui64 off
     if (HeadSize == 0) {
         StartOffset = offset;
         NewHead.Offset = offset;
-        //Y_ABORT_UNLESS(StartPartNo == 0);
+        //AFL_ENSURE(StartPartNo == 0);
     }
 }
 
@@ -462,7 +461,7 @@ TString TPartitionedBlob::CompactHead(bool glueHead, THead& head, bool glueNewHe
     valueD.reserve(estimatedSize);
     if (glueHead) {
         for (ui32 pp = 0; pp < head.Batches.size(); ++pp) {
-            Y_ABORT_UNLESS(head.Batches[pp].Packed);
+            AFL_ENSURE(head.Batches[pp].Packed);
             head.Batches[pp].SerializeTo(valueD);
         }
     }
@@ -471,12 +470,12 @@ TString TPartitionedBlob::CompactHead(bool glueHead, THead& head, bool glueNewHe
             TBatch *b = &newHead.Batches[pp];
             TBatch batch;
             if (!b->Packed) {
-                Y_ABORT_UNLESS(pp + 1 == newHead.Batches.size());
+                AFL_ENSURE(pp + 1 == newHead.Batches.size());
                 batch = newHead.Batches[pp];
                 batch.Pack();
                 b = &batch;
             }
-            Y_ABORT_UNLESS(b->Packed);
+            AFL_ENSURE(b->Packed);
             b->SerializeTo(valueD);
         }
     }
@@ -488,9 +487,9 @@ auto TPartitionedBlob::CreateFormedBlob(ui32 size, bool useRename) -> std::optio
     HeadPartNo = NextPartNo;
     ui32 count = (GlueHead ? Head.GetCount() : 0) + (GlueNewHead ? NewHead.GetCount() : 0);
 
-    Y_ABORT_UNLESS(Offset >= (GlueHead ? Head.Offset : NewHead.Offset));
+    AFL_ENSURE(Offset >= (GlueHead ? Head.Offset : NewHead.Offset));
 
-    Y_ABORT_UNLESS(NewHead.GetNextOffset() >= (GlueHead ? Head.Offset : NewHead.Offset));
+    AFL_ENSURE(NewHead.GetNextOffset() >= (GlueHead ? Head.Offset : NewHead.Offset));
 
     TKey tmpKey, dataKey;
 
@@ -513,11 +512,11 @@ auto TPartitionedBlob::CreateFormedBlob(ui32 size, bool useRename) -> std::optio
         auto batch = TBatch::FromBlobs(Offset, std::move(Blobs));
         Blobs.clear();
         batch.Pack();
-        Y_ABORT_UNLESS(batch.Packed);
+        AFL_ENSURE(batch.Packed);
         batch.SerializeTo(valueD);
     }
 
-    Y_ABORT_UNLESS(valueD.size() <= MaxBlobSize && (valueD.size() + size + 1_MB > MaxBlobSize || HeadSize + BlobsSize + size + GetMaxHeaderSize() <= MaxBlobSize));
+    AFL_ENSURE(valueD.size() <= MaxBlobSize && (valueD.size() + size + 1_MB > MaxBlobSize || HeadSize + BlobsSize + size + GetMaxHeaderSize() <= MaxBlobSize));
     HeadSize = 0;
     BlobsSize = 0;
     TClientBlob::CheckBlob(tmpKey, valueD);
@@ -531,11 +530,9 @@ auto TPartitionedBlob::CreateFormedBlob(ui32 size, bool useRename) -> std::optio
 
 auto TPartitionedBlob::Add(TClientBlob&& blob) -> std::optional<TFormedBlobInfo>
 {
-    Y_ABORT_UNLESS(NewHead.Offset >= Head.Offset,
-                   "Head.Offset=%" PRIu64 ", NewHead.Offset=%" PRIu64,
-                   Head.Offset, NewHead.Offset);
+    AFL_ENSURE(NewHead.Offset >= Head.Offset)("Head.Offset", Head.Offset)("NewHead.Offset", NewHead.Offset);
     ui32 size = blob.GetSerializedSize();
-    Y_ABORT_UNLESS(InternalPartsCount < 1000); //just check for future packing
+    AFL_ENSURE(InternalPartsCount < 1000); //just check for future packing
     if (HeadSize + BlobsSize + size + GetMaxHeaderSize() > MaxBlobSize) {
         NeedCompactHead = true;
     }
@@ -617,22 +614,20 @@ TBlobIterator::TBlobIterator(const TKey& key, const TString& blob)
     , Count(0)
     , InternalPartsCount(0)
 {
-    Y_ABORT_UNLESS(Data != End,
-                   "Key=%s, blob.size=%" PRISZT,
-                   Key.ToString().data(), blob.size());
+    AFL_ENSURE(Data != End)("Key", Key.ToString())("blob.size", blob.size());
     ParseBatch();
-    Y_ABORT_UNLESS(Header.GetPartNo() == Key.GetPartNo());
+    AFL_ENSURE(Header.GetPartNo() == Key.GetPartNo());
 }
 
 void TBlobIterator::ParseBatch() {
-    Y_ABORT_UNLESS(Data < End);
+    AFL_ENSURE(Data < End);
     Header = ExtractHeader(Data, End - Data);
-    //Y_ABORT_UNLESS(Header.GetOffset() == Offset);
+    //AFL_ENSURE(Header.GetOffset() == Offset);
     Count += Header.GetCount();
     Offset += Header.GetCount();
     InternalPartsCount += Header.GetInternalPartsCount();
-    Y_ABORT_UNLESS(Count <= Key.GetCount());
-    Y_ABORT_UNLESS(InternalPartsCount <= Key.GetInternalPartsCount());
+    AFL_ENSURE(Count <= Key.GetCount());
+    AFL_ENSURE(InternalPartsCount <= Key.GetInternalPartsCount());
 }
 
 bool TBlobIterator::IsValid()
@@ -642,11 +637,11 @@ bool TBlobIterator::IsValid()
 
 bool TBlobIterator::Next()
 {
-    Y_ABORT_UNLESS(IsValid());
+    AFL_ENSURE(IsValid());
     Data += Header.GetPayloadSize() + sizeof(ui16) + Header.ByteSize();
     if (Data == End) { //this was last batch
-        Y_ABORT_UNLESS(Count == Key.GetCount());
-        Y_ABORT_UNLESS(InternalPartsCount == Key.GetInternalPartsCount());
+        AFL_ENSURE(Count == Key.GetCount());
+        AFL_ENSURE(InternalPartsCount == Key.GetInternalPartsCount());
         return false;
     }
     ParseBatch();
@@ -655,7 +650,7 @@ bool TBlobIterator::Next()
 
 TBatch TBlobIterator::GetBatch()
 {
-    Y_ABORT_UNLESS(IsValid());
+    AFL_ENSURE(IsValid());
 
     return TBatch(Header, Data + sizeof(ui16) + Header.ByteSize());
 }