You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by bo...@apache.org on 2019/05/28 10:26:15 UTC
[impala] 06/08: IMPALA-8473: Publish lineage info via hook
This is an automated email from the ASF dual-hosted git repository.
boroknagyz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 31195eb8119ac6a557486a10dc24692bb0202f85
Author: Radford Nguyen <ra...@cloudera.com>
AuthorDate: Wed May 15 19:59:21 2019 -0700
IMPALA-8473: Publish lineage info via hook
This commit introduces a hook mechanism for publishing,
lineage data specifically, but query information more
generally, from Impala.
The legacy behavior of writing the lineage file is
being retained but deprecated.
Hooks can be implemented by downstream consumers (i.e.
runtime dependencies) to hook into supported places during
Impala query execution:
- impalad startup
- query completion
- see IMPALA-8572 for caveat/details
The consumers are to be frontend Java dependencies
intiated at runtime. 2 backend flags configure this
behavior:
- `query_event_hook_classes` specifies a comma-separated
list of hook consumer implementation classes that
are instantiated and registered at impala start up.
- `query_event_hook_nthreads`
specifies the number of threads to use for asynchronous
hook execution. (Relevant if multiple hooks are
registered.)
Lineage information is passed from the backend after
a query completes (but before it returns) and given
to every hook to execute asynchronously. In other words,
a query may complete and return to the user before any
or all hooks have completed executing. An exception
during hook on-query-complete execution will simply be logged
and will not be (directly) fatal to the system.
Tests:
- added unit tests for FE hook execution
- added E2E tests for hook configuration, execution, error
- ran full build, tests
Change-Id: I23a896537a98bfef07fb27c70e9a87c105cd77a1
Reviewed-on: http://gerrit.cloudera.org:8080/13352
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/service/frontend.cc | 7 +
be/src/service/frontend.h | 5 +
be/src/service/impala-server.cc | 57 ++++-
be/src/service/impala-server.h | 6 +
be/src/util/backend-gflag-util.cc | 4 +
common/thrift/BackendGflags.thrift | 4 +
common/thrift/Frontend.thrift | 10 +
.../apache/impala/hooks/QueryCompleteContext.java | 56 +++++
.../org/apache/impala/hooks/QueryEventHook.java | 116 +++++++++++
.../apache/impala/hooks/QueryEventHookManager.java | 229 +++++++++++++++++++++
.../org/apache/impala/service/BackendConfig.java | 8 +
.../java/org/apache/impala/service/Frontend.java | 71 ++++++-
.../org/apache/impala/service/JniFrontend.java | 43 ++--
.../impala/hooks/QueryEventHookManagerTest.java | 146 +++++++++++++
.../impala/testutil/AlwaysErrorQueryEventHook.java | 33 +++
.../impala/testutil/CountingQueryEventHook.java | 52 +++++
.../impala/testutil/DummyQueryEventHook.java | 53 +++++
.../impala/testutil/PostQueryErrorEventHook.java | 32 +++
tests/authorization/test_provider.py | 4 +-
tests/custom_cluster/test_query_event_hooks.py | 202 ++++++++++++++++++
20 files changed, 1112 insertions(+), 26 deletions(-)
diff --git a/be/src/service/frontend.cc b/be/src/service/frontend.cc
index 616d8e4..6ef000a 100644
--- a/be/src/service/frontend.cc
+++ b/be/src/service/frontend.cc
@@ -109,6 +109,7 @@ Frontend::Frontend() {
{"getTableFiles", "([B)[B", &get_table_files_id_},
{"showCreateFunction", "([B)Ljava/lang/String;", &show_create_function_id_},
{"buildTestDescriptorTable", "([B)[B", &build_test_descriptor_table_id_},
+ {"callQueryCompleteHooks", "([B)V", &call_query_complete_hooks_id_}
};
JNIEnv* jni_env = JniUtil::GetJNIEnv();
@@ -293,3 +294,9 @@ Status Frontend::BuildTestDescriptorTable(const TBuildTestDescriptorTableParams&
TDescriptorTable* result) {
return JniUtil::CallJniMethod(fe_, build_test_descriptor_table_id_, params, result);
}
+
+// Call FE post-query execution hook
+Status Frontend::CallQueryCompleteHooks(const TQueryCompleteContext& context) {
+ return JniUtil::CallJniMethod(fe_, call_query_complete_hooks_id_, context);
+}
+
diff --git a/be/src/service/frontend.h b/be/src/service/frontend.h
index abcc6c3..f063fce 100644
--- a/be/src/service/frontend.h
+++ b/be/src/service/frontend.h
@@ -24,6 +24,7 @@
#include "gen-cpp/ImpalaHiveServer2Service.h"
#include "gen-cpp/ImpalaInternalService.h"
#include "gen-cpp/Frontend_types.h"
+#include "gen-cpp/LineageGraph_types.h"
#include "common/status.h"
namespace impala {
@@ -185,6 +186,9 @@ class Frontend {
Status BuildTestDescriptorTable(const TBuildTestDescriptorTableParams& params,
TDescriptorTable* result);
+ // Call FE post-query execution hook
+ Status CallQueryCompleteHooks(const TQueryCompleteContext& context);
+
private:
jobject fe_; // instance of org.apache.impala.service.JniFrontend
jmethodID create_exec_request_id_; // JniFrontend.createExecRequest()
@@ -213,6 +217,7 @@ class Frontend {
jmethodID wait_for_catalog_id_; // JniFrontend.waitForCatalog
jmethodID get_table_files_id_; // JniFrontend.getTableFiles
jmethodID show_create_function_id_; // JniFrontend.showCreateFunction
+ jmethodID call_query_complete_hooks_id_; // JniFrontend.callQueryCompleteHooks
// Only used for testing.
jmethodID build_test_descriptor_table_id_; // JniFrontend.buildTestDescriptorTable()
diff --git a/be/src/service/impala-server.cc b/be/src/service/impala-server.cc
index d584f6f..f8d25b5 100644
--- a/be/src/service/impala-server.cc
+++ b/be/src/service/impala-server.cc
@@ -92,6 +92,7 @@
#include "gen-cpp/ImpalaService_types.h"
#include "gen-cpp/ImpalaInternalService.h"
#include "gen-cpp/LineageGraph_types.h"
+#include "gen-cpp/Frontend_types.h"
#include "common/names.h"
@@ -256,6 +257,14 @@ DEFINE_int64(accepted_client_cnxn_timeout, 300000,
"the post-accept, pre-setup connection queue before it is timed out and the "
"connection request is rejected. A value of 0 means there is no timeout.");
+DEFINE_string(query_event_hook_classes, "", "Comma-separated list of java QueryEventHook "
+ "implementation classes to load and register at Impala startup. Class names should "
+ "be fully-qualified and on the classpath. Whitespace acceptable around delimiters.");
+
+DEFINE_int32(query_event_hook_nthreads, 1, "Number of threads to use for "
+ "QueryEventHook execution. If this number is >1 then hooks will execute "
+ "concurrently.");
+
DECLARE_bool(compact_catalog_topic);
namespace impala {
@@ -485,17 +494,42 @@ Status ImpalaServer::LogLineageRecord(const ClientRequestState& client_request_s
// Set the query end time in TLineageGraph. Must use UNIX time directly rather than
// e.g. converting from client_request_state.end_time() (IMPALA-4440).
lineage_graph.__set_ended(UnixMillis() / 1000);
+
string lineage_record;
LineageUtil::TLineageToJSON(lineage_graph, &lineage_record);
- const Status& status = lineage_logger_->AppendEntry(lineage_record);
- if (!status.ok()) {
- LOG(ERROR) << "Unable to record query lineage record: " << status.GetDetail();
- if (FLAGS_abort_on_failed_lineage_event) {
- CLEAN_EXIT_WITH_ERROR("Shutting down Impala Server due to "
- "abort_on_failed_lineage_event=true");
+
+ if (AreQueryHooksEnabled()) {
+ // invoke QueryEventHooks
+ TQueryCompleteContext query_complete_context;
+ query_complete_context.__set_lineage_string(lineage_record);
+ const Status& status = exec_env_->frontend()->CallQueryCompleteHooks(
+ query_complete_context);
+
+ if (!status.ok()) {
+ LOG(ERROR) << "Failed to send query lineage info to FE CallQueryCompleteHooks"
+ << status.GetDetail();
+ if (FLAGS_abort_on_failed_lineage_event) {
+ CLEAN_EXIT_WITH_ERROR("Shutting down Impala Server due to "
+ "abort_on_failed_lineage_event=true");
+ }
}
}
- return status;
+
+ // lineage logfile writing is deprecated in favor of the
+ // QueryEventHooks (see FE). this behavior is being retained
+ // for now but may be removed in the future.
+ if (IsLineageLoggingEnabled()) {
+ const Status& status = lineage_logger_->AppendEntry(lineage_record);
+ if (!status.ok()) {
+ LOG(ERROR) << "Unable to record query lineage record: " << status.GetDetail();
+ if (FLAGS_abort_on_failed_lineage_event) {
+ CLEAN_EXIT_WITH_ERROR("Shutting down Impala Server due to "
+ "abort_on_failed_lineage_event=true");
+ }
+ }
+ return status;
+ }
+ return Status::OK();
}
bool ImpalaServer::IsCoordinator() { return is_coordinator_; }
@@ -529,6 +563,10 @@ bool ImpalaServer::IsLineageLoggingEnabled() {
return !FLAGS_lineage_event_log_dir.empty();
}
+bool ImpalaServer::AreQueryHooksEnabled() {
+ return !FLAGS_query_event_hook_classes.empty();
+}
+
Status ImpalaServer::InitLineageLogging() {
if (!IsLineageLoggingEnabled()) {
LOG(INFO) << "Lineage logging is disabled";
@@ -673,7 +711,9 @@ void ImpalaServer::LogQueryEvents(const ClientRequestState& request_state) {
// TODO: deal with an error status
discard_result(LogAuditRecord(request_state, request_state.exec_request()));
}
- if (IsLineageLoggingEnabled() && log_events) {
+
+ if (log_events &&
+ (AreQueryHooksEnabled() || IsLineageLoggingEnabled())) {
// TODO: deal with an error status
discard_result(LogLineageRecord(request_state));
}
@@ -1217,6 +1257,7 @@ Status ImpalaServer::UnregisterQuery(const TUniqueId& query_id, bool check_infli
ImpaladMetrics::QUERY_DURATIONS->Update(duration_ms);
}
}
+ // TODO (IMPALA-8572): move LogQueryEvents to before query unregistration
LogQueryEvents(*request_state.get());
{
diff --git a/be/src/service/impala-server.h b/be/src/service/impala-server.h
index 5987de0..b23a518 100644
--- a/be/src/service/impala-server.h
+++ b/be/src/service/impala-server.h
@@ -377,8 +377,14 @@ class ImpalaServer : public ImpalaServiceIf,
void WaitForCatalogUpdateTopicPropagation(const TUniqueId& catalog_service_id);
/// Returns true if lineage logging is enabled, false otherwise.
+ ///
+ /// DEPRECATED: lineage file logging has been deprecated in favor of
+ /// query execution hooks (FE)
bool IsLineageLoggingEnabled();
+ /// Returns true if query execution (FE) hooks are enabled, false otherwise.
+ bool AreQueryHooksEnabled();
+
/// Retuns true if this is a coordinator, false otherwise.
bool IsCoordinator();
diff --git a/be/src/util/backend-gflag-util.cc b/be/src/util/backend-gflag-util.cc
index e6b1570..99b2926 100644
--- a/be/src/util/backend-gflag-util.cc
+++ b/be/src/util/backend-gflag-util.cc
@@ -77,6 +77,8 @@ DECLARE_string(ranger_service_type);
DECLARE_string(ranger_app_id);
DECLARE_string(authorization_provider);
DECLARE_bool(recursively_list_partitions);
+DECLARE_string(query_event_hook_classes);
+DECLARE_int32(query_event_hook_nthreads);
namespace impala {
@@ -153,6 +155,8 @@ Status GetThriftBackendGflags(JNIEnv* jni_env, jbyteArray* cfg_bytes) {
cfg.__set_ranger_app_id(FLAGS_ranger_app_id);
cfg.__set_authorization_provider(FLAGS_authorization_provider);
cfg.__set_recursively_list_partitions(FLAGS_recursively_list_partitions);
+ cfg.__set_query_event_hook_classes(FLAGS_query_event_hook_classes);
+ cfg.__set_query_event_hook_nthreads(FLAGS_query_event_hook_nthreads);
RETURN_IF_ERROR(SerializeThriftMsg(jni_env, &cfg, cfg_bytes));
return Status::OK();
}
diff --git a/common/thrift/BackendGflags.thrift b/common/thrift/BackendGflags.thrift
index c242c3e..6574faf 100644
--- a/common/thrift/BackendGflags.thrift
+++ b/common/thrift/BackendGflags.thrift
@@ -129,4 +129,8 @@ struct TBackendGflags {
51: required string authorization_provider
52: required bool recursively_list_partitions
+
+ 53: required string query_event_hook_classes
+
+ 54: required i32 query_event_hook_nthreads
}
diff --git a/common/thrift/Frontend.thrift b/common/thrift/Frontend.thrift
index 2599910..fe26ac6 100644
--- a/common/thrift/Frontend.thrift
+++ b/common/thrift/Frontend.thrift
@@ -932,3 +932,13 @@ struct TTestCaseData {
// underlying thrift layout changes.
5: required string impala_version
}
+
+// Information about a query sent to the FE QueryEventHooks
+// after query execution
+struct TQueryCompleteContext {
+ // the serialized lineage graph of the query, with optional BE-populated information
+ //
+ // this is an experimental feature and the format will likely change
+ // in a future version
+ 1: required string lineage_string
+}
diff --git a/fe/src/main/java/org/apache/impala/hooks/QueryCompleteContext.java b/fe/src/main/java/org/apache/impala/hooks/QueryCompleteContext.java
new file mode 100644
index 0000000..23ea562
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/hooks/QueryCompleteContext.java
@@ -0,0 +1,56 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.impala.hooks;
+
+import java.util.Objects;
+
+/**
+ * {@link QueryCompleteContext} encapsulates immutable information sent from the
+ * BE to a post-query hook.
+ */
+public class QueryCompleteContext {
+ private final String lineageGraph_;
+
+ public QueryCompleteContext(String lineageGraph) {
+ lineageGraph_ = Objects.requireNonNull(lineageGraph);
+ }
+
+ /**
+ * Returns the lineage graph sent from the backend during
+ * {@link QueryEventHook#onQueryComplete(QueryCompleteContext)}. This graph
+ * object will generally contain more information than it did when it was
+ * first constructed in the frontend, because the backend will have filled
+ * in additional information.
+ * <p>
+ * The returned object is a JSON representation of the lineage graph object
+ * for the query. The details of the JSON translation are not provided here
+ * as this is meant to be a temporary feature, and the String format will
+ * be changed to something more strongly-typed in the future.
+ * </p>
+ *
+ * @return lineage graph from the query that executed
+ */
+ public String getLineageGraph() { return lineageGraph_; }
+
+ @Override
+ public String toString() {
+ return "QueryCompleteContext{" +
+ "lineageGraph='" + lineageGraph_ + '\'' +
+ '}';
+ }
+}
diff --git a/fe/src/main/java/org/apache/impala/hooks/QueryEventHook.java b/fe/src/main/java/org/apache/impala/hooks/QueryEventHook.java
new file mode 100644
index 0000000..80ee5a5
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/hooks/QueryEventHook.java
@@ -0,0 +1,116 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.impala.hooks;
+
+/**
+ * {@link QueryEventHook} is the interface for implementations that
+ * can hook into supported events in Impala query execution.
+ */
+public interface QueryEventHook {
+ /**
+ * Hook method invoked when the Impala daemon starts up.
+ * <p>
+ * This method will block completion of daemon startup, so you should
+ * execute any long-running actions asynchronously.
+ * </p>
+ * <h3>Error-Handling</h3>
+ * <p>
+ * Any {@link Exception} thrown from this method will effectively fail
+ * Impala startup with an error. Implementations should handle all
+ * exceptions as gracefully as they can, even if the end result is to
+ * throw them.
+ * </p>
+ */
+ void onImpalaStartup();
+
+ /**
+ * Hook method invoked asynchronously when a (qualifying) Impala query
+ * has executed, but before it has returned.
+ * <p>
+ * This method will not block the invoking or subsequent queries,
+ * but may block future hook invocations if it runs for too long
+ * </p>
+ * <h3>Error-Handling</h3>
+ * <p>
+ * Any {@link Throwable} thrown from this method will only be caught
+ * and logged and will not affect the result of any query. Hook implementations
+ * should make a best-effort to handle their own exceptions.
+ * </p>
+ * <h3>Important:</h3>
+ * <p>
+ * This hook is actually invoked when the query is <i>unregistered</i>,
+ * which may happen a long time after the query has executed.
+ * e.g. the following sequence is possible:
+ * <ol>
+ * <li>User executes query from Hue.
+ * <li>User goes home for weekend, leaving Hue tab open in browser
+ * <li>If we're lucky, the session timeout expires after some amount of idle time.
+ * <li>The query gets unregistered, lineage record gets logged
+ * </ol>
+ * </p>
+ * <h3>Service Guarantees</h3>
+ *
+ * Impala makes the following guarantees about how this method is executed
+ * with respect to other implementations that may be registered:
+ *
+ * <h4>Hooks are executed asynchronously</h4>
+ *
+ * All hook execution happens asynchronously of the query that triggered
+ * them. Hooks may still be executing after the query response has returned
+ * to the caller. Additionally, hooks may execute concurrently if the
+ * hook executor thread size is configured appropriately.
+ *
+ * <h4>Hook Invocation is in Configuration Order</h4>
+ *
+ * The <i>submission</i> of the hook execution tasks occurs in the order
+ * that the hooks were defined in configuration. This generally means that
+ * hooks will <i>start</i> executing in order, but there are no guarantees
+ * about finishing order.
+ * <p>
+ * For example, if configured with {@code query_event_hook_classes=hook1,hook2,hook3},
+ * then hook1 will start before hook2, and hook2 will start before hook3.
+ * If you need to guarantee that hook1 <i>completes</i> before hook2 starts, then
+ * you should specify {@code query_event_hook_nthreads=1} for serial hook
+ * execution.
+ * </p>
+ *
+ * <h4>Hook Execution Blocks</h4>
+ *
+ * A hook will block the thread it executes on until it completes. If a hook hangs,
+ * then the thread also hangs. Impala (currently) will not check for hanging hooks to
+ * take any action. This means that if you have {@code query_event_hook_nthreads}
+ * less than the number of hooks, then 1 hook may effectively block others from
+ * executing.
+ *
+ * <h4>Hook Exceptions are non-fatal</h4>
+ *
+ * Any exception thrown from this hook method will be logged and ignored. Therefore,
+ * an exception in 1 hook will not affect another hook (when no shared resources are
+ * involved).
+ *
+ * <h4>Hook Execution may end abruptly at Impala shutdown</h4>
+ *
+ * If a hook is still executing when Impala is shutdown, there are no guarantees
+ * that it will complete execution before being killed.
+ *
+ *
+ * @param context object containing the post execution context
+ * of the query
+ */
+ void onQueryComplete(QueryCompleteContext context);
+}
diff --git a/fe/src/main/java/org/apache/impala/hooks/QueryEventHookManager.java b/fe/src/main/java/org/apache/impala/hooks/QueryEventHookManager.java
new file mode 100644
index 0000000..67ba808
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/hooks/QueryEventHookManager.java
@@ -0,0 +1,229 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.impala.hooks;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.impala.common.InternalException;
+import org.apache.impala.service.BackendConfig;
+import org.apache.impala.thrift.TBackendGflags;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.stream.Collectors;
+
+/**
+ * {@link QueryEventHookManager} manages the registration and execution of
+ * {@link QueryEventHook}s. Each manager instance may manage its own hooks,
+ * though the expected use-case is to have 1 instance per process, usually
+ * owned by the frontend. This class is not thread-safe.
+ *
+ * <h3>Hook Registration</h3>
+ *
+ * The hook implementation(s) to use at runtime are specified through the
+ * backend config flag {@link TBackendGflags#query_event_hook_classes}
+ * at Impala startup. See {@link #createFromConfig(BackendConfig)}.
+ *
+ * <h3>Hook Classloading</h3>
+ *
+ * Each hook implementation is loaded using `this` manager's classloader; no
+ * classloader isolation is performed. Individual hook implementations should
+ * take care to properly handle any dependencies they bring in to avoid shadowing
+ * existing dependencies on the Impala classpath.
+ *
+ * <h3>Hook Execution</h3>
+ *
+ * Hook initialization ({@link QueryEventHook#onImpalaStartup()} is
+ * performed synchronously during {@link #createFromConfig(BackendConfig)}.
+ * <p>
+ * {@link QueryEventHook#onQueryComplete(QueryCompleteContext)} is performed
+ * asynchronously during {@link #executeQueryCompleteHooks(QueryCompleteContext)}.
+ * This execution is performed by a thread-pool executor, whose size is set at
+ * compile-time. This means that hooks may also execute concurrently.
+ * </p>
+ *
+ */
+public class QueryEventHookManager {
+ private static final Logger LOG =
+ LoggerFactory.getLogger(QueryEventHookManager.class);
+
+ // TODO: figure out a way to source these from the defn so
+ // we don't have to manually sync when they change
+ private static final String BE_HOOKS_FLAG = "query_event_hook_classes";
+ private static final String BE_HOOKS_THREADS_FLAG = "query_event_hook_nthreads";
+
+ private final List<QueryEventHook> hooks_;
+ private final ExecutorService hookExecutor_;
+
+ /**
+ * Static factory method to create a manager instance. This will register
+ * all {@link QueryEventHook}s specified by the backend config flag
+ * {@code query_event_hook_classes} and then invoke their
+ * {@link QueryEventHook#onImpalaStartup()} methods synchronously.
+ *
+ * @throws IllegalArgumentException if config is invalid
+ * @throws InternalException if any hook could not be instantiated
+ * @throws InternalException if any hook.onImpalaStartup() throws an exception
+ */
+ public static QueryEventHookManager createFromConfig(BackendConfig config)
+ throws InternalException {
+
+ final int nHookThreads = config.getNumQueryExecHookThreads();
+ final String queryExecHookClasses = config.getQueryExecHookClasses();
+ LOG.info("QueryEventHook config:");
+ LOG.info("- {}={}", BE_HOOKS_THREADS_FLAG, nHookThreads);
+ LOG.info("- {}={}", BE_HOOKS_FLAG, queryExecHookClasses);
+
+ final String[] hookClasses;
+ if (StringUtils.isNotEmpty(queryExecHookClasses)) {
+ hookClasses = queryExecHookClasses.split("\\s*,\\s*");
+ } else {
+ hookClasses = new String[0];
+ }
+
+ return new QueryEventHookManager(nHookThreads, hookClasses);
+ }
+
+ /**
+ * Instantiates a manager with a fixed-size thread-pool executor for
+ * executing {@link QueryEventHook#onQueryComplete(QueryCompleteContext)}.
+ *
+ * @param nHookExecutorThreads
+ * @param hookClasses
+ *
+ * @throws IllegalArgumentException if {@code nHookExecutorThreads <= 0}
+ * @throws InternalException if any hookClass cannot be instantiated
+ * @throws InternalException if any hookClass.onImpalaStartup throws an exception
+ */
+ private QueryEventHookManager(int nHookExecutorThreads, String[] hookClasses)
+ throws InternalException {
+
+ this.hookExecutor_ = Executors.newFixedThreadPool(nHookExecutorThreads);
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> this.cleanUp()));
+
+ final List<QueryEventHook> hooks = new ArrayList<>(hookClasses.length);
+ this.hooks_ = Collections.unmodifiableList(hooks);
+
+ for (String postExecHook : hookClasses) {
+ final QueryEventHook hook;
+ try {
+ final Class<QueryEventHook> clsHook =
+ (Class<QueryEventHook>) Class.forName(postExecHook);
+ hook = clsHook.newInstance();
+ } catch (InstantiationException
+ | IllegalAccessException
+ | ClassNotFoundException e) {
+ final String msg = String.format(
+ "Unable to instantiate query event hook class %s. Please check %s config",
+ postExecHook, BE_HOOKS_FLAG);
+ LOG.error(msg, e);
+ throw new InternalException(msg, e);
+ }
+
+ hooks.add(hook);
+ }
+
+ for (QueryEventHook hook : hooks) {
+ try {
+ LOG.debug("Initiating hook.onImpalaStartup for {}", hook.getClass().getName());
+ hook.onImpalaStartup();
+ }
+ catch (Exception e) {
+ final String msg = String.format(
+ "Exception during onImpalaStartup from QueryEventHook %s instance=%s",
+ hook.getClass(), hook);
+ LOG.error(msg, e);
+ throw new InternalException(msg, e);
+ }
+ }
+ }
+
+ private void cleanUp() {
+ if (!hookExecutor_.isShutdown()) {
+ hookExecutor_.shutdown();
+ }
+ // TODO (IMPALA-8571): we may want to await termination (up to a timeout)
+ // to ensure that hooks have a chance to complete execution. Executor
+ // threads will typically run to completion after executor shutdown, but
+ // there are some instances where this doesnt hold. e.g.
+ //
+ // - executor thread is sleeping when shutdown is called
+ // - system.exit called
+ }
+
+ /**
+ * Returns an unmodifiable view of all the {@link QueryEventHook}s
+ * registered at construction.
+ *
+ * @return unmodifiable view of all currently-registered hooks
+ */
+ public List<QueryEventHook> getHooks() {
+ return hooks_;
+ }
+
+ /**
+ * Hook method to be called after query execution. This implementation
+ * will execute all currently-registered {@link QueryEventHook}s
+ * asynchronously, returning immediately with a List of {@link Future}s
+ * representing each hook's {@link QueryEventHook#onQueryComplete(QueryCompleteContext)}
+ * invocation.
+ *
+ * <h3>Futures</h3>
+ *
+ * This method will return a list of {@link Future}s representing the future results
+ * of each hook's invocation. The {@link Future#get()} method will return the
+ * hook instance whose invocation it represents. The list of futures are in the
+ * same order as the order in which each hook's job was submitted.
+ *
+ * <h3>Error-Handling</h3>
+ *
+ * Exceptions thrown from {@link QueryEventHook#onQueryComplete(QueryCompleteContext)}
+ * will be logged and then rethrown on the executor thread(s), meaning that they
+ * will not halt execution. Rather, they will be encapsulated in the returned
+ * {@link Future}s, meaning that the caller may choose to check or ignore them
+ * at some later time.
+ *
+ * @param context
+ */
+ public List<Future<QueryEventHook>> executeQueryCompleteHooks(
+ QueryCompleteContext context) {
+ LOG.debug("Query complete hook invoked with: {}", context);
+ return hooks_.stream().map(hook -> {
+ LOG.debug("Initiating onQueryComplete: {}", hook.getClass().getName());
+ return hookExecutor_.submit(() -> {
+ try {
+ hook.onQueryComplete(context);
+ } catch (Throwable t) {
+ final String msg = String.format("Exception thrown by QueryEventHook %s"+
+ ".onQueryComplete method. Hook instance %s. This exception is "+
+ "currently being ignored by Impala, "+
+ "but may cause subsequent problems in that hook's execution",
+ hook.getClass().getName(), hook);
+ LOG.error(msg, t);
+ throw t;
+ }
+ return hook;
+ });
+ }).collect(Collectors.toList());
+ }
+}
diff --git a/fe/src/main/java/org/apache/impala/service/BackendConfig.java b/fe/src/main/java/org/apache/impala/service/BackendConfig.java
index 97e0b64..44bf0a9 100644
--- a/fe/src/main/java/org/apache/impala/service/BackendConfig.java
+++ b/fe/src/main/java/org/apache/impala/service/BackendConfig.java
@@ -161,6 +161,14 @@ public class BackendConfig {
return backendCfg_.getAuthorization_provider();
}
+ public String getQueryExecHookClasses() {
+ return backendCfg_.getQuery_event_hook_classes();
+ }
+
+ public int getNumQueryExecHookThreads() {
+ return backendCfg_.getQuery_event_hook_nthreads();
+ }
+
// Inits the auth_to_local configuration in the static KerberosName class.
private static void initAuthToLocal() {
// If auth_to_local is enabled, we read the configuration hadoop.security.auth_to_local
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 267458b..0c1ff46 100644
--- a/fe/src/main/java/org/apache/impala/service/Frontend.java
+++ b/fe/src/main/java/org/apache/impala/service/Frontend.java
@@ -26,6 +26,7 @@ import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
@@ -92,6 +93,9 @@ import org.apache.impala.common.ImpalaException;
import org.apache.impala.common.InternalException;
import org.apache.impala.common.NotImplementedException;
import org.apache.impala.compat.MetastoreShim;
+import org.apache.impala.hooks.QueryCompleteContext;
+import org.apache.impala.hooks.QueryEventHook;
+import org.apache.impala.hooks.QueryEventHookManager;
import org.apache.impala.planner.HdfsScanNode;
import org.apache.impala.planner.PlanFragment;
import org.apache.impala.planner.Planner;
@@ -156,7 +160,7 @@ import com.google.common.util.concurrent.Uninterruptibles;
* Frontend API for the impalad process.
* This class allows the impala daemon to create TQueryExecRequest
* in response to TClientRequests. Also handles management of the authorization
- * policy.
+ * policy and query execution hooks.
*/
public class Frontend {
private final static Logger LOG = LoggerFactory.getLogger(Frontend.class);
@@ -232,6 +236,8 @@ public class Frontend {
private final ImpaladTableUsageTracker impaladTableUsageTracker_;
+ private final QueryEventHookManager queryHookManager_;
+
public Frontend(AuthorizationFactory authzFactory) throws ImpalaException {
this(authzFactory, FeCatalogManager.createFromBackendConfig());
}
@@ -263,6 +269,7 @@ public class Frontend {
authzChecker_::get);
impaladTableUsageTracker_ = ImpaladTableUsageTracker.createFromConfig(
BackendConfig.INSTANCE);
+ queryHookManager_ = QueryEventHookManager.createFromConfig(BackendConfig.INSTANCE);
}
public FeCatalog getCatalog() { return catalogManager_.getOrCreateCatalog(); }
@@ -1511,4 +1518,66 @@ public class Frontend {
"Unsupported table class: " + table.getClass());
}
}
+
+ /**
+ * Executes the {@link QueryEventHook#onQueryComplete(QueryCompleteContext)}
+ * execution hooks for each hook registered in this instance's
+ * {@link QueryEventHookManager}.
+ *
+ * <h3>Service Guarantees</h3>
+ *
+ * Impala makes the following guarantees about how this method executes hooks:
+ *
+ * <h4>Hooks are executed asynchronously</h4>
+ *
+ * All hook execution happens asynchronously of the query that triggered
+ * them. Hooks may still be executing after the query response has returned
+ * to the caller. Additionally, hooks may execute concurrently if the
+ * hook executor thread size is configured appropriately.
+ *
+ * <h4>Hook Invocation is in Configuration Order</h4>
+ *
+ * The <i>submission</i> of the hook execution tasks occurs in the order
+ * that the hooks were defined in configuration. This generally means that
+ * hooks will <i>start</i> executing in order, but there are no guarantees
+ * about finishing order.
+ * <p>
+ * For example, if configured with {@code query_event_hook_classes=hook1,hook2,hook3},
+ * then hook1 will start before hook2, and hook2 will start before hook3.
+ * If you need to guarantee that hook1 <i>completes</i> before hook2 starts, then
+ * you should specify {@code query_event_hook_nthreads=1} for serial hook
+ * execution.
+ * </p>
+ *
+ * <h4>Hook Execution Blocks</h4>
+ *
+ * A hook will block the thread it executes on until it completes. If a hook hangs,
+ * then the thread also hangs. Impala (currently) will not check for hanging hooks to
+ * take any action. This means that if you have {@code query_event_hook_nthreads}
+ * less than the number of hooks, then 1 hook may effectively block others from
+ * executing.
+ *
+ * <h4>Hook Exceptions are non-fatal</h4>
+ *
+ * Any exception thrown from this hook method will be logged and ignored. Therefore,
+ * an exception in 1 hook will not affect another hook (when no shared resources are
+ * involved).
+ *
+ * <h4>Hook Execution may end abruptly at Impala shutdown</h4>
+ *
+ * If a hook is still executing when Impala is shutdown, there are no guarantees
+ * that it will complete execution before being killed.
+ *
+ * @see QueryCompleteContext
+ * @see QueryEventHookManager
+ *
+ * @param context the execution context of the query
+ */
+ public void callQueryCompleteHooks(QueryCompleteContext context) {
+ // TODO (IMPALA-8571): can we make use of the futures to implement better
+ // error-handling? Currently, the queryHookManager simply
+ // logs-then-rethrows any exception thrown from a hook.postQueryExecute
+ final List<Future<QueryEventHook>> futures
+ = this.queryHookManager_.executeQueryCompleteHooks(context);
+ }
}
diff --git a/fe/src/main/java/org/apache/impala/service/JniFrontend.java b/fe/src/main/java/org/apache/impala/service/JniFrontend.java
index a798bba..631f9bd 100644
--- a/fe/src/main/java/org/apache/impala/service/JniFrontend.java
+++ b/fe/src/main/java/org/apache/impala/service/JniFrontend.java
@@ -17,23 +17,19 @@
package org.apache.impala.service;
-import java.io.File;
-import java.io.IOException;
-import java.util.Collections;
-import java.util.Enumeration;
-import java.util.List;
-import java.util.Map;
-
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.CommonConfigurationKeys;
import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.s3a.S3AFileSystem;
+import org.apache.hadoop.fs.adl.AdlFileSystem;
import org.apache.hadoop.fs.azurebfs.AzureBlobFileSystem;
import org.apache.hadoop.fs.azurebfs.SecureAzureBlobFileSystem;
-import org.apache.hadoop.fs.adl.AdlFileSystem;
+import org.apache.hadoop.fs.s3a.S3AFileSystem;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.security.Groups;
@@ -43,11 +39,8 @@ import org.apache.hadoop.security.ShellBasedUnixGroupsMapping;
import org.apache.hadoop.security.ShellBasedUnixGroupsNetgroupMapping;
import org.apache.impala.analysis.DescriptorTable;
import org.apache.impala.analysis.ToSqlUtils;
-import org.apache.impala.authorization.AuthorizationConfig;
import org.apache.impala.authorization.AuthorizationFactory;
-import org.apache.impala.authorization.AuthorizationProvider;
import org.apache.impala.authorization.ImpalaInternalAdminUser;
-import org.apache.impala.authorization.NoopAuthorizationFactory;
import org.apache.impala.authorization.User;
import org.apache.impala.catalog.FeDataSource;
import org.apache.impala.catalog.FeDb;
@@ -58,6 +51,7 @@ import org.apache.impala.common.FileSystemUtil;
import org.apache.impala.common.ImpalaException;
import org.apache.impala.common.InternalException;
import org.apache.impala.common.JniUtil;
+import org.apache.impala.hooks.QueryCompleteContext;
import org.apache.impala.service.Frontend.PlanCtx;
import org.apache.impala.thrift.TBackendGflags;
import org.apache.impala.thrift.TBuildTestDescriptorTableParams;
@@ -88,6 +82,7 @@ import org.apache.impala.thrift.TLoadDataReq;
import org.apache.impala.thrift.TLoadDataResp;
import org.apache.impala.thrift.TLogLevel;
import org.apache.impala.thrift.TMetadataOpRequest;
+import org.apache.impala.thrift.TQueryCompleteContext;
import org.apache.impala.thrift.TQueryCtx;
import org.apache.impala.thrift.TResultSet;
import org.apache.impala.thrift.TShowFilesParams;
@@ -110,9 +105,12 @@ import org.apache.thrift.protocol.TBinaryProtocol;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
+import java.io.File;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Enumeration;
+import java.util.List;
+import java.util.Map;
/**
* JNI-callable interface onto a wrapped Frontend instance. The main point is to serialise
@@ -124,6 +122,7 @@ public class JniFrontend {
new TBinaryProtocol.Factory();
private final Frontend frontend_;
+
/**
* Create a new instance of the Jni Frontend.
*/
@@ -625,6 +624,20 @@ public class JniFrontend {
}
/**
+ * JNI wrapper for {@link Frontend#callQueryCompleteHooks(QueryCompleteContext)}.
+ *
+ * @param serializedRequest
+ */
+ public void callQueryCompleteHooks(byte[] serializedRequest) throws ImpalaException {
+ final TQueryCompleteContext request = new TQueryCompleteContext();
+ JniUtil.deserializeThrift(protocolFactory_, request, serializedRequest);
+
+ final QueryCompleteContext context =
+ new QueryCompleteContext(request.getLineage_string());
+ this.frontend_.callQueryCompleteHooks(context);
+ }
+
+ /**
* Returns an error string describing configuration issue with the groups mapping
* provider implementation.
*/
diff --git a/fe/src/test/java/org/apache/impala/hooks/QueryEventHookManagerTest.java b/fe/src/test/java/org/apache/impala/hooks/QueryEventHookManagerTest.java
new file mode 100644
index 0000000..efbb0fb
--- /dev/null
+++ b/fe/src/test/java/org/apache/impala/hooks/QueryEventHookManagerTest.java
@@ -0,0 +1,146 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.impala.hooks;
+
+import org.apache.impala.common.InternalException;
+import org.apache.impala.service.BackendConfig;
+import org.apache.impala.testutil.AlwaysErrorQueryEventHook;
+import org.apache.impala.testutil.CountingQueryEventHook;
+import org.apache.impala.testutil.PostQueryErrorEventHook;
+import org.apache.impala.thrift.TBackendGflags;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+import java.util.List;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertEquals;
+
+public class QueryEventHookManagerTest {
+ private TBackendGflags origFlags;
+
+ @Rule
+ public ExpectedException expectedException = ExpectedException.none();
+ private QueryCompleteContext mockQueryCompleteContext =
+ new QueryCompleteContext("unit-test lineage");
+
+ @Before
+ public void setUp() {
+ // since some test cases will need to modify the (static)
+ // be flags, we need to save the original values so they
+ // can be restored and not break other tests
+ if (BackendConfig.INSTANCE == null) {
+ BackendConfig.create(new TBackendGflags());
+ }
+ origFlags = BackendConfig.INSTANCE.getBackendCfg();
+ }
+
+ @After
+ public void tearDown() {
+ BackendConfig.create(origFlags);
+ }
+
+ private static QueryEventHookManager createQueryEventHookManager(int nThreads,
+ String... hooks) throws Exception {
+ if (hooks.length == 0) {
+ BackendConfig.INSTANCE.getBackendCfg().setQuery_event_hook_classes("");
+ } else {
+ BackendConfig.INSTANCE.getBackendCfg().setQuery_event_hook_classes(
+ String.join(",", hooks));
+ }
+
+ BackendConfig.INSTANCE.getBackendCfg().setQuery_event_hook_nthreads(nThreads);
+
+ return QueryEventHookManager.createFromConfig(BackendConfig.INSTANCE);
+ }
+
+ @Test
+ public void testHookRegistration() throws Exception {
+ final QueryEventHookManager mgr = createQueryEventHookManager(1,
+ CountingQueryEventHook.class.getCanonicalName(),
+ CountingQueryEventHook.class.getCanonicalName());
+
+ final List<QueryEventHook> hooks = mgr.getHooks();
+ assertEquals(2, hooks.size());
+ hooks.forEach(h -> assertEquals(CountingQueryEventHook.class, h.getClass()));
+ }
+
+ @Test
+ public void testHookPostQueryExecuteErrorsDoNotKillExecution() throws Exception {
+ // a hook that exceptions should not prevent a subsequent hook from executing
+ final QueryEventHookManager mgr = createQueryEventHookManager(1,
+ PostQueryErrorEventHook.class.getCanonicalName(),
+ CountingQueryEventHook.class.getCanonicalName());
+
+ // make sure error hook will execute first
+ assertEquals(mgr.getHooks().get(0).getClass(), PostQueryErrorEventHook.class);
+
+ final List<Future<QueryEventHook>> futures =
+ mgr.executeQueryCompleteHooks(mockQueryCompleteContext);
+
+ // this should not exception
+ final QueryEventHook hookImpl = futures.get(1).get(2, TimeUnit.SECONDS);
+
+ assertEquals(hookImpl.getClass(), CountingQueryEventHook.class);
+ }
+
+ @Test
+ public void testHookExceptionDuringStartupKillsStartup() throws Exception {
+ expectedException.expect(InternalException.class);
+
+ createQueryEventHookManager(1,
+ AlwaysErrorQueryEventHook.class.getCanonicalName(),
+ CountingQueryEventHook.class.getCanonicalName());
+ }
+
+ @Test
+ public void testHookPostQueryExecuteInvokedCorrectly() throws Exception {
+ final QueryEventHookManager mgr = createQueryEventHookManager(1,
+ CountingQueryEventHook.class.getCanonicalName(),
+ CountingQueryEventHook.class.getCanonicalName());
+
+ List<Future<QueryEventHook>> futures =
+ mgr.executeQueryCompleteHooks(mockQueryCompleteContext);
+
+ assertEquals(
+ futures.size(),
+ mgr.getHooks().size());
+
+ for (Future<QueryEventHook> f : futures) {
+ CountingQueryEventHook hook = (CountingQueryEventHook) f.get(2, TimeUnit.SECONDS);
+ assertEquals(1, hook.getPostQueryExecuteInvocations());
+ }
+
+ futures = mgr.executeQueryCompleteHooks(mockQueryCompleteContext);
+
+ assertEquals(
+ futures.size(),
+ mgr.getHooks().size());
+
+ for (Future<QueryEventHook> f : futures) {
+ CountingQueryEventHook hook = (CountingQueryEventHook) f.get(2, TimeUnit.SECONDS);
+ assertEquals(2, hook.getPostQueryExecuteInvocations());
+ }
+ }
+
+}
+
diff --git a/fe/src/test/java/org/apache/impala/testutil/AlwaysErrorQueryEventHook.java b/fe/src/test/java/org/apache/impala/testutil/AlwaysErrorQueryEventHook.java
new file mode 100644
index 0000000..cbdfb1d
--- /dev/null
+++ b/fe/src/test/java/org/apache/impala/testutil/AlwaysErrorQueryEventHook.java
@@ -0,0 +1,33 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.impala.testutil;
+
+import org.apache.impala.hooks.QueryCompleteContext;
+import org.apache.impala.hooks.QueryEventHook;
+
+public class AlwaysErrorQueryEventHook implements QueryEventHook {
+ @Override
+ public void onImpalaStartup() {
+ throw new RuntimeException("intentional error");
+ }
+
+ @Override
+ public void onQueryComplete(QueryCompleteContext context) {
+ throw new RuntimeException("intentional error");
+ }
+}
diff --git a/fe/src/test/java/org/apache/impala/testutil/CountingQueryEventHook.java b/fe/src/test/java/org/apache/impala/testutil/CountingQueryEventHook.java
new file mode 100644
index 0000000..711255a
--- /dev/null
+++ b/fe/src/test/java/org/apache/impala/testutil/CountingQueryEventHook.java
@@ -0,0 +1,52 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.impala.testutil;
+
+import org.apache.impala.hooks.QueryCompleteContext;
+import org.apache.impala.hooks.QueryEventHook;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class CountingQueryEventHook implements QueryEventHook {
+ private final AtomicInteger startupCount = new AtomicInteger(0);
+ private final AtomicInteger postQueryCount = new AtomicInteger(0);
+
+ @Override
+ public void onImpalaStartup() {
+ startupCount.incrementAndGet();
+ }
+
+ @Override
+ public void onQueryComplete(QueryCompleteContext context) {
+ postQueryCount.incrementAndGet();
+ }
+
+ /**
+ * @return # of times postQueryExecute has been invoked since construction
+ */
+ public int getImpalaStartupInvocations() {
+ return startupCount.get();
+ }
+
+ /**
+ * @return # of times postQueryExecute has been invoked since construction
+ */
+ public int getPostQueryExecuteInvocations() {
+ return postQueryCount.get();
+ }
+}
diff --git a/fe/src/test/java/org/apache/impala/testutil/DummyQueryEventHook.java b/fe/src/test/java/org/apache/impala/testutil/DummyQueryEventHook.java
new file mode 100644
index 0000000..72b224f
--- /dev/null
+++ b/fe/src/test/java/org/apache/impala/testutil/DummyQueryEventHook.java
@@ -0,0 +1,53 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.impala.testutil;
+
+import org.apache.impala.hooks.QueryCompleteContext;
+import org.apache.impala.hooks.QueryEventHook;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+
+public class DummyQueryEventHook implements QueryEventHook {
+ private static final Logger LOG = LoggerFactory.getLogger(DummyQueryEventHook.class);
+
+ @Override
+ public void onImpalaStartup() {
+ LOG.info("{}.onImpalaStartup", this.getClass().getName());
+ try {
+ Files.write(Paths.get("/tmp/" + this.getClass().getName() + ".onImpalaStartup"),
+ "onImpalaStartup invoked".getBytes());
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public void onQueryComplete(QueryCompleteContext context) {
+ LOG.info("{}.onQueryComplete", this.getClass().getName());
+ try {
+ Files.write(Paths.get("/tmp/" + this.getClass().getName() + ".onQueryComplete"),
+ "onQueryComplete invoked".getBytes());
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+}
diff --git a/fe/src/test/java/org/apache/impala/testutil/PostQueryErrorEventHook.java b/fe/src/test/java/org/apache/impala/testutil/PostQueryErrorEventHook.java
new file mode 100644
index 0000000..6712956
--- /dev/null
+++ b/fe/src/test/java/org/apache/impala/testutil/PostQueryErrorEventHook.java
@@ -0,0 +1,32 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.impala.testutil;
+
+import org.apache.impala.hooks.QueryCompleteContext;
+import org.apache.impala.hooks.QueryEventHook;
+
+public class PostQueryErrorEventHook implements QueryEventHook {
+ @Override
+ public void onImpalaStartup() {
+ }
+
+ @Override
+ public void onQueryComplete(QueryCompleteContext context) {
+ throw new RuntimeException("intentional error");
+ }
+}
diff --git a/tests/authorization/test_provider.py b/tests/authorization/test_provider.py
index 9c7ad2d..4d0e671 100644
--- a/tests/authorization/test_provider.py
+++ b/tests/authorization/test_provider.py
@@ -65,7 +65,7 @@ class TestAuthorizationProvider(CustomClusterTestSuite):
try:
super(TestAuthorizationProvider, self).setup_method(method)
except Exception:
- pass
+ self._stop_impala_cluster()
def teardown_method(self, method):
# Explicitly override CustomClusterTestSuite.teardown_method() to
@@ -74,4 +74,4 @@ class TestAuthorizationProvider(CustomClusterTestSuite):
try:
super(TestAuthorizationProvider, self).teardown_method(method)
except Exception:
- pass
+ self._stop_impala_cluster()
diff --git a/tests/custom_cluster/test_query_event_hooks.py b/tests/custom_cluster/test_query_event_hooks.py
new file mode 100644
index 0000000..56ad5e4
--- /dev/null
+++ b/tests/custom_cluster/test_query_event_hooks.py
@@ -0,0 +1,202 @@
+# 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.
+#
+# Client tests for Query Event Hooks
+
+import os
+import time
+import pytest
+import tempfile
+import logging
+
+from getpass import getuser
+from tests.common.custom_cluster_test_suite import CustomClusterTestSuite
+from tests.common.file_utils import assert_file_in_dir_contains
+
+LOG = logging.getLogger(__name__)
+
+
+class TestHooks(CustomClusterTestSuite):
+ """
+ Tests for FE QueryEventHook invocations.
+
+ All test cases in this test suite share an impala log dir, so keep that in mind
+ if parsing any logs during your test assertions.
+ """
+ DUMMY_HOOK = "org.apache.impala.testutil.DummyQueryEventHook"
+ HOOK_POSTEXEC_FILE = "/tmp/{0}.onQueryComplete".format(DUMMY_HOOK)
+ HOOK_START_FILE = "/tmp/{0}.onImpalaStartup".format(DUMMY_HOOK)
+ MINIDUMP_PATH = tempfile.mkdtemp()
+ IMPALA_LOG_DIR = tempfile.mkdtemp(prefix="test_hooks_", dir=os.getenv("LOG_DIR"))
+
+ def teardown(self):
+ try:
+ os.remove(TestHooks.HOOK_START_FILE)
+ except OSError:
+ pass
+
+ def setup_method(self, method):
+ # Explicitly override CustomClusterTestSuite.setup_method() to
+ # clean up the dummy hook's onQueryComplete file
+ try:
+ os.remove(TestHooks.HOOK_START_FILE)
+ os.remove(TestHooks.HOOK_POSTEXEC_FILE)
+ except OSError:
+ pass
+ super(TestHooks, self).setup_method(method)
+
+ def teardown_method(self, method):
+ # Explicitly override CustomClusterTestSuite.teardown_method() to
+ # clean up the dummy hook's onQueryComplete file
+ super(TestHooks, self).teardown_method(method)
+ try:
+ os.remove(TestHooks.HOOK_START_FILE)
+ os.remove(TestHooks.HOOK_POSTEXEC_FILE)
+ except OSError:
+ pass
+
+ @pytest.mark.xfail(run=False, reason="IMPALA-8589")
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(
+ impala_log_dir=IMPALA_LOG_DIR,
+ impalad_args="--query_event_hook_classes={0} "
+ "--minidump_path={1}"
+ .format(DUMMY_HOOK, MINIDUMP_PATH),
+ catalogd_args="--minidump_path={0}".format(MINIDUMP_PATH))
+ def test_query_event_hooks_execute(self, unique_database):
+ """
+ Tests that the post query execution hook actually executes by using a
+ dummy hook implementation that writes a file and testing for existence
+ of that file.
+
+ This test will perform queries needed to induce a hook event but should
+ clean up the db before completion.
+ """
+ user = getuser()
+ impala_client = self.create_impala_client()
+
+ # create closure over variables so we don't have to repeat them
+ def query(sql):
+ return impala_client.execute(sql, user=user)
+
+ # hook should be invoked at daemon startup
+ assert self.__wait_for_file(TestHooks.HOOK_START_FILE, timeout_s=10)
+
+ query("create table {0}.recipes (recipe_id int, recipe_name string)"
+ .format(unique_database))
+ query("insert into {0}.recipes (recipe_id, recipe_name) values "
+ "(1,'Tacos'), (2,'Tomato Soup'), (3,'Grilled Cheese')".format(unique_database))
+
+ # TODO (IMPALA-8572): need to end the session to trigger
+ # query unregister and therefore hook invocation. can possibly remove
+ # after IMPALA-8572 completes
+ impala_client.close()
+
+ # dummy hook should create a file
+ assert self.__wait_for_file(TestHooks.HOOK_POSTEXEC_FILE, timeout_s=10)
+
+ def __wait_for_file(self, filepath, timeout_s=10):
+ if timeout_s < 0: return False
+ LOG.info("Waiting {0} s for file {1}".format(timeout_s, filepath))
+ for i in range(0, timeout_s):
+ LOG.info("Poll #{0} for file {1}".format(i, filepath))
+ if os.path.isfile(filepath):
+ LOG.info("Found file {0}".format(filepath))
+ return True
+ else:
+ time.sleep(1)
+ LOG.info("File {0} not found".format(filepath))
+ return False
+
+
+class TestHooksStartupFail(CustomClusterTestSuite):
+ """
+ Tests for failed startup due to bad QueryEventHook startup or registration
+
+ All test cases in this testsuite@pytest.mark.xfail(run=False, reason="IMPALA-8589")
+ are expected to fail cluster startup and will swallow exceptions thrown during
+ setup_method().
+ """
+
+ FAILING_HOOK = "org.apache.impala.testutil.AlwaysErrorQueryEventHook"
+ NONEXIST_HOOK = "captain.hook"
+ LOG_DIR1 = tempfile.mkdtemp(prefix="test_hooks_startup_fail_", dir=os.getenv("LOG_DIR"))
+ LOG_DIR2 = tempfile.mkdtemp(prefix="test_hooks_startup_fail_", dir=os.getenv("LOG_DIR"))
+ MINIDUMP_PATH = tempfile.mkdtemp()
+
+ @pytest.mark.xfail(run=False, reason="IMPALA-8589")
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(
+ impala_log_dir=LOG_DIR1,
+ impalad_args="--query_event_hook_classes={0} "
+ "--minidump_path={1}"
+ .format(FAILING_HOOK, MINIDUMP_PATH),
+ catalogd_args="--minidump_path={0}".format(MINIDUMP_PATH))
+ def test_hook_startup_fail(self):
+ """
+ Tests that exception during QueryEventHook.onImpalaStart will prevent
+ successful daemon startup.
+
+ This is done by parsing the impala log file for a specific message
+ so is kind of brittle and maybe even prone to flakiness, depending
+ on how the log file is flushed.
+ """
+ # parse log file for expected exception
+ assert_file_in_dir_contains(TestHooksStartupFail.LOG_DIR1,
+ "Exception during onImpalaStartup from "
+ "QueryEventHook class {0}"
+ .format(TestHooksStartupFail.FAILING_HOOK))
+
+ @pytest.mark.xfail(run=False, reason="IMPALA-8589")
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(
+ impala_log_dir=LOG_DIR2,
+ impalad_args="--query_event_hook_classes={0} "
+ "--minidump_path={1}"
+ .format(NONEXIST_HOOK, MINIDUMP_PATH),
+ catalogd_args="--minidump_path={0}".format(MINIDUMP_PATH))
+ def test_hook_instantiation_fail(self):
+ """
+ Tests that failure to instantiate a QueryEventHook will prevent
+ successful daemon startup.
+
+ This is done by parsing the impala log file for a specific message
+ so is kind of brittle and maybe even prone to flakiness, depending
+ on how the log file is flushed.
+ """
+ # parse log file for expected exception
+ assert_file_in_dir_contains(TestHooksStartupFail.LOG_DIR2,
+ "Unable to instantiate query event hook class {0}"
+ .format(TestHooksStartupFail.NONEXIST_HOOK))
+
+ def setup_method(self, method):
+ # Explicitly override CustomClusterTestSuite.setup_method() to
+ # allow it to exception, since this test suite is for cases where
+ # startup fails
+ try:
+ super(TestHooksStartupFail, self).setup_method(method)
+ except Exception:
+ self._stop_impala_cluster()
+
+ def teardown_method(self, method):
+ # Explicitly override CustomClusterTestSuite.teardown_method() to
+ # allow it to exception, since it relies on setup_method() having
+ # completed successfully
+ try:
+ super(TestHooksStartupFail, self).teardown_method(method)
+ except Exception:
+ self._stop_impala_cluster()