Duplication in the code of the partition actor (#17595)

Author: Alek5andr-Kotov

ydb/core/persqueue/partition.cpp

@@ -1446,6 +1446,44 @@ void TPartition::Handle(TEvPQ::TEvGetMaxSeqNoRequest::TPtr& ev, const TActorCont
     ctx.Send(Tablet, response.Release());
 }
 
+void TPartition::OnReadComplete(TReadInfo& info,
+                                TUserInfo* userInfo,
+                                const TEvPQ::TEvBlobResponse* blobResponse,
+                                const TActorContext& ctx)
+{
+    TReadAnswer answer = info.FormAnswer(
+        ctx, blobResponse, StartOffset, EndOffset, Partition, userInfo,
+        info.Destination, GetSizeLag(info.Offset), Tablet, Config.GetMeteringMode(), IsActive()
+    );
+    const auto& resp = dynamic_cast<TEvPQ::TEvProxyResponse*>(answer.Event.Get())->Response;
+
+    if (blobResponse && HasError(*blobResponse)) {
+        if (info.IsSubscription) {
+            TabletCounters.Cumulative()[COUNTER_PQ_READ_SUBSCRIPTION_ERROR].Increment(1);
+        }
+        TabletCounters.Cumulative()[COUNTER_PQ_READ_ERROR].Increment(1);
+        TabletCounters.Percentile()[COUNTER_LATENCY_PQ_READ_ERROR].IncrementFor((ctx.Now() - info.Timestamp).MilliSeconds());
+    } else {
+        if (info.IsSubscription) {
+            TabletCounters.Cumulative()[COUNTER_PQ_READ_SUBSCRIPTION_OK].Increment(1);
+        }
+
+        if (blobResponse) {
+            TabletCounters.Cumulative()[COUNTER_PQ_READ_OK].Increment(1);
+            TabletCounters.Percentile()[COUNTER_LATENCY_PQ_READ_OK].IncrementFor((ctx.Now() - info.Timestamp).MilliSeconds());
+        } else {
+            TabletCounters.Cumulative()[COUNTER_PQ_READ_HEAD_ONLY_OK].Increment(1);
+            TabletCounters.Percentile()[COUNTER_LATENCY_PQ_READ_HEAD_ONLY].IncrementFor((ctx.Now() - info.Timestamp).MilliSeconds());
+        }
+
+        TabletCounters.Cumulative()[COUNTER_PQ_READ_BYTES].Increment(resp->ByteSize());
+    }
+
+    ctx.Send(info.Destination != 0 ? Tablet : ctx.SelfID, answer.Event.Release());
+
+    OnReadRequestFinished(info.Destination, answer.Size, info.User, ctx);
+}
+
 void TPartition::Handle(TEvPQ::TEvBlobResponse::TPtr& ev, const TActorContext& ctx) {
     const ui64 cookie = ev->Get()->GetCookie();
     auto it = ReadInfo.find(cookie);
@@ -1465,28 +1503,7 @@ void TPartition::Handle(TEvPQ::TEvBlobResponse::TPtr& ev, const TActorContext& c
         OnReadRequestFinished(info.Destination, 0, info.User, ctx);
     }
 
-    TReadAnswer answer(info.FormAnswer(
-        ctx, *ev->Get(), StartOffset, EndOffset, Partition, userInfo,
-        info.Destination, GetSizeLag(info.Offset), Tablet, Config.GetMeteringMode(), IsActive()
-    ));
-    const auto& resp = dynamic_cast<TEvPQ::TEvProxyResponse*>(answer.Event.Get())->Response;
-
-    if (HasError(*ev->Get())) {
-        if (info.IsSubscription) {
-            TabletCounters.Cumulative()[COUNTER_PQ_READ_SUBSCRIPTION_ERROR].Increment(1);
-        }
-        TabletCounters.Cumulative()[COUNTER_PQ_READ_ERROR].Increment(1);
-        TabletCounters.Percentile()[COUNTER_LATENCY_PQ_READ_ERROR].IncrementFor((ctx.Now() - info.Timestamp).MilliSeconds());
-    } else {
-        if (info.IsSubscription) {
-            TabletCounters.Cumulative()[COUNTER_PQ_READ_SUBSCRIPTION_OK].Increment(1);
-        }
-        TabletCounters.Cumulative()[COUNTER_PQ_READ_OK].Increment(1);
-        TabletCounters.Percentile()[COUNTER_LATENCY_PQ_READ_OK].IncrementFor((ctx.Now() - info.Timestamp).MilliSeconds());
-        TabletCounters.Cumulative()[COUNTER_PQ_READ_BYTES].Increment(resp->ByteSize());
-    }
-    ctx.Send(info.Destination != 0 ? Tablet : ctx.SelfID, answer.Event.Release());
-    OnReadRequestFinished(info.Destination, answer.Size, info.User, ctx);
+    OnReadComplete(info, userInfo, ev->Get(), ctx);
 }
 
 void TPartition::Handle(TEvPQ::TEvError::TPtr& ev, const TActorContext& ctx) {

ydb/core/persqueue/partition.h

@@ -1013,6 +1013,11 @@ class TPartition : public TActorBootstrapped<TPartition> {
     void DumpKeyValueRequest(const NKikimrClient::TKeyValueRequest& request);
 
     TBlobKeyTokenPtr MakeBlobKeyToken(const TString& key);
+
+    void OnReadComplete(TReadInfo& info,
+                        TUserInfo* userInfo,
+                        const TEvPQ::TEvBlobResponse* blobResponse,
+                        const TActorContext& ctx);
 };
 
 } // namespace NKikimr::NPQ

ydb/core/persqueue/partition_read.cpp

@@ -592,9 +592,9 @@ void TPartition::Handle(TEvPQ::TEvReadTimeout::TPtr& ev, const TActorContext& ct
     auto res = Subscriber.OnTimeout(ev);
     if (!res)
         return;
-    TReadAnswer answer(res->FormAnswer(
-            ctx, StartOffset, res->Offset, Partition, nullptr, res->Destination, 0, Tablet, Config.GetMeteringMode(), IsActive()
-    ));
+    TReadAnswer answer = res->FormAnswer(
+            ctx, nullptr, StartOffset, res->Offset, Partition, nullptr, res->Destination, 0, Tablet, Config.GetMeteringMode(), IsActive()
+    );
     ctx.Send(Tablet, answer.Event.Release());
     PQ_LOG_D(" waiting read cookie " << ev->Get()->Cookie
         << " partition " << Partition << " read timeout for " << res->User << " offset " << res->Offset);
@@ -1041,24 +1041,7 @@ void TPartition::ProcessRead(const TActorContext& ctx, TReadInfo&& info, const u
     if (info.Blobs.empty()) { //all from head, answer right now
         PQ_LOG_D("Reading cookie " << cookie << ". All data is from uncompacted head.");
 
-        TReadAnswer answer = info.FormAnswer(
-            ctx, StartOffset, EndOffset, Partition, &UsersInfoStorage->GetOrCreate(info.User, ctx),
-            info.Destination, GetSizeLag(info.Offset), Tablet, Config.GetMeteringMode(), IsActive()
-        );
-        const auto* ev = dynamic_cast<TEvPQ::TEvProxyResponse*>(answer.Event.Get());
-        Y_ABORT_UNLESS(ev);
-        const auto& resp = ev->Response;
-        if (info.IsSubscription) {
-            TabletCounters.Cumulative()[COUNTER_PQ_READ_SUBSCRIPTION_OK].Increment(1);
-        }
-        TabletCounters.Cumulative()[COUNTER_PQ_READ_HEAD_ONLY_OK].Increment(1);
-        TabletCounters.Percentile()[COUNTER_LATENCY_PQ_READ_HEAD_ONLY].IncrementFor((ctx.Now() - info.Timestamp).MilliSeconds());
-
-        TabletCounters.Cumulative()[COUNTER_PQ_READ_BYTES].Increment(resp->ByteSize());
-
-        ctx.Send(info.Destination != 0 ? Tablet : ctx.SelfID, answer.Event.Release());
-
-        OnReadRequestFinished(info.Destination, answer.Size, info.User, ctx);
+        OnReadComplete(info, &UsersInfoStorage->GetOrCreate(info.User, ctx), nullptr, ctx);
         return;
     }
 

ydb/core/persqueue/subscriber.h

@@ -94,6 +94,7 @@ struct TReadInfo {
 
     TReadAnswer FormAnswer(
         const TActorContext& ctx,
+        const TEvPQ::TEvBlobResponse* response,
         const ui64 startOffset,
         const ui64 endOffset,
         const TPartitionId& partition,
@@ -104,8 +105,11 @@ struct TReadInfo {
         const NKikimrPQ::TPQTabletConfig::EMeteringMode meteringMode,
         const bool isActive
     ) {
-        TEvPQ::TEvBlobResponse response(0, TVector<TRequestedBlob>());
-        return FormAnswer(ctx, response, startOffset, endOffset, partition, ui, dst, sizeLag, tablet, meteringMode, isActive);
+        static TEvPQ::TEvBlobResponse fakeBlobResponse(0, TVector<TRequestedBlob>());
+        if (!response) {
+            response = &fakeBlobResponse;
+        }
+        return FormAnswer(ctx, *response, startOffset, endOffset, partition, ui, dst, sizeLag, tablet, meteringMode, isActive);
     }
 };