You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by ta...@apache.org on 2019/04/30 15:35:59 UTC
[impala] 01/07: IMPALA-2990: timeout unresponsive queries in
coordinator
This is an automated email from the ASF dual-hosted git repository.
tarmstrong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit a73ef68745da6542657892d256e5b1bd29c68803
Author: Thomas Tauber-Marshall <tm...@cloudera.com>
AuthorDate: Tue Jan 29 22:05:56 2019 +0000
IMPALA-2990: timeout unresponsive queries in coordinator
The coordinator currently waits indefinitely if it does not receive a
status report from a backend. This could cause a query to hang
indefinitely in certain situations, for example if the backend decides
to cancel itself as a result of failed status report rpcs.
This patch adds a thread to ImpalaServer which periodically iterates
over all queries for which that server is the coordinator and cancels
any that haven't had a report from a backend in a certain amount of
time.
This patch adds two flags:
--status_report_max_retry_s: the maximum number of seconds a backend
will attempt to send status reports before giving up. This is used
in place of --status_report_max_retries which is now deprecated.
--status_report_cancellation_padding: the coordinator will wait
--status_report_max_retry_s *
(1 + --status_report_cancellation_padding / 100)
before concluding a backend is not responding and cancelling the
query.
Testing:
- Added a functional test that runs a query that is cancelled through
the new mechanism.
- Passed a full set of exhaustive tests.
Ran tests on a 10 node cluster loaded with tpch 500:
- Ran the stress test for 1000 queries with the debug actions:
'REPORT_EXEC_STATUS_DELAY:JITTER@1000'
Prior to this patch, this setup results in hanging queries. With
this patch, no hangs were observed.
- Ran perf tests with 4 concurrent streams, 3 iterations per query.
Found no change in performance.
Change-Id: I196c8c6a5633b1960e2c3a3884777be9b3824987
Reviewed-on: http://gerrit.cloudera.org:8080/12299
Reviewed-by: Thomas Marshall <tm...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/common/global-flags.cc | 1 +
be/src/runtime/coordinator-backend-state.cc | 9 +++-
be/src/runtime/coordinator-backend-state.h | 12 ++++-
be/src/runtime/coordinator.cc | 47 +++++++++++++------
be/src/runtime/coordinator.h | 13 ++++--
be/src/runtime/query-state.cc | 35 +++++++--------
be/src/runtime/query-state.h | 9 ++--
be/src/service/impala-server.cc | 70 +++++++++++++++++++++++++++++
be/src/service/impala-server.h | 7 +++
common/thrift/ImpalaInternalService.thrift | 6 +++
common/thrift/generate_error_codes.py | 6 ++-
tests/custom_cluster/test_rpc_timeout.py | 23 ++++++++--
12 files changed, 191 insertions(+), 47 deletions(-)
diff --git a/be/src/common/global-flags.cc b/be/src/common/global-flags.cc
index cb6ea18..2684fa9 100644
--- a/be/src/common/global-flags.cc
+++ b/be/src/common/global-flags.cc
@@ -332,6 +332,7 @@ REMOVED_FLAG(rpc_cnxn_retry_interval_ms);
REMOVED_FLAG(skip_lzo_version_check);
REMOVED_FLAG(staging_cgroup);
REMOVED_FLAG(status_report_interval);
+REMOVED_FLAG(status_report_max_retries);
REMOVED_FLAG(suppress_unknown_disk_id_warnings);
REMOVED_FLAG(use_krpc);
REMOVED_FLAG(use_kudu_kinit);
diff --git a/be/src/runtime/coordinator-backend-state.cc b/be/src/runtime/coordinator-backend-state.cc
index c0dfd8e..625ee5a 100644
--- a/be/src/runtime/coordinator-backend-state.cc
+++ b/be/src/runtime/coordinator-backend-state.cc
@@ -37,6 +37,7 @@
#include "util/counting-barrier.h"
#include "util/error-util-internal.h"
#include "util/network-util.h"
+#include "util/scope-exit-trigger.h"
#include "util/uid-util.h"
#include "common/names.h"
@@ -168,7 +169,11 @@ void Coordinator::BackendState::Exec(
const DebugOptions& debug_options,
const FilterRoutingTable& filter_routing_table,
CountingBarrier* exec_complete_barrier) {
- NotifyBarrierOnExit notifier(exec_complete_barrier);
+ const auto trigger = MakeScopeExitTrigger([&]() {
+ // Ensure that 'last_report_time_ms_' is set prior to the barrier being notified.
+ last_report_time_ms_ = GenerateReportTimestamp();
+ exec_complete_barrier->Notify();
+ });
TExecQueryFInstancesParams rpc_params;
rpc_params.__set_query_ctx(query_ctx());
SetRpcParams(debug_options, filter_routing_table, &rpc_params);
@@ -304,7 +309,7 @@ bool Coordinator::BackendState::ApplyExecStatusReport(
// the update loop below.
lock_guard<SpinLock> l1(exec_summary->lock);
unique_lock<mutex> lock(lock_);
- last_report_time_ms_ = MonotonicMillis();
+ last_report_time_ms_ = GenerateReportTimestamp();
// If this backend completed previously, don't apply the update.
if (IsDoneLocked(lock)) return false;
diff --git a/be/src/runtime/coordinator-backend-state.h b/be/src/runtime/coordinator-backend-state.h
index 1cc67c1..13a7a35 100644
--- a/be/src/runtime/coordinator-backend-state.h
+++ b/be/src/runtime/coordinator-backend-state.h
@@ -129,6 +129,7 @@ class Coordinator::BackendState {
void MergeErrorLog(ErrorLogMap* merged);
const TNetworkAddress& impalad_address() const { return host_; }
+ const TNetworkAddress& krpc_impalad_address() const { return krpc_host_; }
int state_idx() const { return state_idx_; }
/// Valid after Init().
@@ -137,6 +138,11 @@ class Coordinator::BackendState {
/// Only valid after Exec().
int64_t rpc_latency() const { return rpc_latency_; }
+ int64_t last_report_time_ms() {
+ boost::lock_guard<boost::mutex> l(lock_);
+ return last_report_time_ms_;
+ }
+
/// Print host/port info for the first backend that's still in progress as a
/// debugging aid for backend deadlocks.
static void LogFirstInProgress(std::vector<BackendState*> backend_states);
@@ -149,6 +155,9 @@ class Coordinator::BackendState {
/// adding members to 'value', including the remote host name.
void InstanceStatsToJson(rapidjson::Value* value, rapidjson::Document* doc);
+ /// Returns a timestamp using monotonic time for tracking arrival of status reports.
+ static int64_t GenerateReportTimestamp() { return MonotonicMillis(); }
+
private:
/// Execution stats for a single fragment instance.
/// Not thread-safe.
@@ -290,7 +299,8 @@ class Coordinator::BackendState {
/// successful.
bool rpc_sent_ = false;
- /// Set in ApplyExecStatusReport(). Uses MonotonicMillis().
+ /// Initialized in Init(), then set in each call to ApplyExecStatusReport().
+ /// Uses GenerateReportTimeout().
int64_t last_report_time_ms_ = 0;
const TQueryCtx& query_ctx() const { return coord_.query_ctx(); }
diff --git a/be/src/runtime/coordinator.cc b/be/src/runtime/coordinator.cc
index d04d6bc..5d10857 100644
--- a/be/src/runtime/coordinator.cc
+++ b/be/src/runtime/coordinator.cc
@@ -45,7 +45,6 @@
#include "scheduling/query-schedule.h"
#include "service/client-request-state.h"
#include "util/bloom-filter.h"
-#include "util/counting-barrier.h"
#include "util/hdfs-bulk-ops.h"
#include "util/hdfs-util.h"
#include "util/histogram-metric.h"
@@ -76,7 +75,8 @@ Coordinator::Coordinator(ClientRequestState* parent, const QuerySchedule& schedu
schedule_(schedule),
filter_mode_(schedule.query_options().runtime_filter_mode),
obj_pool_(new ObjectPool()),
- query_events_(events) {}
+ query_events_(events),
+ exec_rpcs_complete_barrier_(schedule_.per_backend_exec_params().size()) {}
Coordinator::~Coordinator() {
// Must have entered a terminal exec state guaranteeing resources were released.
@@ -348,7 +348,6 @@ void Coordinator::InitFilterRoutingTable() {
void Coordinator::StartBackendExec() {
int num_backends = backend_states_.size();
- exec_rpcs_complete_barrier_.reset(new CountingBarrier(num_backends));
backend_exec_complete_barrier_.reset(new CountingBarrier(num_backends));
DebugOptions debug_options(schedule_.query_options());
@@ -361,11 +360,11 @@ void Coordinator::StartBackendExec() {
ExecEnv::GetInstance()->exec_rpc_thread_pool()->Offer(
[backend_state, this, &debug_options]() {
DebugActionNoFail(schedule_.query_options(), "COORD_BEFORE_EXEC_RPC");
- backend_state->Exec(debug_options, filter_routing_table_,
- exec_rpcs_complete_barrier_.get());
+ backend_state->Exec(
+ debug_options, filter_routing_table_, &exec_rpcs_complete_barrier_);
});
}
- exec_rpcs_complete_barrier_->Wait();
+ exec_rpcs_complete_barrier_.Wait();
VLOG_QUERY << "started execution on " << num_backends << " backends for query_id="
<< PrintId(query_id());
@@ -547,8 +546,7 @@ void Coordinator::HandleExecStateTransition(
// Can't transition until the exec RPCs are no longer in progress. Otherwise, a
// cancel RPC could be missed, and resources freed before a backend has had a chance
// to take a resource reference.
- DCHECK(exec_rpcs_complete_barrier_ != nullptr &&
- exec_rpcs_complete_barrier_->pending() <= 0) << "exec rpcs not completed";
+ DCHECK_LE(exec_rpcs_complete_barrier_.pending(), 0) << "exec rpcs not completed";
query_events_->MarkEvent(exec_state_to_event.at(new_state));
// This thread won the race to transitioning into a terminal state - terminate
@@ -663,8 +661,7 @@ Status Coordinator::GetNext(QueryResultSet* results, int max_rows, bool* eos) {
void Coordinator::Cancel() {
// Illegal to call Cancel() before Exec() returns, so there's no danger of the cancel
// RPC passing the exec RPC.
- DCHECK(exec_rpcs_complete_barrier_ != nullptr &&
- exec_rpcs_complete_barrier_->pending() <= 0) << "Exec() must be called first";
+ DCHECK_LE(exec_rpcs_complete_barrier_.pending(), 0) << "Exec() must be called first";
discard_result(SetNonErrorTerminalState(ExecState::CANCELLED));
// CancelBackends() is called for all transitions into a terminal state except
// for RETURNED_RESULTS. We need to call it now because after Cancel() is called
@@ -725,7 +722,7 @@ Status Coordinator::UpdateBackendExecStatus(const ReportExecStatusRequestPB& req
if (!status.ok()) {
// We may start receiving status reports before all exec rpcs are complete.
// Can't apply state transition until no more exec rpcs will be sent.
- exec_rpcs_complete_barrier_->Wait();
+ exec_rpcs_complete_barrier_.Wait();
// Transition the status if we're not already in a terminal state. This won't block
// because either this transitions to an ERROR state or the query is already in
// a terminal state.
@@ -741,6 +738,30 @@ Status Coordinator::UpdateBackendExecStatus(const ReportExecStatusRequestPB& req
return IsExecuting() ? Status::OK() : Status::CANCELLED;
}
+int64_t Coordinator::GetMaxBackendStateLagMs(TNetworkAddress* address) {
+ if (exec_rpcs_complete_barrier_.pending() > 0) {
+ // Exec() hadn't completed for all the backends, so we can't rely on
+ // 'last_report_time_ms_' being set yet.
+ return 0;
+ }
+ DCHECK_GT(backend_states_.size(), 0);
+ int64_t current_time = BackendState::GenerateReportTimestamp();
+ int64_t min_last_report_time_ms = current_time;
+ BackendState* min_state = nullptr;
+ for (BackendState* backend_state : backend_states_) {
+ if (backend_state->IsDone()) continue;
+ int64_t last_report_time_ms = backend_state->last_report_time_ms();
+ DCHECK_GT(last_report_time_ms, 0);
+ if (last_report_time_ms < min_last_report_time_ms) {
+ min_last_report_time_ms = last_report_time_ms;
+ min_state = backend_state;
+ }
+ }
+ if (min_state == nullptr) return 0;
+ *address = min_state->krpc_impalad_address();
+ return current_time - min_last_report_time_ms;
+}
+
// TODO: add histogram/percentile
void Coordinator::ComputeQuerySummary() {
// In this case, the query did not even get to start all fragment instances.
@@ -891,10 +912,10 @@ void Coordinator::UpdateFilter(const TUpdateFilterParams& params) {
shared_lock<shared_mutex> lock(filter_lock_);
DCHECK_NE(filter_mode_, TRuntimeFilterMode::OFF)
<< "UpdateFilter() called although runtime filters are disabled";
- DCHECK(exec_rpcs_complete_barrier_.get() != nullptr)
+ DCHECK(backend_exec_complete_barrier_.get() != nullptr)
<< "Filters received before fragments started!";
- exec_rpcs_complete_barrier_->Wait();
+ exec_rpcs_complete_barrier_.Wait();
DCHECK(filter_routing_table_complete_)
<< "Filter received before routing table complete";
diff --git a/be/src/runtime/coordinator.h b/be/src/runtime/coordinator.h
index 2135b92..044bdd8 100644
--- a/be/src/runtime/coordinator.h
+++ b/be/src/runtime/coordinator.h
@@ -30,13 +30,13 @@
#include "gen-cpp/Frontend_types.h"
#include "gen-cpp/Types_types.h"
#include "runtime/dml-exec-state.h"
+#include "util/counting-barrier.h"
#include "util/progress-updater.h"
#include "util/runtime-profile-counters.h"
#include "util/spinlock.h"
namespace impala {
-class CountingBarrier;
class ClientRequestState;
class FragmentInstanceState;
class MemTracker;
@@ -135,6 +135,12 @@ class Coordinator { // NOLINT: The member variables could be re-ordered to save
Status UpdateBackendExecStatus(const ReportExecStatusRequestPB& request,
const TRuntimeProfileForest& thrift_profiles) WARN_UNUSED_RESULT;
+ /// Returns the time in ms since the latest report was received for the backend which
+ /// has gone the longest without a report being received, and sets 'address' to the host
+ /// for that backend. May return 0, for example if the backends are not initialized yet
+ /// or if all of them have already completed, in which case 'address' will not be set.
+ int64_t GetMaxBackendStateLagMs(TNetworkAddress* address);
+
/// Get cumulative profile aggregated over all fragments of the query.
/// This is a snapshot of the current state of execution and will change in
/// the future if not all fragments have finished execution.
@@ -327,9 +333,8 @@ class Coordinator { // NOLINT: The member variables could be re-ordered to save
/// sequentially, without synchronization.
std::vector<FragmentStats*> fragment_stats_;
- /// Barrier that is released when all calls to BackendState::Exec() have
- /// returned. Initialized in StartBackendExec().
- boost::scoped_ptr<CountingBarrier> exec_rpcs_complete_barrier_;
+ /// Barrier that is released when all calls to BackendState::Exec() have returned.
+ CountingBarrier exec_rpcs_complete_barrier_;
/// Barrier that is released when all backends have indicated execution completion,
/// or when all backends are cancelled due to an execution error or client requested
diff --git a/be/src/runtime/query-state.cc b/be/src/runtime/query-state.cc
index 134b35e..bdbe983 100644
--- a/be/src/runtime/query-state.cc
+++ b/be/src/runtime/query-state.cc
@@ -56,11 +56,6 @@ using kudu::rpc::RpcSidecar;
static const int DEFAULT_REPORT_WAIT_TIME_MS = 5000;
-DEFINE_int32(status_report_interval_ms, DEFAULT_REPORT_WAIT_TIME_MS,
- "Interval between profile reports in milliseconds. If set to <= 0, periodic "
- "reporting is disabled.");
-DEFINE_int32(status_report_max_retries, 3,
- "Max number of times to retry sending the status report before cancelling");
DECLARE_int32(backend_client_rpc_timeout_ms);
DECLARE_int64(rpc_max_message_size);
@@ -384,8 +379,8 @@ bool QueryState::ReportExecStatus() {
// Note that the sidecar is created with faststring so the ownership of the Thrift
// profile buffer is transferred to RPC layer and it is freed after the RPC payload
// is sent. If serialization of the profile to RPC sidecar fails, we will proceed
- // without the profile so that the coordinator can still get the status instead of
- // hitting IMPALA-2990.
+ // without the profile so that the coordinator can still get the status and won't
+ // conclude that the backend has hung and cancel the query.
if (profile_buf != nullptr) {
unique_ptr<kudu::faststring> sidecar_buf = make_unique<kudu::faststring>();
sidecar_buf->assign_copy(profile_buf, profile_len);
@@ -401,19 +396,25 @@ bool QueryState::ReportExecStatus() {
}
}
+ // TODO: --backend_client_rpc_timeout_ms was originally intended as a socket timeout for
+ // Thrift. We should rethink how we set backend rpc timeouts for krpc.
rpc_controller.set_timeout(
MonoDelta::FromMilliseconds(FLAGS_backend_client_rpc_timeout_ms));
ReportExecStatusResponsePB resp;
rpc_status = FromKuduStatus(proxy_->ReportExecStatus(report, &resp, &rpc_controller),
"ReportExecStatus() RPC failed");
result_status = Status(resp.status());
+ int64_t retry_time_ms = 0;
if (rpc_status.ok()) {
num_failed_reports_ = 0;
+ failed_report_time_ms_ = 0;
} else {
++num_failed_reports_;
+ if (failed_report_time_ms_ == 0) failed_report_time_ms_ = MonotonicMillis();
+ retry_time_ms = MonotonicMillis() - failed_report_time_ms_;
LOG(WARNING) << Substitute("Failed to send ReportExecStatus() RPC for query $0. "
- "Consecutive failed reports = $1",
- PrintId(query_id()), num_failed_reports_);
+ "Consecutive failed reports = $1. Time spent retrying = $2ms.",
+ PrintId(query_id()), num_failed_reports_, retry_time_ms);
}
// Notify the fragment instances of the report's status.
@@ -428,20 +429,16 @@ bool QueryState::ReportExecStatus() {
}
}
- if (((!rpc_status.ok() && num_failed_reports_ >= FLAGS_status_report_max_retries)
+ if (((!rpc_status.ok() && retry_time_ms >= query_ctx().status_report_max_retry_s * 1000)
|| !result_status.ok())
&& instances_started) {
// TODO: should we try to keep rpc_status for the final report? (but the final
// report, following this Cancel(), may not succeed anyway.)
// TODO: not keeping an error status here means that all instances might
// abort with CANCELLED status, despite there being an error
- // TODO: Fix IMPALA-2990. Cancelling fragment instances without sending the
- // ReporExecStatus RPC may cause query to hang as the coordinator may not be aware
- // of the cancellation. Remove the log statements once IMPALA-2990 is fixed.
if (!rpc_status.ok()) {
LOG(ERROR) << "Cancelling fragment instances due to failure to reach the "
- << "coordinator. (" << rpc_status.GetDetail()
- << "). Query " << PrintId(query_id()) << " may hang. See IMPALA-2990.";
+ << "coordinator. (" << rpc_status.GetDetail() << ").";
} else if (!result_status.ok()) {
// If the ReportExecStatus RPC succeeded in reaching the coordinator and we get
// back a non-OK status, it means that the coordinator expects us to cancel the
@@ -457,8 +454,10 @@ bool QueryState::ReportExecStatus() {
}
int64_t QueryState::GetReportWaitTimeMs() const {
- return (FLAGS_status_report_interval_ms > 0 ? FLAGS_status_report_interval_ms :
- DEFAULT_REPORT_WAIT_TIME_MS) * (num_failed_reports_ + 1);
+ int64_t report_interval = query_ctx().status_report_interval_ms > 0 ?
+ query_ctx().status_report_interval_ms :
+ DEFAULT_REPORT_WAIT_TIME_MS;
+ return report_interval * (num_failed_reports_ + 1);
}
void QueryState::ErrorDuringPrepare(const Status& status, const TUniqueId& finst_id) {
@@ -603,7 +602,7 @@ void QueryState::MonitorFInstances() {
// reporting back to the coordinator.
DCHECK(backend_exec_state_ == BackendExecState::EXECUTING)
<< BackendExecStateToString(backend_exec_state_);
- if (FLAGS_status_report_interval_ms > 0) {
+ if (query_ctx().status_report_interval_ms > 0) {
while (!WaitForFinishOrTimeout(GetReportWaitTimeMs())) {
ReportExecStatus();
}
diff --git a/be/src/runtime/query-state.h b/be/src/runtime/query-state.h
index 7aae063..3502ae0 100644
--- a/be/src/runtime/query-state.h
+++ b/be/src/runtime/query-state.h
@@ -94,10 +94,6 @@ class ThriftSerializer;
///
/// TODO:
/// - set up kudu clients in Init(), remove related locking
-/// - IMPALA-2990: when ReportExecStatus() encounters an error, query execution at this
-/// node gets aborted, but it's possible for the coordinator not to find out about that;
-/// fix the coordinator to periodically ping the backends (should the coordinator
-/// simply poll for the status reports?)
class QueryState {
public:
/// Use this class to obtain a QueryState for the duration of a function/block,
@@ -390,6 +386,11 @@ class QueryState {
/// The number of failed intermediate reports since the last successfully sent report.
int64_t num_failed_reports_ = 0;
+ /// If a status report fails, set to the current time using MonotonicMillis(). Reset to
+ /// 0 on a successful report. Used to track how long we've been trying unsuccessfully to
+ /// send a status report so that we can cancel after a configurable timeout.
+ int64_t failed_report_time_ms_ = 0;
+
/// Create QueryState w/ a refcnt of 0 and a memory limit of 'mem_limit' bytes applied
/// to the query mem tracker. The query is associated with the resource pool set in
/// 'query_ctx.request_pool' or from 'request_pool', if the former is not set (needed
diff --git a/be/src/service/impala-server.cc b/be/src/service/impala-server.cc
index df09d21..4b50455 100644
--- a/be/src/service/impala-server.cc
+++ b/be/src/service/impala-server.cc
@@ -212,6 +212,17 @@ DEFINE_int32(idle_query_timeout, 0, "The time, in seconds, that a query may be i
"QUERY_TIMEOUT_S overrides this setting, but, if set, --idle_query_timeout represents"
" the maximum allowable timeout.");
+DEFINE_int32(status_report_interval_ms, 5000, "(Advanced) Interval between profile "
+ "reports in milliseconds. If set to <= 0, periodic reporting is disabled and only "
+ "the final report is sent.");
+DEFINE_int32(status_report_max_retry_s, 600, "(Advanced) Max amount of time in seconds "
+ "for a backend to attempt to send a status report before cancelling. This must be > "
+ "--status_report_interval_ms. Effective only if --status_report_interval_ms > 0.");
+DEFINE_int32(status_report_cancellation_padding, 20, "(Advanced) The coordinator will "
+ "wait --status_report_max_retry_s * (1 + --status_report_cancellation_padding / 100) "
+ "without receiving a status report before deciding that a backend is unresponsive "
+ "and the query should be cancelled. This must be > 0.");
+
DEFINE_bool(is_coordinator, true, "If true, this Impala daemon can accept and coordinate "
"queries from clients. If false, it will refuse client connections.");
DEFINE_bool(is_executor, true, "If true, this Impala daemon will execute query "
@@ -394,6 +405,28 @@ ImpalaServer::ImpalaServer(ExecEnv* exec_env)
ABORT_IF_ERROR(Thread::Create("impala-server", "query-expirer",
bind<void>(&ImpalaServer::ExpireQueries, this), &query_expiration_thread_));
+ // Only enable the unresponsive backend thread if periodic status reporting is enabled.
+ if (FLAGS_status_report_interval_ms > 0) {
+ if (FLAGS_status_report_max_retry_s * 1000 <= FLAGS_status_report_interval_ms) {
+ const string& err = "Since --status_report_max_retry_s <= "
+ "--status_report_interval_ms, most queries will likely be cancelled.";
+ LOG(ERROR) << err;
+ if (FLAGS_abort_on_config_error) {
+ CLEAN_EXIT_WITH_ERROR(Substitute("Aborting Impala Server startup: $0", err));
+ }
+ }
+ if (FLAGS_status_report_cancellation_padding <= 0) {
+ const string& err = "--status_report_cancellationn_padding should be > 0.";
+ LOG(ERROR) << err;
+ if (FLAGS_abort_on_config_error) {
+ CLEAN_EXIT_WITH_ERROR(Substitute("Aborting Impala Server startup: $0", err));
+ }
+ }
+ ABORT_IF_ERROR(Thread::Create("impala-server", "unresponsive-backend-thread",
+ bind<void>(&ImpalaServer::UnresponsiveBackendThread, this),
+ &unresponsive_backend_thread_));
+ }
+
is_coordinator_ = FLAGS_is_coordinator;
is_executor_ = FLAGS_is_executor;
}
@@ -1021,6 +1054,8 @@ void ImpalaServer::PrepareQueryContext(const TNetworkAddress& backend_addr,
query_ctx->__set_coord_address(backend_addr);
query_ctx->__set_coord_krpc_address(krpc_addr);
query_ctx->__set_local_time_zone(local_tz_name);
+ query_ctx->__set_status_report_interval_ms(FLAGS_status_report_interval_ms);
+ query_ctx->__set_status_report_max_retry_s(FLAGS_status_report_max_retry_s);
// Creating a random_generator every time is not free, but
// benchmarks show it to be slightly cheaper than contending for a
@@ -2215,6 +2250,41 @@ void ImpalaServer::UnregisterSessionTimeout(int32_t session_timeout) {
}
}
+[[noreturn]] void ImpalaServer::UnresponsiveBackendThread() {
+ int64_t max_lag_ms = FLAGS_status_report_max_retry_s * 1000
+ * (1 + FLAGS_status_report_cancellation_padding / 100.0);
+ DCHECK_GT(max_lag_ms, 0);
+ VLOG(1) << "Queries will be cancelled if a backend has not reported its status in "
+ << "more than " << max_lag_ms << "ms.";
+ while (true) {
+ vector<CancellationWork> to_cancel;
+ client_request_state_map_.DoFuncForAllEntries(
+ [&](const std::shared_ptr<ClientRequestState>& request_state) {
+ Coordinator* coord = request_state->GetCoordinator();
+ if (coord != nullptr) {
+ TNetworkAddress address;
+ int64_t lag_time_ms = coord->GetMaxBackendStateLagMs(&address);
+ if (lag_time_ms > max_lag_ms) {
+ to_cancel.push_back(
+ CancellationWork::TerminatedByServer(request_state->query_id(),
+ Status(TErrorCode::UNRESPONSIVE_BACKEND,
+ PrintId(request_state->query_id()),
+ TNetworkAddressToString(address), lag_time_ms, max_lag_ms),
+ false /* unregister */));
+ }
+ }
+ });
+
+ // We call Offer() outside of DoFuncForAllEntries() to ensure that if the
+ // cancellation_thread_pool_ queue is full, we're not blocked while holding one of the
+ // 'client_request_state_map_' shard locks.
+ for (auto cancellation_work : to_cancel) {
+ cancellation_thread_pool_->Offer(cancellation_work);
+ }
+ SleepForMs(max_lag_ms * 0.1);
+ }
+}
+
Status ImpalaServer::CheckResourceLimits(ClientRequestState* crs) {
Coordinator* coord = crs->GetCoordinator();
// Coordinator may be null if query has not started executing, check again later.
diff --git a/be/src/service/impala-server.h b/be/src/service/impala-server.h
index b576323..5987de0 100644
--- a/be/src/service/impala-server.h
+++ b/be/src/service/impala-server.h
@@ -898,6 +898,10 @@ class ImpalaServer : public ImpalaServiceIf,
/// FLAGS_idle_query_timeout seconds.
[[noreturn]] void ExpireQueries();
+ /// Periodically iterates over all queries and cancels any where a backend hasn't sent a
+ /// status report in greater than GetMaxReportRetryMs().
+ [[noreturn]] void UnresponsiveBackendThread();
+
/// Called from ExpireQueries() to check query resource limits for 'crs'. If the query
/// exceeded a resource limit, returns a non-OK status with information about what
/// limit was exceeded. Returns OK if the query will continue running and expiration
@@ -978,6 +982,9 @@ class ImpalaServer : public ImpalaServiceIf,
/// when there are sessions that have a timeout.
ConditionVariable session_timeout_cv_;
+ /// Thread that runs UnresponsiveBackendThread().
+ std::unique_ptr<Thread> unresponsive_backend_thread_;
+
/// maps from query id to exec state; ClientRequestState is owned by us and referenced
/// as a shared_ptr to allow asynchronous deletion
typedef class ShardedQueryMap<std::shared_ptr<ClientRequestState>>
diff --git a/common/thrift/ImpalaInternalService.thrift b/common/thrift/ImpalaInternalService.thrift
index 6aa0cea..ea05d4a 100644
--- a/common/thrift/ImpalaInternalService.thrift
+++ b/common/thrift/ImpalaInternalService.thrift
@@ -485,6 +485,12 @@ struct TQueryCtx {
// Flag to enable tracing of resource usage consumption for all fragment instances of a
// query. Set in ImpalaServer::PrepareQueryContext().
21: required bool trace_resource_usage = false
+
+ // Taken from the flags of the same name. The coordinator uses these to decide how long
+ // to wait for a report before cancelling a backend, so we want to ensure that the
+ // coordinator and executors for a given query always agree this value.
+ 22: optional i32 status_report_interval_ms
+ 23: optional i32 status_report_max_retry_s
}
// Specification of one output destination of a plan fragment
diff --git a/common/thrift/generate_error_codes.py b/common/thrift/generate_error_codes.py
index 519a7a8..a5d8378 100755
--- a/common/thrift/generate_error_codes.py
+++ b/common/thrift/generate_error_codes.py
@@ -399,7 +399,11 @@ error_codes = (
"Unset or increase NUM_ROWS_PRODUCED_LIMIT query option to produce more rows."),
("EXPR_REWRITE_RESULT_LIMIT_EXCEEDED", 132,
- "Expression rewrite rejected due to result size ($0) exceeding the limit ($1).")
+ "Expression rewrite rejected due to result size ($0) exceeding the limit ($1)."),
+
+ ("UNRESPONSIVE_BACKEND", 133,
+ "Query $0 cancelled due to unresponsive backend: $1 has not sent a report in $2ms "
+ "(max allowed lag is $3ms)")
)
import sys
diff --git a/tests/custom_cluster/test_rpc_timeout.py b/tests/custom_cluster/test_rpc_timeout.py
index 63aca11..40b02fc 100644
--- a/tests/custom_cluster/test_rpc_timeout.py
+++ b/tests/custom_cluster/test_rpc_timeout.py
@@ -28,6 +28,8 @@ class TestRPCTimeout(CustomClusterTestSuite):
resource should be all released."""
TEST_QUERY = "select count(c2.string_col) from \
functional.alltypestiny join functional.alltypessmall c2"
+ # Designed to take approx. 30s.
+ SLOW_TEST_QUERY = TEST_QUERY + " where c2.int_col = sleep(1000)"
@classmethod
def get_workload(self):
@@ -39,12 +41,15 @@ class TestRPCTimeout(CustomClusterTestSuite):
pytest.skip('runs only in exhaustive')
super(TestRPCTimeout, cls).setup_class()
- def execute_query_verify_metrics(self, query, query_options=None, repeat=1):
+ def execute_query_verify_metrics(self, query, query_options=None, repeat=1,
+ expected_exception=None):
for i in range(repeat):
try:
self.execute_query(query, query_options)
- except ImpalaBeeswaxException:
- pass
+ assert expected_exception is None
+ except ImpalaBeeswaxException as e:
+ if expected_exception is not None:
+ assert expected_exception in str(e)
verifiers = [MetricVerifier(i.service)
for i in ImpalaCluster.get_e2e_test_cluster().impalads]
@@ -160,7 +165,7 @@ class TestRPCTimeout(CustomClusterTestSuite):
@pytest.mark.execute_serially
@CustomClusterTestSuite.with_args("--backend_client_rpc_timeout_ms=100"
- " --status_report_interval_ms=1000 --status_report_max_retries=100000")
+ " --status_report_interval_ms=1000 --status_report_max_retry_s=1000")
def test_reportexecstatus_retries(self, unique_database):
tbl = "%s.kudu_test" % unique_database
self.execute_query("create table %s (a int primary key) stored as kudu" % tbl)
@@ -174,3 +179,13 @@ class TestRPCTimeout(CustomClusterTestSuite):
# Ensure that the error log was tracked correctly - all but the first row inserted
# should result in a 'key already present' insert error.
assert "(1 of 99999 similar)" in result.log, str(result)
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args("--status_report_interval_ms=100000 "
+ "--status_report_max_retry_s=1 --abort_on_config_error=false")
+ def test_unresponsive_backend(self, unique_database):
+ """Test the UnresponsiveBackendThread by setting a status report retry time that is
+ much lower than the status report interval, ensuring that the coordinator will
+ conclude that the backend is unresponsive."""
+ self.execute_query_verify_metrics(self.SLOW_TEST_QUERY,
+ expected_exception="cancelled due to unresponsive backend")