Fix message cookie and add more logging (ydb-platform#5897)

pixcc · Jun 25, 2024 · 1078162 · 1078162
1 parent 323bf1b
commit 1078162
Show file tree

Hide file tree

Showing 6 changed files with 22 additions and 16 deletions.
diff --git a/ydb/core/persqueue/ownerinfo.cpp b/ydb/core/persqueue/ownerinfo.cpp
@@ -19,7 +19,7 @@ namespace NPQ {
             THolder<TEvPersQueue::TEvResponse> response = MakeHolder<TEvPersQueue::TEvResponse>();
             response->Record.SetStatus(NMsgBusProxy::MSTATUS_OK);
             response->Record.SetErrorCode(NPersQueue::NErrorCode::BAD_REQUEST);
-            response->Record.SetErrorReason("ownership session is killed by another session with id " + OwnerCookie);
+            response->Record.SetErrorReason("ownership session is killed by another session with id " + OwnerCookie + " partition id " + partition.OriginalPartitionId);
             ctx.Send(Sender, response.Release());
         }
         Sender = sender;

diff --git a/ydb/core/persqueue/partition.cpp b/ydb/core/persqueue/partition.cpp
@@ -627,6 +627,7 @@ void TPartition::Handle(TEvPQ::TEvPartitionStatus::TPtr& ev, const TActorContext
     NKikimrPQ::TStatusResponse::TPartResult result;
     result.SetPartition(Partition.InternalPartitionId);
     result.SetGeneration(TabletGeneration);
+    result.SetCookie(++PQRBCookie);
 
     if (DiskIsFull || WaitingForSubDomainQuota(ctx)) {
         result.SetStatus(NKikimrPQ::TStatusResponse::STATUS_DISK_IS_FULL);

diff --git a/ydb/core/persqueue/partition_write.cpp b/ydb/core/persqueue/partition_write.cpp
@@ -607,7 +607,7 @@ void TPartition::HandleOnWrite(TEvPQ::TEvWrite::TPtr& ev, const TActorContext& c
 
     if (!CanEnqueue()) {
         ReplyError(ctx, ev->Get()->Cookie, InactivePartitionErrorCode,
-            TStringBuilder() << "Write to inactive partition");
+            TStringBuilder() << "Write to inactive partition " << Partition.OriginalPartitionId);
         return;
     }
 
@@ -884,7 +884,7 @@ void TPartition::CancelOneWriteOnWrite(const TActorContext& ctx,
 TPartition::EProcessResult TPartition::PreProcessRequest(TRegisterMessageGroupMsg& msg) {
     if (!CanWrite()) {
         ScheduleReplyError(msg.Cookie, InactivePartitionErrorCode,
-            TStringBuilder() << "Write to inactive partition");
+            TStringBuilder() << "Write to inactive partition " << Partition.OriginalPartitionId);
         return EProcessResult::ContinueDrop;
     }
     if (DiskIsFull) {
@@ -916,7 +916,7 @@ void TPartition::ExecRequest(TRegisterMessageGroupMsg& msg, ProcessParameters& p
 TPartition::EProcessResult TPartition::PreProcessRequest(TDeregisterMessageGroupMsg& msg) {
     if (!CanWrite()) {
         ScheduleReplyError(msg.Cookie, InactivePartitionErrorCode,
-            TStringBuilder() << "Write to inactive partition");
+            TStringBuilder() << "Write to inactive partition " << Partition.OriginalPartitionId);
         return EProcessResult::ContinueDrop;
     }
     if (DiskIsFull) {
@@ -940,7 +940,7 @@ void TPartition::ExecRequest(TDeregisterMessageGroupMsg& msg, ProcessParameters&
 TPartition::EProcessResult TPartition::PreProcessRequest(TSplitMessageGroupMsg& msg) {
     if (!CanWrite()) {
         ScheduleReplyError(msg.Cookie, InactivePartitionErrorCode,
-            TStringBuilder() << "Write to inactive partition");
+            TStringBuilder() << "Write to inactive partition " << Partition.OriginalPartitionId);
         return EProcessResult::ContinueDrop;
     }
     if (DiskIsFull) {
@@ -985,7 +985,7 @@ void TPartition::ExecRequest(TSplitMessageGroupMsg& msg, ProcessParameters& para
 TPartition::EProcessResult TPartition::PreProcessRequest(TWriteMsg& p) {
     if (!CanWrite()) {
         ScheduleReplyError(p.Cookie, InactivePartitionErrorCode,
-            TStringBuilder() << "Write to inactive partition");
+            TStringBuilder() << "Write to inactive partition " << Partition.OriginalPartitionId);
         return EProcessResult::ContinueDrop;
     }
     if (DiskIsFull) {
@@ -1004,7 +1004,7 @@ TPartition::EProcessResult TPartition::PreProcessRequest(TWriteMsg& p) {
 bool TPartition::ExecRequest(TWriteMsg& p, ProcessParameters& parameters, TEvKeyValue::TEvRequest* request) {
     if (!CanWrite()) {
         ScheduleReplyError(p.Cookie, InactivePartitionErrorCode,
-            TStringBuilder() << "Write to inactive partition");
+            TStringBuilder() << "Write to inactive partition " << Partition.OriginalPartitionId);
         return false;
     }
     if (DiskIsFull) {

diff --git a/ydb/core/persqueue/writer/writer.cpp b/ydb/core/persqueue/writer/writer.cpp
@@ -219,8 +219,8 @@ class TPartitionWriter: public TActorBootstrapped<TPartitionWriter>, private TRl
     }
 
     void Disconnected(EErrorCode errorCode) {
-        Send(Client, new TEvPartitionWriter::TEvDisconnected());
         BecomeZombie(errorCode, "Disconnected");
+        Send(Client, new TEvPartitionWriter::TEvDisconnected(errorCode));
     }
 
     /// GetWriteId

diff --git a/ydb/core/persqueue/writer/writer.h b/ydb/core/persqueue/writer/writer.h
@@ -151,6 +151,11 @@ struct TEvPartitionWriter {
     };
 
     struct TEvDisconnected: public TEventLocal<TEvDisconnected, EvDisconnected> {
+        TEvDisconnected(TEvWriteResponse::EErrorCode errorCode)
+            : ErrorCode(errorCode) {
+        }
+
+        const TEvWriteResponse::EErrorCode ErrorCode;
     };
 
     struct TEvTxWriteRequest : public TEventLocal<TEvTxWriteRequest, EvTxWriteRequest> {

diff --git a/ydb/services/persqueue_v1/actors/write_session_actor.cpp b/ydb/services/persqueue_v1/actors/write_session_actor.cpp
@@ -1034,10 +1034,6 @@ void TWriteSessionActor<UseMigrationProtocol>::ProcessWriteResponse(
 
 template<bool UseMigrationProtocol>
 void TWriteSessionActor<UseMigrationProtocol>::Handle(NPQ::TEvPartitionWriter::TEvWriteResponse::TPtr& ev, const TActorContext& ctx) {
-    if (State != ES_INITED) {
-        return CloseSession("got write response but not wait for it", PersQueue::ErrorCode::ERROR, ctx);
-    }
-
     const auto& result = *ev->Get();
     if (!result.IsSuccess()) {
         const auto& record = result.Record;
@@ -1048,9 +1044,12 @@ void TWriteSessionActor<UseMigrationProtocol>::Handle(NPQ::TEvPartitionWriter::T
         }
     }
 
+    if (State != ES_INITED) {
+        return CloseSession(TStringBuilder() << "got write response but not wait for it (" << static_cast<int>(State) << ")", PersQueue::ErrorCode::BAD_REQUEST, ctx);
+    }
+
     if (AcceptedRequests.empty()) {
-        CloseSession("got too many replies from server, internal error", PersQueue::ErrorCode::ERROR, ctx);
-        return;
+        return CloseSession("got too many replies from server, internal error", PersQueue::ErrorCode::ERROR, ctx);
     }
 
     const auto& writeRequest = AcceptedRequests.front();
@@ -1064,8 +1063,9 @@ void TWriteSessionActor<UseMigrationProtocol>::Handle(NPQ::TEvPartitionWriter::T
 }
 
 template<bool UseMigrationProtocol>
-void TWriteSessionActor<UseMigrationProtocol>::Handle(NPQ::TEvPartitionWriter::TEvDisconnected::TPtr&, const TActorContext& ctx) {
-    CloseSession("pipe to partition's tablet is dead", PersQueue::ErrorCode::TABLET_PIPE_DISCONNECTED, ctx);
+void TWriteSessionActor<UseMigrationProtocol>::Handle(NPQ::TEvPartitionWriter::TEvDisconnected::TPtr& ev, const TActorContext& ctx) {
+    CloseSession(TStringBuilder() << "pipe to partition's " << Partition << " tablet is dead #" << static_cast<int>(ev->Get()->ErrorCode),
+             PersQueue::ErrorCode::TABLET_PIPE_DISCONNECTED, ctx);
 }
 
 template<bool UseMigrationProtocol>