[BUG] add upstream for abort req

src/storage/tx/ob_trans_part_ctx.cpp

@@ -3471,8 +3471,9 @@ bool ObPartTransCtx::is_root() const { return !exec_info_.upstream_.is_valid();
 bool ObPartTransCtx::is_leaf() const
 {
   return exec_info_.participants_.empty()
-         // TODO(handora.qc): is root also a leaf?
-         && !is_root();
+    // root must not be leaf, because the distributed txn must be composed by
+    // more than one participants.
+    && !is_root();
 }
 
 //***************************** for 4.0

src/storage/tx/ob_tx_2pc_msg_handler.cpp

@@ -155,6 +155,7 @@ int ObPartTransCtx::post_msg_(const ObTwoPhaseCommitMsgType& msg_type,
   case ObTwoPhaseCommitMsgType::OB_MSG_TX_ABORT_REQ: {
     Ob2pcAbortReqMsg abort_req;
     build_tx_common_msg_(receiver, abort_req);
+    abort_req.upstream_ = ls_id_;
     if (OB_FAIL(post_msg_(receiver, abort_req))) {
       TRANS_LOG(WARN, "rpc post msg failed", K(ret), K(*this), K(receiver), K(msg_type));
     }
@@ -704,6 +705,9 @@ int ObPartTransCtx::handle_tx_2pc_abort_req(const Ob2pcAbortReqMsg &msg)
 
   if (OB_FAIL(set_2pc_request_id_(msg.request_id_))) {
     TRANS_LOG(WARN, "set request id failed", KR(ret), K(msg), K(*this));
+  } else if (msg.upstream_.is_valid() && // upstream may be invalid for orphan msg
+             OB_FAIL(set_2pc_upstream_(msg.upstream_))) {
+    TRANS_LOG(WARN, "set upstream failed", KR(ret), K(msg), K(*this));
   } else if (OB_FAIL(handle_2pc_req(msg_type))) {
     TRANS_LOG(WARN, "handle 2pc request failed", KR(ret), K(msg), K(*this));
   }

src/storage/tx/ob_tx_msg.cpp

@@ -45,7 +45,7 @@ OB_SERIALIZE_MEMBER_INHERIT(Ob2pcPreCommitReqMsg, ObTxMsg, commit_version_);
 OB_SERIALIZE_MEMBER_INHERIT(Ob2pcPreCommitRespMsg, ObTxMsg, commit_version_);
 OB_SERIALIZE_MEMBER_INHERIT(Ob2pcCommitReqMsg, ObTxMsg, commit_version_, prepare_info_array_);
 OB_SERIALIZE_MEMBER_INHERIT(Ob2pcCommitRespMsg, ObTxMsg, commit_version_);
-OB_SERIALIZE_MEMBER_INHERIT(Ob2pcAbortReqMsg, ObTxMsg);
+OB_SERIALIZE_MEMBER_INHERIT(Ob2pcAbortReqMsg, ObTxMsg, upstream_);
 OB_SERIALIZE_MEMBER_INHERIT(Ob2pcAbortRespMsg, ObTxMsg);
 OB_SERIALIZE_MEMBER_INHERIT(Ob2pcClearReqMsg, ObTxMsg);
 OB_SERIALIZE_MEMBER_INHERIT(Ob2pcClearRespMsg, ObTxMsg);

src/storage/tx/ob_tx_msg.h

@@ -363,11 +363,13 @@ namespace transaction
     {
     public:
       Ob2pcAbortReqMsg() :
-          ObTxMsg(TX_2PC_ABORT_REQ)
+        ObTxMsg(TX_2PC_ABORT_REQ),
+        upstream_(share::ObLSID::INVALID_LS_ID)
       {}
     public:
       bool is_valid() const;
-      // INHERIT_TO_STRING_KV("txMsg", ObTxMsg);
+      share::ObLSID upstream_;
+      INHERIT_TO_STRING_KV("txMsg", ObTxMsg, K_(upstream));
       OB_UNIS_VERSION(1);
     };