More useful error for different coordinators (#15058)

Author: nikvas0

ydb/core/kqp/common/kqp_tx_manager.cpp

@@ -307,6 +307,10 @@ class TKqpTransactionManager : public IKqpTransactionManager {
         return !dontNeedCommit;
     }
 
+    virtual ui64 GetCoordinator() const override {
+        return Coordinator;
+    }
+
     void StartPrepare() override {
         AFL_ENSURE(!CollectOnly);
         AFL_ENSURE(State == ETransactionState::COLLECTING);

ydb/core/kqp/common/kqp_tx_manager.h

@@ -95,6 +95,8 @@ class IKqpTransactionManager {
 
     virtual bool NeedCommit() const = 0;
 
+    virtual ui64 GetCoordinator() const = 0;
+
     virtual void StartPrepare() = 0;
 
     struct TPrepareInfo {

ydb/core/kqp/runtime/kqp_write_actor.cpp

@@ -2581,11 +2581,21 @@ class TKqpBufferWriteActor :public TActorBootstrapped<TKqpBufferWriteActor>, pub
         Process();
     }
 
-    void OnPrepared(IKqpTransactionManager::TPrepareResult&& preparedInfo, ui64 dataSize) override {
+    void OnPrepared(IKqpTransactionManager::TPrepareResult&& preparedInfo, ui64) override {
         if (State != EState::PREPARING) {
             return;
         }
-        Y_UNUSED(preparedInfo, dataSize);
+        if (!preparedInfo.Coordinator || (TxManager->GetCoordinator() && preparedInfo.Coordinator != TxManager->GetCoordinator())) {
+            CA_LOG_E("Handle TEvWriteResult: unable to select coordinator. Tx canceled, actorId: " << SelfId()
+                << ", previously selected coordinator: " << TxManager->GetCoordinator()
+                << ", coordinator selected at propose result: " << preparedInfo.Coordinator);
+
+            TxProxyMon->TxResultAborted->Inc();
+            ReplyErrorAndDie(NYql::NDqProto::StatusIds::CANCELLED,
+                NKikimrIssues::TIssuesIds::TX_DECLINED_IMPLICIT_COORDINATOR,
+                "Unable to choose coordinator.");
+            return;
+        }
         if (TxManager->ConsumePrepareTransactionResult(std::move(preparedInfo))) {
             OnOperationFinished(Counters->BufferActorPrepareLatencyHistogram);
             TxManager->StartExecute();
@@ -2596,11 +2606,10 @@ class TKqpBufferWriteActor :public TActorBootstrapped<TKqpBufferWriteActor>, pub
         Process();
     }
 
-    void OnCommitted(ui64 shardId, ui64 dataSize) override {
+    void OnCommitted(ui64 shardId, ui64) override {
         if (State != EState::COMMITTING) {
             return;
         }
-        Y_UNUSED(dataSize);
         if (TxManager->ConsumeCommitResult(shardId)) {
             CA_LOG_D("Committed TxId=" << TxId.value_or(0));
             OnOperationFinished(Counters->BufferActorCommitLatencyHistogram);
@@ -2615,8 +2624,7 @@ class TKqpBufferWriteActor :public TActorBootstrapped<TKqpBufferWriteActor>, pub
         }
     }
 
-    void OnMessageAcknowledged(ui64 dataSize) override {
-        Y_UNUSED(dataSize);
+    void OnMessageAcknowledged(ui64) override {
         Process();
     }
 
@@ -2641,7 +2649,7 @@ class TKqpBufferWriteActor :public TActorBootstrapped<TKqpBufferWriteActor>, pub
         ReplyErrorAndDie(statusCode, std::move(issues));
     }
 
-    void ReplyErrorAndDie(NYql::NDqProto::StatusIds::StatusCode statusCode, NYql::EYqlIssueCode id, const TString& message, const NYql::TIssues& subIssues = {}) {
+    void ReplyErrorAndDie(NYql::NDqProto::StatusIds::StatusCode statusCode, auto id, const TString& message, const NYql::TIssues& subIssues = {}) {
         BufferWriteActorState.EndError(message);
         BufferWriteActor.EndError(message);