diff --git a/ydb/core/tx/columnshard/transactions/operators/ev_write/primary.h b/ydb/core/tx/columnshard/transactions/operators/ev_write/primary.h index 3bb8dffa156f..92408091c320 100644 --- a/ydb/core/tx/columnshard/transactions/operators/ev_write/primary.h +++ b/ydb/core/tx/columnshard/transactions/operators/ev_write/primary.h @@ -131,6 +131,7 @@ class TEvWriteCommitPrimaryTransactionOperator: public TEvWriteCommitSyncTransac using TBase = NOlap::NDataSharing::TExtendedTransactionBase; const ui64 TxId; const ui64 TabletId; + bool CheckFinishedFlag = false; virtual bool DoExecute(NTabletFlatExecutor::TTransactionContext& txc, const NActors::TActorContext& /*ctx*/) override { auto op = Self->GetProgressTxController().GetTxOperatorVerifiedAs(TxId, true); @@ -141,6 +142,7 @@ class TEvWriteCommitPrimaryTransactionOperator: public TEvWriteCommitSyncTransac AFL_WARN(NKikimrServices::TX_COLUMNSHARD_TX)("event", "ack_tablet_duplication")("wait", JoinSeq(",", op->WaitShardsResultAck))( "receive", TabletId); } else { + CheckFinishedFlag = true; AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD_TX)("event", "ack_tablet")("wait", JoinSeq(",", op->WaitShardsResultAck))( "receive", TabletId); Self->GetProgressTxController().WriteTxOperatorInfo(txc, TxId, op->SerializeToProto().SerializeAsString()); diff --git a/ydb/core/tx/columnshard/transactions/operators/ev_write/secondary.h b/ydb/core/tx/columnshard/transactions/operators/ev_write/secondary.h index b91826a74138..f60d7c0b2f4b 100644 --- a/ydb/core/tx/columnshard/transactions/operators/ev_write/secondary.h +++ b/ydb/core/tx/columnshard/transactions/operators/ev_write/secondary.h @@ -62,6 +62,7 @@ class TEvWriteCommitSecondaryTransactionOperator: public TEvWriteCommitSyncTrans private: using TBase = NOlap::NDataSharing::TExtendedTransactionBase; const ui64 TxId; + bool NeedContinueFlag = false; virtual bool DoExecute(NTabletFlatExecutor::TTransactionContext& txc, const NActors::TActorContext& ctx) override { auto op = Self->GetProgressTxController().GetTxOperatorVerifiedAs(TxId, true); @@ -74,12 +75,15 @@ class TEvWriteCommitSecondaryTransactionOperator: public TEvWriteCommitSyncTrans } Self->GetProgressTxController().WriteTxOperatorInfo(txc, TxId, op->SerializeToProto().SerializeAsString()); if (!op->NeedReceiveBroken) { - Self->EnqueueProgressTx(ctx, TxId); + NeedContinueFlag = true; } } return true; } - virtual void DoComplete(const NActors::TActorContext& /*ctx*/) override { + virtual void DoComplete(const NActors::TActorContext& ctx) override { + if (NeedContinueFlag) { + Self->EnqueueProgressTx(ctx, TxId); + } } public: @@ -111,7 +115,6 @@ class TEvWriteCommitSecondaryTransactionOperator: public TEvWriteCommitSyncTrans if (BrokenFlag) { Self->GetProgressTxController().ExecuteOnCancel(TxId, txc); } - Self->EnqueueProgressTx(ctx, TxId); } return true; } @@ -124,6 +127,7 @@ class TEvWriteCommitSecondaryTransactionOperator: public TEvWriteCommitSyncTrans if (BrokenFlag) { Self->GetProgressTxController().CompleteOnCancel(TxId, ctx); } + Self->EnqueueProgressTx(ctx, TxId); } }