You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by jo...@apache.org on 2018/12/07 04:30:16 UTC
[1/4] impala git commit: Print query id when timing out waiting for
state
Repository: impala
Updated Branches:
refs/heads/master 7dbeb6950 -> f266d2398
Print query id when timing out waiting for state
This makes it easier to find queries that caused failures in the logs.
Testing: I triggered this by reducing the timeout in
test_restart_services and observing that the query id was printed.
Change-Id: Ided03536b8031b8699ba65e1bf81a473b762dac7
Reviewed-on: http://gerrit.cloudera.org:8080/12039
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/2572a78b
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/2572a78b
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/2572a78b
Branch: refs/heads/master
Commit: 2572a78bdaa1d51b44ae3d90b0324dd4d54fa1cb
Parents: 7dbeb69
Author: Lars Volker <lv...@cloudera.com>
Authored: Wed Dec 5 18:41:19 2018 -0800
Committer: Impala Public Jenkins <im...@cloudera.com>
Committed: Fri Dec 7 02:26:18 2018 +0000
----------------------------------------------------------------------
tests/common/impala_service.py | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/2572a78b/tests/common/impala_service.py
----------------------------------------------------------------------
diff --git a/tests/common/impala_service.py b/tests/common/impala_service.py
index cfd4b55..5b35556 100644
--- a/tests/common/impala_service.py
+++ b/tests/common/impala_service.py
@@ -263,8 +263,8 @@ class ImpaladService(BaseImpalaService):
return
sleep(interval)
assert target_state == query_state, \
- 'Did not reach query state in time target={0} actual={1}'.format(
- target_state, query_state)
+ 'Query {0} did not reach query state in time target={1} actual={2}'.format(
+ query_handle.get_handle().id, target_state, query_state)
return
def wait_for_query_status(self, client, query_id, expected_content,
[4/4] impala git commit: IMPALA-5031: Extra large shifts are
undefined behavior
Posted by jo...@apache.org.
IMPALA-5031: Extra large shifts are undefined behavior
According to [expr.shift] in the C++14 standard, an expression in
which the shift value is "greater than or equal to the length in bits
of the promoted left operand" has undefined behavior.
Change-Id: Ie9cec6b2a887e62da594975bc971d38c1d38b0a0
Reviewed-on: http://gerrit.cloudera.org:8080/11951
Reviewed-by: Jim Apple <jb...@apache.org>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/f266d239
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/f266d239
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/f266d239
Branch: refs/heads/master
Commit: f266d2398ad507a466dd0cfe42db204ae1917f55
Parents: be88ae4
Author: Jim Apple <jb...@apache.org>
Authored: Sat Nov 17 19:26:34 2018 -0800
Committer: Impala Public Jenkins <im...@cloudera.com>
Committed: Fri Dec 7 03:55:11 2018 +0000
----------------------------------------------------------------------
be/src/util/bit-packing-test.cc | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/f266d239/be/src/util/bit-packing-test.cc
----------------------------------------------------------------------
diff --git a/be/src/util/bit-packing-test.cc b/be/src/util/bit-packing-test.cc
index f1ec10a..d65bb38 100644
--- a/be/src/util/bit-packing-test.cc
+++ b/be/src/util/bit-packing-test.cc
@@ -33,7 +33,7 @@ namespace impala {
namespace {
uint32_t ComputeMask(int bit_width) {
- return (1U << bit_width) - 1;
+ return (bit_width < 32) ? ((1U << bit_width) - 1) : ~0U;
}
}
[3/4] impala git commit: IMPALA-7821: Refactor ExprRewriteRulesTest
for easier debugging
Posted by jo...@apache.org.
IMPALA-7821: Refactor ExprRewriteRulesTest for easier debugging
Impala provides a single test class ExprRewriteRulesTest for all rewrite
rules. Each rule class has an associated test function. The
SimplifyConditionalRule class is complex, and the result test function
is quite large.
When doing work to modify a particular conditional rewrite, it became
clear that debugging would be much easier if each detailed rewrite had
its own test rather than using one huge test function. This ticket asks
to break up the big function. (JUnit does not care about small vs. large
functions.)
This patch splits the function separate from later work that will modify
the tests themselves.
Also unified test function naming, using the standard "javaStyle" rather
than the unusual "CppStyle" that was used.
Testing: this is a test, ran the test to ensure it still passes.
Change-Id: I99a28b170bc0132424041e51f61ebe5c848c4083
Reviewed-on: http://gerrit.cloudera.org:8080/11890
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Reviewed-by: Bharath Vissapragada <bh...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/be88ae4f
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/be88ae4f
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/be88ae4f
Branch: refs/heads/master
Commit: be88ae4f0b614971724dafd481ec6b7dc2334936
Parents: 938be0e
Author: Paul Rogers <pr...@cloudera.com>
Authored: Tue Nov 6 12:59:05 2018 -0800
Committer: Bharath Vissapragada <bh...@cloudera.com>
Committed: Fri Dec 7 03:53:00 2018 +0000
----------------------------------------------------------------------
.../impala/analysis/ExprRewriteRulesTest.java | 101 ++++++++++++++-----
1 file changed, 75 insertions(+), 26 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/be88ae4f/fe/src/test/java/org/apache/impala/analysis/ExprRewriteRulesTest.java
----------------------------------------------------------------------
diff --git a/fe/src/test/java/org/apache/impala/analysis/ExprRewriteRulesTest.java b/fe/src/test/java/org/apache/impala/analysis/ExprRewriteRulesTest.java
index 547e7a2..9ac444a 100644
--- a/fe/src/test/java/org/apache/impala/analysis/ExprRewriteRulesTest.java
+++ b/fe/src/test/java/org/apache/impala/analysis/ExprRewriteRulesTest.java
@@ -152,7 +152,7 @@ public class ExprRewriteRulesTest extends FrontendTestBase {
}
@Test
- public void TestBetweenToCompoundRule() throws ImpalaException {
+ public void testBetweenToCompoundRule() throws ImpalaException {
ExprRewriteRule rule = BetweenToCompoundRule.INSTANCE;
// Basic BETWEEN predicates.
@@ -191,7 +191,7 @@ public class ExprRewriteRulesTest extends FrontendTestBase {
}
@Test
- public void TestExtractCommonConjunctsRule() throws ImpalaException {
+ public void testExtractCommonConjunctsRule() throws ImpalaException {
ExprRewriteRule rule = ExtractCommonConjunctRule.INSTANCE;
// One common conjunct: int_col < 10
@@ -286,7 +286,7 @@ public class ExprRewriteRulesTest extends FrontendTestBase {
* testing is done in expr-test.cc.
*/
@Test
- public void TestFoldConstantsRule() throws ImpalaException {
+ public void testFoldConstantsRule() throws ImpalaException {
ExprRewriteRule rule = FoldConstantsRule.INSTANCE;
RewritesOk("1 + 1", rule, "2");
@@ -311,19 +311,26 @@ public class ExprRewriteRulesTest extends FrontendTestBase {
RewritesOk("rand()", rule, null);
RewritesOk("random()", rule, null);
RewritesOk("uuid()", rule, null);
+
+ RewritesOk("null + 1", rule, "NULL");
+ RewritesOk("(1 + 1) is null", rule, "FALSE");
+ RewritesOk("(null + 1) is null", rule, "TRUE");
}
@Test
- public void TestSimplifyConditionalsRule() throws ImpalaException {
+ public void testIf() throws ImpalaException {
ExprRewriteRule rule = SimplifyConditionalsRule.INSTANCE;
- // IF
RewritesOk("if(true, id, id+1)", rule, "id");
RewritesOk("if(false, id, id+1)", rule, "id + 1");
RewritesOk("if(null, id, id+1)", rule, "id + 1");
RewritesOk("if(id = 0, true, false)", rule, null);
+ }
+
+ @Test
+ public void testIfNull() throws ImpalaException {
+ ExprRewriteRule rule = SimplifyConditionalsRule.INSTANCE;
- // IFNULL and its aliases
for (String f : ImmutableList.of("ifnull", "isnull", "nvl")) {
RewritesOk(f + "(null, id)", rule, "id");
RewritesOk(f + "(null, null)", rule, "NULL");
@@ -331,24 +338,31 @@ public class ExprRewriteRulesTest extends FrontendTestBase {
RewritesOk(f + "(1, 2)", rule, "1");
RewritesOk(f + "(0, id)", rule, "0");
- // non literal constants shouldn't be simplified by the rule
- RewritesOk(f + "(1 + 1, id)", rule, null);
- RewritesOk(f + "(NULL + 1, id)", rule, null);
+
+ // TODO: IMPALA-7769
+ //RewritesOk(f + "(1 + 1, id)", rule, "2");
+ //RewritesOk(f + "(NULL + 1, id)", rule, "id");
+ //RewritesOk(f + "(cast(null as int), id)", rule, "id");
}
+ }
+
+ @Test
+ public void testCompoundPredicate() throws ImpalaException {
+ ExprRewriteRule rule = SimplifyConditionalsRule.INSTANCE;
- // CompoundPredicate
RewritesOk("false || id = 0", rule, "id = 0");
RewritesOk("true || id = 0", rule, "TRUE");
RewritesOk("false && id = 0", rule, "FALSE");
RewritesOk("true && id = 0", rule, "id = 0");
- // NULL with a non-constant other child doesn't get rewritten.
- RewritesOk("null && id = 0", rule, null);
- RewritesOk("null || id = 0", rule, null);
+ }
+ @Test
+ public void testCaseWithExpr() throws ImpalaException {
+ ExprRewriteRule rule = SimplifyConditionalsRule.INSTANCE;
List<ExprRewriteRule> rules = Lists.newArrayList();
rules.add(FoldConstantsRule.INSTANCE);
rules.add(rule);
- // CASE with caseExpr
+
// Single TRUE case with no preceding non-constant cases.
RewritesOk("case 1 when 0 then id when 1 then id + 1 when 2 then id + 2 end", rule,
"id + 1");
@@ -369,8 +383,15 @@ public class ExprRewriteRulesTest extends FrontendTestBase {
RewritesOk("case 0 when null then id else 1 end", rule, "1");
// All non-constant, don't rewrite.
RewritesOk("case id when 1 then 1 when 2 then 2 else 3 end", rule, null);
+ }
+
+ @Test
+ public void testCaseWithoutExpr() throws ImpalaException {
+ ExprRewriteRule rule = SimplifyConditionalsRule.INSTANCE;
+ List<ExprRewriteRule> rules = Lists.newArrayList();
+ rules.add(FoldConstantsRule.INSTANCE);
+ rules.add(rule);
- // CASE without caseExpr
// Single TRUE case with no predecing non-constant case.
RewritesOk("case when FALSE then 0 when TRUE then 1 end", rule, "1");
// Single TRUE case with preceding non-constant case.
@@ -392,8 +413,18 @@ public class ExprRewriteRulesTest extends FrontendTestBase {
"CASE WHEN id = 0 THEN 0 ELSE 2 END");
// All non-constant, don't rewrite.
RewritesOk("case when id = 0 then 0 when id = 1 then 1 end", rule, null);
+ RewritesOk("case when id = 1 then 10 when false then 20 " +
+ "when true then 30 else 40 end", rule,
+ "CASE WHEN id = 1 THEN 10 ELSE 30 END");
+ }
+
+ @Test
+ public void testDecode() throws ImpalaException {
+ ExprRewriteRule rule = SimplifyConditionalsRule.INSTANCE;
+ List<ExprRewriteRule> rules = Lists.newArrayList();
+ rules.add(FoldConstantsRule.INSTANCE);
+ rules.add(rule);
- // DECODE
// Single TRUE case with no preceding non-constant case.
RewritesOk("decode(1, 0, id, 1, id + 1, 2, id + 2)", rules, "id + 1");
// Single TRUE case with predecing non-constant case.
@@ -414,9 +445,16 @@ public class ExprRewriteRulesTest extends FrontendTestBase {
RewritesOk("decode(id, null, 0, 1)", rules, null);
// All non-constant, don't rewrite.
RewritesOk("decode(id, 1, 1, 2, 2)", rules, null);
+ }
+
+ /**
+ * IMPALA-5125: Exprs containing aggregates should not be rewritten if the rewrite
+ * eliminates all aggregates.
+ */
+ @Test
+ public void testExcludeAggregates() throws ImpalaException {
+ ExprRewriteRule rule = SimplifyConditionalsRule.INSTANCE;
- // IMPALA-5125: Exprs containing aggregates should not be rewritten if the rewrite
- // eliminates all aggregates.
RewritesOk("if(true, 0, sum(id))", rule, null);
RewritesOk("if(false, max(id), min(id))", rule, "min(id)");
RewritesOk("true || sum(id) = 0", rule, null);
@@ -424,7 +462,18 @@ public class ExprRewriteRulesTest extends FrontendTestBase {
RewritesOk("ifnull(1, max(id))", rule, null);
RewritesOk("case when true then 0 when false then sum(id) end", rule, null);
RewritesOk(
- "case when true then count(id) when false then sum(id) end", rule, "count(id)");
+ "case when true then count(id) when false then sum(id) end",
+ rule, "count(id)");
+ RewritesOk("sum(id) is distinct from null", rule, null);
+ RewritesOk("sum(id) is distinct from sum(id)", rule, null);
+ }
+
+ @Test
+ public void testCoalesce() throws ImpalaException {
+ ExprRewriteRule rule = SimplifyConditionalsRule.INSTANCE;
+ List<ExprRewriteRule> rules = Lists.newArrayList();
+ rules.add(FoldConstantsRule.INSTANCE);
+ rules.add(rule);
// IMPALA-5016: Simplify COALESCE function
// Test skipping leading nulls.
@@ -453,7 +502,7 @@ public class ExprRewriteRulesTest extends FrontendTestBase {
}
@Test
- public void TestNormalizeExprsRule() throws ImpalaException {
+ public void testNormalizeExprsRule() throws ImpalaException {
ExprRewriteRule rule = NormalizeExprsRule.INSTANCE;
// CompoundPredicate
@@ -466,7 +515,7 @@ public class ExprRewriteRulesTest extends FrontendTestBase {
}
@Test
- public void TestNormalizeBinaryPredicatesRule() throws ImpalaException {
+ public void testNormalizeBinaryPredicatesRule() throws ImpalaException {
ExprRewriteRule rule = NormalizeBinaryPredicatesRule.INSTANCE;
RewritesOk("0 = id", rule, "id = 0");
@@ -488,7 +537,7 @@ public class ExprRewriteRulesTest extends FrontendTestBase {
}
@Test
- public void TestEqualityDisjunctsToInRule() throws ImpalaException {
+ public void testEqualityDisjunctsToInRule() throws ImpalaException {
ExprRewriteRule edToInrule = EqualityDisjunctsToInRule.INSTANCE;
ExprRewriteRule normalizeRule = NormalizeBinaryPredicatesRule.INSTANCE;
List<ExprRewriteRule> comboRules = Lists.newArrayList(normalizeRule,
@@ -551,7 +600,7 @@ public class ExprRewriteRulesTest extends FrontendTestBase {
}
@Test
- public void TestNormalizeCountStarRule() throws ImpalaException {
+ public void testNormalizeCountStarRule() throws ImpalaException {
ExprRewriteRule rule = NormalizeCountStarRule.INSTANCE;
RewritesOk("count(1)", rule, "count(*)");
@@ -565,7 +614,7 @@ public class ExprRewriteRulesTest extends FrontendTestBase {
}
@Test
- public void TestSimplifyDistinctFromRule() throws ImpalaException {
+ public void testSimplifyDistinctFromRule() throws ImpalaException {
ExprRewriteRule rule = SimplifyDistinctFromRule.INSTANCE;
// Can be simplified
@@ -588,7 +637,7 @@ public class ExprRewriteRulesTest extends FrontendTestBase {
}
@Test
- public void TestRemoveRedundantStringCastRule() throws ImpalaException {
+ public void testRemoveRedundantStringCastRule() throws ImpalaException {
ExprRewriteRule removeRule = RemoveRedundantStringCast.INSTANCE;
ExprRewriteRule foldConstantRule = FoldConstantsRule.INSTANCE;
List<ExprRewriteRule> comboRules = Lists.newArrayList(removeRule, foldConstantRule);
@@ -682,7 +731,7 @@ public class ExprRewriteRulesTest extends FrontendTestBase {
* it can be further simplified via SimplifyDistinctFromRule.
*/
@Test
- public void TestNullif() throws ImpalaException {
+ public void testNullif() throws ImpalaException {
List<ExprRewriteRule> rules = Lists.newArrayList(
SimplifyConditionalsRule.INSTANCE,
SimplifyDistinctFromRule.INSTANCE);
[2/4] impala git commit: IMPALA-7738: Implement timeouts for HDFS
open calls
Posted by jo...@apache.org.
IMPALA-7738: Implement timeouts for HDFS open calls
This is part 1 of a push to add timeouts for all HDFS operations.
It adds timeouts for opening an HDFS file handle.
It introduces a new SynchronousThreadPool, which executes
an operation in a thread pool and waits up to a specified
timeout for the operation to complete. This type of thread
pool can accept any subclass of SynchronousWorkItem, and
a single thread pool can process different types of work
items. It is tested by a new test case in thread-pool-test.
This also introduces a new HdfsMonitor which implements
timeouts for HDFS operations, currently limited to
hdfsOpenFile(). This is implemented using a SynchronousThreadPool.
The timeout for hdfs operations is specified by
hdfs_operation_timeout_sec, which defaults to 5 minutes.
Testing:
1. Added a test to thread-pool-test for the new
SynchronousThreadPool.
2. Core tests
3. Added a custom cluster test that does "kill -STOP"
for the NameNode and verifies that a subsequent
hdfsOpenFile operation times out.
Change-Id: Ia14403ca5f3f19c6d5f61b9ab2306b0ad3267454
Reviewed-on: http://gerrit.cloudera.org:8080/11874
Reviewed-by: Joe McDonnell <jo...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/938be0e8
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/938be0e8
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/938be0e8
Branch: refs/heads/master
Commit: 938be0e840c84263a2b47fb89e655d998363b819
Parents: 2572a78
Author: Joe McDonnell <jo...@cloudera.com>
Authored: Fri Nov 2 18:03:34 2018 -0700
Committer: Joe McDonnell <jo...@cloudera.com>
Committed: Fri Dec 7 03:50:11 2018 +0000
----------------------------------------------------------------------
be/src/common/status.h | 6 ++
be/src/exec/base-sequence-scanner.cc | 3 +-
be/src/runtime/io/CMakeLists.txt | 1 +
be/src/runtime/io/disk-io-mgr.cc | 48 +++++-----
be/src/runtime/io/disk-io-mgr.h | 31 +++---
be/src/runtime/io/handle-cache.h | 32 ++++---
be/src/runtime/io/handle-cache.inline.h | 45 +++++----
be/src/runtime/io/hdfs-file-reader.cc | 25 ++---
be/src/runtime/io/hdfs-file-reader.h | 2 +-
be/src/runtime/io/hdfs-monitored-ops.cc | 99 +++++++++++++++++++
be/src/runtime/io/hdfs-monitored-ops.h | 62 ++++++++++++
be/src/runtime/runtime-state.cc | 4 +-
be/src/util/thread-pool-test.cc | 87 +++++++++++++++++
be/src/util/thread-pool.h | 127 +++++++++++++++++++++++++
common/thrift/generate_error_codes.py | 6 ++
tests/custom_cluster/test_hdfs_timeout.py | 90 ++++++++++++++++++
16 files changed, 582 insertions(+), 86 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/be/src/common/status.h
----------------------------------------------------------------------
diff --git a/be/src/common/status.h b/be/src/common/status.h
index d60efa7..6c14ad4 100644
--- a/be/src/common/status.h
+++ b/be/src/common/status.h
@@ -227,6 +227,12 @@ class NODISCARD Status {
&& msg_->error() == TErrorCode::DISK_IO_ERROR;
}
+ bool IsThreadPoolError() const {
+ return msg_ != NULL
+ && (msg_->error() == TErrorCode::THREAD_POOL_SUBMIT_FAILED ||
+ msg_->error() == TErrorCode::THREAD_POOL_TASK_TIMED_OUT);
+ }
+
/// Returns the error message associated with a non-successful status.
const ErrorMsg& msg() const {
DCHECK(msg_ != NULL);
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/be/src/exec/base-sequence-scanner.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/base-sequence-scanner.cc b/be/src/exec/base-sequence-scanner.cc
index 6757ec3..b06602a 100644
--- a/be/src/exec/base-sequence-scanner.cc
+++ b/be/src/exec/base-sequence-scanner.cc
@@ -185,7 +185,8 @@ Status BaseSequenceScanner::GetNextInternal(RowBatch* row_batch) {
if (!status.IsCancelled() &&
!status.IsMemLimitExceeded() &&
!status.IsInternalError() &&
- !status.IsDiskIoError()) {
+ !status.IsDiskIoError() &&
+ !status.IsThreadPoolError()) {
state_->LogError(ErrorMsg(TErrorCode::SEQUENCE_SCANNER_PARSE_ERROR,
stream_->filename(), stream_->file_offset(),
(stream_->eof() ? "(EOF)" : "")));
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/be/src/runtime/io/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/CMakeLists.txt b/be/src/runtime/io/CMakeLists.txt
index 675bd6e..2215678 100644
--- a/be/src/runtime/io/CMakeLists.txt
+++ b/be/src/runtime/io/CMakeLists.txt
@@ -31,6 +31,7 @@ add_library(Io
scan-range.cc
hdfs-file-reader.cc
local-file-reader.cc
+ hdfs-monitored-ops.cc
)
add_dependencies(Io gen-deps)
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/be/src/runtime/io/disk-io-mgr.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/disk-io-mgr.cc b/be/src/runtime/io/disk-io-mgr.cc
index fb0c876..ce6fd45 100644
--- a/be/src/runtime/io/disk-io-mgr.cc
+++ b/be/src/runtime/io/disk-io-mgr.cc
@@ -183,7 +183,7 @@ DiskIoMgr::DiskIoMgr() :
file_handle_cache_(min(FLAGS_max_cached_file_handles,
FileSystemUtil::MaxNumFileHandles()),
FLAGS_num_file_handle_cache_partitions,
- FLAGS_unused_file_handle_timeout_sec) {
+ FLAGS_unused_file_handle_timeout_sec, &hdfs_monitor_) {
DCHECK_LE(READ_SIZE_MIN_VALUE, FLAGS_read_size);
int num_local_disks = DiskInfo::num_disks();
if (FLAGS_num_disks < 0 || FLAGS_num_disks > DiskInfo::num_disks()) {
@@ -207,7 +207,7 @@ DiskIoMgr::DiskIoMgr(int num_local_disks, int threads_per_rotational_disk,
file_handle_cache_(min(FLAGS_max_cached_file_handles,
FileSystemUtil::MaxNumFileHandles()),
FLAGS_num_file_handle_cache_partitions,
- FLAGS_unused_file_handle_timeout_sec) {
+ FLAGS_unused_file_handle_timeout_sec, &hdfs_monitor_) {
if (num_local_disks == 0) num_local_disks = DiskInfo::num_disks();
disk_queues_.resize(num_local_disks + REMOTE_NUM_DISKS);
CheckSseSupport();
@@ -259,6 +259,9 @@ Status DiskIoMgr::Init() {
disk_thread_group_.AddThread(move(t));
}
}
+ // The file handle cache depends on the HDFS monitor, so initialize it first.
+ // Use the same number of threads for the HDFS monitor as there are Disk IO threads.
+ RETURN_IF_ERROR(hdfs_monitor_.Init(disk_thread_group_.Size()));
RETURN_IF_ERROR(file_handle_cache_.Init());
cached_read_options_ = hadoopRzOptionsAlloc();
@@ -478,36 +481,35 @@ int DiskIoMgr::AssignQueue(const char* file, int disk_id, bool expected_local) {
return disk_id % num_local_disks();
}
-ExclusiveHdfsFileHandle* DiskIoMgr::GetExclusiveHdfsFileHandle(const hdfsFS& fs,
- std::string* fname, int64_t mtime, RequestContext *reader) {
+Status DiskIoMgr::GetExclusiveHdfsFileHandle(const hdfsFS& fs,
+ std::string* fname, int64_t mtime, RequestContext *reader,
+ unique_ptr<ExclusiveHdfsFileHandle>& fid_out) {
SCOPED_TIMER(reader->open_file_timer_);
- ExclusiveHdfsFileHandle* fid = new ExclusiveHdfsFileHandle(fs, fname->data(), mtime);
- if (!fid->ok()) {
- VLOG_FILE << "Opening the file " << fname << " failed.";
- delete fid;
- return nullptr;
- }
+ unique_ptr<ExclusiveHdfsFileHandle> fid;
+ fid.reset(new ExclusiveHdfsFileHandle(fs, fname, mtime));
+ RETURN_IF_ERROR(fid->Init(&hdfs_monitor_));
+ fid_out.swap(fid);
ImpaladMetrics::IO_MGR_NUM_FILE_HANDLES_OUTSTANDING->Increment(1L);
// Every exclusive file handle is considered a cache miss
ImpaladMetrics::IO_MGR_CACHED_FILE_HANDLES_HIT_RATIO->Update(0L);
ImpaladMetrics::IO_MGR_CACHED_FILE_HANDLES_MISS_COUNT->Increment(1L);
reader->cached_file_handles_miss_count_.Add(1L);
- return fid;
+ return Status::OK();
}
-void DiskIoMgr::ReleaseExclusiveHdfsFileHandle(ExclusiveHdfsFileHandle* fid) {
+void DiskIoMgr::ReleaseExclusiveHdfsFileHandle(unique_ptr<ExclusiveHdfsFileHandle> fid) {
DCHECK(fid != nullptr);
ImpaladMetrics::IO_MGR_NUM_FILE_HANDLES_OUTSTANDING->Increment(-1L);
- delete fid;
+ fid.reset();
}
-CachedHdfsFileHandle* DiskIoMgr::GetCachedHdfsFileHandle(const hdfsFS& fs,
- std::string* fname, int64_t mtime, RequestContext *reader) {
+Status DiskIoMgr::GetCachedHdfsFileHandle(const hdfsFS& fs,
+ std::string* fname, int64_t mtime, RequestContext *reader,
+ CachedHdfsFileHandle** handle_out) {
bool cache_hit;
SCOPED_TIMER(reader->open_file_timer_);
- CachedHdfsFileHandle* fh = file_handle_cache_.GetFileHandle(fs, fname, mtime, false,
- &cache_hit);
- if (fh == nullptr) return nullptr;
+ RETURN_IF_ERROR(file_handle_cache_.GetFileHandle(fs, fname, mtime, false, handle_out,
+ &cache_hit));
ImpaladMetrics::IO_MGR_NUM_FILE_HANDLES_OUTSTANDING->Increment(1L);
if (cache_hit) {
ImpaladMetrics::IO_MGR_CACHED_FILE_HANDLES_HIT_RATIO->Update(1L);
@@ -518,7 +520,7 @@ CachedHdfsFileHandle* DiskIoMgr::GetCachedHdfsFileHandle(const hdfsFS& fs,
ImpaladMetrics::IO_MGR_CACHED_FILE_HANDLES_MISS_COUNT->Increment(1L);
reader->cached_file_handles_miss_count_.Add(1L);
}
- return fh;
+ return Status::OK();
}
void DiskIoMgr::ReleaseCachedHdfsFileHandle(std::string* fname,
@@ -534,12 +536,12 @@ Status DiskIoMgr::ReopenCachedHdfsFileHandle(const hdfsFS& fs, std::string* fnam
ImpaladMetrics::IO_MGR_CACHED_FILE_HANDLES_REOPENED->Increment(1L);
file_handle_cache_.ReleaseFileHandle(fname, *fid, true);
// The old handle has been destroyed, so *fid must be overwritten before returning.
- *fid = file_handle_cache_.GetFileHandle(fs, fname, mtime, true,
+ *fid = nullptr;
+ Status status = file_handle_cache_.GetFileHandle(fs, fname, mtime, true, fid,
&cache_hit);
- if (*fid == nullptr) {
+ if (!status.ok()) {
ImpaladMetrics::IO_MGR_NUM_FILE_HANDLES_OUTSTANDING->Increment(-1L);
- return Status(TErrorCode::DISK_IO_ERROR,
- GetHdfsErrorMsg("Failed to open HDFS file ", fname->data()));
+ return status;
}
DCHECK(!cache_hit);
return Status::OK();
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/be/src/runtime/io/disk-io-mgr.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/disk-io-mgr.h b/be/src/runtime/io/disk-io-mgr.h
index d36906d..ac93c75 100644
--- a/be/src/runtime/io/disk-io-mgr.h
+++ b/be/src/runtime/io/disk-io-mgr.h
@@ -27,6 +27,7 @@
#include "common/status.h"
#include "runtime/bufferpool/buffer-pool.h"
#include "runtime/io/handle-cache.h"
+#include "runtime/io/hdfs-monitored-ops.h"
#include "runtime/io/local-file-system.h"
#include "runtime/io/request-ranges.h"
#include "util/aligned-new.h"
@@ -300,21 +301,24 @@ class DiskIoMgr : public CacheLineAligned {
bool Validate() const;
/// Given a FS handle, name and last modified time of the file, construct a new
- /// ExclusiveHdfsFileHandle. This records the time spent opening the handle in
- /// 'reader' and counts this as a cache miss. In the case of an error, returns nullptr.
- ExclusiveHdfsFileHandle* GetExclusiveHdfsFileHandle(const hdfsFS& fs,
- std::string* fname, int64_t mtime, RequestContext* reader);
+ /// ExclusiveHdfsFileHandle and return it via 'fid'. This records the time spent
+ /// opening the handle in 'reader' and counts this as a cache miss. In case of an
+ /// error, returns status and 'fid' is untouched.
+ Status GetExclusiveHdfsFileHandle(const hdfsFS& fs,
+ std::string* fname, int64_t mtime, RequestContext* reader,
+ std::unique_ptr<ExclusiveHdfsFileHandle>& fid) WARN_UNUSED_RESULT;
/// Releases an exclusive file handle, destroying it
- void ReleaseExclusiveHdfsFileHandle(ExclusiveHdfsFileHandle* fid);
+ void ReleaseExclusiveHdfsFileHandle(std::unique_ptr<ExclusiveHdfsFileHandle> fid);
/// Given a FS handle, name and last modified time of the file, gets a
- /// CachedHdfsFileHandle from the file handle cache. Records the time spent
- /// opening the handle in 'reader'. On success, records statistics
- /// about whether this was a cache hit or miss in the 'reader' as well as at the
- /// system level. In case of an error returns nullptr.
- CachedHdfsFileHandle* GetCachedHdfsFileHandle(const hdfsFS& fs,
- std::string* fname, int64_t mtime, RequestContext* reader);
+ /// CachedHdfsFileHandle from the file handle cache and returns it via 'fid'.
+ /// Records the time spent opening the handle in 'reader'. On success, records
+ /// statistics about whether this was a cache hit or miss in the 'reader' as well as
+ /// at the system level. In case of an error, returns status and 'fid' is untouched.
+ Status GetCachedHdfsFileHandle(const hdfsFS& fs,
+ std::string* fname, int64_t mtime, RequestContext* reader,
+ CachedHdfsFileHandle** fid) WARN_UNUSED_RESULT;
/// Releases a file handle back to the file handle cache when it is no longer in use.
void ReleaseCachedHdfsFileHandle(std::string* fname, CachedHdfsFileHandle* fid);
@@ -323,7 +327,7 @@ class DiskIoMgr : public CacheLineAligned {
/// file handle from the cache. Records the time spent reopening the handle
/// in 'reader'. Returns an error if the file could not be reopened.
Status ReopenCachedHdfsFileHandle(const hdfsFS& fs, std::string* fname, int64_t mtime,
- RequestContext* reader, CachedHdfsFileHandle** fid);
+ RequestContext* reader, CachedHdfsFileHandle** fid) WARN_UNUSED_RESULT;
// Function to change the underlying LocalFileSystem object used for disk I/O.
// DiskIoMgr will also take responsibility of the received LocalFileSystem pointer.
@@ -408,6 +412,9 @@ class DiskIoMgr : public CacheLineAligned {
/// round-robin assignment for that case.
static AtomicInt32 next_disk_id_;
+ /// Thread pool used to implement timeouts for HDFS operations.
+ HdfsMonitor hdfs_monitor_;
+
// Caching structure that maps file names to cached file handles. The cache has an upper
// limit of entries defined by FLAGS_max_cached_file_handles. Evicted cached file
// handles are closed.
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/be/src/runtime/io/handle-cache.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/handle-cache.h b/be/src/runtime/io/handle-cache.h
index 232b630..d4cff37 100644
--- a/be/src/runtime/io/handle-cache.h
+++ b/be/src/runtime/io/handle-cache.h
@@ -35,6 +35,8 @@
namespace impala {
namespace io {
+class HdfsMonitor;
+
/// This abstract class is a small wrapper around the hdfsFile handle and the file system
/// instance which is needed to close the file handle. The handle incorporates
/// the last modified time of the file when it was opened. This is used to distinguish
@@ -47,17 +49,21 @@ class HdfsFileHandle {
/// Destructor will close the file handle
~HdfsFileHandle();
+ /// Init opens the file handle
+ Status Init(HdfsMonitor* monitor);
+
hdfsFile file() const { return hdfs_file_; }
int64_t mtime() const { return mtime_; }
- bool ok() const { return hdfs_file_ != nullptr; }
protected:
- /// Constructor will open the file
- HdfsFileHandle(const hdfsFS& fs, const char* fname, int64_t mtime);
+ HdfsFileHandle(const hdfsFS& fs, const std::string* fname, int64_t mtime)
+ : fs_(fs), fname_(fname), mtime_(mtime) {}
private:
hdfsFS fs_;
- hdfsFile hdfs_file_;
+ // fname_ has a limited lifetime. It is only valid from construction until Init().
+ const std::string* fname_;
+ hdfsFile hdfs_file_ = nullptr;
int64_t mtime_;
};
@@ -65,7 +71,7 @@ class HdfsFileHandle {
/// other purpose.
class CachedHdfsFileHandle : public HdfsFileHandle {
public:
- CachedHdfsFileHandle(const hdfsFS& fs, const char* fname, int64_t mtime);
+ CachedHdfsFileHandle(const hdfsFS& fs, const std::string* fname, int64_t mtime);
~CachedHdfsFileHandle();
};
@@ -73,7 +79,7 @@ class CachedHdfsFileHandle : public HdfsFileHandle {
/// is not appropriate.
class ExclusiveHdfsFileHandle : public HdfsFileHandle {
public:
- ExclusiveHdfsFileHandle(const hdfsFS& fs, const char* fname, int64_t mtime)
+ ExclusiveHdfsFileHandle(const hdfsFS& fs, const std::string* fname, int64_t mtime)
: HdfsFileHandle(fs, fname, mtime) {}
};
@@ -110,7 +116,7 @@ class FileHandleCache {
/// up. The cache will age out any file handle that is unused for
/// `unused_handle_timeout_secs` seconds. Age out is disabled if this is set to zero.
FileHandleCache(size_t capacity, size_t num_partitions,
- uint64_t unused_handle_timeout_secs);
+ uint64_t unused_handle_timeout_secs, HdfsMonitor* hdfs_monitor);
/// Destructor is only called for backend tests
~FileHandleCache();
@@ -132,8 +138,9 @@ class FileHandleCache {
///
/// This obtains exclusive control over the returned file handle. It must be paired
/// with a call to ReleaseFileHandle to release exclusive control.
- CachedHdfsFileHandle* GetFileHandle(const hdfsFS& fs, std::string* fname,
- int64_t mtime, bool require_new_handle, bool* cache_hit);
+ Status GetFileHandle(const hdfsFS& fs, std::string* fname,
+ int64_t mtime, bool require_new_handle, CachedHdfsFileHandle** handle_out,
+ bool* cache_hit) WARN_UNUSED_RESULT;
/// Release the exclusive hold on the specified file handle (which was obtained
/// by calling GetFileHandle). The cache may evict a file handle if the cache is
@@ -154,8 +161,8 @@ class FileHandleCache {
typedef std::list<LruListEntry> LruListType;
struct FileHandleEntry {
- FileHandleEntry(CachedHdfsFileHandle* fh_in, LruListType& lru_list)
- : fh(fh_in), lru_entry(lru_list.end()) {}
+ FileHandleEntry(std::unique_ptr<CachedHdfsFileHandle> fh_in, LruListType& lru_list)
+ : fh(std::move(fh_in)), lru_entry(lru_list.end()) {}
std::unique_ptr<CachedHdfsFileHandle> fh;
/// in_use is true for a file handle checked out via GetFileHandle() that has not
@@ -210,6 +217,9 @@ class FileHandleCache {
/// the shut_down_promise_ is set.
std::unique_ptr<Thread> eviction_thread_;
Promise<bool> shut_down_promise_;
+
+ /// Thread pool used to implement timeouts for HDFS operations
+ HdfsMonitor* hdfs_monitor_;
};
}
}
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/be/src/runtime/io/handle-cache.inline.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/handle-cache.inline.h b/be/src/runtime/io/handle-cache.inline.h
index 99b106c..8db7730 100644
--- a/be/src/runtime/io/handle-cache.inline.h
+++ b/be/src/runtime/io/handle-cache.inline.h
@@ -18,6 +18,7 @@
#include <tuple>
#include "runtime/io/handle-cache.h"
+#include "runtime/io/hdfs-monitored-ops.h"
#include "util/hash-util.h"
#include "util/time.h"
@@ -27,22 +28,25 @@
namespace impala {
namespace io {
-HdfsFileHandle::HdfsFileHandle(const hdfsFS& fs, const char* fname,
- int64_t mtime)
- : fs_(fs), hdfs_file_(hdfsOpenFile(fs, fname, O_RDONLY, 0, 0, 0)), mtime_(mtime) {
- VLOG_FILE << "hdfsOpenFile() file=" << fname << " fid=" << hdfs_file_;
-}
-
HdfsFileHandle::~HdfsFileHandle() {
if (hdfs_file_ != nullptr && fs_ != nullptr) {
VLOG_FILE << "hdfsCloseFile() fid=" << hdfs_file_;
- hdfsCloseFile(fs_, hdfs_file_);
+ hdfsCloseFile(fs_, hdfs_file_); // TODO: check return code
}
fs_ = nullptr;
+ fname_ = nullptr;
hdfs_file_ = nullptr;
}
-CachedHdfsFileHandle::CachedHdfsFileHandle(const hdfsFS& fs, const char* fname,
+Status HdfsFileHandle::Init(HdfsMonitor* monitor) {
+ Status status = monitor->OpenHdfsFileWithTimeout(fs_, fname_, O_RDONLY, 0,
+ &hdfs_file_);
+ // fname_ is no longer needed, null it out
+ fname_ = nullptr;
+ return status;
+}
+
+CachedHdfsFileHandle::CachedHdfsFileHandle(const hdfsFS& fs, const string* fname,
int64_t mtime)
: HdfsFileHandle(fs, fname, mtime) {
ImpaladMetrics::IO_MGR_NUM_CACHED_FILE_HANDLES->Increment(1L);
@@ -53,9 +57,10 @@ CachedHdfsFileHandle::~CachedHdfsFileHandle() {
}
FileHandleCache::FileHandleCache(size_t capacity,
- size_t num_partitions, uint64_t unused_handle_timeout_secs)
+ size_t num_partitions, uint64_t unused_handle_timeout_secs, HdfsMonitor* hdfs_monitor)
: cache_partitions_(num_partitions),
- unused_handle_timeout_secs_(unused_handle_timeout_secs) {
+ unused_handle_timeout_secs_(unused_handle_timeout_secs),
+ hdfs_monitor_(hdfs_monitor) {
DCHECK_GT(num_partitions, 0);
size_t remainder = capacity % num_partitions;
size_t base_capacity = capacity / num_partitions;
@@ -80,9 +85,9 @@ Status FileHandleCache::Init() {
&FileHandleCache::EvictHandlesLoop, this, &eviction_thread_);
}
-CachedHdfsFileHandle* FileHandleCache::GetFileHandle(
+Status FileHandleCache::GetFileHandle(
const hdfsFS& fs, std::string* fname, int64_t mtime, bool require_new_handle,
- bool* cache_hit) {
+ CachedHdfsFileHandle** handle_out, bool* cache_hit) {
// Hash the key and get appropriate partition
int index = HashUtil::Hash(fname->data(), fname->size(), 0) % cache_partitions_.size();
FileHandleCachePartition& p = cache_partitions_[index];
@@ -112,7 +117,8 @@ CachedHdfsFileHandle* FileHandleCache::GetFileHandle(
elem->lru_entry = p.lru_list.end();
*cache_hit = true;
elem->in_use = true;
- return elem->fh.get();
+ *handle_out = elem->fh.get();
+ return Status::OK();
}
++range.first;
}
@@ -123,11 +129,9 @@ CachedHdfsFileHandle* FileHandleCache::GetFileHandle(
// the file handle without holding the lock to reduce contention.
*cache_hit = false;
// Create a new file handle
- CachedHdfsFileHandle* new_fh = new CachedHdfsFileHandle(fs, fname->data(), mtime);
- if (!new_fh->ok()) {
- delete new_fh;
- return nullptr;
- }
+ std::unique_ptr<CachedHdfsFileHandle> new_fh;
+ new_fh.reset(new CachedHdfsFileHandle(fs, fname, mtime));
+ RETURN_IF_ERROR(new_fh->Init(hdfs_monitor_));
// Get the lock and create/move the new entry into the map
// This entry is new and will be immediately used. Place it as the first entry
@@ -137,7 +141,7 @@ CachedHdfsFileHandle* FileHandleCache::GetFileHandle(
boost::lock_guard<SpinLock> g(p.lock);
pair<typename MapType::iterator, typename MapType::iterator> range =
p.cache.equal_range(*fname);
- FileHandleEntry entry(new_fh, p.lru_list);
+ FileHandleEntry entry(std::move(new_fh), p.lru_list);
typename MapType::iterator new_it = p.cache.emplace_hint(range.first,
*fname, std::move(entry));
++p.size;
@@ -145,7 +149,8 @@ CachedHdfsFileHandle* FileHandleCache::GetFileHandle(
FileHandleEntry* new_elem = &new_it->second;
DCHECK(!new_elem->in_use);
new_elem->in_use = true;
- return new_elem->fh.get();
+ *handle_out = new_elem->fh.get();
+ return Status::OK();
}
void FileHandleCache::ReleaseFileHandle(std::string* fname,
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/be/src/runtime/io/hdfs-file-reader.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/hdfs-file-reader.cc b/be/src/runtime/io/hdfs-file-reader.cc
index c6a6cec..6fb0508 100644
--- a/be/src/runtime/io/hdfs-file-reader.cc
+++ b/be/src/runtime/io/hdfs-file-reader.cc
@@ -55,16 +55,12 @@ Status HdfsFileReader::Open(bool use_file_handle_cache) {
if (use_file_handle_cache && expected_local_) return Status::OK();
auto io_mgr = scan_range_->io_mgr_;
// Get a new exclusive file handle.
- exclusive_hdfs_fh_ = io_mgr->GetExclusiveHdfsFileHandle(hdfs_fs_,
- scan_range_->file_string(), scan_range_->mtime(), scan_range_->reader_);
- if (exclusive_hdfs_fh_ == nullptr) {
- return Status(TErrorCode::DISK_IO_ERROR,
- GetHdfsErrorMsg("Failed to open HDFS file ", *scan_range_->file_string()));
- }
+ RETURN_IF_ERROR(io_mgr->GetExclusiveHdfsFileHandle(hdfs_fs_,
+ scan_range_->file_string(), scan_range_->mtime(), scan_range_->reader_,
+ exclusive_hdfs_fh_));
if (hdfsSeek(hdfs_fs_, exclusive_hdfs_fh_->file(), scan_range_->offset_) != 0) {
// Destroy the file handle
- io_mgr->ReleaseExclusiveHdfsFileHandle(exclusive_hdfs_fh_);
- exclusive_hdfs_fh_ = nullptr;
+ io_mgr->ReleaseExclusiveHdfsFileHandle(std::move(exclusive_hdfs_fh_));
return Status(TErrorCode::DISK_IO_ERROR,
Substitute("Error seeking to $0 in file: $1 $2", scan_range_->offset(),
*scan_range_->file_string(), GetHdfsErrorMsg("")));
@@ -99,13 +95,9 @@ Status HdfsFileReader::ReadFromPos(int64_t file_offset, uint8_t* buffer,
if (exclusive_hdfs_fh_ != nullptr) {
hdfs_file = exclusive_hdfs_fh_->file();
} else {
- borrowed_hdfs_fh = io_mgr->GetCachedHdfsFileHandle(hdfs_fs_,
+ RETURN_IF_ERROR(io_mgr->GetCachedHdfsFileHandle(hdfs_fs_,
scan_range_->file_string(),
- scan_range_->mtime(), request_context);
- if (borrowed_hdfs_fh == nullptr) {
- return Status(TErrorCode::DISK_IO_ERROR,
- GetHdfsErrorMsg("Failed to open HDFS file ", *scan_range_->file_string()));
- }
+ scan_range_->mtime(), request_context, &borrowed_hdfs_fh));
hdfs_file = borrowed_hdfs_fh->file();
}
@@ -228,8 +220,7 @@ void HdfsFileReader::Close() {
}
// Destroy the file handle.
- scan_range_->io_mgr_->ReleaseExclusiveHdfsFileHandle(exclusive_hdfs_fh_);
- exclusive_hdfs_fh_ = nullptr;
+ scan_range_->io_mgr_->ReleaseExclusiveHdfsFileHandle(std::move(exclusive_hdfs_fh_));
ImpaladMetrics::IO_MGR_NUM_OPEN_FILES->Increment(-1L);
}
@@ -287,7 +278,7 @@ void HdfsFileReader::ResetState() {
string HdfsFileReader::DebugString() const {
return FileReader::DebugString() + Substitute(
" exclusive_hdfs_fh=$0 num_remote_bytes=$1",
- exclusive_hdfs_fh_, num_remote_bytes_);
+ exclusive_hdfs_fh_.get(), num_remote_bytes_);
}
}
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/be/src/runtime/io/hdfs-file-reader.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/hdfs-file-reader.h b/be/src/runtime/io/hdfs-file-reader.h
index ba95095..f8dbf93 100644
--- a/be/src/runtime/io/hdfs-file-reader.h
+++ b/be/src/runtime/io/hdfs-file-reader.h
@@ -57,7 +57,7 @@ private:
/// 3. The hdfs file is expected to be remote (expected_local_ == false)
/// In each case, the scan range gets a new ExclusiveHdfsFileHandle at Open(),
/// owns it exclusively, and destroys it in Close().
- ExclusiveHdfsFileHandle* exclusive_hdfs_fh_ = nullptr;
+ std::unique_ptr<ExclusiveHdfsFileHandle> exclusive_hdfs_fh_;
/// If true, we expect the reads to be a local read. Note that if this is false,
/// it does not necessarily mean we expect the read to be remote, and that we never
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/be/src/runtime/io/hdfs-monitored-ops.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/hdfs-monitored-ops.cc b/be/src/runtime/io/hdfs-monitored-ops.cc
new file mode 100644
index 0000000..3ea5c86
--- /dev/null
+++ b/be/src/runtime/io/hdfs-monitored-ops.cc
@@ -0,0 +1,99 @@
+// 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.
+
+#include "gutil/strings/substitute.h"
+
+#include "common/names.h"
+#include "common/status.h"
+#include "runtime/io/hdfs-monitored-ops.h"
+#include "util/hdfs-util.h"
+#include "util/time.h"
+
+namespace impala {
+
+namespace io {
+
+// Timeout for HDFS operations. This defaults to 5 minutes
+DEFINE_uint64(hdfs_operation_timeout_sec, 300, "Maximum time, in seconds, that an "
+ "HDFS operation should wait before timing out and failing.");
+
+Status HdfsMonitor::Init(int32_t num_threads) {
+ // The thread pool sets its queue size to be equal to the number of threads.
+ // TODO: is there a better queue size?
+ hdfs_worker_pool_.reset(new SynchronousThreadPool("hdfs monitor",
+ "hdfs_monitor_", num_threads, num_threads));
+ return hdfs_worker_pool_->Init();
+}
+
+class OpenHdfsFileOp : public SynchronousWorkItem {
+ public:
+ // To guarantee the appropriate lifetime, the 'fname' argument is copied. The 'fs'
+ // is safe, because it is kept open until the process terminates (see hdfs-fs-cache.h).
+ OpenHdfsFileOp(const hdfsFS& fs, const std::string* fname, int flags,
+ uint64_t blocksize)
+ : fs_(fs), fname_(*fname), flags_(flags), blocksize_(blocksize) {}
+
+ // Run hdfsOpenFile() and return error status if hdfsOpenFile() fails. If hdfsOpenFile()
+ // succeeds, it is valid to call GetFileHandle().
+ virtual Status Execute() override;
+
+ virtual std::string GetDescription() override;
+
+ // If the handle is successfully opened, GetHandle() will return the handle.
+ hdfsFile GetFileHandle() { return hdfs_file_; }
+
+ private:
+ const hdfsFS& fs_;
+ const std::string fname_;
+ int flags_;
+ uint64_t blocksize_;
+ hdfsFile hdfs_file_;
+};
+
+Status OpenHdfsFileOp::Execute() {
+ hdfs_file_ = hdfsOpenFile(fs_, fname_.data(), flags_, 0, 0, blocksize_);
+ if (hdfs_file_ == nullptr) {
+ // GetHdfsErrorMsg references thread local state to get error information, so it
+ // must happen in the same thread as the hdfsOpenFile().
+ return Status(TErrorCode::DISK_IO_ERROR,
+ GetHdfsErrorMsg("Failed to open HDFS file ", fname_));
+ }
+ return Status::OK();
+}
+
+std::string OpenHdfsFileOp::GetDescription() {
+ return Substitute("hdfsOpenFile() for $0", fname_);
+}
+
+Status HdfsMonitor::OpenHdfsFileWithTimeout(const hdfsFS& fs, const std::string* fname,
+ int flags, uint64_t blocksize, hdfsFile* fid_out) {
+ std::shared_ptr<OpenHdfsFileOp> op(new OpenHdfsFileOp(fs, fname, flags, blocksize));
+
+ // Submit the operation to the synchronous thread pool. If this times out, it returns
+ // an error status.
+ Status status = hdfs_worker_pool_->SynchronousOffer(op,
+ FLAGS_hdfs_operation_timeout_sec * MILLIS_PER_SEC);
+ RETURN_IF_ERROR(status);
+
+ // The operation didn't time out, and there wasn't an error. Get the handle.
+ *fid_out = op->GetFileHandle();
+ return Status::OK();
+}
+
+}
+
+}
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/be/src/runtime/io/hdfs-monitored-ops.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/hdfs-monitored-ops.h b/be/src/runtime/io/hdfs-monitored-ops.h
new file mode 100644
index 0000000..d20341a
--- /dev/null
+++ b/be/src/runtime/io/hdfs-monitored-ops.h
@@ -0,0 +1,62 @@
+// 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.
+
+#ifndef IMPALA_RUNTIME_IO_HDFS_MONITORED_OPS_H
+#define IMPALA_RUNTIME_IO_HDFS_MONITORED_OPS_H
+
+#include "runtime/io/handle-cache.h"
+#include "util/thread-pool.h"
+
+namespace impala {
+
+namespace io {
+
+/// The HdfsMonitor implements timeouts on HDFS operations that otherwise would block
+/// indefinitely. It submits the operations to a thread pool and waits with a timeout
+/// for a response.
+class HdfsMonitor {
+ public:
+ HdfsMonitor() {}
+
+ // Initialize the thread pool with 'num_threads'
+ Status Init(int32_t num_threads) WARN_UNUSED_RESULT;
+
+ // Open the specified HDFS file. If the operation times out, returns an error status
+ // and leaves 'hdfs_file_out' untouched. If the operation does not time out, then
+ // the operation can still fail if the file does not exist. In this case,
+ // 'hdfs_file_out' is null and an error status is returned. If the operation succeeds,
+ // then 'hdfs_file_out' contains the file handle.
+ //
+ // This is a thin wrapper around hdfsOpenFile() that implements a timeout.
+ // The differences from hdfsOpenFile() are:
+ // - The filename is taken as a const string* rather than a const char*
+ // - The bufferSize and replication are omitted, since no Impala code sets
+ // these to anything other than 0.
+ // - It has be rearranged to return Status.
+ Status OpenHdfsFileWithTimeout(const hdfsFS& fs, const std::string* fname, int flags,
+ uint64_t blocksize, hdfsFile* hdfs_file_out) WARN_UNUSED_RESULT;
+
+ private:
+ // Pool of threads handling HDFS operations.
+ std::unique_ptr<SynchronousThreadPool> hdfs_worker_pool_;
+};
+
+}
+
+}
+
+#endif
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/be/src/runtime/runtime-state.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/runtime-state.cc b/be/src/runtime/runtime-state.cc
index 9681aac..0899964 100644
--- a/be/src/runtime/runtime-state.cc
+++ b/be/src/runtime/runtime-state.cc
@@ -219,7 +219,9 @@ Status RuntimeState::LogOrReturnError(const ErrorMsg& message) {
|| message.error() == TErrorCode::CANCELLED_INTERNALLY
|| message.error() == TErrorCode::MEM_LIMIT_EXCEEDED
|| message.error() == TErrorCode::INTERNAL_ERROR
- || message.error() == TErrorCode::DISK_IO_ERROR) {
+ || message.error() == TErrorCode::DISK_IO_ERROR
+ || message.error() == TErrorCode::THREAD_POOL_SUBMIT_FAILED
+ || message.error() == TErrorCode::THREAD_POOL_TASK_TIMED_OUT) {
return Status(message);
}
// Otherwise, add the error to the error log and continue.
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/be/src/util/thread-pool-test.cc
----------------------------------------------------------------------
diff --git a/be/src/util/thread-pool-test.cc b/be/src/util/thread-pool-test.cc
index a7e4df7..d0e9287 100644
--- a/be/src/util/thread-pool-test.cc
+++ b/be/src/util/thread-pool-test.cc
@@ -68,6 +68,93 @@ TEST(ThreadPoolTest, BasicTest) {
EXPECT_EQ(expected_count, count);
}
+class SleepWorkItem : public SynchronousWorkItem {
+public:
+ SleepWorkItem(int64_t timeout_ms, bool* destructor_called)
+ : timeout_ms_(timeout_ms), destructor_called_ptr_(destructor_called) {
+ *destructor_called_ptr_ = false;
+ }
+
+ ~SleepWorkItem() {
+ *destructor_called_ptr_ = true;
+ }
+
+ virtual Status Execute() override {
+ if (timeout_ms_ > 0) SleepForMs(timeout_ms_);
+ message_ = "Done";
+ return Status::OK();
+ }
+ virtual std::string GetDescription() override {
+ return Substitute("Simple task with $0 millisecond timeout", timeout_ms_);
+ }
+
+ std::string GetMessage() { return message_; }
+private:
+ std::string message_ = "Not done";
+ int64_t timeout_ms_;
+ bool* destructor_called_ptr_;
+};
+
+TEST(ThreadPoolTest, SynchronousThreadPoolTest) {
+ // Create a synchronous pool with one thread and a queue size of one.
+ SynchronousThreadPool pool("sync-thread-pool", "worker", 1, 1);
+ ASSERT_OK(pool.Init());
+
+ // Base case: work item takes no time, run it with a timeout of 5 milliseconds
+ unique_ptr<bool> no_sleep_destroyed(new bool);
+ std::shared_ptr<SleepWorkItem> no_sleep(new SleepWorkItem(0, no_sleep_destroyed.get()));
+ ASSERT_OK(pool.SynchronousOffer(no_sleep, 5));
+ ASSERT_EQ(no_sleep->GetMessage(), "Done");
+ // If the SynchronousOffer() completed successfully, the thread pool does not have any
+ // shared_ptr to the work item. The caller is the only holder, so when it calls
+ // reset, the destructor must be called.
+ no_sleep.reset();
+ ASSERT_TRUE(*no_sleep_destroyed);
+
+ // Timeout case #1: Submit one task that takes 100 milliseconds. Offer it with a timeout
+ // of 1 millisecond so that the caller immediately times out.
+ unique_ptr<bool> long_sleep_destroyed(new bool);
+ std::shared_ptr<SleepWorkItem> long_sleep(
+ new SleepWorkItem(100, long_sleep_destroyed.get()));
+ Status timeout_status = pool.SynchronousOffer(long_sleep, 1);
+ ASSERT_EQ(timeout_status.code(), TErrorCode::THREAD_POOL_TASK_TIMED_OUT);
+ // The work item is still running, and even if the caller releases its shared_ptr,
+ // the work item is not destroyed.
+ long_sleep.reset();
+ ASSERT_FALSE(*long_sleep_destroyed);
+
+ // The single thread in the thread pool is still running. Submit another task
+ // that will queue. The task doesn't matter.
+ unique_ptr<bool> queued_task_destroyed(new bool);
+ std::shared_ptr<SleepWorkItem> queued_task(
+ new SleepWorkItem(0, queued_task_destroyed.get()));
+ Status queued_task_status = pool.SynchronousOffer(queued_task, 1);
+ ASSERT_EQ(queued_task_status.code(), TErrorCode::THREAD_POOL_TASK_TIMED_OUT);
+ // The work item is queued, and even if the caller releases its shared_ptr, the work
+ // item is not destroyed.
+ queued_task.reset();
+ ASSERT_FALSE(*queued_task_destroyed);
+
+ // Now, the queue is full. Any new task will fail to submit.
+ unique_ptr<bool> fail_to_submit_destroyed(new bool);
+ std::shared_ptr<SleepWorkItem> fail_to_submit(
+ new SleepWorkItem(0, fail_to_submit_destroyed.get()));
+ Status fail_to_submit_status = pool.SynchronousOffer(fail_to_submit, 1);
+ ASSERT_EQ(fail_to_submit_status.code(), TErrorCode::THREAD_POOL_SUBMIT_FAILED);
+ // When the submit fails, the thread pool does not keep any shared_ptr to the work
+ // item. When the caller releases its shared_ptr, the work item is immediately
+ // destroyed.
+ fail_to_submit.reset();
+ ASSERT_TRUE(*fail_to_submit_destroyed);
+
+ // The tasks will still complete
+ pool.DrainAndShutdown();
+ // The work items that the thread pool had running and in the queue are destroyed
+ // when they complete (even though the caller long since released its shared_ptr).
+ ASSERT_TRUE(*long_sleep_destroyed);
+ ASSERT_TRUE(*queued_task_destroyed);
+}
+
}
IMPALA_TEST_MAIN();
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/be/src/util/thread-pool.h
----------------------------------------------------------------------
diff --git a/be/src/util/thread-pool.h b/be/src/util/thread-pool.h
index fedba58..c64285c 100644
--- a/be/src/util/thread-pool.h
+++ b/be/src/util/thread-pool.h
@@ -105,6 +105,20 @@ class ThreadPool : public CacheLineAligned {
return work_queue_.BlockingPut(std::forward<V>(work));
}
+ /// Blocks until the work item is placed on the queue or the timeout expires. The
+ /// ThreadPool must be initialized before calling this method. The same requirements
+ /// about the lifetime of 'work' applies as in Offer() above. If the operation times
+ /// out, the work item can be safely freed.
+ ///
+ /// Returns true if the work item was successfully added to the queue, false otherwise
+ /// (which means the operation timed out or the thread pool has already been shut down).
+ template <typename V>
+ bool Offer(V&& work, int64_t timeout_millis) {
+ DCHECK(initialized_);
+ int64_t timeout_micros = timeout_millis * MICROS_PER_MILLI;
+ return work_queue_.BlockingPutWithTimeout(std::forward<V>(work), timeout_micros);
+ }
+
/// Shuts the thread pool down, causing the work queue to cease accepting offered work
/// and the worker threads to terminate once they have processed their current work item.
/// Returns once the shutdown flag has been set, does not wait for the threads to
@@ -219,6 +233,119 @@ class CallableThreadPool : public ThreadPool<boost::function<void()>> {
}
};
+/// Parent class for all synchronous work items
+///
+/// Important note for all subclasses:
+/// All fields need to have a lifetime that matches this operation's lifetime.
+/// In particular, caller-provided arguments need to be moved, copied, or need to have a
+/// lifetime independent of the caller. The monitored operation must survive the caller
+/// timing out and potentially deallocating memory.
+class SynchronousWorkItem {
+ public:
+ virtual ~SynchronousWorkItem() {}
+
+ /// Customized implementation for each operation. Subclasses must override this.
+ /// The status is conveyed back to the original caller.
+ virtual Status Execute() {
+ DCHECK(false) << "Execute() must be implemented";
+ return Status("Execute() must be implemented");
+ }
+
+ virtual std::string GetDescription() {
+ DCHECK(false) << "GetDescription() must be implemented";
+ return "";
+ }
+
+ private:
+ friend class SynchronousThreadPool;
+
+ /// This is called by the worker thread and handles the mechanics of notifying
+ /// the caller and conveying status when Execute() completes.
+ void WorkerExecute() {
+ Status status = Execute();
+ DCHECK(!done_promise_.IsSet());
+ discard_result(done_promise_.Set(status));
+ }
+
+ /// Wait for the operation to complete or time out with the specified limit.
+ /// If the operation times out, it returns TErrorCode::THREAD_POOL_TASK_TIMED_OUT.
+ /// Otherwise, it returns the status returned by Execute().
+ Status Wait(int64_t timeout_millis) {
+ bool timed_out;
+ Status status = done_promise_.Get(timeout_millis, &timed_out);
+ if (timed_out) {
+ Status timeout_status = Status(TErrorCode::THREAD_POOL_TASK_TIMED_OUT,
+ GetDescription(), timeout_millis / MILLIS_PER_SEC);
+ LOG(WARNING) << timeout_status.GetDetail();
+ return timeout_status;
+ }
+ return status;
+ }
+
+ // Set to the return status of ExecuteImpl() upon completion
+ Promise<Status> done_promise_;
+};
+
+/// Synchronous thread pool can run any subclass of SynchronousWorkItem
+/// Ownership is shared between the caller side and the worker side:
+/// 1. The caller accesses the operation to wait for its completion (or timeout) and
+/// retrieve any result.
+/// 2. The ThreadPool worker calls Execute() on the operation and needs to maintain
+/// ownership until the operation completes. The blocking queue inside the ThreadPool
+/// also needs to maintain ownership until a thread removes the operation for
+/// processing.
+/// This is an awkward circumstance to have exclusive ownership. The caller can time
+/// out and leave while the worker is processing. When the worker completes and could
+/// release ownership, the caller might still need to retrieve the result or the caller
+/// might be gone. shared_ptr does what we want: the HdfsMonitorOp will survive until
+/// neither thread needs it anymore.
+class SynchronousThreadPool : public ThreadPool<std::shared_ptr<SynchronousWorkItem>> {
+ public:
+ SynchronousThreadPool(const std::string& group, const std::string& thread_prefix,
+ uint32_t num_threads, uint32_t queue_size) :
+ ThreadPool<std::shared_ptr<SynchronousWorkItem>>(group, thread_prefix, num_threads,
+ queue_size, &SynchronousThreadPool::Worker) {}
+
+ /// Run the provided work item and wait up to 'timeout_milliseconds' for the
+ /// operation to complete. If it completes, return the status from the work
+ /// item's Execute() function. Otherwise, return an error status:
+ /// - THREAD_POOL_TASK_TIMED_OUT if the individual task timed out
+ /// - THREAD_POOL_SUBMIT_FAILED if all the threads are busy and the task did not
+ /// even start
+ Status SynchronousOffer(std::shared_ptr<SynchronousWorkItem> work,
+ int64_t timeout_milliseconds) {
+ MonotonicStopWatch offer_timer;
+ offer_timer.Start();
+ bool offer_success = Offer(work, timeout_milliseconds);
+ offer_timer.Stop();
+ if (!offer_success) {
+ // This scenario only happens when all threads are occupied and the queue
+ // is full. This means the system is in a catastrophic state. Log to ERROR.
+ Status failed_to_submit_status =
+ Status(TErrorCode::THREAD_POOL_SUBMIT_FAILED, work->GetDescription(),
+ timeout_milliseconds / MILLIS_PER_SEC);
+ LOG(ERROR) << failed_to_submit_status.GetDetail();
+ return failed_to_submit_status;
+ }
+
+ int64_t offer_time_millis =
+ offer_timer.ElapsedTime() / (NANOS_PER_MICRO * MICROS_PER_MILLI);
+ int64_t timeout_left = timeout_milliseconds - offer_time_millis;
+ if (timeout_left <= 0) {
+ Status timeout_status = Status(TErrorCode::THREAD_POOL_TASK_TIMED_OUT,
+ work->GetDescription(), timeout_milliseconds / MILLIS_PER_SEC);
+ LOG(WARNING) << timeout_status.GetDetail();
+ return timeout_status;
+ }
+ return work->Wait(timeout_left);
+ }
+
+ private:
+ static void Worker(int thread_id, const std::shared_ptr<SynchronousWorkItem>& work) {
+ work->WorkerExecute();
+ }
+};
+
}
#endif
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/common/thrift/generate_error_codes.py
----------------------------------------------------------------------
diff --git a/common/thrift/generate_error_codes.py b/common/thrift/generate_error_codes.py
index d8d40b9..dd8f4f6 100755
--- a/common/thrift/generate_error_codes.py
+++ b/common/thrift/generate_error_codes.py
@@ -373,6 +373,12 @@ error_codes = (
("PARQUET_CORRUPT_BOOL_VALUE", 122, "File '$0' is corrupt: error decoding BOOLEAN "
"value with encoding $1 at offset $2"),
+
+ ("THREAD_POOL_SUBMIT_FAILED", 123,
+ "Failed to submit $0 to thread pool after waiting $1 seconds"),
+
+ ("THREAD_POOL_TASK_TIMED_OUT", 124,
+ "$0 failed to finish before the $1 second timeout"),
)
import sys
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/tests/custom_cluster/test_hdfs_timeout.py
----------------------------------------------------------------------
diff --git a/tests/custom_cluster/test_hdfs_timeout.py b/tests/custom_cluster/test_hdfs_timeout.py
new file mode 100644
index 0000000..a20285a
--- /dev/null
+++ b/tests/custom_cluster/test_hdfs_timeout.py
@@ -0,0 +1,90 @@
+# 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.
+
+import pytest
+import re
+import time
+
+from tests.common.custom_cluster_test_suite import CustomClusterTestSuite
+from tests.common.skip import SkipIfNotHdfsMinicluster
+from subprocess import check_call, check_output
+
+
+@SkipIfNotHdfsMinicluster.tuned_for_minicluster
+class TestHdfsTimeouts(CustomClusterTestSuite):
+ """Test to verify that HDFS operations time out correctly."""
+
+ @classmethod
+ def get_workload(self):
+ return 'functional-query'
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(
+ impalad_args="--hdfs_operation_timeout_sec=5 --max_cached_file_handles=0")
+ def test_hdfs_open_timeout(self, vector):
+ """This verifies that hdfsOpenFile times out appropriately. It tests this by
+ halting the NameNode, running a query that needs to do hdfsOpenFile,
+ and verifying that it times out and throws an error."""
+
+ # Find the NameNode's pid via pgrep. This would raise an error if it did not
+ # find a pid, so there is at least one match.
+ pgrep_output = check_output(["pgrep", "-f", "namenode.NameNode"])
+ # In our test environment, this should only match one pid
+ assert(pgrep_output.count("\n") == 1)
+ namenode_pid = pgrep_output.strip()
+
+ # Run a query successfully. This fetches metadata from the NameNode,
+ # and since this will be cached, a subsequent run will not ask the NameNode
+ # for metadata. This means a subsequent execution will only talk to the NameNode
+ # for file open.
+ self.execute_query_expect_success(self.client,
+ "select count(*) from functional.alltypes", vector=vector)
+
+ # Stop the NameNode and execute the query again. Since the file handle cache is off,
+ # the query will do hdfsOpenFile calls and talk to the NameNode. Since the NameNode
+ # is stopped, those calls will hang, testing the timeout functionality.
+ ex = None
+ result = None
+ try:
+ # Stop the NameNode
+ check_call(["kill", "-STOP", namenode_pid])
+ start_time = time.time()
+ result = self.execute_query("select count(*) from functional.alltypes",
+ vector=vector)
+ end_time = time.time()
+ except Exception, e:
+ ex = e
+ finally:
+ end_time = time.time()
+ # Always resume the NameNode
+ check_call(["kill", "-CONT", namenode_pid])
+
+ # The query should have failed, which raises an exception
+ if ex is None:
+ assert False, "Query should have failed, but instead returned {0}".format(result)
+
+ # The exception should contain the appropriate error message
+ error_pattern = "hdfsOpenFile\(\) for.*failed to finish before the 5 second timeout"
+ assert len(re.findall(error_pattern, str(ex))) > 0
+
+ # The timeout is 5 seconds and seems to be enforced within about 20 seconds, so it
+ # would be unusual if this query does not finish in 60 seconds.
+ assert (end_time - start_time) < 60.0
+
+ # Execute the query a final time to verify that the system recovers.
+ self.execute_query_expect_success(self.client,
+ "select count(*) from functional.alltypes", vector=vector)