Delayed message loses `Sender` (#13066)

2 files changed, 9 insertions, 2 deletions

diff --git a/ydb/core/persqueue/events/internal.h b/ydb/core/persqueue/events/internal.h
index 6d23f435d95..bc7e3df210d 100644
--- a/ydb/core/persqueue/events/internal.h
+++ b/ydb/core/persqueue/events/internal.h
@@ -1066,6 +1066,7 @@ struct TEvPQ {
     };
 
     struct TEvGetWriteInfoRequest : public TEventLocal<TEvGetWriteInfoRequest, EvGetWriteInfoRequest> {
+        TActorId OriginalPartition;
     };
 
     struct TEvGetWriteInfoResponse : public TEventLocal<TEvGetWriteInfoResponse, EvGetWriteInfoResponse> {
diff --git a/ydb/core/persqueue/partition.cpp b/ydb/core/persqueue/partition.cpp
index 656d4ed8d9a..95b908737f5 100644
--- a/ydb/core/persqueue/partition.cpp
+++ b/ydb/core/persqueue/partition.cpp
@@ -1015,6 +1015,7 @@ void TPartition::HandleOnInit(TEvPQ::TEvGetWriteInfoRequest::TPtr& ev, const TAc
 
     Y_ABORT_UNLESS(IsSupportive());
 
+    ev->Get()->OriginalPartition = ev->Sender;
     PendingEvents.emplace_back(ev->ReleaseBase().Release());
 }
 
@@ -1132,11 +1133,16 @@ void TPartition::Handle(TEvPQ::TEvTxRollback::TPtr& ev, const TActorContext& ctx
 
 void TPartition::Handle(TEvPQ::TEvGetWriteInfoRequest::TPtr& ev, const TActorContext& ctx) {
     PQ_LOG_D("Handle TEvPQ::TEvGetWriteInfoRequest");
+    TActorId originalPartition = ev->Get()->OriginalPartition;
+    if (!originalPartition) {
+        // original message
+        originalPartition = ev->Sender;
+    }
     if (ClosedInternalPartition || WaitingForPreviousBlobQuota() || (CurrentStateFunc() != &TThis::StateIdle)) {
         PQ_LOG_D("Send TEvPQ::TEvGetWriteInfoError");
         auto* response = new TEvPQ::TEvGetWriteInfoError(Partition.InternalPartitionId,
                                                          "Write info requested while writes are not complete");
-        ctx.Send(ev->Sender, response);
+        ctx.Send(originalPartition, response);
         ClosedInternalPartition = true;
         return;
     }
@@ -1160,7 +1166,7 @@ void TPartition::Handle(TEvPQ::TEvGetWriteInfoRequest::TPtr& ev, const TActorCon
     response->InputLags = std::move(SupportivePartitionTimeLag);
 
     PQ_LOG_D("Send TEvPQ::TEvGetWriteInfoResponse");
-    ctx.Send(ev->Sender, response);
+    ctx.Send(originalPartition, response);
 }
 
 void TPartition::WriteInfoResponseHandler(