You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by mi...@apache.org on 2018/01/08 17:04:03 UTC
[2/3] impala git commit: IMPALA-2248: Make idle_session_timeout a
query option
IMPALA-2248: Make idle_session_timeout a query option
This commit makes idle_session_timeout a query option.
idle_session_timeout currently can be set as a command line
option, which will be the default timeout for sessions.
HS2 sessions can override it with a smaller value by setting
it in the configuration overlay of HS2 OpenSession().
However, we can't override idle_session_timeout for JDBC/ODBC
connections, because we cannot put this in the connection string.
This commit is a workaround for this problem, it allows JDBC/ODBC
connections to set the session timeout as a query option
with the SET statement.
After this commit, the session timeout can be overridden to
any value, i.e. the command line flag idle_session_timeout
doesn't limit this option anymore.
I created an automated test case in JdbcTest.java based on
test_hs2.py::test_concurrent_session_mixed_idle_timeout. I also
extended the test_session_expiration and test_set_and_unset
test suites.
Change-Id: I32e2775f80da387b0df4195fe2c5435b3f8e585e
Reviewed-on: http://gerrit.cloudera.org:8080/8490
Reviewed-by: Tim Armstrong <ta...@cloudera.com>
Tested-by: Impala Public Jenkins
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/ce65b43d
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/ce65b43d
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/ce65b43d
Branch: refs/heads/master
Commit: ce65b43d47d16840998eb5fab5333695050ac436
Parents: 302ec25
Author: Zoltan Borok-Nagy <bo...@cloudera.com>
Authored: Tue Nov 7 14:51:36 2017 +0100
Committer: Impala Public Jenkins <im...@gerrit.cloudera.org>
Committed: Sat Jan 6 01:47:47 2018 +0000
----------------------------------------------------------------------
be/src/service/client-request-state.cc | 18 +++-
be/src/service/impala-hs2-server.cc | 24 ++---
be/src/service/impala-server.cc | 47 +++++---
be/src/service/impala-server.h | 30 ++++--
be/src/service/query-options.cc | 23 +++-
be/src/service/query-options.h | 3 +-
common/thrift/ImpalaInternalService.thrift | 5 +
common/thrift/ImpalaService.thrift | 4 +
.../org/apache/impala/service/JdbcTest.java | 107 +++++++++++++++++--
.../java/org/apache/impala/util/Metrics.java | 82 ++++++++++++++
tests/custom_cluster/test_session_expiration.py | 49 ++++++++-
tests/custom_cluster/test_set_and_unset.py | 47 ++++++++
tests/hs2/hs2_test_suite.py | 10 +-
tests/hs2/test_hs2.py | 2 +-
14 files changed, 391 insertions(+), 60 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/ce65b43d/be/src/service/client-request-state.cc
----------------------------------------------------------------------
diff --git a/be/src/service/client-request-state.cc b/be/src/service/client-request-state.cc
index 76e2557..5e95bb2 100644
--- a/be/src/service/client-request-state.cc
+++ b/be/src/service/client-request-state.cc
@@ -17,6 +17,7 @@
#include "service/client-request-state.h"
+#include <boost/algorithm/string/predicate.hpp>
#include <limits>
#include <gutil/strings/substitute.h>
@@ -42,6 +43,7 @@
#include "common/names.h"
+using boost::algorithm::iequals;
using boost::algorithm::join;
using namespace apache::hive::service::cli::thrift;
using namespace apache::thrift;
@@ -200,12 +202,18 @@ Status ClientRequestState::Exec(TExecRequest* exec_request) {
if (exec_request_.set_query_option_request.__isset.key) {
// "SET key=value" updates the session query options.
DCHECK(exec_request_.set_query_option_request.__isset.value);
- RETURN_IF_ERROR(SetQueryOption(
- exec_request_.set_query_option_request.key,
- exec_request_.set_query_option_request.value,
- &session_->set_query_options,
- &session_->set_query_options_mask));
+ const auto& key = exec_request_.set_query_option_request.key;
+ const auto& value = exec_request_.set_query_option_request.value;
+ RETURN_IF_ERROR(SetQueryOption(key, value, &session_->set_query_options,
+ &session_->set_query_options_mask));
SetResultSet({}, {}, {});
+ if (iequals(key, "idle_session_timeout")) {
+ // IMPALA-2248: Session timeout is set as a query option
+ session_->last_accessed_ms = UnixMillis(); // do not expire session immediately
+ session_->UpdateTimeout();
+ VLOG_QUERY << "ClientRequestState::Exec() SET: idle_session_timeout="
+ << PrettyPrinter::Print(session_->session_timeout, TUnit::TIME_S);
+ }
} else {
// "SET" or "SET ALL"
bool is_set_all = exec_request_.set_query_option_request.__isset.is_set_all &&
http://git-wip-us.apache.org/repos/asf/impala/blob/ce65b43d/be/src/service/impala-hs2-server.cc
----------------------------------------------------------------------
diff --git a/be/src/service/impala-hs2-server.cc b/be/src/service/impala-hs2-server.cc
index 20608b7..2f2ec40 100644
--- a/be/src/service/impala-hs2-server.cc
+++ b/be/src/service/impala-hs2-server.cc
@@ -296,7 +296,7 @@ void ImpalaServer::OpenSession(TOpenSessionResp& return_val,
// query options.
// TODO: put secret in session state map and check it
// TODO: Fix duplication of code between here and ConnectionStart().
- shared_ptr<SessionState> state = make_shared<SessionState>();
+ shared_ptr<SessionState> state = make_shared<SessionState>(this);
state->closed = false;
state->start_time_ms = UnixMillis();
state->session_type = TSessionType::HIVESERVER2;
@@ -316,8 +316,8 @@ void ImpalaServer::OpenSession(TOpenSessionResp& return_val,
// Process the supplied configuration map.
state->database = "default";
- state->session_timeout = FLAGS_idle_session_timeout;
state->server_default_query_options = &default_query_options_;
+ state->session_timeout = FLAGS_idle_session_timeout;
if (request.__isset.configuration) {
typedef map<string, string> ConfigurationMap;
for (const ConfigurationMap::value_type& v: request.configuration) {
@@ -330,22 +330,16 @@ void ImpalaServer::OpenSession(TOpenSessionResp& return_val,
HS2_RETURN_IF_ERROR(return_val, status, SQLSTATE_GENERAL_ERROR);
} else if (iequals(v.first, "use:database")) {
state->database = v.second;
- } else if (iequals(v.first, "idle_session_timeout")) {
- int32_t requested_timeout = atoi(v.second.c_str());
- if (requested_timeout > 0) {
- if (FLAGS_idle_session_timeout > 0) {
- state->session_timeout = min(FLAGS_idle_session_timeout, requested_timeout);
- } else {
- state->session_timeout = requested_timeout;
- }
- }
- VLOG_QUERY << "OpenSession(): idle_session_timeout="
- << PrettyPrinter::Print(state->session_timeout, TUnit::TIME_S);
} else {
// Normal configuration key. Use it to set session default query options.
// Ignore failure (failures will be logged in SetQueryOption()).
- discard_result(SetQueryOption(v.first, v.second, &state->set_query_options,
- &state->set_query_options_mask));
+ Status status = SetQueryOption(v.first, v.second, &state->set_query_options,
+ &state->set_query_options_mask);
+ if (status.ok() && iequals(v.first, "idle_session_timeout")) {
+ state->session_timeout = state->set_query_options.idle_session_timeout;
+ VLOG_QUERY << "OpenSession(): idle_session_timeout="
+ << PrettyPrinter::Print(state->session_timeout, TUnit::TIME_S);
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/impala/blob/ce65b43d/be/src/service/impala-server.cc
----------------------------------------------------------------------
diff --git a/be/src/service/impala-server.cc b/be/src/service/impala-server.cc
index 2222802..49009b6 100644
--- a/be/src/service/impala-server.cc
+++ b/be/src/service/impala-server.cc
@@ -188,7 +188,8 @@ DEFINE_string(ssl_minimum_version, "tlsv1", SSL_MIN_VERSION_HELP.c_str());
DEFINE_int32(idle_session_timeout, 0, "The time, in seconds, that a session may be idle"
" for before it is closed (and all running queries cancelled) by Impala. If 0, idle"
- " sessions are never expired.");
+ " sessions are never expired. It can be overridden by the query option"
+ " 'idle_session_timeout' for specific sessions");
DEFINE_int32(idle_query_timeout, 0, "The time, in seconds, that a query may be idle for"
" (i.e. no processing work is done and no updates are received from the client) "
"before it is cancelled. If 0, idle queries are never expired. The query option "
@@ -1118,14 +1119,7 @@ Status ImpalaServer::CloseSessionInternal(const TUniqueId& session_id,
discard_result(UnregisterQuery(query_id, false, &status));
}
// Reconfigure the poll period of session_timeout_thread_ if necessary.
- int32_t session_timeout = session_state->session_timeout;
- if (session_timeout > 0) {
- lock_guard<mutex> l(session_timeout_lock_);
- multiset<int32_t>::const_iterator itr = session_timeout_set_.find(session_timeout);
- DCHECK(itr != session_timeout_set_.end());
- session_timeout_set_.erase(itr);
- session_timeout_cv_.NotifyOne();
- }
+ UnregisterSessionTimeout(session_state->session_timeout);
return Status::OK();
}
@@ -1203,6 +1197,9 @@ void ImpalaServer::TransmitData(
}
void ImpalaServer::InitializeConfigVariables() {
+ // Set idle_session_timeout here to let the SET command return the value of
+ // the command line option FLAGS_idle_session_timeout
+ default_query_options_.__set_idle_session_timeout(FLAGS_idle_session_timeout);
QueryOptionsMask set_query_options; // unused
Status status = ParseQueryOptions(FLAGS_default_query_options,
&default_query_options_, &set_query_options);
@@ -1227,6 +1224,20 @@ void ImpalaServer::InitializeConfigVariables() {
}
}
+void ImpalaServer::SessionState::UpdateTimeout() {
+ DCHECK(impala_server != nullptr);
+ int32_t old_timeout = session_timeout;
+ if (set_query_options.__isset.idle_session_timeout) {
+ session_timeout = set_query_options.idle_session_timeout;
+ } else {
+ session_timeout = server_default_query_options->idle_session_timeout;
+ }
+ if (old_timeout != session_timeout) {
+ impala_server->UnregisterSessionTimeout(old_timeout);
+ impala_server->RegisterSessionTimeout(session_timeout);
+ }
+}
+
void ImpalaServer::AddOptionLevelToConfig(ConfigVariable* config,
const string& option_key) const {
const auto query_option_level = query_option_levels_.find(option_key);
@@ -1742,8 +1753,7 @@ void ImpalaServer::ConnectionStart(
// Beeswax only allows for one session per connection, so we can share the session ID
// with the connection ID
const TUniqueId& session_id = connection_context.connection_id;
- shared_ptr<SessionState> session_state;
- session_state.reset(new SessionState);
+ shared_ptr<SessionState> session_state = make_shared<SessionState>(this);
session_state->closed = false;
session_state->start_time_ms = UnixMillis();
session_state->last_accessed_ms = UnixMillis();
@@ -1807,11 +1817,22 @@ void ImpalaServer::ConnectionEnd(
void ImpalaServer::RegisterSessionTimeout(int32_t session_timeout) {
if (session_timeout <= 0) return;
- lock_guard<mutex> l(session_timeout_lock_);
- session_timeout_set_.insert(session_timeout);
+ {
+ lock_guard<mutex> l(session_timeout_lock_);
+ session_timeout_set_.insert(session_timeout);
+ }
session_timeout_cv_.NotifyOne();
}
+void ImpalaServer::UnregisterSessionTimeout(int32_t session_timeout) {
+ if (session_timeout > 0) {
+ lock_guard<mutex> l(session_timeout_lock_);
+ auto itr = session_timeout_set_.find(session_timeout);
+ DCHECK(itr != session_timeout_set_.end());
+ session_timeout_set_.erase(itr);
+ }
+}
+
[[noreturn]] void ImpalaServer::ExpireSessions() {
while (true) {
{
http://git-wip-us.apache.org/repos/asf/impala/blob/ce65b43d/be/src/service/impala-server.h
----------------------------------------------------------------------
diff --git a/be/src/service/impala-server.h b/be/src/service/impala-server.h
index 9e6c830..868ae54 100644
--- a/be/src/service/impala-server.h
+++ b/be/src/service/impala-server.h
@@ -350,11 +350,15 @@ class ImpalaServer : public ImpalaServiceIf,
/// run as children of Beeswax sessions) get results back in the expected format -
/// child queries inherit the HS2 version from their parents, and a Beeswax session
/// will never update the HS2 version from the default.
- SessionState() : closed(false), expired(false),
- hs2_version(apache::hive::service::cli::thrift::
+ SessionState(ImpalaServer* impala_server) : impala_server(impala_server),
+ closed(false), expired(false), hs2_version(apache::hive::service::cli::thrift::
TProtocolVersion::HIVE_CLI_SERVICE_PROTOCOL_V1), total_queries(0), ref_count(0) {
+ DCHECK(this->impala_server != nullptr);
}
+ /// Pointer to the Impala server of this session
+ ImpalaServer* impala_server;
+
TSessionType::type session_type;
/// Time the session was created, in ms since epoch (UTC).
@@ -421,9 +425,14 @@ class ImpalaServer : public ImpalaServiceIf,
uint32_t ref_count;
/// Per-session idle timeout in seconds. Default value is FLAGS_idle_session_timeout.
- /// It can be overridden with a smaller value via the option "idle_session_timeout"
- /// when opening a HS2 session.
- int32_t session_timeout;
+ /// It can be overridden with the query option "idle_session_timeout" when opening a
+ /// HS2 session, or using the SET command.
+ int32_t session_timeout = 0;
+
+ /// Updates the session timeout based on the query option idle_session_timeout.
+ /// It registers/unregisters the session timeout to the Impala server.
+ /// The lock must be owned by the caller of this function.
+ void UpdateTimeout();
/// Builds a Thrift representation of this SessionState for serialisation to
/// the frontend.
@@ -437,6 +446,7 @@ class ImpalaServer : public ImpalaServiceIf,
private:
friend class ChildQuery;
friend class ImpalaHttpHandler;
+ friend class SessionState;
boost::scoped_ptr<ImpalaHttpHandler> http_handler_;
@@ -766,11 +776,13 @@ class ImpalaServer : public ImpalaServiceIf,
const QueryOptionsMask& override_options_mask);
/// Register timeout value upon opening a new session. This will wake up
- /// session_timeout_thread_ to update its poll period.
+ /// session_timeout_thread_.
void RegisterSessionTimeout(int32_t timeout);
- /// To be run in a thread which wakes up every x / 2 seconds in which x is the minimum
- /// non-zero idle session timeout value of all sessions. This function checks all
+ /// Unregister timeout value.
+ void UnregisterSessionTimeout(int32_t timeout);
+
+ /// To be run in a thread which wakes up every second. This function checks all
/// sessions for their last-idle times. Those that have been idle for longer than
/// their configured timeout values are 'expired': they will no longer accept queries
/// and any running queries associated with those sessions are unregistered.
@@ -831,7 +843,7 @@ class ImpalaServer : public ImpalaServiceIf,
boost::mutex session_timeout_lock_;
/// session_timeout_thread_ relies on the following conditional variable to wake up
- /// on every poll period expiration or when the poll period changes.
+ /// when there are sessions that have a timeout.
ConditionVariable session_timeout_cv_;
/// map from query id to exec state; ClientRequestState is owned by us and referenced
http://git-wip-us.apache.org/repos/asf/impala/blob/ce65b43d/be/src/service/query-options.cc
----------------------------------------------------------------------
diff --git a/be/src/service/query-options.cc b/be/src/service/query-options.cc
index 49f1698..e8e8c7b 100644
--- a/be/src/service/query-options.cc
+++ b/be/src/service/query-options.cc
@@ -40,6 +40,8 @@ using beeswax::TQueryOptionLevel;
using namespace impala;
using namespace strings;
+DECLARE_int32(idle_session_timeout);
+
// Utility method to wrap ParseUtil::ParseMemSpec() by returning a Status instead of an
// int.
static Status ParseMemValue(const string& value, const string& key, int64_t* result) {
@@ -94,8 +96,15 @@ static void ResetQueryOption(const int option, TQueryOptions* query_options) {
}
}
+static TQueryOptions DefaultQueryOptions() {
+ TQueryOptions defaults;
+ // default value of idle_session_timeout is set by a command line flag.
+ defaults.__set_idle_session_timeout(FLAGS_idle_session_timeout);
+ return defaults;
+}
+
string impala::DebugQueryOptions(const TQueryOptions& query_options) {
- const static TQueryOptions defaults;
+ const static TQueryOptions defaults = DefaultQueryOptions();
int i = 0;
stringstream ss;
#define QUERY_OPT_FN(NAME, ENUM, LEVEL)\
@@ -569,6 +578,18 @@ Status impala::SetQueryOption(const string& key, const string& value,
query_options->__set_max_row_size(max_row_size_bytes);
break;
}
+ case TImpalaQueryOptions::IDLE_SESSION_TIMEOUT: {
+ StringParser::ParseResult result;
+ const int32_t requested_timeout =
+ StringParser::StringToInt<int32_t>(value.c_str(), value.length(), &result);
+ if (result != StringParser::PARSE_SUCCESS || requested_timeout < 0) {
+ return Status(
+ Substitute("Invalid idle session timeout: '$0'. "
+ "Only positive numbers are allowed.", value));
+ }
+ query_options->__set_idle_session_timeout(requested_timeout);
+ break;
+ }
default:
// We hit this DCHECK(false) if we forgot to add the corresponding entry here
// when we add a new query option.
http://git-wip-us.apache.org/repos/asf/impala/blob/ce65b43d/be/src/service/query-options.h
----------------------------------------------------------------------
diff --git a/be/src/service/query-options.h b/be/src/service/query-options.h
index 64e0647..b7065f4 100644
--- a/be/src/service/query-options.h
+++ b/be/src/service/query-options.h
@@ -41,7 +41,7 @@ typedef std::unordered_map<string, beeswax::TQueryOptionLevel::type>
// the DCHECK.
#define QUERY_OPTS_TABLE\
DCHECK_EQ(_TImpalaQueryOptions_VALUES_TO_NAMES.size(),\
- TImpalaQueryOptions::MAX_ROW_SIZE + 1);\
+ TImpalaQueryOptions::IDLE_SESSION_TIMEOUT + 1);\
QUERY_OPT_FN(abort_on_default_limit_exceeded, ABORT_ON_DEFAULT_LIMIT_EXCEEDED,\
TQueryOptionLevel::DEPRECATED)\
QUERY_OPT_FN(abort_on_error, ABORT_ON_ERROR, TQueryOptionLevel::REGULAR)\
@@ -129,6 +129,7 @@ typedef std::unordered_map<string, beeswax::TQueryOptionLevel::type>
QUERY_OPT_FN(min_spillable_buffer_size, MIN_SPILLABLE_BUFFER_SIZE,\
TQueryOptionLevel::ADVANCED)\
QUERY_OPT_FN(max_row_size, MAX_ROW_SIZE, TQueryOptionLevel::REGULAR)\
+ QUERY_OPT_FN(idle_session_timeout, IDLE_SESSION_TIMEOUT, TQueryOptionLevel::REGULAR)\
;
http://git-wip-us.apache.org/repos/asf/impala/blob/ce65b43d/common/thrift/ImpalaInternalService.thrift
----------------------------------------------------------------------
diff --git a/common/thrift/ImpalaInternalService.thrift b/common/thrift/ImpalaInternalService.thrift
index 1fd9c25..a584ce1 100644
--- a/common/thrift/ImpalaInternalService.thrift
+++ b/common/thrift/ImpalaInternalService.thrift
@@ -287,6 +287,11 @@ struct TQueryOptions {
// sophisticated strategies - e.g. reserving a small number of buffers large enough to
// fit maximum-sized rows.
60: optional i64 max_row_size = 524288;
+
+ // The time, in seconds, that a session may be idle for before it is closed (and all
+ // running queries cancelled) by Impala. If 0, idle sessions never expire.
+ // The default session timeout is set by the command line flag of the same name.
+ 61: optional i32 idle_session_timeout;
}
// Impala currently has two types of sessions: Beeswax and HiveServer2
http://git-wip-us.apache.org/repos/asf/impala/blob/ce65b43d/common/thrift/ImpalaService.thrift
----------------------------------------------------------------------
diff --git a/common/thrift/ImpalaService.thrift b/common/thrift/ImpalaService.thrift
index 061da00..53f6e81 100644
--- a/common/thrift/ImpalaService.thrift
+++ b/common/thrift/ImpalaService.thrift
@@ -288,6 +288,10 @@ enum TImpalaQueryOptions {
// The maximum row size that memory is reserved for, in bytes.
MAX_ROW_SIZE,
+
+ // The time, in seconds, that a session may be idle for before it is closed (and all
+ // running queries cancelled) by Impala. If 0, idle sessions never expire.
+ IDLE_SESSION_TIMEOUT,
}
// The summary of a DML statement.
http://git-wip-us.apache.org/repos/asf/impala/blob/ce65b43d/fe/src/test/java/org/apache/impala/service/JdbcTest.java
----------------------------------------------------------------------
diff --git a/fe/src/test/java/org/apache/impala/service/JdbcTest.java b/fe/src/test/java/org/apache/impala/service/JdbcTest.java
index 8ee8fba..1a907f8 100644
--- a/fe/src/test/java/org/apache/impala/service/JdbcTest.java
+++ b/fe/src/test/java/org/apache/impala/service/JdbcTest.java
@@ -30,6 +30,8 @@ import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Types;
+import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -38,11 +40,12 @@ import org.junit.After;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
-
import org.apache.impala.analysis.CreateTableStmt;
import org.apache.impala.analysis.SqlParser;
import org.apache.impala.analysis.SqlScanner;
import org.apache.impala.testutil.ImpalaJdbcClient;
+import org.apache.impala.util.Metrics;
+
import com.google.common.collect.Lists;
/**
@@ -60,13 +63,7 @@ public class JdbcTest {
@BeforeClass
public static void setUp() throws Exception {
- ImpalaJdbcClient client = ImpalaJdbcClient.createClientUsingHiveJdbcDriver();
- client.connect();
- con_ = client.getConnection();
- assertNotNull("Connection is null", con_);
- assertFalse("Connection should not be closed", con_.isClosed());
- Statement stmt = con_.createStatement();
- assertNotNull("Statement is null", stmt);
+ con_ = createConnection();
}
@AfterClass
@@ -85,6 +82,19 @@ public class JdbcTest {
expectedException);
}
+ private static Connection createConnection() throws Exception {
+ ImpalaJdbcClient client = ImpalaJdbcClient.createClientUsingHiveJdbcDriver();
+ client.connect();
+ Connection connection = client.getConnection();
+
+ assertNotNull("Connection is null", connection);
+ assertFalse("Connection should not be closed", connection.isClosed());
+ Statement stmt = connection.createStatement();
+ assertNotNull("Statement is null", stmt);
+
+ return connection;
+ }
+
protected void addTestTable(String createTableSql) throws Exception {
// Parse the stmt to extract the table name. We do this first to ensure
// that we do not execute arbitrary SQL here and pollute the test setup.
@@ -576,4 +586,85 @@ public class JdbcTest {
rs.close();
}
}
+
+ @Test
+ public void testConcurrentSessionMixedIdleTimeout() throws Exception {
+ // Test for concurrent idle sessions' expiration with mixed timeout durations.
+ Metrics metrics = new Metrics();
+
+ List<Integer> timeoutPeriods = Arrays.asList(0, 3, 15);
+ List<Connection> connections = new ArrayList<>();
+ List<Long> lastTimeSessionActive = new ArrayList<>();
+
+ for (int timeout : timeoutPeriods) {
+ connections.add(createConnection());
+ }
+
+ Long numOpenSessions = (Long)metrics.getMetric(
+ "impala-server.num-open-hiveserver2-sessions");
+ Long numExpiredSessions = (Long)metrics.getMetric(
+ "impala-server.num-sessions-expired");
+
+ for (int i = 0; i < connections.size(); ++i) {
+ Connection connection = connections.get(i);
+ Integer timeout = timeoutPeriods.get(i);
+
+ connection.createStatement().executeQuery("SELECT 1+2");
+ connection.createStatement().executeQuery("SET IDLE_SESSION_TIMEOUT=" +
+ Integer.toString(timeout));
+
+ lastTimeSessionActive.add(System.currentTimeMillis() / 1000);
+ }
+
+ assertEquals(numOpenSessions, (Long)metrics.getMetric(
+ "impala-server.num-open-hiveserver2-sessions"));
+ assertEquals(numExpiredSessions, (Long)metrics.getMetric(
+ "impala-server.num-sessions-expired"));
+
+ for (int timeout : timeoutPeriods) {
+ // Let's expire a session by sleeping,
+ // while renewing the remainders by issuing a query
+ // (except for the 0 timeout, which should never expire)
+ int timeoutToleranceMs = 1500;
+ int sleepPeriodMs = timeout * 1000 + timeoutToleranceMs;
+ Thread.sleep(sleepPeriodMs);
+
+ for (int i = 0; i < connections.size(); ++i) {
+ Connection connection = connections.get(i);
+
+ if (connection != null) {
+ Integer timeoutPeriod = timeoutPeriods.get(i);
+ long now = System.currentTimeMillis() / 1000;
+
+ boolean sessionIsValid = timeoutPeriod == 0 ||
+ timeoutPeriod > now - lastTimeSessionActive.get(i);
+
+ try (ResultSet rs = connection.createStatement().executeQuery("SELECT 1+2")) {
+ assertTrue(sessionIsValid);
+ lastTimeSessionActive.set(i, now);
+ }
+ catch (SQLException exception) {
+ assertFalse(sessionIsValid);
+ connection.close();
+ connections.set(i, null);
+ numOpenSessions -= 1;
+ numExpiredSessions += 1;
+ }
+ }
+ }
+
+ assertEquals(numOpenSessions, (Long)metrics.getMetric(
+ "impala-server.num-open-hiveserver2-sessions"));
+ assertEquals(numExpiredSessions, (Long)metrics.getMetric(
+ "impala-server.num-sessions-expired"));
+ }
+
+ assertNotNull("Connection with 0 timeout should not be null", connections.get(0));
+ connections.get(0).close();
+ connections.set(0, null);
+
+ for (Connection connection : connections) {
+ assertNull("Connection is not null", connection);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/impala/blob/ce65b43d/fe/src/test/java/org/apache/impala/util/Metrics.java
----------------------------------------------------------------------
diff --git a/fe/src/test/java/org/apache/impala/util/Metrics.java b/fe/src/test/java/org/apache/impala/util/Metrics.java
new file mode 100644
index 0000000..537f406
--- /dev/null
+++ b/fe/src/test/java/org/apache/impala/util/Metrics.java
@@ -0,0 +1,82 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.impala.util;
+
+import java.net.URL;
+import java.util.Scanner;
+
+import org.json.simple.JSONObject;
+import org.json.simple.parser.JSONParser;
+
+/**
+ * Utility class for retrieving metrics from the Impala webserver.
+ */
+public class Metrics {
+ private final static String DEFAULT_WEBSERVER_BASE_URL = "http://localhost:25000/";
+ private final static String JSON_METRICS = "jsonmetrics?json";
+
+ private String webserverBaseUrl;
+
+ public Metrics() {
+ this.webserverBaseUrl = DEFAULT_WEBSERVER_BASE_URL;
+ }
+
+ public Metrics(String webserverUrl) {
+ this.webserverBaseUrl = webserverUrl;
+ }
+
+ /**
+ * Returns the metric for the given metric id from the Impala web server.
+ * @param metricId identifier of the metric we want to retrieve
+ * @return A Java Object that holds the value for the given metric id.
+ * The caller needs to cast it to the appropriate type, e.g. Long, String, etc.
+ */
+ public Object getMetric(String metricId) throws Exception {
+ String content = readContent(new URL(this.webserverBaseUrl + JSON_METRICS));
+ if (content == null) return null;
+
+ JSONObject json = toJson(content);
+ if (json == null) return null;
+
+ return json.get(metricId);
+ }
+
+ private static String readContent(URL url) throws Exception {
+ String ret = null;
+
+ try (Scanner scanner = new Scanner(url.openStream(), "UTF-8")) {
+ scanner.useDelimiter("\\A");
+ if (scanner.hasNext()) {
+ ret = scanner.next();
+ }
+ }
+
+ return ret;
+ }
+
+ private static JSONObject toJson(String text) throws Exception {
+ JSONParser parser = new JSONParser();
+ Object obj = parser.parse(text);
+
+ if (!(obj instanceof JSONObject)) {
+ return null;
+ }
+
+ return (JSONObject)obj;
+ }
+}
http://git-wip-us.apache.org/repos/asf/impala/blob/ce65b43d/tests/custom_cluster/test_session_expiration.py
----------------------------------------------------------------------
diff --git a/tests/custom_cluster/test_session_expiration.py b/tests/custom_cluster/test_session_expiration.py
index 8d47f9c..e89fffd 100644
--- a/tests/custom_cluster/test_session_expiration.py
+++ b/tests/custom_cluster/test_session_expiration.py
@@ -42,9 +42,50 @@ class TestSessionExpiration(CustomClusterTestSuite):
sleep(3)
assert num_expired == impalad.service.get_metric_value(
"impala-server.num-sessions-expired")
- # Wait for session expiration. Session timeout was set for 6 seconds, so Impala
- # will poll the session expiry queue every 3 seconds. So, as long as the sleep in
- # ImpalaSever::ExpireSessions() is not late, the session will expire in at most 9
- # seconds. The test has already waited 3 seconds.
+ # Wait for session expiration. Impala will poll the session expiry queue every second
impalad.service.wait_for_metric_value(
"impala-server.num-sessions-expired", num_expired + 1, 20)
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args("--idle_session_timeout=3")
+ def test_session_expiration_with_set(self, vector):
+ impalad = self.cluster.get_any_impalad()
+ self.client.close()
+ num_expired = impalad.service.get_metric_value("impala-server.num-sessions-expired")
+
+ # Test if we can set a shorter timeout than the process-wide option
+ client = impalad.service.create_beeswax_client()
+ client.execute("SET IDLE_SESSION_TIMEOUT=1")
+ sleep(2.5)
+ assert num_expired + 1 == impalad.service.get_metric_value(
+ "impala-server.num-sessions-expired")
+
+ # Test if we can set a longer timeout than the process-wide option
+ client = impalad.service.create_beeswax_client()
+ client.execute("SET IDLE_SESSION_TIMEOUT=10")
+ sleep(5)
+ assert num_expired + 1 == impalad.service.get_metric_value(
+ "impala-server.num-sessions-expired")
+
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args("--idle_session_timeout=5")
+ def test_unsetting_session_expiration(self, vector):
+ impalad = self.cluster.get_any_impalad()
+ self.client.close()
+ num_expired = impalad.service.get_metric_value("impala-server.num-sessions-expired")
+
+ # Test unsetting IDLE_SESSION_TIMEOUT
+ client = impalad.service.create_beeswax_client()
+ client.execute("SET IDLE_SESSION_TIMEOUT=1")
+
+ # Unset to 5 sec
+ client.execute('SET IDLE_SESSION_TIMEOUT=""')
+ sleep(2)
+ # client session should be alive at this point
+ assert num_expired == impalad.service.get_metric_value(
+ "impala-server.num-sessions-expired")
+ sleep(5)
+ # now client should have expired
+ assert num_expired + 1 == impalad.service.get_metric_value(
+ "impala-server.num-sessions-expired")
http://git-wip-us.apache.org/repos/asf/impala/blob/ce65b43d/tests/custom_cluster/test_set_and_unset.py
----------------------------------------------------------------------
diff --git a/tests/custom_cluster/test_set_and_unset.py b/tests/custom_cluster/test_set_and_unset.py
index 8cc756c..cb596b4 100644
--- a/tests/custom_cluster/test_set_and_unset.py
+++ b/tests/custom_cluster/test_set_and_unset.py
@@ -84,6 +84,53 @@ class TestSetAndUnset(CustomClusterTestSuite, HS2TestSuite):
get_profile_req.sessionHandle = self.session_handle
assert "BATCH_SIZE=999" in self.hs2_client.GetRuntimeProfile(get_profile_req).profile
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(
+ impalad_args="--idle_session_timeout=321")
+ @needs_session(TCLIService.TProtocolVersion.HIVE_CLI_SERVICE_PROTOCOL_V6)
+ def test_set_and_unset_session_timeout(self):
+ """
+ Starts Impala cluster with idle_session_timeout configured, and checks if
+ the SET query displays the correct value.
+
+ The Beeswax API and the HiveServer2 implementations are slightly different,
+ so the same test is run in both contexts.
+ """
+ # Beeswax API:
+ # Default value
+ result = self.execute_query_expect_success(self.client, "set")
+ assert "IDLE_SESSION_TIMEOUT\t321\tREGULAR" in result.data, "baseline"
+ # Session override
+ self.execute_query_expect_success(self.client, "set idle_session_timeout=123")
+ assert "IDLE_SESSION_TIMEOUT\t123\tREGULAR" in \
+ self.execute_query_expect_success(self.client, "set").data, "session override"
+ self.execute_query_expect_success(self.client, 'set idle_session_timeout=1000')
+ assert "IDLE_SESSION_TIMEOUT\t1000\tREGULAR" in \
+ self.execute_query_expect_success(self.client, "set").data, "session override"
+ self.execute_query_expect_success(self.client, 'set idle_session_timeout=0')
+ assert "IDLE_SESSION_TIMEOUT\t0\tREGULAR" in \
+ self.execute_query_expect_success(self.client, "set").data, "session override"
+ # Unset
+ self.execute_query_expect_success(self.client, 'set idle_session_timeout=""')
+ assert "IDLE_SESSION_TIMEOUT\t321\tREGULAR" in \
+ self.execute_query_expect_success(self.client, "set").data, "reset"
+ assert "IDLE_SESSION_TIMEOUT\t321\tREGULAR" in \
+ self.execute_query_expect_success(self.client, "set").data, "after errors"
+
+ # Same dance, but with HS2:
+ # Default value
+ assert ("IDLE_SESSION_TIMEOUT", "321") in self.get_set_results(), "baseline"
+ # Session override
+ self.execute_statement("set idle_session_timeout=123")
+ assert ("IDLE_SESSION_TIMEOUT", "123") in self.get_set_results(), "session override"
+ self.execute_statement("set idle_session_timeout=1000")
+ assert ("IDLE_SESSION_TIMEOUT", "1000") in self.get_set_results(), "session override"
+ self.execute_statement("set idle_session_timeout=0")
+ assert ("IDLE_SESSION_TIMEOUT", "0") in self.get_set_results(), "session override"
+ # Unset
+ self.execute_statement("set idle_session_timeout=''")
+ assert ("IDLE_SESSION_TIMEOUT", "321") in self.get_set_results(), "reset"
+
def get_set_results(self):
"""
Executes a "SET" HiveServer2 query and returns a list
http://git-wip-us.apache.org/repos/asf/impala/blob/ce65b43d/tests/hs2/hs2_test_suite.py
----------------------------------------------------------------------
diff --git a/tests/hs2/hs2_test_suite.py b/tests/hs2/hs2_test_suite.py
index da9f68b..137bd46 100644
--- a/tests/hs2/hs2_test_suite.py
+++ b/tests/hs2/hs2_test_suite.py
@@ -228,13 +228,17 @@ class HS2TestSuite(ImpalaTestSuite):
assert False, 'Did not reach expected operation state %s in time, actual state was ' \
'%s' % (expected_state, get_operation_status_resp.operationState)
- def execute_statement(self, statement, conf_overlay=None):
- """Executes statement and returns response, which is checked."""
+ def execute_statement(self, statement, conf_overlay=None,
+ expected_status_code=TCLIService.TStatusCode.SUCCESS_STATUS,
+ expected_error_prefix=None):
+ """Executes statement and checks if the response meets the expectations.
+ If so, it returns the response."""
execute_statement_req = TCLIService.TExecuteStatementReq()
execute_statement_req.sessionHandle = self.session_handle
execute_statement_req.statement = statement
if conf_overlay:
execute_statement_req.confOverlay = conf_overlay
execute_statement_resp = self.hs2_client.ExecuteStatement(execute_statement_req)
- HS2TestSuite.check_response(execute_statement_resp)
+ HS2TestSuite.check_response(execute_statement_resp, expected_status_code,
+ expected_error_prefix)
return execute_statement_resp
http://git-wip-us.apache.org/repos/asf/impala/blob/ce65b43d/tests/hs2/test_hs2.py
----------------------------------------------------------------------
diff --git a/tests/hs2/test_hs2.py b/tests/hs2/test_hs2.py
index 5fa8fda..040d997 100644
--- a/tests/hs2/test_hs2.py
+++ b/tests/hs2/test_hs2.py
@@ -213,7 +213,7 @@ class TestHS2(HS2TestSuite):
"impala-server.num-sessions-expired")
for timeout in timeout_periods:
- sleep_period = timeout * 1.5
+ sleep_period = timeout + 1.5
time.sleep(sleep_period)
for i, session_handle in enumerate(session_handles):
if session_handle is not None: