You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by bo...@apache.org on 2021/05/12 15:49:27 UTC
[impala] 01/03: IMPALA-10413: fix impalad crashes when canceling
the retrying query
This is an automated email from the ASF dual-hosted git repository.
boroknagyz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 42684b44d3f6c8f3cacfccaafb83cb11f1e90287
Author: xqhe <he...@126.com>
AuthorDate: Wed Dec 30 18:51:04 2020 +0800
IMPALA-10413: fix impalad crashes when canceling the retrying query
The crash happens when canceling the retrying query.
If the original query was unregistered while the new query was
being created, it will call HandleRetryFailure to abort the new
query. But the status is ok, so when calling Status::AddDetail
impalad will crash.
After the WaitAsync interface called and before the
retry_request_state moved to retried_client_request_state_ , if
abort the new retry query, retry_request_state need to call
Finalize, otherwise the wait-thread will leak.
In some cases like canceled the original query or closed the session
we may not create the new query, so we also check whether the query
is retried.
Tests:
Add test in tests/custom_cluster/test_query_retries.py and manually
tested 100 times to make sure that there was no Impalad crash
Change-Id: I4fd7228acd0a70d33859029052239f9b9f795e5d
Reviewed-on: http://gerrit.cloudera.org:8080/16911
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/runtime/query-driver.cc | 29 +++++++++++--
be/src/service/client-request-state.h | 4 ++
be/src/service/impala-server.cc | 20 ++++++---
tests/custom_cluster/test_query_retries.py | 67 +++++++++++++++++++++++++++---
4 files changed, 106 insertions(+), 14 deletions(-)
diff --git a/be/src/runtime/query-driver.cc b/be/src/runtime/query-driver.cc
index 983d599..8a1a4bf 100644
--- a/be/src/runtime/query-driver.cc
+++ b/be/src/runtime/query-driver.cc
@@ -29,6 +29,7 @@
#include "common/names.h"
#include "common/thread-debug-info.h"
+DECLARE_string(debug_actions);
namespace impala {
QueryDriver::QueryDriver(ImpalaServer* parent_server) : parent_server_(parent_server) {}
@@ -104,7 +105,9 @@ ClientRequestState* QueryDriver::GetClientRequestState(const TUniqueId& query_id
return retried_client_request_state_.get();
}
DCHECK(client_request_state_ != nullptr);
- DCHECK(client_request_state_->query_id() == query_id);
+ // If cancel the retrying query the retried_client_request_state_ may not have been
+ // updated. In this case return nullptr
+ if (client_request_state_->query_id() != query_id) return nullptr;
return client_request_state_.get();
}
@@ -287,33 +290,50 @@ void QueryDriver::RetryQueryFromThread(
// 'result_cache_max_size' value was already validated in ImpalaHs2Server, so it does
// not need to be validated again.
if (request_state->IsResultCacheingEnabled()) {
- status = parent_server_->SetupResultsCacheing(
- retry_query_handle, session, request_state->result_cache_max_size());
+ status = DebugAction(FLAGS_debug_actions, "QUERY_RETRY_SET_RESULT_CACHE");
+ if (status.ok()) {
+ status = parent_server_->SetupResultsCacheing(
+ retry_query_handle, session, request_state->result_cache_max_size());
+ }
if (!status.ok()) {
string error_msg = Substitute(
"Setting up results cacheing for query $0 failed", PrintId(retry_query_id));
HandleRetryFailure(&status, &error_msg, request_state, retry_query_id);
+ // 'retried_client_request_state_' hasn't been updated at this point, so the active
+ // ClientRequestState is still the original one. When HandleRetryFailure unregisters
+ // the retried query, it actually finalizes the original ClientRequestState. So we
+ // have to explicitly finalize 'retry_request_state', otherwise we'll hit some
+ // illegal states in destroying it.
+ RETURN_VOID_IF_ERROR(retry_request_state->Finalize(false, nullptr));
return;
}
}
// Mark the new query as "in flight".
- status = parent_server_->SetQueryInflight(session, retry_query_handle);
+ status = DebugAction(FLAGS_debug_actions, "QUERY_RETRY_SET_QUERY_IN_FLIGHT");
+ if (status.ok()) {
+ status = parent_server_->SetQueryInflight(session, retry_query_handle);
+ }
if (!status.ok()) {
string error_msg = Substitute(
"SetQueryInFlight for new query with id $0 failed", PrintId(retry_query_id));
HandleRetryFailure(&status, &error_msg, request_state, retry_query_id);
+ RETURN_VOID_IF_ERROR(retry_request_state->Finalize(false, nullptr));
return;
}
+ DebugActionNoFail(FLAGS_debug_actions, "RETRY_DELAY_CHECKING_ORIGINAL_DRIVER");
// 'client_request_state_' points to the original query and
// 'retried_client_request_state_' points to the retried query.
{
// Before exposing the new query, check if the original query was unregistered while
// the new query was being created. If it was, then abort the new query.
if (parent_server_->GetQueryDriver(query_id) == nullptr) {
+ status = Status(TErrorCode::RUNTIME_ERROR,
+ "Failed to retry query since the original query was unregistered");
string error_msg = Substitute("Query was unregistered");
HandleRetryFailure(&status, &error_msg, request_state, retry_query_id);
+ RETURN_VOID_IF_ERROR(retry_request_state->Finalize(true, nullptr));
return;
}
lock_guard<SpinLock> l(client_request_state_lock_);
@@ -371,6 +391,7 @@ void QueryDriver::CreateRetriedClientRequestState(ClientRequestState* request_st
void QueryDriver::HandleRetryFailure(Status* status, string* error_msg,
ClientRequestState* request_state, const TUniqueId& retry_query_id) {
+ DCHECK(status != nullptr && !status->ok());
status->AddDetail(
Substitute("Failed to retry query $0", PrintId(request_state->query_id())));
status->AddDetail(*error_msg);
diff --git a/be/src/service/client-request-state.h b/be/src/service/client-request-state.h
index c2dd271..5924bb0 100644
--- a/be/src/service/client-request-state.h
+++ b/be/src/service/client-request-state.h
@@ -391,6 +391,10 @@ class ClientRequestState {
return *retried_id_;
}
+ bool IsSetRetriedId() const {
+ return retried_id_ != nullptr;
+ }
+
/// Returns the QueryDriver that owns this ClientRequestState.
QueryDriver* parent_driver() const { return parent_driver_; }
diff --git a/be/src/service/impala-server.cc b/be/src/service/impala-server.cc
index 62698a6..a48f7fe 100644
--- a/be/src/service/impala-server.cc
+++ b/be/src/service/impala-server.cc
@@ -1446,8 +1446,8 @@ void ImpalaServer::UnregisterQueryDiscardResult(
const TUniqueId& query_id, bool check_inflight, const Status* cause) {
Status status = UnregisterQuery(query_id, check_inflight, cause);
if (!status.ok()) {
- LOG(ERROR) << Substitute("Query de-registration for query_id={0} failed: {1}",
- PrintId(query_id, cause->GetDetail()));
+ LOG(ERROR) << Substitute("Query de-registration for query_id=$0 failed: $1",
+ PrintId(query_id), cause->GetDetail());
}
}
@@ -1571,7 +1571,13 @@ Status ImpalaServer::GetQueryHandle(
VLOG(1) << err.GetDetail();
return err;
}
- query_handle->SetHandle(query_driver, query_driver->GetClientRequestState(query_id));
+ ClientRequestState* request_state = query_driver->GetClientRequestState(query_id);
+ if (UNLIKELY(request_state == nullptr)) {
+ Status err = Status::Expected(TErrorCode::INVALID_QUERY_HANDLE, PrintId(query_id));
+ VLOG(1) << err.GetDetail();
+ return err;
+ }
+ query_handle->SetHandle(query_driver, request_state);
return Status::OK();
}
@@ -2321,8 +2327,12 @@ void ImpalaServer::QueryStateRecord::Init(const ClientRequestState& query_handle
}
user_has_profile_access = query_handle.user_has_profile_access();
- was_retried = query_handle.WasRetried();
- if (was_retried) retried_query_id = make_unique<TUniqueId>(query_handle.retried_id());
+ // In some cases like canceling and closing the original query or closing the session
+ // we may not create the new query, we also check whether the retrided query id is set.
+ was_retried = query_handle.WasRetried() && query_handle.IsSetRetriedId();
+ if (was_retried) {
+ retried_query_id = make_unique<TUniqueId>(query_handle.retried_id());
+ }
}
bool ImpalaServer::QueryStateRecordLessThan::operator() (
diff --git a/tests/custom_cluster/test_query_retries.py b/tests/custom_cluster/test_query_retries.py
index 5ce74cd..1b6c3b9 100644
--- a/tests/custom_cluster/test_query_retries.py
+++ b/tests/custom_cluster/test_query_retries.py
@@ -435,7 +435,7 @@ class TestQueryRetries(CustomClusterTestSuite):
killed_impalad.kill()
# Wait until the retry is running.
- self.__wait_until_retried(handle)
+ self.__wait_until_retry_state(handle, 'RETRIED')
# Kill another impalad so that another retry is attempted.
self.cluster.impalads[2].kill()
@@ -718,6 +718,64 @@ class TestQueryRetries(CustomClusterTestSuite):
@pytest.mark.execute_serially
@CustomClusterTestSuite.with_args(
+ impalad_args="--debug_actions=RETRY_DELAY_CHECKING_ORIGINAL_DRIVER:SLEEP@1000",
+ statestored_args="--statestore_heartbeat_frequency_ms=60000")
+ def test_retrying_query_cancel(self):
+ """Trigger a query retry, and then cancel and close the retried query in RETRYING
+ state. Validate that it doesn't crash the impalad. Set a really high statestore
+ heartbeat frequency so that killed impalads are not removed from the cluster
+ membership."""
+
+ # Kill an impalad, and run a query. The query should be retried.
+ self.cluster.impalads[1].kill()
+ query = "select count(*) from tpch_parquet.lineitem"
+ handle = self.execute_query_async(query,
+ query_options={'retry_failed_queries': 'true'})
+ self.__wait_until_retry_state(handle, 'RETRYING')
+
+ # Cancel the query.
+ self.client.cancel(handle)
+
+ # Check the original query retry status.
+ profile = self.__get_original_query_profile(handle.get_handle().id)
+ retry_status = re.search("Retry Status: (.*)", profile)
+ assert retry_status.group(1) == 'RETRYING'
+
+ self.client.close_query(handle)
+ time.sleep(2)
+ assert self.cluster.impalads[0].get_pid() is not None, "Coordinator crashed"
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(
+ impalad_args="--debug_actions=QUERY_RETRY_SET_RESULT_CACHE:FAIL",
+ statestored_args="--statestore_heartbeat_frequency_ms=60000")
+ def test_retry_query_result_cacheing_failed(self):
+ """Test setting up results cacheing failed."""
+
+ self.cluster.impalads[1].kill()
+ query = "select count(*) from tpch_parquet.lineitem"
+ self.hs2_client.set_configuration({'retry_failed_queries': 'true'})
+ self.hs2_client.set_configuration_option('impala.resultset.cache.size', '1024')
+ self.hs2_client.execute_async(query)
+ self.assert_eventually(60, 0.1,
+ lambda: self.cluster.get_first_impalad().service.get_num_in_flight_queries() == 1)
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(
+ impalad_args="--debug_actions=QUERY_RETRY_SET_QUERY_IN_FLIGHT:FAIL",
+ statestored_args="--statestore_heartbeat_frequency_ms=60000")
+ def test_retry_query_set_query_in_flight_failed(self):
+ """Test setting query in flight failed."""
+
+ # Kill an impalad, and run a query. The query should be retried.
+ self.cluster.impalads[1].kill()
+ query = "select count(*) from tpch_parquet.lineitem"
+ self.execute_query_async(query, query_options={'retry_failed_queries': 'true'})
+ self.assert_eventually(60, 0.1,
+ lambda: self.cluster.get_first_impalad().service.get_num_in_flight_queries() == 1)
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(
statestored_args="-statestore_heartbeat_frequency_ms=60000")
def test_retry_query_timeout(self):
"""Trigger a query retry, and then leave the query handle open until the
@@ -842,10 +900,9 @@ class TestQueryRetries(CustomClusterTestSuite):
if not retried_query_id_search: return None
return retried_query_id_search.group(1)
- def __wait_until_retried(self, handle, timeout=300):
+ def __wait_until_retry_state(self, handle, retry_state, timeout=300):
"""Wait until the given query handle has been retried. This is achieved by polling the
runtime profile of the query and checking the 'Retry Status' field."""
- retried_state = "RETRIED"
def __get_retry_status():
profile = self.__get_original_query_profile(handle.get_handle().id)
@@ -854,10 +911,10 @@ class TestQueryRetries(CustomClusterTestSuite):
start_time = time.time()
retry_status = __get_retry_status()
- while retry_status != retried_state and time.time() - start_time < timeout:
+ while retry_status != retry_state and time.time() - start_time < timeout:
retry_status = __get_retry_status()
time.sleep(0.5)
- if retry_status != retried_state:
+ if retry_status != retry_state:
raise Timeout("query {0} was not retried within timeout".format
(handle.get_handle().id))