A careful end to the write session (ydb-platform#2717)

jepett0 · Mar 14, 2024 · e89e3eb · e89e3eb
1 parent fef7607
commit e89e3eb
Show file tree

Hide file tree

Showing 4 changed files with 13 additions and 0 deletions.
diff --git a/ydb/services/deprecated/persqueue_v0/grpc_pq_actor.h b/ydb/services/deprecated/persqueue_v0/grpc_pq_actor.h
@@ -554,6 +554,8 @@ class TWriteSessionActor : public NActors::TActorBootstrapped<TWriteSessionActor
     NPersQueue::TWriteRequest::TInit InitRequest;
 
     TActorId PartitionChooser;
+
+    bool SessionClosed = false;
 };
 
 class TReadSessionActor : public TActorBootstrapped<TReadSessionActor> {

diff --git a/ydb/services/deprecated/persqueue_v0/grpc_pq_write_actor.cpp b/ydb/services/deprecated/persqueue_v0/grpc_pq_write_actor.cpp
@@ -506,6 +506,11 @@ void TWriteSessionActor::ProceedPartition(const ui32 partition, const TActorCont
 }
 
 void TWriteSessionActor::CloseSession(const TString& errorReason, const NPersQueue::NErrorCode::EErrorCode errorCode, const NActors::TActorContext& ctx) {
+    if (SessionClosed) {
+        return;
+    }
+    SessionClosed = true;
+
     if (errorCode != NPersQueue::NErrorCode::OK) {
         if (InternalErrorCode(errorCode)) {
             SLIErrors.Inc();

diff --git a/ydb/services/persqueue_v1/actors/write_session_actor.h b/ydb/services/persqueue_v1/actors/write_session_actor.h
@@ -271,6 +271,8 @@ class TWriteSessionActor
 
     TActorId PartitionWriterCache;
     TActorId PartitionChooser;
+
+    bool SessionClosed = false;
 };
 
 }

diff --git a/ydb/services/persqueue_v1/actors/write_session_actor.ipp b/ydb/services/persqueue_v1/actors/write_session_actor.ipp
@@ -726,6 +726,10 @@ void TWriteSessionActor<UseMigrationProtocol>::DestroyPartitionWriterCache(const
 
 template<bool UseMigrationProtocol>
 void TWriteSessionActor<UseMigrationProtocol>::CloseSession(const TString& errorReason, const PersQueue::ErrorCode::ErrorCode errorCode, const NActors::TActorContext& ctx) {
+    if (SessionClosed) {
+        return;
+    }
+    SessionClosed = true;
 
     if (errorCode != PersQueue::ErrorCode::OK) {
-Original file line number
+Diff line change
@@ Expand Up / @@ -271,6 +271,8 @@ class TWriteSessionActor @@
         TActorId PartitionWriterCache;
         TActorId PartitionChooser;
+        bool SessionClosed = false;
     };
     }
@@ Expand Down @@