You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by st...@apache.org on 2020/07/14 00:23:05 UTC
[impala] branch master updated: IMPALA-5534: Fix and enable
experimental failure tests
This is an automated email from the ASF dual-hosted git repository.
stakiar pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
The following commit(s) were added to refs/heads/master by this push:
new fc19e70 IMPALA-5534: Fix and enable experimental failure tests
fc19e70 is described below
commit fc19e70cbc94735c38e64573c1e015878f218281
Author: Sahil Takiar <ta...@gmail.com>
AuthorDate: Tue Apr 7 06:15:18 2020 -0700
IMPALA-5534: Fix and enable experimental failure tests
Moves the test_catalog_hms_failures.py and test_process_failures.py from
the experimental tests to custom cluster tests.
catalog_service/test_hms_failure.py is combined with
custom_cluster/test_catalog_hms_failure.py as well in order to unify all
tests for HMS failures. Several modifications to the tests were
necessary to get them working again, but for the most part, the logic of
the tests remained the same. A few additional fault tolerance tests
(e.g. TestHiveMetaStoreFailure::test_hms_client_retries) were added as
well. The overall goal is to increase the process failure test coverage
for all components: impalads, statestore, catalogd, HMS, etc.
test_restart_catalogd in test_process_failures.py fails due to
IMPALA-9848, so it is skipped for now.
Testing:
* Ran new tests locally
Change-Id: I9dbb98017fb6c40cea349e7c63a35c325cbbc288
Reviewed-on: http://gerrit.cloudera.org:8080/16157
Reviewed-by: Sahil Takiar <st...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../java/org/apache/impala/service/Frontend.java | 4 +-
tests/catalog_service/test_hms_failure.py | 86 --------------
tests/common/impala_service.py | 3 +-
.../test_catalog_hms_failures.py | 132 ++++++++++++++++++---
.../test_process_failures.py | 132 +++++++++++----------
5 files changed, 188 insertions(+), 169 deletions(-)
diff --git a/fe/src/main/java/org/apache/impala/service/Frontend.java b/fe/src/main/java/org/apache/impala/service/Frontend.java
index 7c40f69..eb3bc69 100644
--- a/fe/src/main/java/org/apache/impala/service/Frontend.java
+++ b/fe/src/main/java/org/apache/impala/service/Frontend.java
@@ -121,6 +121,7 @@ import org.apache.impala.planner.PlanFragment;
import org.apache.impala.planner.Planner;
import org.apache.impala.planner.ScanNode;
import org.apache.impala.thrift.TAlterDbParams;
+import org.apache.impala.thrift.TBackendGflags;
import org.apache.impala.thrift.TCatalogOpRequest;
import org.apache.impala.thrift.TCatalogOpType;
import org.apache.impala.thrift.TCatalogServiceRequestHeader;
@@ -328,7 +329,8 @@ public class Frontend {
BackendConfig.INSTANCE);
queryHookManager_ = QueryEventHookManager.createFromConfig(BackendConfig.INSTANCE);
if (!isBackendTest) {
- metaStoreClientPool_ = new MetaStoreClientPool(1, 0);
+ TBackendGflags cfg = BackendConfig.INSTANCE.getBackendCfg();
+ metaStoreClientPool_ = new MetaStoreClientPool(1, cfg.initial_hms_cnxn_timeout_s);
if (MetastoreShim.getMajorVersion() > 2) {
transactionKeepalive_ = new TransactionKeepalive(metaStoreClientPool_);
} else {
diff --git a/tests/catalog_service/test_hms_failure.py b/tests/catalog_service/test_hms_failure.py
deleted file mode 100644
index 4e47b39..0000000
--- a/tests/catalog_service/test_hms_failure.py
+++ /dev/null
@@ -1,86 +0,0 @@
-# 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.
-#
-# Tests to validate the Catalog Service continues to function even if the HMS fails.
-
-import pytest
-import os
-from subprocess import check_call
-
-from tests.beeswax.impala_beeswax import ImpalaBeeswaxException
-from tests.common.impala_test_suite import ImpalaTestSuite
-from tests.common.test_dimensions import create_single_exec_option_dimension
-from tests.util.filesystem_utils import IS_ISILON, IS_LOCAL
-
-class TestHiveMetaStoreFailure(ImpalaTestSuite):
- @classmethod
- def get_workload(cls):
- return 'functional-query'
-
- @classmethod
- def add_test_dimensions(cls):
- super(TestHiveMetaStoreFailure, cls).add_test_dimensions()
- cls.ImpalaTestMatrix.add_dimension(create_single_exec_option_dimension())
-
- # There is no reason to run these tests using all dimensions.
- cls.ImpalaTestMatrix.add_constraint(lambda v:\
- v.get_value('table_format').file_format == 'parquet' and\
- v.get_value('table_format').compression_codec == 'none')
-
- @classmethod
- def setup_class(cls):
- super(TestHiveMetaStoreFailure, cls).setup_class()
-
- @classmethod
- def run_hive_server(cls):
- script = os.path.join(os.environ['IMPALA_HOME'], 'testdata/bin/run-hive-server.sh')
- run_cmd = [script]
- if IS_LOCAL or IS_ISILON:
- run_cmd.append('-only_metastore')
- check_call(run_cmd, close_fds=True)
-
- @classmethod
- def teardown_class(cls):
- # Make sure the metastore is running even if the test aborts somewhere unexpected
- # before restarting the metastore itself.
- cls.run_hive_server()
- super(TestHiveMetaStoreFailure, cls).teardown_class()
-
- @pytest.mark.execute_serially
- def test_hms_service_dies(self, vector):
- """Regression test for IMPALA-823 to verify the catalog service works properly when
- HMS connections fail"""
- # Force the tables to be uncached and then kill the hive metastore.
- tbl_name = "functional.alltypes"
- self.client.execute("invalidate metadata %s" % tbl_name)
- kill_cmd = os.path.join(os.environ['IMPALA_HOME'], 'testdata/bin/kill-hive-server.sh')
- check_call([kill_cmd], close_fds=True)
-
- try:
- self.client.execute("describe %s" % tbl_name)
- pytest.xfail("It is possible (although unlikely) that all tables completed " +
- "loading between the time 'invalidate metadata' was called and the HMS was " +
- "killed. TODO: Make the test case more deterministic to ensure this never " +
- "happens.")
- except ImpalaBeeswaxException as e:
- print str(e)
- assert "Failed to load metadata for table: %s. Running 'invalidate metadata %s' "\
- "may resolve this problem." % (tbl_name, tbl_name) in str(e)
- self.run_hive_server()
-
- self.client.execute("invalidate metadata %s" % tbl_name)
- self.client.execute("describe %s" % tbl_name)
diff --git a/tests/common/impala_service.py b/tests/common/impala_service.py
index 22bdbe4..fe6b0e6 100644
--- a/tests/common/impala_service.py
+++ b/tests/common/impala_service.py
@@ -319,7 +319,8 @@ class ImpaladService(BaseImpalaService):
try:
query_state = client.get_state(query_handle)
except Exception as e:
- pass
+ LOG.error("Exception while getting state of query {0}\n{1}".format(
+ query_handle.get_handle().id, str(e)))
if query_state == target_state:
return
sleep(interval)
diff --git a/tests/experiments/test_catalog_hms_failures.py b/tests/custom_cluster/test_catalog_hms_failures.py
similarity index 56%
rename from tests/experiments/test_catalog_hms_failures.py
rename to tests/custom_cluster/test_catalog_hms_failures.py
index efa4fed..95fdcfd 100644
--- a/tests/experiments/test_catalog_hms_failures.py
+++ b/tests/custom_cluster/test_catalog_hms_failures.py
@@ -1,4 +1,3 @@
-# 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
@@ -14,23 +13,112 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-#
-# Test Catalog behavior when HMS is not present
-import os
import pytest
-from subprocess import check_call
+import os
import time
+import threading
+from subprocess import check_call
from tests.beeswax.impala_beeswax import ImpalaBeeswaxException
from tests.common.custom_cluster_test_suite import (
CustomClusterTestSuite,
- NUM_SUBSCRIBERS)
+ DEFAULT_CLUSTER_SIZE)
from tests.util.filesystem_utils import IS_ISILON, IS_LOCAL
+
+NUM_SUBSCRIBERS = DEFAULT_CLUSTER_SIZE + 1
+
+
+class TestHiveMetaStoreFailure(CustomClusterTestSuite):
+ """Tests to validate the Catalog Service continues to function even if the HMS
+ fails."""
+
+ @classmethod
+ def get_workload(cls):
+ return 'functional-query'
+
+ @classmethod
+ def setup_class(cls):
+ if cls.exploration_strategy() != 'exhaustive':
+ pytest.skip('These tests only run in exhaustive')
+ super(TestHiveMetaStoreFailure, cls).setup_class()
+
+ @classmethod
+ def run_hive_server(cls):
+ script = os.path.join(os.environ['IMPALA_HOME'], 'testdata/bin/run-hive-server.sh')
+ run_cmd = [script]
+ if IS_LOCAL or IS_ISILON:
+ run_cmd.append('-only_metastore')
+ check_call(run_cmd, close_fds=True)
+
+ @classmethod
+ def teardown_class(cls):
+ # Make sure the metastore is running even if the test aborts somewhere unexpected
+ # before restarting the metastore itself.
+ cls.run_hive_server()
+ super(TestHiveMetaStoreFailure, cls).teardown_class()
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(
+ impalad_args='--use_local_catalog --catalog_topic_mode=minimal',
+ catalogd_args='--catalog_topic_mode=minimal')
+ def test_hms_service_dies(self):
+ """Regression test for IMPALA-823 to verify the catalog service works properly when
+ HMS connections fail"""
+ # Force the tables to be uncached and then kill the hive metastore.
+ tbl_name = "functional.alltypes"
+ self.client.execute("invalidate metadata %s" % tbl_name)
+ kill_cmd = os.path.join(os.environ['IMPALA_HOME'], 'testdata/bin/kill-hive-server.sh')
+ check_call([kill_cmd], close_fds=True)
+
+ try:
+ self.client.execute("describe %s" % tbl_name)
+ except ImpalaBeeswaxException as e:
+ print str(e)
+ assert "Failed to load metadata for table: %s. Running 'invalidate metadata %s' "\
+ "may resolve this problem." % (tbl_name, tbl_name) in str(e)
+ self.run_hive_server()
+
+ self.client.execute("invalidate metadata %s" % tbl_name)
+ self.client.execute("describe %s" % tbl_name)
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(
+ impalad_args='--use_local_catalog --catalog_topic_mode=minimal',
+ catalogd_args='--catalog_topic_mode=minimal')
+ def test_hms_client_retries(self):
+ """Test that a running query will trigger the retry logic in
+ RetryingMetaStoreClient."""
+ # Force the tables to be uncached and then kill the hive metastore.
+ tbl_name = "functional.alltypes"
+ self.client.execute("invalidate metadata %s" % tbl_name)
+ kill_cmd = os.path.join(os.environ['IMPALA_HOME'], 'testdata/bin/kill-hive-server.sh')
+ check_call([kill_cmd], close_fds=True)
+
+ # Run a query asynchronously.
+ query = "select * from {0} limit 1".format(tbl_name)
+ thread = threading.Thread(target=lambda:
+ self.execute_query_expect_success(self.client, query))
+ thread.start()
+
+ # Wait 1 second for the catalogd to start contacting HMS, then start HMS.
+ time.sleep(1)
+ self.run_hive_server()
+
+ # Wait for the query to complete, assert that the HMS client retried the connection.
+ thread.join()
+ self.assert_catalogd_log_contains("INFO",
+ "MetaStoreClient lost connection. Attempting to reconnect")
+
+
class TestCatalogHMSFailures(CustomClusterTestSuite):
+ """Test Catalog behavior when HMS is not present."""
+
@classmethod
def setup_class(cls):
+ if cls.exploration_strategy() != 'exhaustive':
+ pytest.skip('These tests only run in exhaustive')
super(TestCatalogHMSFailures, cls).setup_class()
@classmethod
@@ -45,11 +133,11 @@ class TestCatalogHMSFailures(CustomClusterTestSuite):
def cleanup_process(cls, proc):
try:
proc.kill()
- except:
+ except Exception:
pass
try:
proc.wait()
- except:
+ except Exception:
pass
@classmethod
@@ -65,8 +153,10 @@ class TestCatalogHMSFailures(CustomClusterTestSuite):
client.execute('show databases')
@pytest.mark.execute_serially
- @CustomClusterTestSuite.with_args(catalogd_args='--initial_hms_cnxn_timeout_s=120')
- def test_kill_hms_after_catalog_init(self, vector):
+ @CustomClusterTestSuite.with_args(
+ impalad_args='--use_local_catalog --catalog_topic_mode=minimal',
+ catalogd_args='--initial_hms_cnxn_timeout_s=120 --catalog_topic_mode=minimal')
+ def test_kill_hms_after_catalog_init(self):
"""IMPALA-4278: If HMS dies after catalogd initialization, SQL statements that force
metadata load should fail quickly. After HMS restart, metadata load should work
again"""
@@ -97,8 +187,10 @@ class TestCatalogHMSFailures(CustomClusterTestSuite):
self.reload_metadata(client)
@pytest.mark.execute_serially
- @CustomClusterTestSuite.with_args(catalogd_args='--initial_hms_cnxn_timeout_s=120')
- def test_start_catalog_before_hms(self, vector):
+ @CustomClusterTestSuite.with_args(
+ impalad_args='--use_local_catalog --catalog_topic_mode=minimal',
+ catalogd_args='--initial_hms_cnxn_timeout_s=120 --catalog_topic_mode=minimal')
+ def test_start_catalog_before_hms(self):
"""IMPALA-4278: If catalogd is started with initial_hms_cnxn_timeout_s set to a value
greater than HMS startup time, it will manage to establish connection to HMS even if
HMS is started a little later"""
@@ -121,7 +213,7 @@ class TestCatalogHMSFailures(CustomClusterTestSuite):
try:
# Start the catalog service asynchronously.
- catalogd.start()
+ catalogd.start(wait_until_ready=False)
# Wait 10s to be sure that the catalogd is in the 'trying to connect' phase of its
# startup.
time.sleep(10)
@@ -129,7 +221,7 @@ class TestCatalogHMSFailures(CustomClusterTestSuite):
# Start Hive and wait for catalogd to come up
self.run_hive_server()
statestored.service.wait_for_live_subscribers(NUM_SUBSCRIBERS, timeout=60)
- impalad.service.wait_for_metric_value('catalog.ready', 1, timeout=60)
+ impalad.service.wait_for_metric_value('catalog.ready', True, timeout=60)
# Metadata load should work now
self.reload_metadata(client)
@@ -138,8 +230,10 @@ class TestCatalogHMSFailures(CustomClusterTestSuite):
self.cleanup_process(catalogd)
@pytest.mark.execute_serially
- @CustomClusterTestSuite.with_args(catalogd_args='--initial_hms_cnxn_timeout_s=30')
- def test_catalogd_fails_if_hms_started_late(self, vector):
+ @CustomClusterTestSuite.with_args(
+ impalad_args='--use_local_catalog --catalog_topic_mode=minimal',
+ catalogd_args='--initial_hms_cnxn_timeout_s=30 --catalog_topic_mode=minimal')
+ def test_catalogd_fails_if_hms_started_late(self):
"""IMPALA-4278: If the HMS is not started within initial_hms_cnxn_timeout_s, then the
catalogd fails"""
# Make sure that catalogd is connected to HMS
@@ -161,7 +255,7 @@ class TestCatalogHMSFailures(CustomClusterTestSuite):
try:
# Start the catalog service asynchronously.
- catalogd.start()
+ catalogd.start(wait_until_ready=False)
# Wait 40s to be sure that the catalogd has been trying to connect to HMS longer
# than initial_hms_cnxn_timeout_s.
time.sleep(40)
@@ -170,7 +264,7 @@ class TestCatalogHMSFailures(CustomClusterTestSuite):
self.run_hive_server()
# catalogd has terminated by now
- assert catalogd.get_pid() == None, "catalogd should have terminated"
+ assert not catalogd.get_pid(), "catalogd should have terminated"
finally:
# Make sure to clean up the catalogd process that we started
self.cleanup_process(catalogd)
@@ -179,7 +273,7 @@ class TestCatalogHMSFailures(CustomClusterTestSuite):
# Start the catalog service again and wait for it to come up.
catalogd.start()
statestored.service.wait_for_live_subscribers(NUM_SUBSCRIBERS, timeout=60)
- impalad.service.wait_for_metric_value('catalog.ready', 1, timeout=60)
+ impalad.service.wait_for_metric_value('catalog.ready', True, timeout=60)
# Metadata load should work now
self.reload_metadata(client)
diff --git a/tests/experiments/test_process_failures.py b/tests/custom_cluster/test_process_failures.py
similarity index 57%
rename from tests/experiments/test_process_failures.py
rename to tests/custom_cluster/test_process_failures.py
index 928159e..4fb58a4 100644
--- a/tests/experiments/test_process_failures.py
+++ b/tests/custom_cluster/test_process_failures.py
@@ -14,88 +14,99 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-#
-# Impala process failure test suite
import pytest
from tests.common.custom_cluster_test_suite import (
- NUM_SUBSCRIBERS,
- CLUSTER_SIZE,
+ DEFAULT_CLUSTER_SIZE,
CustomClusterTestSuite)
# The exact query doesn't matter much for these tests, just want a query that touches
# data on all nodes.
-QUERY = "select count(l_comment) from lineitem"
+QUERY = "select count(l_comment) from tpch.lineitem"
+
+# The default number of statestore subscribers should be the default cluster size plus
+# one for the catalogd.
+DEFAULT_NUM_SUBSCRIBERS = DEFAULT_CLUSTER_SIZE + 1
+
-# Validates killing and restarting impalad processes between query executions
class TestProcessFailures(CustomClusterTestSuite):
+ """Validates killing and restarting impalad processes between query executions."""
+
+ @classmethod
+ def setup_class(cls):
+ if cls.exploration_strategy() != 'exhaustive':
+ pytest.skip('These tests only run in exhaustive')
+ super(TestProcessFailures, cls).setup_class()
+
@pytest.mark.execute_serially
- def test_restart_coordinator(self, vector):
- """Restarts the coordinator between queries"""
+ def test_restart_coordinator(self):
+ """Restarts the coordinator between queries."""
impalad = self.cluster.get_any_impalad()
client = impalad.service.create_beeswax_client()
- self.execute_query_using_client(client, QUERY, vector)
+ self.execute_query_expect_success(client, QUERY)
statestored = self.cluster.statestored
impalad.restart()
- statestored.service.wait_for_live_subscribers(NUM_SUBSCRIBERS, timeout=60)
+ statestored.service.wait_for_live_subscribers(DEFAULT_NUM_SUBSCRIBERS, timeout=60)
# Reconnect
client = impalad.service.create_beeswax_client()
impalad.service.wait_for_metric_value('catalog.ready', 1, timeout=60)
- self.execute_query_using_client(client, QUERY, vector)
+ self.execute_query_expect_success(client, QUERY)
@pytest.mark.execute_serially
- def test_restart_statestore(self, vector):
- """Tests the cluster still functions when the statestore dies"""
+ @CustomClusterTestSuite.with_args(
+ impalad_args='--use_local_catalog',
+ catalogd_args='--catalog_topic_mode=minimal')
+ def test_restart_statestore(self):
+ """Tests the cluster still functions when the statestore dies."""
impalad = self.cluster.get_any_impalad()
client = impalad.service.create_beeswax_client()
statestored = self.cluster.statestored
statestored.kill()
impalad.service.wait_for_metric_value(
- 'statestore-subscriber.connected', 0, timeout=60)
+ 'statestore-subscriber.connected', False, timeout=60)
# impalad should still see the same number of live backends
- assert impalad.service.get_num_known_live_backends() == CLUSTER_SIZE
+ assert impalad.service.get_num_known_live_backends() == DEFAULT_CLUSTER_SIZE
- self.execute_query_using_client(client, QUERY, vector)
+ self.execute_query_expect_success(client, QUERY)
# Reconnect
statestored.start()
impalad.service.wait_for_metric_value(
- 'statestore-subscriber.connected', 1, timeout=60)
- statestored.service.wait_for_live_subscribers(NUM_SUBSCRIBERS, timeout=60)
+ 'statestore-subscriber.connected', True, timeout=60)
+ statestored.service.wait_for_live_subscribers(DEFAULT_NUM_SUBSCRIBERS, timeout=60)
# Wait for the number of live backends to reach the cluster size. Even though
# all backends have subscribed to the statestore, this impalad may not have
# received the update yet.
- impalad.service.wait_for_num_known_live_backends(CLUSTER_SIZE, timeout=60)
+ impalad.service.wait_for_num_known_live_backends(DEFAULT_CLUSTER_SIZE, timeout=60)
- self.execute_query_using_client(client, QUERY, vector)
+ self.execute_query_expect_success(client, QUERY)
@pytest.mark.execute_serially
- def test_kill_restart_worker(self, vector):
- """Verifies a worker is able to be killed"""
+ def test_kill_restart_worker(self):
+ """Verifies a worker is able to be killed."""
impalad = self.cluster.get_any_impalad()
client = impalad.service.create_beeswax_client()
- self.execute_query_using_client(client, QUERY, vector)
+ self.execute_query_expect_success(client, QUERY)
# select a different impalad and restart it
worker_impalad = self.cluster.get_different_impalad(impalad)
- print "Coordinator impalad: %s Worker impalad: %s" % (impalad, worker_impalad)
# Start executing a query. It will be cancelled due to a killed worker.
- handle = self.execute_query_async_using_client(client, QUERY, vector)
+ handle = client.execute_async(QUERY)
statestored = self.cluster.statestored
worker_impalad.kill()
# First wait until the the statestore realizes the impalad has gone down.
- statestored.service.wait_for_live_subscribers(NUM_SUBSCRIBERS - 1, timeout=60)
+ statestored.service.wait_for_live_subscribers(DEFAULT_NUM_SUBSCRIBERS - 1, timeout=60)
# Wait until the impalad registers another instance went down.
- impalad.service.wait_for_num_known_live_backends(CLUSTER_SIZE - 1, timeout=60)
+ impalad.service.wait_for_num_known_live_backends(DEFAULT_CLUSTER_SIZE - 1, timeout=60)
# Wait until the in-flight query has been cancelled.
# The in-flight query should have been cancelled, reporting a failed worker as the
@@ -103,39 +114,38 @@ class TestProcessFailures(CustomClusterTestSuite):
# node, or because a stream sender failed to establish a thrift connection. It is
# non-deterministic which of those paths will initiate cancellation, but in either
# case the query status should include the failed (or unreachable) worker.
- impalad.service.wait_for_query_state(client, handle,\
- client.QUERY_STATES['EXCEPTION'])
-
- # Wait for the query status on the query profile web page to contain the
- # expected failed hostport.
query_id = handle.get_handle().id
- failed_hostport = "%s:%s" % (worker_impalad.service.hostname,\
+ error_state = "Failed due to unreachable impalad"
+ assert impalad.service.wait_for_query_status(client, query_id, error_state)
+
+ # Assert that the query status on the query profile web page contains the expected
+ # failed hostport.
+ failed_hostport = "%s:%s" % (worker_impalad.service.hostname,
worker_impalad.service.be_port)
- query_status_match =\
- impalad.service.wait_for_query_status(client, query_id, failed_hostport)
- try:
- if not query_status_match:
- query_profile_page = impalad.service.read_query_profile_page(query_id)
- assert False, "Query status did not contain expected hostport %s\n\n%s"\
- % (failed_hostport, query_profile_page)
- finally:
- self.close_query_using_client(client, handle)
+ query_profile_page = impalad.service.read_query_profile_page(query_id)
+ assert failed_hostport in query_profile_page,\
+ "Query status did not contain expected hostport %s\n\n%s" % (failed_hostport,
+ query_profile_page)
# Should work fine even if a worker is down.
- self.execute_query_using_client(client, QUERY, vector)
+ self.execute_query_expect_success(client, QUERY)
# Bring the worker back online and validate queries still work.
worker_impalad.start()
- statestored.service.wait_for_live_subscribers(NUM_SUBSCRIBERS, timeout=60)
- worker_impalad.service.wait_for_metric_value('catalog.ready', 1, timeout=60)
- self.execute_query_using_client(client, QUERY, vector)
+ statestored.service.wait_for_live_subscribers(DEFAULT_NUM_SUBSCRIBERS, timeout=60)
+ worker_impalad.service.wait_for_metric_value('catalog.ready', True, timeout=60)
+ self.execute_query_expect_success(client, QUERY)
@pytest.mark.execute_serially
- def test_restart_catalogd(self, vector):
+ @CustomClusterTestSuite.with_args(
+ impalad_args='--use_local_catalog',
+ catalogd_args='--catalog_topic_mode=minimal')
+ @pytest.mark.xfail(run=False, reason="IMPALA-9848")
+ def test_restart_catalogd(self):
# Choose a random impalad verify a query can run against it.
impalad = self.cluster.get_any_impalad()
client = impalad.service.create_beeswax_client()
- self.execute_query_using_client(client, QUERY, vector)
+ self.execute_query_expect_success(client, QUERY)
# Kill the catalogd.
catalogd = self.cluster.catalogd
@@ -143,26 +153,25 @@ class TestProcessFailures(CustomClusterTestSuite):
# The statestore should detect the catalog service has gone down.
statestored = self.cluster.statestored
- statestored.service.wait_for_live_subscribers(NUM_SUBSCRIBERS - 1, timeout=60)
+ statestored.service.wait_for_live_subscribers(DEFAULT_NUM_SUBSCRIBERS - 1, timeout=60)
# We should still be able to execute queries using the impalad.
- self.execute_query_using_client(client, QUERY, vector)
+ self.execute_query_expect_success(client, QUERY)
# Start the catalog service back up.
catalogd.start()
- statestored.service.wait_for_live_subscribers(NUM_SUBSCRIBERS, timeout=60)
+ statestored.service.wait_for_live_subscribers(DEFAULT_NUM_SUBSCRIBERS, timeout=60)
# Execute a query against the catalog service.
- impalad.service.wait_for_metric_value('catalog.ready', 1, timeout=60)
- self.execute_query_using_client(client, "refresh lineitem", vector)
- self.execute_query_using_client(client, QUERY, vector)
+ impalad.service.wait_for_metric_value('catalog.ready', True, timeout=60)
+ self.execute_query_expect_success(client, QUERY)
@pytest.mark.execute_serially
- def test_restart_all_impalad(self, vector):
+ def test_restart_all_impalad(self):
"""Restarts all the impalads and runs a query"""
impalad = self.cluster.get_any_impalad()
client = impalad.service.create_beeswax_client()
- self.execute_query_using_client(client, QUERY, vector)
+ self.execute_query_expect_success(client, QUERY)
# Kill each impalad and wait for the statestore to register the failures.
for impalad_proc in self.cluster.impalads:
@@ -178,13 +187,12 @@ class TestProcessFailures(CustomClusterTestSuite):
# The impalads should re-register with the statestore on restart at which point they
# can execute queries.
- statestored.service.wait_for_live_subscribers(NUM_SUBSCRIBERS, timeout=60)
+ statestored.service.wait_for_live_subscribers(DEFAULT_NUM_SUBSCRIBERS, timeout=60)
for impalad in self.cluster.impalads:
- impalad.service.wait_for_num_known_live_backends(CLUSTER_SIZE, timeout=60)
- impalad.service.wait_for_metric_value('catalog.ready', 1, timeout=60)
+ impalad.service.wait_for_num_known_live_backends(DEFAULT_CLUSTER_SIZE, timeout=60)
+ impalad.service.wait_for_metric_value('catalog.ready', True, timeout=60)
client = impalad.service.create_beeswax_client()
- self.execute_query_using_client(client, QUERY, vector)
+ self.execute_query_expect_success(client, QUERY)
# Make sure the catalog service is actually back up by executing an operation
# against it.
- self.execute_query_using_client(client, "refresh lineitem", vector)
- self.execute_query_using_client(client, QUERY, vector)
+ self.execute_query_expect_success(client, QUERY)