Merge ccc6a4e into 80754f0

Author: nshestakov

ydb/services/deprecated/persqueue_v0/grpc_pq_write_actor.cpp

@@ -310,7 +310,10 @@ void TWriteSessionActor::SetupCounters(const TString& cloudId, const TString& db
 
 
 void TWriteSessionActor::Handle(TEvDescribeTopicsResponse::TPtr& ev, const TActorContext& ctx) {
-    Y_ABORT_UNLESS(State == ES_WAIT_SCHEME || State == ES_INITED);
+    if (State != ES_WAIT_SCHEME && State != ES_INITED) {
+        return CloseSession("erroneous internal state", NPersQueue::NErrorCode::ERROR, ctx);
+    }
+
     auto& res = ev->Get()->Result;
     Y_ABORT_UNLESS(res->ResultSet.size() == 1);
 
@@ -865,7 +868,7 @@ void TWriteSessionActor::LogSession(const TActorContext& ctx) {
 
 void TWriteSessionActor::HandleWakeup(const TActorContext& ctx) {
     if (State != ES_INITED) {
-        return;
+        return CloseSession("erroneous internal state", NPersQueue::NErrorCode::ERROR, ctx);
     }
 
     auto now = ctx.Now();

ydb/services/persqueue_v1/actors/write_session_actor.ipp

@@ -1505,7 +1505,10 @@ void TWriteSessionActor<UseMigrationProtocol>::Handle(TEvents::TEvWakeup::TPtr&
 
 template<bool UseMigrationProtocol>
 void TWriteSessionActor<UseMigrationProtocol>::RecheckACL(const TActorContext& ctx) {
-    Y_ABORT_UNLESS(State == ES_INITED);
+    if (State != ES_INITED) {
+        LOG_ERROR_S(ctx, NKikimrServices::PQ_WRITE_PROXY, "WriteSessionActor state is wrong. Actual state '" << (int)State << "'");
+        return CloseSession("erroneous internal state", PersQueue::ErrorCode::ERROR, ctx);
+    }
 
     auto now = ctx.Now();