[Stable-25-3] Fix evwrite rollback wait (#29153)

Author: nikvas0

ydb/core/kqp/common/kqp_tx_manager.cpp

@@ -36,7 +36,7 @@ class TKqpTransactionManager : public IKqpTransactionManager {
         : CollectOnly(collectOnly) {}
 
     void AddShard(ui64 shardId, bool isOlap, const TString& path) override {
-        Y_ABORT_UNLESS(State == ETransactionState::COLLECTING || State == ETransactionState::ERROR);
+        AFL_ENSURE(State == ETransactionState::COLLECTING || State == ETransactionState::ERROR);
         ShardsIds.insert(shardId);
         auto& shardInfo = ShardsInfo[shardId];
         shardInfo.IsOlap = isOlap;
@@ -48,7 +48,7 @@ class TKqpTransactionManager : public IKqpTransactionManager {
     }
 
     void AddAction(ui64 shardId, ui8 action) override {
-        Y_ABORT_UNLESS(State == ETransactionState::COLLECTING || State == ETransactionState::ERROR);
+        AFL_ENSURE(State == ETransactionState::COLLECTING || State == ETransactionState::ERROR);
         ShardsInfo.at(shardId).Flags |= action;
         if (action & EAction::WRITE) {
             ReadOnly = false;
@@ -57,7 +57,7 @@ class TKqpTransactionManager : public IKqpTransactionManager {
     }
 
     void AddTopic(ui64 topicId, const TString& path) override {
-        Y_ABORT_UNLESS(State == ETransactionState::COLLECTING || State == ETransactionState::ERROR);
+        AFL_ENSURE(State == ETransactionState::COLLECTING || State == ETransactionState::ERROR);
         ShardsIds.insert(topicId);
         auto& shardInfo = ShardsInfo[topicId];
 
@@ -83,7 +83,7 @@ class TKqpTransactionManager : public IKqpTransactionManager {
     }
 
     bool AddLock(ui64 shardId, const NKikimrDataEvents::TLock& lockProto) override {
-        Y_ABORT_UNLESS(State == ETransactionState::COLLECTING || State == ETransactionState::ERROR);
+        AFL_ENSURE(State == ETransactionState::COLLECTING || State == ETransactionState::ERROR);
         TKqpLock lock(lockProto);
         bool isError = (lock.Proto.GetCounter() >= NKikimr::TSysTables::TLocksTable::TLock::ErrorMin);
         bool isInvalidated = (lock.Proto.GetCounter() == NKikimr::TSysTables::TLocksTable::TLock::ErrorAlreadyBroken)

ydb/core/kqp/runtime/kqp_write_actor.cpp

@@ -2985,12 +2985,22 @@ class TKqpBufferWriteActor :public TActorBootstrapped<TKqpBufferWriteActor>, pub
             }
         }
 
+        Clear();
+
+        Send(MakePipePerNodeCacheID(false), new TEvPipeCache::TEvUnlink(0));
+        TActorBootstrapped<TKqpBufferWriteActor>::PassAway();
+    }
+
+    void Clear() {
         ForEachWriteActor([](TKqpTableWriteActor* actor, const TActorId) {
             actor->Terminate();
         });
 
-        Send(MakePipePerNodeCacheID(false), new TEvPipeCache::TEvUnlink(0));
-        TActorBootstrapped<TKqpBufferWriteActor>::PassAway();
+        {
+            Y_ABORT_UNLESS(Alloc);
+            TGuard<NMiniKQL::TScopedAlloc> allocGuard(*Alloc);
+            WriteInfos.clear();
+        }
     }
 
     void Handle(TEvTxProxy::TEvProposeTransactionStatus::TPtr &ev) {
@@ -3849,6 +3859,8 @@ class TKqpBufferWriteActor :public TActorBootstrapped<TKqpBufferWriteActor>, pub
             std::move(issues),
             BuildStats()
         });
+
+        Clear();
     }
 
     TString GetPathes(ui64 shardId) const {