You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by vi...@apache.org on 2020/04/17 13:22:47 UTC
[impala] branch master updated: IMPALA-9663: Fix for NPE in fire
listener events
This is an automated email from the ASF dual-hosted git repository.
vihangk1 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 6fcc758 IMPALA-9663: Fix for NPE in fire listener events
6fcc758 is described below
commit 6fcc758c1aa0b6075738edb76371a0cc43d04388
Author: Vihang Karajgaonkar <vi...@apache.org>
AuthorDate: Thu Apr 16 18:09:23 2020 -0700
IMPALA-9663: Fix for NPE in fire listener events
IMPALA-8632 introduced a improvement to Events processor which issues a
new HMS API call to fire INSERT events when table and partitions are
inserted into. However, if there is no new events generated HMS does
not set the eventIds field in the response. This fix adds a check to
not fire the listener events in such cases. Also, adds a check to make
sure that eventIds are set in the response.
Testing:
1. Added a test sql in test_event_processing.py which was failing with
a NPE before the fix.
Change-Id: Ibfcc5acd598fb0354a5a8288df7c495359f9e53d
Reviewed-on: http://gerrit.cloudera.org:8080/15745
Reviewed-by: Anurag Mantripragada <an...@cloudera.com>
Reviewed-by: Xiaomeng Zhang <xi...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Reviewed-by: Quanlong Huang <hu...@gmail.com>
---
.../java/org/apache/impala/compat/MetastoreShim.java | 17 ++++++++++++-----
.../org/apache/impala/service/CatalogOpExecutor.java | 2 ++
tests/custom_cluster/test_event_processing.py | 16 +++++++++++++++-
3 files changed, 29 insertions(+), 6 deletions(-)
diff --git a/fe/src/compat-hive-3/java/org/apache/impala/compat/MetastoreShim.java b/fe/src/compat-hive-3/java/org/apache/impala/compat/MetastoreShim.java
index 5981b2c..2091fbc 100644
--- a/fe/src/compat-hive-3/java/org/apache/impala/compat/MetastoreShim.java
+++ b/fe/src/compat-hive-3/java/org/apache/impala/compat/MetastoreShim.java
@@ -88,8 +88,6 @@ import org.apache.hive.service.rpc.thrift.TGetFunctionsReq;
import org.apache.hive.service.rpc.thrift.TGetSchemasReq;
import org.apache.hive.service.rpc.thrift.TGetTablesReq;
import org.apache.impala.authorization.User;
-import org.apache.impala.catalog.CatalogServiceCatalog;
-import org.apache.impala.catalog.HdfsPartition;
import org.apache.impala.catalog.MetaStoreClientPool.MetaStoreClient;
import org.apache.impala.common.ImpalaException;
import org.apache.impala.common.ImpalaRuntimeException;
@@ -103,17 +101,19 @@ import org.apache.impala.thrift.TResultSet;
import org.apache.impala.util.AcidUtils;
import org.apache.impala.util.AcidUtils.TblTransaction;
import org.apache.impala.util.MetaStoreUtil.InsertEventInfo;
-import org.apache.log4j.Logger;
import org.apache.thrift.TException;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
/**
* A wrapper around some of Hive's Metastore API's to abstract away differences
* between major versions of Hive. This implements the shimmed methods for Hive 3.
*/
public class MetastoreShim {
- private static final Logger LOG = Logger.getLogger(MetastoreShim.class);
+ private static final Logger LOG = LoggerFactory.getLogger(MetastoreShim.class);
private static final String EXTWRITE = "EXTWRITE";
private static final String EXTREAD = "EXTREAD";
@@ -1029,6 +1029,8 @@ public class MetastoreShim {
Preconditions.checkNotNull(msClient);
Preconditions.checkNotNull(dbName);
Preconditions.checkNotNull(tableName);
+ Preconditions.checkState(!insertEventInfos.isEmpty(), "Atleast one insert event "
+ + "info must be provided.");
LOG.debug(String.format(
"Firing %s insert event for %s", insertEventInfos.size(), tableName));
FireEventRequestData data = new FireEventRequestData();
@@ -1053,7 +1055,12 @@ public class MetastoreShim {
data.setInsertDatas(insertDatas);
}
FireEventResponse response = msClient.fireListenerEvent(rqst);
-
+ if (!response.isSetEventIds()) {
+ LOG.error("FireEventResponse does not have event ids set for table {}.{}. This "
+ + "may cause the table to unnecessarily be refreshed when the insert event "
+ + "is received.", dbName, tableName);
+ return Collections.EMPTY_LIST;
+ }
return response.getEventIds();
}
}
diff --git a/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java b/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
index 3ac2a05..174b122 100644
--- a/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
+++ b/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
@@ -4511,6 +4511,8 @@ public class CatalogOpExecutor {
}
}
+ if (insertEventInfos.isEmpty()) return;
+
MetaStoreClient metaStoreClient = catalog_.getMetaStoreClient();
List<Long> eventIds = MetastoreShim.fireInsertEvents(metaStoreClient,
insertEventInfos, table.getDb().getName(), table.getName());
diff --git a/tests/custom_cluster/test_event_processing.py b/tests/custom_cluster/test_event_processing.py
index cdd1a48..cc381ba 100644
--- a/tests/custom_cluster/test_event_processing.py
+++ b/tests/custom_cluster/test_event_processing.py
@@ -237,6 +237,15 @@ class TestEventProcessing(CustomClusterTestSuite):
tbl_name = self.__get_random_name("tbl_")
tbl2 = self.__get_random_name("tbl_")
view_name = self.__get_random_name("view_")
+ # create a empty table for both partitioned and unpartitioned case for testing insert
+ # events
+ empty_unpartitioned_tbl = self.__get_random_name("insert_test_tbl_")
+ empty_partitioned_tbl = self.__get_random_name("insert_test_parttbl_")
+ self.client.execute(
+ "create table {0}.{1} (c1 int)".format(db_name, empty_unpartitioned_tbl))
+ self.client.execute(
+ "create table {0}.{1} (c1 int) partitioned by (part int)".format(db_name,
+ empty_partitioned_tbl))
self_event_test_queries = {
# Queries which will increment the self-events-skipped counter
True: [
@@ -289,7 +298,12 @@ class TestEventProcessing(CustomClusterTestSuite):
db_name, tbl_name),
# drop non-existing partition; essentially this is a no-op
"alter table {0}.{1} drop if exists partition (year=2100, month=1)".format(
- db_name, tbl_name)]
+ db_name, tbl_name),
+ "insert overwrite {0}.{1} select * from {0}.{1}".format(
+ db_name, empty_unpartitioned_tbl),
+ "insert overwrite {0}.{1} partition(part) select * from {0}.{1}".format(
+ db_name, empty_partitioned_tbl),
+ ]
}
if HIVE_MAJOR_VERSION >= 3:
# insert into a existing partition; generates INSERT events