You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by ta...@apache.org on 2018/11/05 21:57:53 UTC
[1/3] impala git commit: IMPALA-5031: memcpy cannot take null
arguments
Repository: impala
Updated Branches:
refs/heads/master f876b320a -> 572722e35
IMPALA-5031: memcpy cannot take null arguments
This patch fixes UBSAN "null pointer passed as argument" errors in
data loading. These are undefined behavior according to "7.1.4 Use of
library functions" in the C99 standard (which is included in C++14 in
section [intro.refs]):
If an argument to a function has an invalid value (such as a value
outside the domain of the function, or a pointer outside the
address space of the program, or a null pointer, or a pointer to
non-modifiable storage when the corresponding parameter is not
const-qualified) or a type (after promotion) not expected by a
function with variable number of arguments, the behavior is
undefined.
The interesting parts of the backtraces for the errors fixed in this
patch are below:
runtime/string-buffer.h:54:12: runtime error: null pointer passed as argument 1, which is declared to never be null
/usr/include/string.h:43:45: note: nonnull attribute specified here
StringBuffer::Append(char const*, long) runtime/string-buffer.h:54:5
ColumnStatsBase::CopyToBuffer(StringBuffer*, StringValue*) exec/parquet-column-stats.cc:151:51
ColumnStats<StringValue>::MaterializeStringValuesToInternalBuffers() exec/parquet-column-stats.inline.h:237:70
HdfsParquetTableWriter::BaseColumnWriter::MaterializeStatsValues() exec/hdfs-parquet-table-writer.cc:149:63
HdfsParquetTableWriter::AppendRows(RowBatch*, vector<int> const&, bool*) exec/hdfs-parquet-table-writer.cc:1129:53
HdfsTableSink::WriteRowsToPartition(RuntimeState*, RowBatch*, pair<unique_ptr<OutputPartition, default_delete<OutputPartition> >, vector<int, all> > >*) exec/hdfs-table-sink.cc:256:71
HdfsTableSink::Send(RuntimeState*, RowBatch*) exec/hdfs-table-sink.cc:591:45
util/streaming-sampler.h:111:22: runtime error: null pointer passed as argument 2, which is declared to never be null
/usr/include/string.h:43:45: note: nonnull attribute specified here
StreamingSampler<long, 64>::SetSamples(int, vector<long> const&) util/streaming-sampler.h:111:5
RuntimeProfile::Update(vector<TRuntimeProfileNode> const&, int*) util/runtime-profile.cc:313:30
RuntimeProfile::Update(TRuntimeProfileTree const&) util/runtime-profile.cc:246:3
Coordinator::BackendState::InstanceStats::Update(TFragmentInstanceExecStatus const&, Coordinator::ExecSummary*, ProgressUpdater*) runtime/coordinator-backend-state.cc:474:13
Coordinator::BackendState::ApplyExecStatusReport(TReportExecStatusParams const&, Coordinator::ExecSummary*, ProgressUpdater*) runtime/coordinator-backend-state.cc:287:21
Coordinator::UpdateBackendExecStatus(TReportExecStatusParams const&) runtime/coordinator.cc:679:22
ClientRequestState::UpdateBackendExecStatus(TReportExecStatusParams const&) service/client-request-state.cc:1254:18
ImpalaServer::ReportExecStatus(TReportExecStatusResult&, TReportExecStatusParams const&) service/impala-server.cc:1343:18
ImpalaInternalService::ReportExecStatus(TReportExecStatusResult&, TReportExecStatusParams const&) service/impala-internal-service.cc:87:19
Change-Id: Ib9acc8c32409e67253a987eb3d1fd7d921efcb51
Reviewed-on: http://gerrit.cloudera.org:8080/11812
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/a03e2201
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/a03e2201
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/a03e2201
Branch: refs/heads/master
Commit: a03e22011dfc6c93818e923d0dc29e6f61fefbe9
Parents: f876b32
Author: Jim Apple <jb...@apache.org>
Authored: Sat Oct 27 16:57:55 2018 -0700
Committer: Impala Public Jenkins <im...@cloudera.com>
Committed: Sat Nov 3 22:25:29 2018 +0000
----------------------------------------------------------------------
be/src/runtime/string-buffer.h | 3 ++-
be/src/util/streaming-sampler.h | 2 +-
2 files changed, 3 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/a03e2201/be/src/runtime/string-buffer.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/string-buffer.h b/be/src/runtime/string-buffer.h
index 2188e4d..8416405 100644
--- a/be/src/runtime/string-buffer.h
+++ b/be/src/runtime/string-buffer.h
@@ -23,6 +23,7 @@
#include "runtime/mem-pool.h"
#include "runtime/mem-tracker.h"
#include "runtime/string-value.h"
+#include "util/ubsan.h"
namespace impala {
@@ -51,7 +52,7 @@ class StringBuffer {
Status Append(const char* str, int64_t str_len) WARN_UNUSED_RESULT {
int64_t new_len = len_ + str_len;
if (UNLIKELY(new_len > buffer_size_)) RETURN_IF_ERROR(GrowBuffer(new_len));
- memcpy(buffer_ + len_, str, str_len);
+ Ubsan::MemCpy(buffer_ + len_, str, str_len);
len_ += str_len;
return Status::OK();
}
http://git-wip-us.apache.org/repos/asf/impala/blob/a03e2201/be/src/util/streaming-sampler.h
----------------------------------------------------------------------
diff --git a/be/src/util/streaming-sampler.h b/be/src/util/streaming-sampler.h
index dfa3dab..027f330 100644
--- a/be/src/util/streaming-sampler.h
+++ b/be/src/util/streaming-sampler.h
@@ -108,7 +108,7 @@ class StreamingSampler {
boost::lock_guard<SpinLock> l(lock_);
period_ = period;
samples_collected_ = samples.size();
- memcpy(samples_, samples.data(), sizeof(T) * samples_collected_);
+ Ubsan::MemCpy(samples_, samples.data(), sizeof(T) * samples_collected_);
current_sample_sum_ = 0;
current_sample_count_ = 0;
current_sample_total_time_ = 0;
[2/3] impala git commit: IMPALA-6436: exit instead of abort for
catalog startup failure
Posted by ta...@apache.org.
IMPALA-6436: exit instead of abort for catalog startup failure
Rename EXIT_WITH_EXC to ABORT_WITH_EXC to make the behaviour more
obvious at callsites.
Handle exceptions from Catalog constructor by logging the backtrace and
exiting cleanly, rather than aborting. This will prevent generation of a
coredump or minidump.
Testing:
Tested starting the catalogd locally without the HMS running and a
low connection timeout:
start-impala-cluster.py --catalogd_args=--initial_hms_cnxn_timeout_s=2
Confirmed that the backtrace was logged to catalogd.ERROR and that no
core or minidump was generated.
Change-Id: I4026dccb39843b847426112fc0fe9ba897e48dcc
Reviewed-on: http://gerrit.cloudera.org:8080/11871
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/f08642bf
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/f08642bf
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/f08642bf
Branch: refs/heads/master
Commit: f08642bf43102cf326f4f00e9b9e3536d6906b2c
Parents: a03e220
Author: Tim Armstrong <ta...@cloudera.com>
Authored: Fri Nov 2 09:46:29 2018 -0700
Committer: Impala Public Jenkins <im...@cloudera.com>
Committed: Mon Nov 5 21:19:06 2018 +0000
----------------------------------------------------------------------
be/src/catalog/catalog.cc | 4 +--
be/src/scheduling/request-pool-service.cc | 10 ++++----
be/src/service/fe-support.cc | 2 +-
be/src/service/frontend.cc | 4 +--
be/src/util/jni-util.h | 35 ++++++++++++++++++++------
be/src/util/logging-support.cc | 4 +--
be/src/util/zip-util.cc | 2 +-
7 files changed, 40 insertions(+), 21 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/f08642bf/be/src/catalog/catalog.cc
----------------------------------------------------------------------
diff --git a/be/src/catalog/catalog.cc b/be/src/catalog/catalog.cc
index a2d5f4b..e5bf35c 100644
--- a/be/src/catalog/catalog.cc
+++ b/be/src/catalog/catalog.cc
@@ -76,7 +76,7 @@ Catalog::Catalog() {
JNIEnv* jni_env = getJNIEnv();
// Create an instance of the java class JniCatalog
catalog_class_ = jni_env->FindClass("org/apache/impala/service/JniCatalog");
- EXIT_IF_EXC(jni_env);
+ ABORT_IF_EXC(jni_env);
uint32_t num_methods = sizeof(methods) / sizeof(methods[0]);
for (int i = 0; i < num_methods; ++i) {
@@ -87,7 +87,7 @@ Catalog::Catalog() {
ABORT_IF_ERROR(GetThriftBackendGflags(jni_env, &cfg_bytes));
jobject catalog = jni_env->NewObject(catalog_class_, catalog_ctor_, cfg_bytes);
- EXIT_IF_EXC(jni_env);
+ CLEAN_EXIT_IF_EXC(jni_env);
ABORT_IF_ERROR(JniUtil::LocalToGlobalRef(jni_env, catalog, &catalog_));
}
http://git-wip-us.apache.org/repos/asf/impala/blob/f08642bf/be/src/scheduling/request-pool-service.cc
----------------------------------------------------------------------
diff --git a/be/src/scheduling/request-pool-service.cc b/be/src/scheduling/request-pool-service.cc
index f9fdf88..c3c9750 100644
--- a/be/src/scheduling/request-pool-service.cc
+++ b/be/src/scheduling/request-pool-service.cc
@@ -124,7 +124,7 @@ RequestPoolService::RequestPoolService(MetricGroup* metrics) :
JNIEnv* jni_env = getJNIEnv();
request_pool_service_class_ =
jni_env->FindClass("org/apache/impala/util/RequestPoolService");
- EXIT_IF_EXC(jni_env);
+ ABORT_IF_EXC(jni_env);
uint32_t num_methods = sizeof(methods) / sizeof(methods[0]);
for (int i = 0; i < num_methods; ++i) {
ABORT_IF_ERROR(JniUtil::LoadJniMethod(jni_env, request_pool_service_class_,
@@ -133,18 +133,18 @@ RequestPoolService::RequestPoolService(MetricGroup* metrics) :
jstring fair_scheduler_config_path =
jni_env->NewStringUTF(FLAGS_fair_scheduler_allocation_path.c_str());
- EXIT_IF_EXC(jni_env);
+ ABORT_IF_EXC(jni_env);
jstring llama_site_path =
jni_env->NewStringUTF(FLAGS_llama_site_path.c_str());
- EXIT_IF_EXC(jni_env);
+ ABORT_IF_EXC(jni_env);
jobject request_pool_service = jni_env->NewObject(request_pool_service_class_, ctor_,
fair_scheduler_config_path, llama_site_path);
- EXIT_IF_EXC(jni_env);
+ ABORT_IF_EXC(jni_env);
ABORT_IF_ERROR(JniUtil::LocalToGlobalRef(jni_env, request_pool_service,
&request_pool_service_));
jni_env->CallObjectMethod(request_pool_service_, start_id);
- EXIT_IF_EXC(jni_env);
+ ABORT_IF_EXC(jni_env);
}
Status RequestPoolService::ResolveRequestPool(const TQueryCtx& ctx,
http://git-wip-us.apache.org/repos/asf/impala/blob/f08642bf/be/src/service/fe-support.cc
----------------------------------------------------------------------
diff --git a/be/src/service/fe-support.cc b/be/src/service/fe-support.cc
index 5dd06de..4a25250 100644
--- a/be/src/service/fe-support.cc
+++ b/be/src/service/fe-support.cc
@@ -699,7 +699,7 @@ void InitFeSupport(bool disable_codegen) {
jclass native_backend_cl = env->FindClass("org/apache/impala/service/FeSupport");
env->RegisterNatives(native_backend_cl, native_methods,
sizeof(native_methods) / sizeof(native_methods[0]));
- EXIT_IF_EXC(env);
+ ABORT_IF_EXC(env);
}
}
http://git-wip-us.apache.org/repos/asf/impala/blob/f08642bf/be/src/service/frontend.cc
----------------------------------------------------------------------
diff --git a/be/src/service/frontend.cc b/be/src/service/frontend.cc
index af45d96..c4ad4c7 100644
--- a/be/src/service/frontend.cc
+++ b/be/src/service/frontend.cc
@@ -106,7 +106,7 @@ Frontend::Frontend() {
JNIEnv* jni_env = getJNIEnv();
// create instance of java class JniFrontend
fe_class_ = jni_env->FindClass("org/apache/impala/service/JniFrontend");
- EXIT_IF_EXC(jni_env);
+ ABORT_IF_EXC(jni_env);
uint32_t num_methods = sizeof(methods) / sizeof(methods[0]);
for (int i = 0; i < num_methods; ++i) {
@@ -117,7 +117,7 @@ Frontend::Frontend() {
ABORT_IF_ERROR(GetThriftBackendGflags(jni_env, &cfg_bytes));
jobject fe = jni_env->NewObject(fe_class_, fe_ctor_, cfg_bytes);
- EXIT_IF_EXC(jni_env);
+ ABORT_IF_EXC(jni_env);
ABORT_IF_ERROR(JniUtil::LocalToGlobalRef(jni_env, fe, &fe_));
}
http://git-wip-us.apache.org/repos/asf/impala/blob/f08642bf/be/src/util/jni-util.h
----------------------------------------------------------------------
diff --git a/be/src/util/jni-util.h b/be/src/util/jni-util.h
index 576f943..4b18f20 100644
--- a/be/src/util/jni-util.h
+++ b/be/src/util/jni-util.h
@@ -59,8 +59,8 @@
#define THROW_IF_EXC(env, exc_class) \
do { \
jthrowable exc = (env)->ExceptionOccurred(); \
- if (exc != NULL) { \
- DCHECK((throwable_to_string_id_) != NULL); \
+ if (exc != nullptr) { \
+ DCHECK((throwable_to_string_id_) != nullptr); \
jstring stack = (jstring) env->CallStaticObjectMethod(JniUtil::jni_util_class(), \
(JniUtil::throwable_to_stack_trace_id()), exc); \
jboolean is_copy; \
@@ -75,7 +75,7 @@
#define RETURN_IF_EXC(env) \
do { \
jthrowable exc = (env)->ExceptionOccurred(); \
- if (exc != NULL) { \
+ if (exc != nullptr) { \
jstring stack = (jstring) env->CallStaticObjectMethod(JniUtil::jni_util_class(), \
(JniUtil::throwable_to_stack_trace_id()), exc); \
jboolean is_copy; \
@@ -86,10 +86,14 @@
} \
} while (false)
-#define EXIT_IF_EXC(env) \
+// If there's an exception in 'env', log the backtrace at FATAL level and abort the
+// process. This will generate a core dump if core dumps are enabled, so this should
+// generally only be called for internal errors where the coredump is useful for
+// diagnostics.
+#define ABORT_IF_EXC(env) \
do { \
jthrowable exc = (env)->ExceptionOccurred(); \
- if (exc != NULL) { \
+ if (exc != nullptr) { \
jstring stack = (jstring) env->CallStaticObjectMethod(JniUtil::jni_util_class(), \
(JniUtil::throwable_to_stack_trace_id()), exc); \
jboolean is_copy; \
@@ -99,10 +103,25 @@
} \
} while (false)
+// If there's an exception in 'env', log the backtrace at ERROR level and exit the process
+// cleanly with status 1.
+#define CLEAN_EXIT_IF_EXC(env) \
+ do { \
+ jthrowable exc = (env)->ExceptionOccurred(); \
+ if (exc != nullptr) { \
+ jstring stack = (jstring) env->CallStaticObjectMethod(JniUtil::jni_util_class(), \
+ (JniUtil::throwable_to_stack_trace_id()), exc); \
+ jboolean is_copy; \
+ const char* c_stack = \
+ reinterpret_cast<const char*>((env)->GetStringUTFChars(stack, &is_copy)); \
+ CLEAN_EXIT_WITH_ERROR(c_stack); \
+ } \
+ } while (false)
+
#define RETURN_ERROR_IF_EXC(env) \
do { \
jthrowable exc = (env)->ExceptionOccurred(); \
- if (exc != NULL) return JniUtil::GetJniExceptionMsg(env);\
+ if (exc != nullptr) return JniUtil::GetJniExceptionMsg(env);\
} while (false)
/// C linkage for helper functions in hdfsJniHelper.h
@@ -118,8 +137,8 @@ class Status;
/// the corresponding objects.
class JniLocalFrame {
public:
- JniLocalFrame(): env_(NULL) {}
- ~JniLocalFrame() { if (env_ != NULL) env_->PopLocalFrame(NULL); }
+ JniLocalFrame(): env_(nullptr) {}
+ ~JniLocalFrame() { if (env_ != nullptr) env_->PopLocalFrame(nullptr); }
JniLocalFrame(JniLocalFrame&& other) noexcept
: env_(other.env_) {
http://git-wip-us.apache.org/repos/asf/impala/blob/f08642bf/be/src/util/logging-support.cc
----------------------------------------------------------------------
diff --git a/be/src/util/logging-support.cc b/be/src/util/logging-support.cc
index 457c8f9..b050f37 100644
--- a/be/src/util/logging-support.cc
+++ b/be/src/util/logging-support.cc
@@ -110,7 +110,7 @@ Webserver::UrlCallback MakeCallback(const F& fnc, bool display_log4j_handlers) {
void InitDynamicLoggingSupport() {
JNIEnv* env = getJNIEnv();
log4j_logger_class_ = env->FindClass("org/apache/impala/util/GlogAppender");
- EXIT_IF_EXC(env);
+ ABORT_IF_EXC(env);
JniMethodDescriptor get_log_level_method_desc =
{"getLogLevel", "([B)Ljava/lang/String;", &get_log_level_method};
JniMethodDescriptor set_log_level_method_desc =
@@ -240,7 +240,7 @@ void InitJvmLoggingSupport() {
nm.signature = const_cast<char*>("(ILjava/lang/String;Ljava/lang/String;I)V");
nm.fnPtr = reinterpret_cast<void*>(::Java_org_apache_impala_util_NativeLogger_Log);
env->RegisterNatives(native_backend_cl, &nm, 1);
- EXIT_IF_EXC(env);
+ ABORT_IF_EXC(env);
InitDynamicLoggingSupport();
}
http://git-wip-us.apache.org/repos/asf/impala/blob/f08642bf/be/src/util/zip-util.cc
----------------------------------------------------------------------
diff --git a/be/src/util/zip-util.cc b/be/src/util/zip-util.cc
index f6e1f15..43c46df 100644
--- a/be/src/util/zip-util.cc
+++ b/be/src/util/zip-util.cc
@@ -30,7 +30,7 @@ jmethodID ZipUtil::extract_files_method; // ZipUtil.extractFiles()
void ZipUtil::InitJvm() {
JNIEnv* env = getJNIEnv();
zip_util_class_ = env->FindClass("org/apache/impala/util/ZipUtil");
- EXIT_IF_EXC(env);
+ ABORT_IF_EXC(env);
JniMethodDescriptor extract_files_method_desc =
{"extractFiles", "([B)V", &extract_files_method};
[3/3] impala git commit: IMPALA-7775: fix some lifecycle issues in
statestore/session tests
Posted by ta...@apache.org.
IMPALA-7775: fix some lifecycle issues in statestore/session tests
Background threads from the Statestore's thread pool continued
running in the background and could dereference invalid memory.
We make sure these threads are cleaned up before moving onto
the next test. Note that we don't clean up all background
threads, just the ones that had caused issues here.
I refactored the memory management a bit to put all objects
that we can't safely free into a single ObjectPool.
The statestore tests also had an issue with the lifetime of the
string flags FLAGS_ssl_*_certificate. Those were overwritten
with new values while the thread pool threads were still running,
which could cause use-after-free bugs.
Testing:
Looped the tests under ASAN with the "stress" utility running at the
same time to flush out races.
Ran core tests.
Change-Id: I3b25c8b8a96bfa1183ce273b3bb4debde234dd01
Reviewed-on: http://gerrit.cloudera.org:8080/11864
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/572722e3
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/572722e3
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/572722e3
Branch: refs/heads/master
Commit: 572722e3505a663560e3fff1fe04d8fe932c0959
Parents: f08642b
Author: Tim Armstrong <ta...@cloudera.com>
Authored: Thu Nov 1 12:00:37 2018 -0700
Committer: Impala Public Jenkins <im...@cloudera.com>
Committed: Mon Nov 5 21:41:00 2018 +0000
----------------------------------------------------------------------
be/src/service/session-expiry-test.cc | 13 ++++-
be/src/statestore/statestore-test.cc | 89 ++++++++++++++++++------------
be/src/statestore/statestore.cc | 11 ++++
be/src/statestore/statestore.h | 7 ++-
4 files changed, 81 insertions(+), 39 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/572722e3/be/src/service/session-expiry-test.cc
----------------------------------------------------------------------
diff --git a/be/src/service/session-expiry-test.cc b/be/src/service/session-expiry-test.cc
index 89ae842..7e5b86a 100644
--- a/be/src/service/session-expiry-test.cc
+++ b/be/src/service/session-expiry-test.cc
@@ -44,14 +44,20 @@ DECLARE_int32(beeswax_port);
// TODO: Come up with a short-running test that confirms a session will keep itself alive
// that doesn't depend upon being rescheduled in a timely fashion.
+// Object pool containing all objects that must live for the duration of the process.
+// E.g. objects that are singletons and never destroyed in a real daemon (so don't support
+// tear-down logic), but which we create multiple times in unit tests. We leak this pool
+// instead of destroying it to avoid destroying the contained objects.
+static ObjectPool* perm_objects;
+
TEST(SessionTest, TestExpiry) {
const int NUM_SESSIONS = 5;
const int MAX_IDLE_TIMEOUT_MS = 4000;
FLAGS_idle_session_timeout = 1;
// Skip validation checks for in-process backend.
FLAGS_abort_on_config_error = false;
- scoped_ptr<MetricGroup> metrics(new MetricGroup("statestore"));
- Statestore* statestore = new Statestore(metrics.get());
+ MetricGroup* metrics = perm_objects->Add(new MetricGroup("statestore"));
+ Statestore* statestore = perm_objects->Add(new Statestore(metrics));
IGNORE_LEAKING_OBJECT(statestore);
// Pass in 0 to have the statestore use an ephemeral port for the service.
ABORT_IF_ERROR(statestore->Init(0));
@@ -111,11 +117,14 @@ TEST(SessionTest, TestExpiry) {
// work). Sleep to allow the threads closing the session to complete before tearing down
// the server.
SleepForMs(1000);
+ statestore->ShutdownForTesting();
}
int main(int argc, char** argv) {
::testing::InitGoogleTest(&argc, argv);
impala::InitCommonRuntime(argc, argv, true, impala::TestInfo::BE_TEST);
InitFeSupport();
+ perm_objects = new ObjectPool;
+ IGNORE_LEAKING_OBJECT(perm_objects);
return RUN_ALL_TESTS();
}
http://git-wip-us.apache.org/repos/asf/impala/blob/572722e3/be/src/statestore/statestore-test.cc
----------------------------------------------------------------------
diff --git a/be/src/statestore/statestore-test.cc b/be/src/statestore/statestore-test.cc
index a9ee095..f517a56 100644
--- a/be/src/statestore/statestore-test.cc
+++ b/be/src/statestore/statestore-test.cc
@@ -34,77 +34,94 @@ DECLARE_int32(state_store_port);
namespace impala {
+// Object pool containing all objects that must live for the duration of the process.
+// E.g. objects that are singletons and never destroyed in a real daemon (so don't support
+// tear-down logic), but which we create multiple times in unit tests. We leak this pool
+// instead of destroying it to avoid destroying the contained objects.
+static ObjectPool* perm_objects;
+
TEST(StatestoreTest, SmokeTest) {
// All allocations done by 'new' to avoid problems shutting down Thrift servers
// gracefully.
- scoped_ptr<MetricGroup> metrics(new MetricGroup("statestore"));
- Statestore* statestore = new Statestore(metrics.get());
+ MetricGroup* metrics = perm_objects->Add(new MetricGroup("statestore"));
+ Statestore* statestore = perm_objects->Add(new Statestore(metrics));
// Thrift will internally pick an ephemeral port if we pass in 0 as the port.
int statestore_port = 0;
- IGNORE_LEAKING_OBJECT(statestore);
ASSERT_OK(statestore->Init(statestore_port));
- scoped_ptr<MetricGroup> metrics_2(new MetricGroup("statestore_2"));
+ MetricGroup* metrics_2 = perm_objects->Add(new MetricGroup("statestore_2"));
// Port already in use
- Statestore* statestore_wont_start = new Statestore(metrics_2.get());
+ Statestore* statestore_wont_start = perm_objects->Add(new Statestore(metrics_2));
ASSERT_FALSE(statestore_wont_start->Init(statestore->port()).ok());
- StatestoreSubscriber* sub_will_start =
+ StatestoreSubscriber* sub_will_start = perm_objects->Add(
new StatestoreSubscriber("sub1", MakeNetworkAddress("localhost", 0),
- MakeNetworkAddress("localhost", statestore->port()), new MetricGroup(""));
- IGNORE_LEAKING_OBJECT(sub_will_start);
+ MakeNetworkAddress("localhost", statestore->port()), new MetricGroup("")));
ASSERT_OK(sub_will_start->Start());
// Confirm that a subscriber trying to use an in-use port will fail to start.
- StatestoreSubscriber* sub_will_not_start = new StatestoreSubscriber("sub3",
- MakeNetworkAddress("localhost", sub_will_start->heartbeat_port()),
- MakeNetworkAddress("localhost", statestore->port()), new MetricGroup(""));
- IGNORE_LEAKING_OBJECT(sub_will_not_start);
+ StatestoreSubscriber* sub_will_not_start = perm_objects->Add(new StatestoreSubscriber(
+ "sub3", MakeNetworkAddress("localhost", sub_will_start->heartbeat_port()),
+ MakeNetworkAddress("localhost", statestore->port()), new MetricGroup("")));
ASSERT_FALSE(sub_will_not_start->Start().ok());
+
+ statestore->ShutdownForTesting();
}
-TEST(StatestoreSslTest, SmokeTest) {
+// Runs an SSL smoke test with provided parameters.
+void SslSmokeTestHelper(const string& server_ca_certificate,
+ const string& client_ca_certificate, bool sub_should_start) {
string impala_home(getenv("IMPALA_HOME"));
stringstream server_cert;
server_cert << impala_home << "/be/src/testutil/server-cert.pem";
- FLAGS_ssl_server_certificate = server_cert.str();
- FLAGS_ssl_client_ca_certificate = server_cert.str();
+ // Override flags for the duration of this test. Modifying them while the statestore
+ // is running is unsafe.
+ FLAGS_ssl_server_certificate = server_ca_certificate;
+ FLAGS_ssl_client_ca_certificate = client_ca_certificate;
stringstream server_key;
server_key << impala_home << "/be/src/testutil/server-key.pem";
FLAGS_ssl_private_key = server_key.str();
// Thrift will internally pick an ephemeral port if we pass in 0 as the port.
int statestore_port = 0;
- scoped_ptr<MetricGroup> metrics(new MetricGroup("statestore"));
- Statestore* statestore = new Statestore(metrics.get());
- IGNORE_LEAKING_OBJECT(statestore);
+ MetricGroup* metrics = perm_objects->Add(new MetricGroup("statestore"));
+ Statestore* statestore = perm_objects->Add(new Statestore(metrics));
ASSERT_OK(statestore->Init(statestore_port));
- StatestoreSubscriber* sub_will_start = new StatestoreSubscriber("smoke_sub1",
- MakeNetworkAddress("localhost", 0),
- MakeNetworkAddress("localhost", statestore->port()), new MetricGroup(""));
- IGNORE_LEAKING_OBJECT(sub_will_start);
- ASSERT_OK(sub_will_start->Start());
+ StatestoreSubscriber* sub = perm_objects->Add(
+ new StatestoreSubscriber("smoke_sub", MakeNetworkAddress("localhost", 0),
+ MakeNetworkAddress("localhost", statestore->port()), new MetricGroup("")));
+ Status sub_status = sub->Start();
+ ASSERT_EQ(sub_should_start, sub_status.ok());
- stringstream invalid_server_cert;
- invalid_server_cert << impala_home << "/be/src/testutil/invalid-server-cert.pem";
- FLAGS_ssl_client_ca_certificate = invalid_server_cert.str();
+ statestore->ShutdownForTesting();
+}
- StatestoreSubscriber* sub_will_not_start = new StatestoreSubscriber("smoke_sub2",
- MakeNetworkAddress("localhost", 0),
- MakeNetworkAddress("localhost", statestore->port()), new MetricGroup(""));
- IGNORE_LEAKING_OBJECT(sub_will_not_start);
- ASSERT_FALSE(sub_will_not_start->Start().ok());
+string GetValidServerCert() {
+ string impala_home(getenv("IMPALA_HOME"));
+ stringstream server_cert;
+ server_cert << impala_home << "/be/src/testutil/server-cert.pem";
+ return server_cert.str();
+}
+
+TEST(StatestoreSslTest, ValidCertSmokeTest) {
+ string valid_cert = GetValidServerCert();
+ SslSmokeTestHelper(valid_cert, valid_cert, true);
}
+TEST(StatestoreSslTest, InvalidCertSmokeTest) {
+ string impala_home(getenv("IMPALA_HOME"));
+ stringstream invalid_server_cert;
+ invalid_server_cert << impala_home << "/be/src/testutil/invalid-server-cert.pem";
+ SslSmokeTestHelper(GetValidServerCert(), invalid_server_cert.str(), false);
}
+} // namespace impala
+
int main(int argc, char** argv) {
::testing::InitGoogleTest(&argc, argv);
impala::InitCommonRuntime(argc, argv, false, impala::TestInfo::BE_TEST);
- int rc = RUN_ALL_TESTS();
- // IMPALA-5291: statestore services and subscribers may still be running at this point
- // and accessing global state. Exit without running global destructors to avoid
- // races with other threads when tearing down the proces.
- _exit(rc);
+ perm_objects = new ObjectPool;
+ IGNORE_LEAKING_OBJECT(perm_objects);
+ return RUN_ALL_TESTS();
}
http://git-wip-us.apache.org/repos/asf/impala/blob/572722e3/be/src/statestore/statestore.cc
----------------------------------------------------------------------
diff --git a/be/src/statestore/statestore.cc b/be/src/statestore/statestore.cc
index 4e63dad..70183d7 100644
--- a/be/src/statestore/statestore.cc
+++ b/be/src/statestore/statestore.cc
@@ -36,6 +36,7 @@
#include "util/debug-util.h"
#include "util/logging-support.h"
#include "util/openssl-util.h"
+#include "util/test-info.h"
#include "util/time.h"
#include "util/uid-util.h"
#include "util/webserver.h"
@@ -1050,3 +1051,13 @@ void Statestore::MainLoop() {
subscriber_priority_topic_update_threadpool_.Join();
subscriber_heartbeat_threadpool_.Join();
}
+
+void Statestore::ShutdownForTesting() {
+ CHECK(TestInfo::is_be_test()) << "Only valid to call in backend tests.";
+ subscriber_topic_update_threadpool_.Shutdown();
+ subscriber_priority_topic_update_threadpool_.Shutdown();
+ subscriber_heartbeat_threadpool_.Shutdown();
+ subscriber_topic_update_threadpool_.Join();
+ subscriber_priority_topic_update_threadpool_.Join();
+ subscriber_heartbeat_threadpool_.Join();
+}
http://git-wip-us.apache.org/repos/asf/impala/blob/572722e3/be/src/statestore/statestore.h
----------------------------------------------------------------------
diff --git a/be/src/statestore/statestore.h b/be/src/statestore/statestore.h
index 52f7d68..871494c 100644
--- a/be/src/statestore/statestore.h
+++ b/be/src/statestore/statestore.h
@@ -157,6 +157,11 @@ class Statestore : public CacheLineAligned {
/// The main processing loop. Runs infinitely.
void MainLoop();
+ /// Shut down some background threads. Only used for testing. Note that this is not
+ /// a clean shutdown because we can't correctly tear down 'thrift_server_', so
+ /// not all background threads are stopped and this object cannot be destroyed.
+ void ShutdownForTesting();
+
/// Returns the Thrift API interface that proxies requests onto the local Statestore.
const boost::shared_ptr<StatestoreServiceIf>& thrift_iface() const {
return thrift_iface_;
@@ -713,6 +718,6 @@ class Statestore : public CacheLineAligned {
[[noreturn]] void MonitorSubscriberHeartbeat();
};
-}
+} // namespace impala
#endif