| 2667 | } |
| 2668 | |
| 2669 | Status ClientRequestState::TryKillQueryRemotely( |
| 2670 | const TUniqueId& query_id, const KillQueryRequestPB& request) { |
| 2671 | // The initial status should be INVALID_QUERY_HANDLE so that if there is no other |
| 2672 | // coordinator in the cluster, it will be the status to return. |
| 2673 | Status status = Status::Expected(TErrorCode::INVALID_QUERY_HANDLE, PrintId(query_id)); |
| 2674 | ExecutorGroup all_coordinators = |
| 2675 | ExecEnv::GetInstance()->cluster_membership_mgr()->GetSnapshot()->all_coordinators; |
| 2676 | // Skipping the current impalad. |
| 2677 | unique_ptr<ExecutorGroup> other_coordinators{ExecutorGroup::GetFilteredExecutorGroup( |
| 2678 | &all_coordinators, {ExecEnv::GetInstance()->krpc_address()})}; |
| 2679 | // If we get an RPC error, instead of returning immediately, we record it and move |
| 2680 | // on to the next coordinator. |
| 2681 | Status rpc_errors = Status::OK(); |
| 2682 | for (const auto& backend : other_coordinators->GetAllExecutorDescriptors()) { |
| 2683 | // The logic here is similar to ExecShutdownRequest() |
| 2684 | NetworkAddressPB krpc_addr = MakeNetworkAddressPB(backend.ip_address(), |
| 2685 | backend.address().port(), backend.backend_id(), |
| 2686 | ExecEnv::GetInstance()->rpc_mgr()->GetUdsAddressUniqueId()); |
| 2687 | VLOG_QUERY << "Sending KillQuery() RPC to " << NetworkAddressPBToString(krpc_addr); |
| 2688 | unique_ptr<ControlServiceProxy> proxy; |
| 2689 | Status get_proxy_status = |
| 2690 | ControlService::GetProxy(krpc_addr, backend.address().hostname(), &proxy); |
| 2691 | if (!get_proxy_status.ok()) { |
| 2692 | Status get_proxy_status_to_report{Substitute( |
| 2693 | "KillQuery: Could not get Proxy to ControlService at $0 with error: $1.", |
| 2694 | NetworkAddressPBToString(krpc_addr), get_proxy_status.msg().msg())}; |
| 2695 | rpc_errors.MergeStatus(get_proxy_status_to_report); |
| 2696 | LOG(ERROR) << get_proxy_status_to_report.GetDetail(); |
| 2697 | continue; |
| 2698 | } |
| 2699 | KillQueryResponsePB response; |
| 2700 | const int num_retries = 3; |
| 2701 | const int64_t timeout_ms = 10 * MILLIS_PER_SEC; |
| 2702 | const int64_t backoff_time_ms = 3 * MILLIS_PER_SEC; |
| 2703 | // Currently, a KILL QUERY statement is not interruptible. |
| 2704 | Status rpc_status = RpcMgr::DoRpcWithRetry(proxy, &ControlServiceProxy::KillQuery, |
| 2705 | request, &response, query_ctx_, "KillQuery() RPC failed", num_retries, timeout_ms, |
| 2706 | backoff_time_ms, "CRS_KILL_QUERY_RPC"); |
| 2707 | if (!rpc_status.ok()) { |
| 2708 | LOG(ERROR) << rpc_status.GetDetail(); |
| 2709 | rpc_errors.MergeStatus(rpc_status); |
| 2710 | continue; |
| 2711 | } |
| 2712 | // Currently, we only support killing one query in one KILL QUERY statement. |
| 2713 | DCHECK_EQ(response.statuses_size(), 1); |
| 2714 | status = StatusFromProto(response.statuses(0)); |
| 2715 | if (status.ok()) { |
| 2716 | // Kill succeeded. |
| 2717 | VLOG_QUERY << "KillQuery: Found the coordinator at " |
| 2718 | << NetworkAddressPBToString(krpc_addr); |
| 2719 | SetResultSet({Substitute("Query $0 is killed.", PrintId(query_id))}); |
| 2720 | return query_status_; |
| 2721 | } else if (status.code() != TErrorCode::INVALID_QUERY_HANDLE) { |
| 2722 | LOG(ERROR) << "KillQuery: Found the coordinator at " |
| 2723 | << NetworkAddressPBToString(krpc_addr) |
| 2724 | << " but failed to kill the query: " |
| 2725 | << status.GetDetail(); |
| 2726 | // Kill failed, but we found the coordinator of the query. |
nothing calls this directly
no test coverage detected