hcq/oceanbase
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

not retry if init_sqc message may be sent out successfully

Browse Source
This commit is contained in:
obdev
2022-10-28 02:35:52 +00:00
committed by wangzelin.wzl
parent a7f8810345
commit cf98047e85
6 changed files with 39 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
src/sql/engine/px/ob_px_rpc_processor.cpp
View File

@ -476,7 +476,10 @@ int ObFastInitSqcCB::deal_with_rpc_timeout_err_safely()
{
int ret = OB_SUCCESS;
ObDealWithRpcTimeoutCall call(addr_, retry_info_, timeout_ts_, trace_id_);
// only if it's sure init_sqc msg is not sent to sqc successfully, we can retry the query.
bool init_sqc_not_send_out = (get_error() == EASY_TIMEOUT_NOT_SENT_OUT
|| get_error() == EASY_DISCONNECT_NOT_SENT_OUT);
ObDealWithRpcTimeoutCall call(addr_, retry_info_, timeout_ts_, trace_id_, init_sqc_not_send_out);
call.ret_ = OB_TIMEOUT;
ObGlobalInterruptManager *manager = ObGlobalInterruptManager::getInstance();
if (OB_NOT_NULL(manager)) {
@ -529,7 +532,12 @@ void ObDealWithRpcTimeoutCall::deal_with_rpc_timeout_err()
LOG_WARN("fail to add invalid server distinctly", K_(trace_id), K(a_ret), K_(addr));
}
}
ret_ = OB_RPC_CONNECT_ERROR;
if (can_retry_) {
// return OB_RPC_CONNECT_ERROR to retry.
ret_ = OB_RPC_CONNECT_ERROR;
} else {
ret_ = OB_PACKET_STATUS_UNKNOWN;
}
} else {
LOG_DEBUG("rpc return OB_TIMEOUT, and it is actually timeout, "
"do not change error code", K(ret_),

6
src/sql/engine/px/ob_px_rpc_processor.h
View File

@ -120,8 +120,9 @@ public:
ObDealWithRpcTimeoutCall(common::ObAddr addr,
ObQueryRetryInfo *retry_info,
int64_t timeout_ts,
common::ObCurTraceId::TraceId &trace_id) : addr_(addr), retry_info_(retry_info),
timeout_ts_(timeout_ts), trace_id_(trace_id), ret_(common::OB_TIMEOUT) {}
common::ObCurTraceId::TraceId &trace_id,
bool retry) : addr_(addr), retry_info_(retry_info),
timeout_ts_(timeout_ts), trace_id_(trace_id), ret_(common::OB_TIMEOUT), can_retry_(retry) {}
~ObDealWithRpcTimeoutCall() = default;
void operator() (hash::HashMapPair<ObInterruptibleTaskID,
ObInterruptCheckerNode *> &entry);
@ -132,6 +133,7 @@ public:
int64_t timeout_ts_;
common::ObCurTraceId::TraceId trace_id_;
int ret_;
bool can_retry_;
};
class ObFastInitSqcCB

9
src/sql/engine/px/ob_px_sqc_async_proxy.cpp
View File

@ -185,7 +185,14 @@ int ObPxSqcAsyncProxy::wait_all() {
if (phy_plan_ctx_->get_timeout_timestamp() -
ObTimeUtility::current_time() > 0) {
error_index_ = idx;
ret = OB_RPC_CONNECT_ERROR;
bool init_sqc_not_send_out = (callback.get_error() == EASY_TIMEOUT_NOT_SENT_OUT
|| callback.get_error() == EASY_DISCONNECT_NOT_SENT_OUT);
if (init_sqc_not_send_out) {
// only if it's sure init_sqc msg is not sent to sqc successfully, return OB_RPC_CONNECT_ERROR to retry.
ret = OB_RPC_CONNECT_ERROR;
} else {
ret = OB_PACKET_STATUS_UNKNOWN;
}
} else {
ret = OB_TIMEOUT;
}