IMPALA-8847: Ignore add partition events with empty partition list

Certain Hive queries like "alter table <table> add if not exists
partition (<part_spec>)" generate a add_partition event even if the
partition did not really exists. Such events have a empty partition list
in the event message which trips on the Precondition check in the
AddPartitionEvent. This causes event processor to go into error state.
The only way to recover is to issue invalidate metadata in such a case.

The patch adds logic to ignore such events.

Testing:
1. Added a test case which reproduces the issue. The test case works
after the patch is applied.

Change-Id: I877ce6233934e7090cd18e497f748bc6479838cb
Reviewed-on: http://gerrit.cloudera.org:8080/14049
Tested-by: Impala Public Jenkins <impala-public-jenkins@cloudera.com>
Reviewed-by: Vihang Karajgaonkar <vihang@cloudera.com>

Author: Vihang Karajgaonkar

fe/src/main/java/org/apache/impala/catalog/events/MetastoreEvents.java

@@ -1331,7 +1331,7 @@ protected static String getStringProperty(
   }
 
   public static class AddPartitionEvent extends TableInvalidatingEvent {
-    private final Partition lastAddedPartition_;
+    private Partition lastAddedPartition_;
     private final List<Partition> addedPartitions_;
 
     /**
@@ -1350,12 +1350,16 @@ private AddPartitionEvent(CatalogServiceCatalog catalog, Metrics metrics,
                 .getAddPartitionMessage(event.getMessage());
         addedPartitions_ =
             Lists.newArrayList(addPartitionMessage_.getPartitionObjs());
-        Preconditions.checkState(addedPartitions_.size() > 0);
-        // when multiple partitions are added in HMS they are all added as one transaction
-        // Hence all the partitions which are present in the message must have the same
-        // serviceId and version if it is set. hence it is fine to just look at the
-        // last added partition in the list and use it for the self-event ids
-        lastAddedPartition_ = addedPartitions_.get(addedPartitions_.size() - 1);
+        // it is possible that the added partitions is empty in certain cases. See
+        // IMPALA-8847 for example
+        if (!addedPartitions_.isEmpty()) {
+          // when multiple partitions are added in HMS they are all added as one
+          // transaction Hence all the partitions which are present in the message must
+          // have the same serviceId and version if it is set. hence it is fine to just
+          // look at the last added partition in the list and use it for the self-event
+          // ids
+          lastAddedPartition_ = addedPartitions_.get(addedPartitions_.size() - 1);
+        }
         msTbl_ = addPartitionMessage_.getTableObj();
       } catch (Exception ex) {
         throw new MetastoreNotificationException(ex);
@@ -1364,12 +1368,15 @@ private AddPartitionEvent(CatalogServiceCatalog catalog, Metrics metrics,
 
     @Override
     public void process() throws MetastoreNotificationException, CatalogException {
+      // bail out early if there are not partitions to process
+      if (addedPartitions_.isEmpty()) {
+        infoLog("Partition list is empty. Ignoring this event.");
+        return;
+      }
       if (isSelfEvent()) {
         infoLog("Not processing the event as it is a self-event");
         return;
       }
-      // Notification is created for newly created partitions only. We need not worry
-      // about "IF NOT EXISTS".
       try {
         // Reload the whole table if it's a transactional table.
         if (AcidUtils.isTransactionalTable(msTbl_.getParameters())) {
@@ -1541,7 +1548,6 @@ private DropPartitionEvent(CatalogServiceCatalog catalog, Metrics metrics,
         msTbl_ = Preconditions.checkNotNull(dropPartitionMessage.getTableObj());
         droppedPartitions_ = dropPartitionMessage.getPartitions();
         Preconditions.checkNotNull(droppedPartitions_);
-        Preconditions.checkState(droppedPartitions_.size() > 0);
       } catch (Exception ex) {
         throw new MetastoreNotificationException(
             debugString("Could not parse event message. "
@@ -1553,6 +1559,12 @@ private DropPartitionEvent(CatalogServiceCatalog catalog, Metrics metrics,
 
     @Override
     public void process() throws MetastoreNotificationException {
+      // we have seen cases where a add_partition event is generated with empty
+      // partition list (see IMPALA-8547 for details. Make sure that droppedPartitions
+      // list is not empty
+      if (droppedPartitions_.isEmpty()) {
+        infoLog("Partition list is empty. Ignoring this event.");
+      }
       // We do not need self event as dropPartition() call is a no-op if the directory
       // doesn't exist.
       try {

tests/custom_cluster/test_event_processing.py

@@ -26,6 +26,7 @@
 from tests.common.custom_cluster_test_suite import CustomClusterTestSuite
 from tests.common.skip import SkipIfS3, SkipIfABFS, SkipIfADLS, SkipIfIsilon, SkipIfLocal
 from tests.util.hive_utils import HiveDbWrapper
+from tests.util.event_processor_utils import EventProcessorUtils
 
 
 @SkipIfS3.hive
@@ -129,6 +130,54 @@ def run_test_insert_events(self, is_transactional=False):
          " year=2019 and id=101" % (db_name, TBL_INSERT_PART))
      assert data.split('\t') == ['101', 'z', '28', '3', '2019']
 
+  @CustomClusterTestSuite.with_args(
+    catalogd_args="--hms_event_polling_interval_s=1"
+  )
+  @SkipIfHive2.acid
+  def test_empty_partition_events_transactional(self, unique_database):
+    self._run_test_empty_partition_events(unique_database, True)
+
+  @CustomClusterTestSuite.with_args(
+    catalogd_args="--hms_event_polling_interval_s=1"
+  )
+  def test_empty_partition_events(self, unique_database):
+    self._run_test_empty_partition_events(unique_database, False)
+
+  def _run_test_empty_partition_events(self, unique_database, is_transactional):
+    TBLPROPERTIES = ""
+    if is_transactional:
+       TBLPROPERTIES = "TBLPROPERTIES ('transactional'='true'," \
+           "'transactional_properties'='insert_only')"
+    test_tbl = unique_database + ".test_events"
+    self.run_stmt_in_hive("create table {0} (key string, value string) \
+      partitioned by (year int) {1} stored as parquet".format(test_tbl, TBLPROPERTIES))
+    EventProcessorUtils.wait_for_event_processing(self.hive_client)
+    self.client.execute("describe {0}".format(test_tbl))
+
+    self.run_stmt_in_hive(
+      "alter table {0} add partition (year=2019)".format(test_tbl))
+    EventProcessorUtils.wait_for_event_processing(self.hive_client)
+    assert [('2019',)] == self.get_impala_partition_info(test_tbl, 'year')
+
+    self.run_stmt_in_hive(
+      "alter table {0} add if not exists partition (year=2019)".format(test_tbl))
+    EventProcessorUtils.wait_for_event_processing(self.hive_client)
+    assert [('2019',)] == self.get_impala_partition_info(test_tbl, 'year')
+    assert EventProcessorUtils.get_event_processor_status() == "ACTIVE"
+
+    self.run_stmt_in_hive(
+      "alter table {0} drop partition (year=2019)".format(test_tbl))
+    EventProcessorUtils.wait_for_event_processing(self.hive_client)
+    assert ('2019') not in self.get_impala_partition_info(test_tbl, 'year')
+    assert EventProcessorUtils.get_event_processor_status() == "ACTIVE"
+
+    self.run_stmt_in_hive(
+      "alter table {0} drop if exists partition (year=2019)".format(test_tbl))
+    EventProcessorUtils.wait_for_event_processing(self.hive_client)
+    assert ('2019') not in self.get_impala_partition_info(test_tbl, 'year')
+    assert EventProcessorUtils.get_event_processor_status() == "ACTIVE"
+
+
   def wait_for_insert_event_processing(self, previous_event_id):
     """Waits until the event processor has finished processing insert events. Since two
     events are created for every insert done through hive, we wait until the event id is

tests/util/event_processor_utils.py

@@ -0,0 +1,91 @@
+# 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.
+#
+# Impala tests for Hive Metastore, covering the expected propagation
+# of metadata from Hive to Impala or Impala to Hive. Each test
+# modifies the metadata via Hive and checks that the modification
+# succeeded by querying Impala, or vice versa.
+
+import requests
+import time
+import json
+from tests.common.environ import build_flavor_timeout
+
+
+class EventProcessorUtils(object):
+
+  DEFAULT_CATALOG_URL = "http://localhost:25020"
+
+  @staticmethod
+  def wait_for_event_processing(hive_client, timeout=10):
+      """Waits till the event processor has synced to the latest event id from metastore
+         or the timeout value in seconds whichever is earlier"""
+      success = False
+      assert timeout > 0
+      assert hive_client is not None
+      current_event_id = EventProcessorUtils.get_current_notification_id(hive_client)
+      end_time = time.time() + timeout
+      while time.time() < end_time:
+        new_event_id = EventProcessorUtils.get_last_synced_event_id()
+        if new_event_id >= current_event_id:
+          success = True
+          break
+        time.sleep(0.1)
+      if not success:
+        raise Exception(
+          "Event processor did not sync till last known event id{0} \
+          within {1} seconds".format(current_event_id, timeout))
+      # Wait for catalog update to be propagated.
+      time.sleep(build_flavor_timeout(6, slow_build_timeout=10))
+      return success
+
+  @staticmethod
+  def get_event_processor_metrics():
+     """Scrapes the catalog's /events webpage and return a dictionary with the event
+     processor metrics."""
+     response = requests.get("%s/events?json" % EventProcessorUtils.DEFAULT_CATALOG_URL)
+     assert response.status_code == requests.codes.ok
+     varz_json = json.loads(response.text)
+     metrics = varz_json["event_processor_metrics"].strip().splitlines()
+
+     # Helper to strip a pair of elements
+     def strip_pair(p):
+       return (p[0].strip(), p[1].strip())
+
+     pairs = [strip_pair(kv.split(':')) for kv in metrics if kv]
+     return dict(pairs)
+
+  @staticmethod
+  def get_last_synced_event_id():
+    """Returns the last_synced_event_id."""
+    metrics = EventProcessorUtils.get_event_processor_metrics()
+    assert 'last-synced-event-id' in metrics.keys()
+    return int(metrics['last-synced-event-id'])
+
+  @staticmethod
+  def get_event_processor_status():
+    """
+    Returns the current status of the EventsProcessor
+    """
+    metrics = EventProcessorUtils.get_event_processor_metrics()
+    assert 'status' in metrics.keys()
+    return metrics['status']
+
+  @staticmethod
+  def get_current_notification_id(hive_client):
+    """Returns the current notification from metastore"""
+    assert hive_client is not None
+    return hive_client.get_current_notificationEventId()