// 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.catalog.events;

import com.codahale.metrics.Gauge;
import com.codahale.metrics.Snapshot;
import com.codahale.metrics.Timer;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import javax.annotation.Nullable;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.hadoop.hive.metastore.IMetaStoreClient;
import org.apache.hadoop.hive.metastore.IMetaStoreClient.NotificationFilter;
import org.apache.hadoop.hive.metastore.api.CurrentNotificationEventId;
import org.apache.hadoop.hive.metastore.api.NotificationEvent;
import org.apache.hadoop.hive.metastore.api.NotificationEventRequest;
import org.apache.hadoop.hive.metastore.api.NotificationEventResponse;
import org.apache.hadoop.hive.metastore.messaging.MessageDeserializer;
import org.apache.impala.catalog.CatalogException;
import org.apache.impala.catalog.CatalogServiceCatalog;
import org.apache.impala.catalog.Db;
import org.apache.impala.catalog.HdfsTable;
import org.apache.impala.catalog.MetaStoreClientPool.MetaStoreClient;
import org.apache.impala.catalog.MetastoreClientInstantiationException;
import org.apache.impala.catalog.Table;
import org.apache.impala.catalog.IncompleteTable;
import org.apache.impala.catalog.events.ConfigValidator.ValidationResult;
import org.apache.impala.catalog.events.MetastoreEvents.DropDatabaseEvent;
import org.apache.impala.catalog.events.MetastoreEvents.MetastoreEvent;
import org.apache.impala.catalog.events.MetastoreEvents.MetastoreEventFactory;
import org.apache.impala.common.Metrics;
import org.apache.impala.common.PrintUtils;
import org.apache.impala.compat.MetastoreShim;
import org.apache.impala.service.CatalogOpExecutor;
import org.apache.impala.thrift.TEventProcessorMetrics;
import org.apache.impala.thrift.TEventProcessorMetricsSummaryResponse;
import org.apache.impala.util.MetaStoreUtil;
import org.apache.impala.util.ThreadNameAnnotator;
import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * A metastore event is a instance of the class
 * <code>org.apache.hadoop.hive.metastore.api.NotificationEvent</code>. Metastore can be
 * configured, to work with Listeners which are called on various DDL operations like
 * create/alter/drop operations on database, table, partition etc. Each event has a unique
 * incremental id and the generated events are be fetched from Metastore to get
 * incremental updates to the metadata stored in Hive metastore using the the public API
 * <code>get_next_notification</code> These events could be generated by external
 * Metastore clients like Apache Hive or Apache Spark as well as other Impala clusters
 * configured to talk with the same metastore.
 *
 * This class is used to poll metastore for such events at a given frequency. By observing
 * such events, we can take appropriate action on the catalogD
 * (refresh/invalidate/add/remove) so that catalog represents the latest information
 * available in metastore. We keep track of the last synced event id in each polling
 * iteration so the next batch can be requested appropriately. The current batch size is
 * constant and set to MAX_EVENTS_PER_RPC.
 *
 * <pre>
 *      +---------------+   +----------------+        +--------------+
 *      |Catalog state  |   |Catalog         |        |              |
 *      |stale          |   |State up-to-date|        |Catalog state |
 *      |(apply event)  |   |(ignore)        |        |is newer than |
 *      |               |   |                |        |event         |
 *      |               |   |                |        |(ignore)      |
 *      +------+--------+   +-----+----------+        +-+------------+
 *             |                  |                     |
 *             |                  |                     |
 *             |                  |                     |
 *             |                  |                     |
 *             |                  |                     |
 * +-----------V------------------V---------------------V----------->  Event Timeline
 *                                ^
 *                                |
 *                                |
 *                                |
 *                                |
 *                                E
 *
 * </pre>
 * Consistency model: Events could be seen as DDLs operations from past either done from
 * this same cluster or some other external system. For example in the events timeline
 * given above, consider a Event E at any given time. The catalog state for the
 * corresponding object of the event could either be stale, exactly-same or at a version
 * which is higher than one provided by event. Catalog state should only be updated when
 * it is stale with respect to the event. In order to determine if the catalog object is
 * stale, we rely on a combination of create EventId and object version.
 * In case of create/drop events on database/table/partitions we use the
 * <code>createEventId</code> field of the corresponding object in the catalogd
 * to determine if the event needs to be processed or ignored. E.g. if Impala creates
 * a table, {@link CatalogOpExecutor} will create the table and assign the createEventId
 * of the table by fetching the CREATE_TABLE event from HMS. Later on when the event
 * is fetched by events processor, it uses the createEventId of the Catalogd's table to
 * ignore the event. Similar approach is used for databases and partition create events.
 *
 * In case of Drop events for database/table/partition events processor looks at the
 * {@link DeleteEventLog} in the CatalogOpExecutor to determine if the table has been
 * dropped already from catalogd.
 *
 * In case of ALTER/INSERT events events processor relies on the object version in the
 * properties of the table to determine if this is a self-event or not.
 *
 * Following table shows the actions to be taken when the given event type is received.
 *
 * <pre>
 *               +------------------------------------------------+ --------------------+
 *               |    Catalog object state                                              |
 * +--------------------------------------+-----------------------+---------------------+
 * | Event type  | Loaded                 | Incomplete            | Not present         |
 * |             |                        |                       |                     |
 * +--------------------------------------------------------------+---------------------+
 * |             |                        |                       |                     |
 * | CREATE EVENT| Ignore                 | Ignore                | addIfNotRemovedLater|
 * |             |                        |                       |                     |
 * |             |                        |                       |                     |
 * | ALTER EVENT | Refresh                | Ignore                | Ignore              |
 * |             |                        |                       |                     |
 * |             |                        |                       |                     |
 * | DROP EVENT  | removeIfNotAddedLater  | removeIfNotAddedLater | Ignore              |
 * |             |                        |                       |                     |
 * |             |                        |                       |                     |
 * | INSERT EVENT| Refresh                | Ignore                | Ignore              |
 * |             |                        |                       |                     |
 * +-------------+------------------------+-----------------------+---------------------+
 * </pre>
 *
 * Currently event handlers rely on createEventId on Database, Table and Partition to
 * uniquely determine if the object from event is same as object in the catalog. This
 * information is used to make sure that we are deleting the right incarnation of the
 * object when compared to Metastore.
 *
 * Self-events:
 * Events could be generated by this Catalog's operations as well. Invalidating table
 * for such events is unnecessary and inefficient. In order to detect such self-events
 * when catalog executes a DDL operation it appends the current catalog version to the
 * list of version numbers for the in-flight events for the table. Events processor
 * clears this version when the corresponding version number identified by serviceId is
 * received in the event. This is needed since it is possible that a external
 * non-Impala system which generates the event presents the same serviceId and version
 * number later on. The algorithm to detect such self-event is as below.
 *
 * 1. Add the service id and expected catalog version to database/table/partition
 * parameters when executing the DDL operation. When the HMS operation is successful, add
 * the version number to the list of version for in-flight events at
 * table/database/partition level.
 * 2. When the event is received, the first time you see the combination of serviceId
 * and version number, event processor clears the version number from table's list and
 * determines the event as self-generated (and hence ignored).
 * 3. If the event data presents a unknown serviceId or if the version number is not
 * present in the list of in-flight versions, event is not a self-event and needs to be
 * processed.
 *
 * In order to limit the total memory footprint, only 100 version numbers are stored at
 * the catalog object. Since the event processor is expected to poll every few seconds
 * this should be a reasonable bound which satisfies most use-cases. Otherwise, event
 * processor may wrongly process a self-event to invalidate the table. In such a case,
 * its a performance penalty not a correctness issue.
 *
 * All the operations which change the state of catalog cache while processing a certain
 * event type must be atomic in nature. We rely on taking a DDL lock in CatalogOpExecutor
 * in case of create/drop events and object (Db or table) level writeLock in case of alter
 * events to make sure that readers are blocked while the metadata
 * update operation is being performed. Since the events are generated post-metastore
 * operations, such catalog updates do not need to update the state in Hive Metastore.
 *
 * Error Handling: The event processor could be in ACTIVE, PAUSED, ERROR states. In case
 * of any errors while processing the events the state of event processor changes to ERROR
 * and no subsequent events are polled. In such a case a invalidate metadata command
 * restarts the event polling which updates the lastSyncedEventId to the latest from
 * metastore.
 *
 * TODO:
 * 1. a global invalidate metadata command to get the events processor out of error state
 * is too heavy weight. We should make it easier to recover from the error state.
 * 2. The createEventId logic can be extended to track the last eventId which the table
 * has synced to and we can then get rid of self-event logic for alter events too.
 */
public class MetastoreEventsProcessor implements ExternalEventsProcessor {

  public static final String HMS_ADD_THRIFT_OBJECTS_IN_EVENTS_CONFIG_KEY =
      "hive.metastore.notifications.add.thrift.objects";

  private static final Logger LOG =
      LoggerFactory.getLogger(MetastoreEventsProcessor.class);

  private static final MessageDeserializer MESSAGE_DESERIALIZER =
      MetastoreShim.getMessageDeserializer();

  private static MetastoreEventsProcessor instance;

  // maximum number of events to poll in each RPC
  private static final int EVENTS_BATCH_SIZE_PER_RPC = 1000;

  // maximum time to wait for a clean shutdown of scheduler in seconds
  private static final int SCHEDULER_SHUTDOWN_TIMEOUT = 10;

  // time taken to fetch events during each poll
  public static final String EVENTS_FETCH_DURATION_METRIC = "events-fetch-duration";
  // time taken to apply all the events received duration each poll
  public static final String EVENTS_PROCESS_DURATION_METRIC = "events-apply-duration";
  // rate of events received per unit time
  public static final String EVENTS_RECEIVED_METRIC = "events-received";
  // total number of events which are skipped because of the flag setting or
  // in case of [CREATE|DROP] events on [DATABASE|TABLE|PARTITION] which were ignored
  // because the [DATABASE|TABLE|PARTITION] was already [PRESENT|ABSENT] in the catalogd.
  public static final String EVENTS_SKIPPED_METRIC = "events-skipped";
  // name of the event processor status metric
  public static final String STATUS_METRIC = "status";
  // last synced event id
  public static final String LAST_SYNCED_ID_METRIC = "last-synced-event-id";
  // last synced event time
  public static final String LAST_SYNCED_EVENT_TIME = "last-synced-event-time";
  // latest event id in Hive metastore
  public static final String LATEST_EVENT_ID = "latest-event-id";
  // event time of the latest event in Hive metastore
  public static final String LATEST_EVENT_TIME = "latest-event-time";
  // delay(secs) in events processing
  public static final String EVENT_PROCESSING_DELAY = "event-processing-delay";
  // metric name for number of tables which are refreshed by event processor so far
  public static final String NUMBER_OF_TABLE_REFRESHES = "tables-refreshed";
  // number of times events processor refreshed a partition
  public static final String NUMBER_OF_PARTITION_REFRESHES = "partitions-refreshed";
  // number of tables which were added to the catalogd based on events.
  public static final String NUMBER_OF_TABLES_ADDED = "tables-added";
  // number of tables which were removed to the catalogd based on events.
  public static final String NUMBER_OF_TABLES_REMOVED = "tables-removed";
  // number of databases which were added to the catalogd based on events.
  public static final String NUMBER_OF_DATABASES_ADDED = "databases-added";
  // number of database which were removed to the catalogd based on events.
  public static final String NUMBER_OF_DATABASES_REMOVED = "databases-removed";
  // number of partitions which were added to the catalogd based on events.
  public static final String NUMBER_OF_PARTITIONS_ADDED = "partitions-added";
  // number of partitions which were removed to the catalogd based on events.
  public static final String NUMBER_OF_PARTITIONS_REMOVED = "partitions-removed";
  // number of entries in the delete event log
  public static final String DELETE_EVENT_LOG_SIZE = "delete-event-log-size";
  // number of batch events generated
  public static final String NUMBER_OF_BATCH_EVENTS = "batch-events-created";

  // metric to measure the delay in msec, between the event created in metastore and time
  // it took to be consumed by the event processor
  public static final String AVG_DELAY_IN_CONSUMING_EVENTS = "events-consuming" +
      "-delay";

  private static final long SECOND_IN_NANOS = 1000 * 1000 * 1000L;

  // List of event types to skip while fetching notification events from metastore
  private static final List<String> EVENT_SKIP_LIST = Arrays.asList("OPEN_TXN");

  /**
   * Wrapper around {@link
   * MetastoreEventsProcessor#getNextMetastoreEventsInBatches(CatalogServiceCatalog,
   * long, NotificationFilter, int)} which passes the default batch size.
   */
  public static List<NotificationEvent> getNextMetastoreEventsInBatches(
      CatalogServiceCatalog catalog, long eventId, NotificationFilter filter)
      throws MetastoreNotificationFetchException {
    return getNextMetastoreEventsInBatches(catalog, eventId, filter,
        EVENTS_BATCH_SIZE_PER_RPC);
  }

  /**
   * Gets the next list of {@link NotificationEvent} from Hive Metastore which are
   * greater than the given eventId and filtered according to the provided filter.
   * @param catalog The CatalogServiceCatalog used to get the metastore client
   * @param eventId The eventId after which the events are needed.
   * @param filter The {@link NotificationFilter} used to filter the list of fetched
   *               events. Note that this is a client side filter not a server side
   *               filter. Unfortunately, HMS doesn't provide a similar mechanism to
   *               do server side filtering.
   * @param eventsBatchSize the batch size for fetching the events from metastore.
   * @return List of {@link NotificationEvent} which are all greater than eventId and
   * satisfy the given filter.
   * @throws MetastoreNotificationFetchException in case of RPC errors to metastore.
   */
  @VisibleForTesting
  public static List<NotificationEvent> getNextMetastoreEventsInBatches(
      CatalogServiceCatalog catalog, long eventId, NotificationFilter filter,
      int eventsBatchSize) throws MetastoreNotificationFetchException {
    Preconditions.checkArgument(eventsBatchSize > 0);
    List<NotificationEvent> result = new ArrayList<>();
    try (MetaStoreClient msc = catalog.getMetaStoreClient()) {
      long toEventId = msc.getHiveClient().getCurrentNotificationEventId()
          .getEventId();
      if (toEventId <= eventId) return result;
      long currentEventId = eventId;
      while (currentEventId < toEventId) {
        int batchSize = Math
            .min(eventsBatchSize, (int)(toEventId - currentEventId));
        // we don't call the HiveMetaStoreClient's getNextNotification()
        // call here because it can throw a IllegalStateException if the eventId
        // which we pass in is very old and metastore has already cleaned up
        // the events since that eventId.
        NotificationEventRequest eventRequest = new NotificationEventRequest();
        eventRequest.setMaxEvents(batchSize);
        eventRequest.setLastEvent(currentEventId);
        NotificationEventResponse notificationEventResponse =
            MetastoreShim.getNextNotification(msc.getHiveClient(), eventRequest, true);
        if (notificationEventResponse.getEvents().isEmpty()) {
          // Possible to receive empty list due to event skip list in request
          return result;
        }
        for (NotificationEvent event : notificationEventResponse.getEvents()) {
          // if no filter is provided we add all the events
          if (filter == null || filter.accept(event)) result.add(event);
          currentEventId = event.getEventId();
        }
      }
      return result;
    } catch (MetastoreClientInstantiationException | TException e) {
      throw new MetastoreNotificationFetchException(String.format(
          CatalogOpExecutor.HMS_RPC_ERROR_FORMAT_STR, "getNextNotification"), e);
    }
  }

  /**
   * Sync table to latest event id starting from last synced
   * event id.
   * @param catalog
   * @param tbl: Catalog table to be synced
   * @param eventFactory
   * @throws CatalogException
   * @throws MetastoreNotificationException
   */
  public static void syncToLatestEventId(CatalogServiceCatalog catalog,
      org.apache.impala.catalog.Table tbl, EventFactory eventFactory, Metrics metrics)
      throws CatalogException, MetastoreNotificationException {
    Preconditions.checkArgument(tbl != null, "tbl is null");
    Preconditions.checkState(!(tbl instanceof IncompleteTable) &&
        tbl.isLoaded(), "table %s is either incomplete or not loaded",
        tbl.getFullName());
    Preconditions.checkState(tbl.isWriteLockedByCurrentThread(),
        String.format("Write lock is not held on table %s by current thread",
            tbl.getFullName()));
    long lastEventId = tbl.getLastSyncedEventId();
    Preconditions.checkArgument(lastEventId > 0, "lastEvent " +
        " Id %s for table %s should be greater than 0", lastEventId, tbl.getFullName());

    String annotation = String.format("sync table %s to latest HMS event id",
        tbl.getFullName());
    try(ThreadNameAnnotator tna = new ThreadNameAnnotator(annotation)) {
      List<NotificationEvent> events = getNextMetastoreEventsInBatches(catalog,
          lastEventId, getTableNotificationEventFilter(tbl));

      if (events.isEmpty()) {
        LOG.debug("table {} synced till event id {}. No new HMS events to process from "
                + "event id: {}", tbl.getFullName(), lastEventId, lastEventId + 1);
        return;
      }
      MetastoreEvents.MetastoreEvent currentEvent = null;
      for (NotificationEvent event : events) {
        currentEvent = eventFactory.get(event, metrics);
        LOG.trace("for table {}, processing event {}", tbl.getFullName(), currentEvent);
        currentEvent.processIfEnabled();
        if (currentEvent.isDropEvent()) {
          // currentEvent can only be DropPartition or DropTable
          Preconditions.checkNotNull(currentEvent.getDbName());
          Preconditions.checkNotNull(currentEvent.getTableName());
          String key = DeleteEventLog.getTblKey(currentEvent.getDbName(),
              currentEvent.getTableName());
          catalog.getMetastoreEventProcessor().getDeleteEventLog()
              .addRemovedObject(currentEvent.getEventId(), key);
        }
        if (currentEvent instanceof MetastoreEvents.DropTableEvent) {
          // return after processing table drop event
          return;
        }
      }
      // setting HMS Event ID after all the events
      // are successfully processed only if table was
      // not dropped
      // Certain events like alter_table, do an incremental reload which sets the event
      // id to the current hms event id at that time. Therefore, check table's last
      // synced event id again before setting currentEvent id
      if (currentEvent.getEventId() > tbl.getLastSyncedEventId()) {
        tbl.setLastSyncedEventId(currentEvent.getEventId());
      }
      LOG.info("Synced table {} till HMS event:  {}", tbl.getFullName(),
          tbl.getLastSyncedEventId());
    }
  }

  /**
   * Sync database to latest event id starting from the last synced
   * event id
   * @param catalog
   * @param db
   * @param eventFactory
   * @throws CatalogException
   * @throws MetastoreNotificationException
   */
  public static void syncToLatestEventId(CatalogServiceCatalog catalog,
      org.apache.impala.catalog.Db db, EventFactory eventFactory, Metrics metrics)
      throws CatalogException, MetastoreNotificationException {
    Preconditions.checkArgument(db != null, "db is null");
    long lastEventId = db.getLastSyncedEventId();
    Preconditions.checkArgument(lastEventId > 0, "Invalid " +
        "last synced event ID %s for db %s ", lastEventId, db.getName());
    Preconditions.checkState(db.isLockHeldByCurrentThread(),
        "Current thread does not hold lock on db: %s", db.getName());

    String annotation = String.format("sync db %s to latest HMS event id", db.getName());
    try(ThreadNameAnnotator tna = new ThreadNameAnnotator(annotation)) {
      List<NotificationEvent> events = getNextMetastoreEventsInBatches(catalog,
          lastEventId, getDbNotificationEventFilter(db));

      if (events.isEmpty()) {
        LOG.debug("db {} already synced till event id: {}, no new hms events from "
            + "event id: {}", db.getName(), lastEventId, lastEventId+1);
        return;
      }

      MetastoreEvents.MetastoreEvent currentEvent = null;
      for (NotificationEvent event : events) {
        currentEvent = eventFactory.get(event, metrics);
        LOG.trace("for db {}, processing event: {}", db.getName(), currentEvent);
        currentEvent.processIfEnabled();
        if (currentEvent.isDropEvent()) {
          Preconditions.checkState(currentEvent instanceof DropDatabaseEvent,
              "invalid drop event {} ", currentEvent);
          Preconditions.checkNotNull(currentEvent.getDbName());
          String key = DeleteEventLog.getDbKey(currentEvent.getDbName());
          catalog.getMetastoreEventProcessor().getDeleteEventLog()
              .addRemovedObject(currentEvent.getEventId(), key);
          // return after processing drop db event
          return;
        }
      }
      // setting HMS Event Id after all the events
      // are successfully processed only if db was not dropped
      db.setLastSyncedEventId(currentEvent.getEventId());
      LOG.info("Synced db {} till HMS event {}", db.getName(),
          currentEvent);
    }
  }

  /*
  This filter is used when syncing events for a table to the latest HMS event id.
  It filters all events except db related ones.
   */
  private static NotificationFilter getTableNotificationEventFilter(Table tbl) {
    NotificationFilter filter = new NotificationFilter() {
      @Override
      public boolean accept(NotificationEvent event) {
        if (event.getDbName() != null && event.getTableName() != null) {
          return tbl.getDb().getName().equalsIgnoreCase(event.getDbName()) &&
              tbl.getName().equalsIgnoreCase(event.getTableName());
        }
        // filter all except db events
        return event.getDbName() == null;
      }
    };
    return filter;
  }

  /*
  This filter is used when syncing db to the latest HMS event id. The
  filter accepts all events except table related ones
   */
  private static NotificationFilter getDbNotificationEventFilter(Db db) {
    NotificationFilter filter = new NotificationFilter() {
      @Override
      public boolean accept(NotificationEvent event) {
        if (event.getDbName() != null && event.getTableName() == null) {
          return db.getName().equalsIgnoreCase(event.getDbName());
        }
        // filter all events except table events
        return event.getTableName() == null;
      }
    };
    return filter;
  }

  // possible status of event processor
  public enum EventProcessorStatus {
    PAUSED, // event processor is paused because catalog is being reset concurrently
    ACTIVE, // event processor is scheduled at a given frequency
    ERROR, // event processor is in error state and event processing has stopped
    NEEDS_INVALIDATE, // event processor could not resolve certain events and needs a
    // manual invalidate command to reset the state (See AlterEvent for a example)
    STOPPED, // event processor is shutdown. No events will be processed
    DISABLED // event processor is not configured to run
  }

  // current status of this event processor
  private EventProcessorStatus eventProcessorStatus_ = EventProcessorStatus.STOPPED;

  // error message when event processor comes into ERROR/NEEDS_INVALIDATE states
  private String eventProcessorErrorMsg_ = null;

  // event factory which is used to get or create MetastoreEvents
  private final MetastoreEventFactory metastoreEventFactory_;

  // keeps track of the current event that we are processing
  private NotificationEvent currentEvent_;

  // keeps track of the last event id which we have synced to
  private final AtomicLong lastSyncedEventId_ = new AtomicLong(-1);
  private final AtomicLong lastSyncedEventTimeSecs_ = new AtomicLong(0);

  // The event id and eventTime of the latest event in HMS. Only used in metrics to show
  // how far we are lagging behind.
  private final AtomicLong latestEventId_ = new AtomicLong(0);
  private final AtomicLong latestEventTimeSecs_ = new AtomicLong(0);

  // The duration in nanoseconds of the processing of the last event batch.
  private final AtomicLong lastEventProcessDurationNs_ = new AtomicLong(0);

  // polling interval in seconds. Note this is a time we wait AFTER each fetch call
  private final long pollingFrequencyInSec_;

  // catalog service instance to be used while processing events
  protected final CatalogServiceCatalog catalog_;

  // scheduler daemon thread executor for processing events at given frequency
  private final ScheduledExecutorService processEventsScheduler_ =
      Executors.newSingleThreadScheduledExecutor(
          new ThreadFactoryBuilder()
              .setDaemon(true)
              .setNameFormat("MetastoreEventsProcessor-ProcessEvents")
              .build());

  // scheduler daemon thread executor to update the latest event id at given frequency
  private final ScheduledExecutorService updateEventIdScheduler_ =
      Executors.newSingleThreadScheduledExecutor(
          new ThreadFactoryBuilder()
              .setDaemon(true)
              .setNameFormat("MetastoreEventsProcessor-UpdateEventId")
              .build());

  // metrics registry to keep track of metrics related to event processing
  //TODO create a separate singleton class which wraps around this so that we don't
  // have to pass it around as a argument in constructor in MetastoreEvents
  private final Metrics metrics_ = new Metrics();

  // When events processing is ACTIVE this delete event log is used to keep track of
  // DROP events for databases, tables and partitions so that the MetastoreEventsProcessor
  // can ignore the drop events when they are received later.
  private final DeleteEventLog deleteEventLog_ = new DeleteEventLog();

  @VisibleForTesting
  MetastoreEventsProcessor(CatalogOpExecutor catalogOpExecutor, long startSyncFromId,
      long pollingFrequencyInSec) throws CatalogException {
    Preconditions.checkState(pollingFrequencyInSec > 0);
    this.catalog_ = Preconditions.checkNotNull(catalogOpExecutor.getCatalog());
    validateConfigs();
    lastSyncedEventId_.set(startSyncFromId);
    lastSyncedEventTimeSecs_.set(getEventTimeFromHMS(startSyncFromId));
    initMetrics();
    metastoreEventFactory_ = new MetastoreEventFactory(catalogOpExecutor);
    pollingFrequencyInSec_ = pollingFrequencyInSec;
  }

  /**
   * Fetches the required metastore config values and validates them against the
   * expected values. The configurations to validate are different for HMS-2 v/s HMS-3
   * and hence it uses MetastoreShim to get the configurations which need to be validated.
   * @throws CatalogException if one or more validations fail or if metastore is not
   * accessible
   */
  @VisibleForTesting
  void validateConfigs() throws CatalogException {
    List<ValidationResult> validationErrors = new ArrayList<>();
    for (MetastoreEventProcessorConfig config : getEventProcessorConfigsToValidate()) {
      String configKey = config.getValidator().getConfigKey();
      try {
        String value = getConfigValueFromMetastore(configKey, "");
        ValidationResult result = config.validate(value);
        if (!result.isValid()) validationErrors.add(result);
      } catch (TException e) {
        String msg = String.format("Unable to get configuration %s from metastore. Check "
            + "if metastore is accessible", configKey);
        LOG.error(msg, e);
        throw new CatalogException(msg);
      }
    }
    if (!validationErrors.isEmpty()) {
      LOG.error("Found {} incorrect metastore configuration(s).",
          validationErrors.size());
      for (ValidationResult invalidConfig: validationErrors) {
        LOG.error(invalidConfig.getReason());
      }
      throw new CatalogException(String.format("Found %d incorrect metastore "
          + "configuration(s). Events processor cannot start. See ERROR log for more "
          + "details.", validationErrors.size()));
    }
  }

  public DeleteEventLog getDeleteEventLog() { return deleteEventLog_; }

  /**
   * Returns the list of Metastore configurations to validate depending on the hive
   * version
   */
  public static List<MetastoreEventProcessorConfig> getEventProcessorConfigsToValidate() {
    return Arrays.asList(MetastoreEventProcessorConfig.FIRE_EVENTS_FOR_DML,
        MetastoreEventProcessorConfig.METASTORE_DEFAULT_CATALOG_NAME);
  }

  private void initMetrics() {
    metrics_.addTimer(EVENTS_FETCH_DURATION_METRIC);
    metrics_.addTimer(EVENTS_PROCESS_DURATION_METRIC);
    metrics_.addMeter(EVENTS_RECEIVED_METRIC);
    metrics_.addCounter(EVENTS_SKIPPED_METRIC);
    metrics_.addGauge(STATUS_METRIC, (Gauge<String>) () -> getStatus().toString());
    metrics_.addGauge(LAST_SYNCED_ID_METRIC, (Gauge<Long>) lastSyncedEventId_::get);
    metrics_.addGauge(LAST_SYNCED_EVENT_TIME,
        (Gauge<Long>) lastSyncedEventTimeSecs_::get);
    metrics_.addGauge(LATEST_EVENT_ID, (Gauge<Long>) latestEventId_::get);
    metrics_.addGauge(LATEST_EVENT_TIME, (Gauge<Long>) latestEventTimeSecs_::get);
    metrics_.addGauge(EVENT_PROCESSING_DELAY,
        (Gauge<Long>) () -> latestEventTimeSecs_.get() - lastSyncedEventTimeSecs_.get());
    metrics_.addCounter(NUMBER_OF_TABLE_REFRESHES);
    metrics_.addCounter(NUMBER_OF_PARTITION_REFRESHES);
    metrics_.addCounter(NUMBER_OF_TABLES_ADDED);
    metrics_.addCounter(NUMBER_OF_TABLES_REMOVED);
    metrics_.addCounter(NUMBER_OF_DATABASES_ADDED);
    metrics_.addCounter(NUMBER_OF_DATABASES_REMOVED);
    metrics_.addCounter(NUMBER_OF_PARTITIONS_ADDED);
    metrics_.addCounter(NUMBER_OF_PARTITIONS_REMOVED);
    metrics_
        .addGauge(DELETE_EVENT_LOG_SIZE, (Gauge<Integer>) deleteEventLog_::size);
    metrics_.addCounter(NUMBER_OF_BATCH_EVENTS);
    metrics_.addTimer(AVG_DELAY_IN_CONSUMING_EVENTS);
  }

  /**
   * Schedules the daemon thread at a given frequency. It is important to note that this
   * method schedules with FixedDelay instead of FixedRate. The reason it is scheduled at
   * a fixedDelay is to make sure that we don't pile up the pending tasks in case each
   * polling operation is taking longer than the given frequency. Because of the fixed
   * delay, the new poll operation is scheduled at the time when previousPoll operation
   * completes + givenDelayInSec
   */
  @Override
  public synchronized void start() {
    Preconditions.checkState(eventProcessorStatus_ != EventProcessorStatus.ACTIVE);
    startScheduler();
    updateStatus(EventProcessorStatus.ACTIVE);
    LOG.info(String.format("Successfully started metastore event processing."
        + " Polling interval: %d seconds.", pollingFrequencyInSec_));
  }

  /**
   * Gets the current event processor status
   */
  public EventProcessorStatus getStatus() {
    return eventProcessorStatus_;
  }

  /**
   * Returns the value for a given config key from Hive Metastore.
   * @param config Hive configuration name
   * @param defaultVal Default value to return if config not present in Hive
   */
  @VisibleForTesting
  public String getConfigValueFromMetastore(String config, String defaultVal)
      throws TException {
    try (MetaStoreClient metaStoreClient = catalog_.getMetaStoreClient()) {
      IMetaStoreClient iMetaStoreClient = metaStoreClient.getHiveClient();
      return MetaStoreUtil.getMetastoreConfigValue(iMetaStoreClient, config, defaultVal);
    }
  }

  /**
   * returns the current value of LastSyncedEventId. This method is not thread-safe and
   * only to be used for testing purposes
   */
  @VisibleForTesting
  public long getLastSyncedEventId() {
    return lastSyncedEventId_.get();
  }

  /**
   * Returns the current value of latestEventId_. This method is not thread-safe and
   * only to be used for testing purposes
   */
  @VisibleForTesting
  public long getLatestEventId() {
    return latestEventId_.get();
  }

  @VisibleForTesting
  void startScheduler() {
    Preconditions.checkState(pollingFrequencyInSec_ > 0);
    LOG.info(String.format("Starting metastore event polling with interval %d seconds.",
        pollingFrequencyInSec_));
    processEventsScheduler_.scheduleAtFixedRate(this ::processEvents,
        pollingFrequencyInSec_, pollingFrequencyInSec_, TimeUnit.SECONDS);
    // Update latestEventId in another thread in case that the processEvents() thread is
    // blocked by slow metadata reloading or waiting for table locks.
    updateEventIdScheduler_.scheduleAtFixedRate(this ::updateLatestEventId,
        pollingFrequencyInSec_, pollingFrequencyInSec_, TimeUnit.SECONDS);
  }

  /**
   * Stops the event processing and changes the status of event processor to
   * <code>EventProcessorStatus.PAUSED</code>. No new events will be processed as long
   * the status is stopped. If this event processor is actively processing events when
   * stop is called, this method blocks until the current processing is complete
   */
  @Override
  public synchronized void pause() {
    // when concurrent invalidate metadata are running, it is possible the we receive
    // a pause method call on a already paused events processor.
    if (eventProcessorStatus_ == EventProcessorStatus.PAUSED) {
      return;
    }
    updateStatus(EventProcessorStatus.PAUSED);
    LOG.info(String.format("Event processing is paused. Last synced event id is %d",
        lastSyncedEventId_.get()));
  }

  /**
   * Get the current notification event id from metastore
   */
  @Override
  public long getCurrentEventId() throws MetastoreNotificationFetchException {
    try (MetaStoreClient metaStoreClient = catalog_.getMetaStoreClient()) {
      return metaStoreClient.getHiveClient().getCurrentNotificationEventId().getEventId();
    } catch (MetastoreClientInstantiationException | TException e) {
      throw new MetastoreNotificationFetchException("Unable to fetch the current " +
          "notification event id. Check if metastore service is accessible", e);
    }
  }

  public static long getCurrentEventIdNoThrow(IMetaStoreClient client) {
    long latestEventId = -1L;
    try {
      latestEventId = client.getCurrentNotificationEventId().getEventId();
    } catch (TException exception) {
      LOG.warn(String.format("Unable to fetch latest event id from HMS: %s",
          exception.getMessage()));
    }
    return latestEventId;
  }

  /**
   * Fetch the event from HMS specified by 'eventId'
   * @return null if the event has been cleaned up or any error occurs.
   */
  private NotificationEvent getEventFromHMS(MetaStoreClient msClient, long eventId) {
    NotificationEventRequest eventRequest = new NotificationEventRequest();
    eventRequest.setLastEvent(eventId - 1);
    eventRequest.setMaxEvents(1);
    try {
      NotificationEventResponse response = MetastoreShim.getNextNotification(
          msClient.getHiveClient(), eventRequest, false);
      Iterator<NotificationEvent> eventIter = response.getEventsIterator();
      if (!eventIter.hasNext()) {
        LOG.warn("Unable to fetch event {}. It has been cleaned up", eventId);
        return null;
      }
      return eventIter.next();
    } catch (TException e) {
      LOG.warn("Unable to fetch event {}", eventId, e);
    }
    return null;
  }

  /**
   * Get the event time by fetching the specified event from HMS.
   * @return 0 if the event has been cleaned up or any error occurs.
   */
  @VisibleForTesting
  public int getEventTimeFromHMS(long eventId) {
    try (MetaStoreClient msClient = catalog_.getMetaStoreClient()) {
      NotificationEvent event = getEventFromHMS(msClient, eventId);
      if (event != null) return event.getEventTime();
    } catch (MetastoreClientInstantiationException e) {
      LOG.error("Failed to get event time from HMS for event {}", eventId, e);
    }
    return 0;
  }

  /**
   * Starts the event processor from a given event id
   */
  @Override
  public synchronized void start(long fromEventId) {
    Preconditions.checkArgument(fromEventId >= 0);
    EventProcessorStatus currentStatus = eventProcessorStatus_;
    long prevLastSyncedEventId = lastSyncedEventId_.get();
    if (currentStatus == EventProcessorStatus.ACTIVE) {
      // if events processor is already active, we should make sure that the
      // start event id provided is not behind the lastSyncedEventId. This could happen
      // when there are concurrent invalidate metadata calls. if we detect such a case
      // we should return here.
      if (prevLastSyncedEventId >= fromEventId) {
        return;
      }
    }
    lastSyncedEventId_.set(fromEventId);
    lastSyncedEventTimeSecs_.set(getEventTimeFromHMS(fromEventId));
    updateStatus(EventProcessorStatus.ACTIVE);
    LOG.info(String.format(
        "Metastore event processing restarted. Last synced event id was updated "
            + "from %d to %d", prevLastSyncedEventId, lastSyncedEventId_.get()));
  }

  /**
   * Stops the event processing and shuts down the scheduler. In case there is a batch of
   * events which is being processed currently, the
   * <code>processEvents</code> method releases lock after every event is processed.
   * Hence, it is possible that at-least 1 event from the batch be
   * processed while shutdown() waits to acquire lock on this object.
   */
  @Override
  public synchronized void shutdown() {
    Preconditions.checkState(eventProcessorStatus_ != EventProcessorStatus.STOPPED,
        "Event processing is already stopped");
    shutdownAndAwaitTermination(processEventsScheduler_);
    shutdownAndAwaitTermination(updateEventIdScheduler_);
    updateStatus(EventProcessorStatus.STOPPED);
    LOG.info("Metastore event processing stopped.");
  }

  /**
   * Attempts to cleanly shutdown the scheduler pool. If the pool does not shutdown
   * within timeout, does a force shutdown which might interrupt currently running tasks.
   */
  private synchronized void shutdownAndAwaitTermination(ScheduledExecutorService ses) {
    Preconditions.checkNotNull(ses);
    ses.shutdown(); // disable new tasks from being submitted
    try {
      // wait for 10 secs for scheduler to complete currently running tasks
      if (!ses.awaitTermination(SCHEDULER_SHUTDOWN_TIMEOUT, TimeUnit.SECONDS)) {
        // executor couldn't terminate and timed-out, force the termination
        LOG.info(String.format("Scheduler pool did not terminate within %d seconds. "
            + "Attempting to stop currently running tasks", SCHEDULER_SHUTDOWN_TIMEOUT));
        ses.shutdownNow();
      }
    } catch (InterruptedException e) {
      // current thread interrupted while pool was waiting for termination
      // issue a shutdownNow before returning to cancel currently running tasks
      LOG.info("Received interruptedException. Terminating currently running tasks.", e);
      ses.shutdownNow();
    }
  }

  /**
   * Gets metastore notification events from the given eventId. The returned list of
   * NotificationEvents are filtered using the NotificationFilter provided if it is not
   * null.
   * @param eventId The returned events are all after this given event id.
   * @param currentEventId Current event id on metastore
   * @param getAllEvents If this is true all the events since eventId are returned.
   *                     Note that Hive MetaStore can limit the response to a specific
   *                     maximum number of limit based on the value of configuration
   *                     {@code hive.metastore.max.event.response}.
   *                     If it is false, only EVENTS_BATCH_SIZE_PER_RPC events are
   *                     returned, caller is expected to issue more calls to this method
   *                     to fetch the remaining events.
   * @param filter This is a nullable argument. If not null, the events are filtered
   *               and then returned using this. Otherwise, all the events are returned.
   * @return List of NotificationEvents from metastore since eventId.
   * @throws MetastoreNotificationFetchException In case of exceptions from HMS.
   */
  public List<NotificationEvent> getNextMetastoreEvents(final long eventId,
      final long currentEventId, final boolean getAllEvents,
      @Nullable final NotificationFilter filter)
      throws MetastoreNotificationFetchException {
    // no new events since we last polled
    if (currentEventId <= eventId) {
      return Collections.emptyList();
    }
    final Timer.Context context = metrics_.getTimer(EVENTS_FETCH_DURATION_METRIC).time();
    try (MetaStoreClient msClient = catalog_.getMetaStoreClient()) {
      int batchSize = getAllEvents ? -1 : EVENTS_BATCH_SIZE_PER_RPC;
      // we use the thrift API directly instead of
      // HiveMetastoreClient#getNextNotification because the HMS client can throw an
      // exception when there is a gap between the eventIds returned.
      NotificationEventRequest eventRequest = new NotificationEventRequest();
      eventRequest.setLastEvent(eventId);
      eventRequest.setMaxEvents(batchSize);
      NotificationEventResponse response =
          MetastoreShim.getNextNotification(msClient.getHiveClient(), eventRequest, true);
      LOG.info(String.format("Received %d events. Start event id : %d",
          response.getEvents().size(), eventId));
      if (filter == null) return response.getEvents();
      List<NotificationEvent> filteredEvents = new ArrayList<>();
      for (NotificationEvent event : response.getEvents()) {
        if (filter.accept(event)) filteredEvents.add(event);
      }
      return filteredEvents;
    } catch (MetastoreClientInstantiationException | TException e) {
      throw new MetastoreNotificationFetchException(
          "Unable to fetch notifications from metastore. Last synced event id is "
              + eventId, e);
    } finally {
      context.stop();
    }
  }

  /**
   * Fetch the next batch of NotificationEvents from metastore. The default batch size is
   * <code>EVENTS_BATCH_SIZE_PER_RPC</code>
   */
  @VisibleForTesting
  protected List<NotificationEvent> getNextMetastoreEvents()
      throws MetastoreNotificationFetchException {
    return getNextMetastoreEvents(getCurrentEventId());
  }

  /**
   * Fetch the next batch of NotificationEvents from metastore. The default batch size is
   * <code>EVENTS_BATCH_SIZE_PER_RPC</code>
   * @param currentEventId Current event id on metastore
   * @return List of NotificationEvents from metastore since lastSyncedEventId
   * @throws MetastoreNotificationFetchException
   */
  @VisibleForTesting
  public List<NotificationEvent> getNextMetastoreEvents(long currentEventId)
      throws MetastoreNotificationFetchException {
    return getNextMetastoreEvents(lastSyncedEventId_.get(), currentEventId, false, null);
  }

  /**
   * This method issues a request to Hive Metastore if needed, based on the current event
   * id in metastore and the last synced event_id. Events are fetched in fixed sized
   * batches. Each NotificationEvent received is processed by its corresponding
   * <code>MetastoreEvent</code>
   */
  @Override
  public void processEvents() {
    currentEvent_ = null;
    try {
      EventProcessorStatus currentStatus = eventProcessorStatus_;
      if (currentStatus != EventProcessorStatus.ACTIVE) {
        LOG.warn(String.format(
            "Event processing is skipped since status is %s. Last synced event id is %d",
            currentStatus, lastSyncedEventId_.get()));
        return;
      }
      // fetch the current notification event id. We assume that the polling interval
      // is small enough that most of these polling operations result in zero new
      // events. In such a case, fetching current notification event id is much faster
      // (and cheaper on HMS side) instead of polling for events directly
      long currentEventId = getCurrentEventId();
      List<NotificationEvent> events = getNextMetastoreEvents(currentEventId);
      processEvents(currentEventId, events);
    } catch (MetastoreNotificationFetchException ex) {
      // No need to change the EventProcessor state to error since we want the
      // EventProcessor to continue getting new events after HMS is back up.
      LOG.error("Unable to fetch the next batch of metastore events. Hive Metastore " +
        "may be unavailable. Will retry.", ex);
    } catch (MetastoreNotificationNeedsInvalidateException ex) {
      updateStatus(EventProcessorStatus.NEEDS_INVALIDATE);
      String msg = "Event processing needs a invalidate command to resolve the state";
      LOG.error(msg, ex);
      eventProcessorErrorMsg_ = LocalDateTime.now().toString() + '\n' + msg + '\n' +
          ExceptionUtils.getFullStackTrace(ex);
    } catch (Exception ex) {
      // There are lot of Preconditions which can throw RuntimeExceptions when we
      // process events this catch all exception block is needed so that the scheduler
      // thread does not die silently
      updateStatus(EventProcessorStatus.ERROR);
      String msg = "Unexpected exception received while processing event";
      LOG.error(msg, ex);
      eventProcessorErrorMsg_ = LocalDateTime.now().toString() + '\n' + msg + '\n' +
          ExceptionUtils.getFullStackTrace(ex);
      dumpEventInfoToLog(currentEvent_);
    }
  }

  /**
   * Update the latest event id regularly so we know how far we are lagging behind.
   */
  @VisibleForTesting
  public void updateLatestEventId() {
    EventProcessorStatus currentStatus = eventProcessorStatus_;
    if (currentStatus != EventProcessorStatus.ACTIVE) {
      return;
    }
    try (MetaStoreClient msClient = catalog_.getMetaStoreClient()) {
      CurrentNotificationEventId currentNotificationEventId =
          msClient.getHiveClient().getCurrentNotificationEventId();
      long currentEventId = currentNotificationEventId.getEventId();
      // no new events since we last polled
      if (currentEventId <= latestEventId_.get()) {
        return;
      }
      // Fetch the last event to get its eventTime.
      NotificationEvent event = getEventFromHMS(msClient, currentEventId);
      // Events could be empty if they are just cleaned up.
      if (event == null) return;
      long lastSyncedEventId = lastSyncedEventId_.get();
      long lastSyncedEventTime = lastSyncedEventTimeSecs_.get();
      long currentEventTime = event.getEventTime();
      latestEventId_.set(currentEventId);
      latestEventTimeSecs_.set(currentEventTime);
      LOG.info("Latest event in HMS: id={}, time={}. Last synced event: id={}, time={}.",
          currentEventId, currentEventTime, lastSyncedEventId, lastSyncedEventTime);
      if (currentEventTime > lastSyncedEventTime) {
        LOG.warn("Lag: {}. {} events pending to be processed.",
            PrintUtils.printTimeMs((currentEventTime - lastSyncedEventTime) * 1000),
            currentEventId - lastSyncedEventId);
      }
    } catch (Exception e) {
      LOG.error("Unable to update current notification event id. Last value: {}",
          latestEventId_, e);
    }
  }

  /**
   * Gets the current event processor metrics along with its status. If the status is
   * not active the metrics are skipped. Only the status is sent
   */
  @Override
  public TEventProcessorMetrics getEventProcessorMetrics() {
    TEventProcessorMetrics eventProcessorMetrics = new TEventProcessorMetrics();
    EventProcessorStatus currentStatus = getStatus();
    eventProcessorMetrics.setStatus(currentStatus.toString());
    eventProcessorMetrics.setLast_synced_event_id(getLastSyncedEventId());
    if (currentStatus != EventProcessorStatus.ACTIVE) return eventProcessorMetrics;
    // The following counters are only updated when event-processor is active.
    eventProcessorMetrics.setLast_synced_event_time(lastSyncedEventTimeSecs_.get());
    eventProcessorMetrics.setLatest_event_id(latestEventId_.get());
    eventProcessorMetrics.setLatest_event_time(latestEventTimeSecs_.get());

    long eventsReceived = metrics_.getMeter(EVENTS_RECEIVED_METRIC).getCount();
    long eventsSkipped = metrics_.getCounter(EVENTS_SKIPPED_METRIC).getCount();
    eventProcessorMetrics.setEvents_received(eventsReceived);
    eventProcessorMetrics.setEvents_skipped(eventsSkipped);

    Snapshot fetchDuration =
        metrics_.getTimer(EVENTS_FETCH_DURATION_METRIC).getSnapshot();
    double avgFetchDuration = fetchDuration.getMean() / SECOND_IN_NANOS;
    double p75FetchDuration = fetchDuration.get75thPercentile() / SECOND_IN_NANOS;
    double p95FetchDuration = fetchDuration.get95thPercentile() / SECOND_IN_NANOS;
    double p99FetchDuration = fetchDuration.get99thPercentile() / SECOND_IN_NANOS;
    eventProcessorMetrics.setEvents_fetch_duration_mean(avgFetchDuration);
    eventProcessorMetrics.setEvents_fetch_duration_p75(p75FetchDuration);
    eventProcessorMetrics.setEvents_fetch_duration_p95(p95FetchDuration);
    eventProcessorMetrics.setEvents_fetch_duration_p99(p99FetchDuration);

    Snapshot processDuration =
        metrics_.getTimer(EVENTS_PROCESS_DURATION_METRIC).getSnapshot();
    double avgProcessDuration = processDuration.getMean() / SECOND_IN_NANOS;
    double p75ProcessDuration = processDuration.get75thPercentile() / SECOND_IN_NANOS;
    double p95ProcessDuration = processDuration.get95thPercentile() / SECOND_IN_NANOS;
    double p99ProcessDuration = processDuration.get99thPercentile() / SECOND_IN_NANOS;
    eventProcessorMetrics.setEvents_process_duration_mean(avgProcessDuration);
    eventProcessorMetrics.setEvents_process_duration_p75(p75ProcessDuration);
    eventProcessorMetrics.setEvents_process_duration_p95(p95ProcessDuration);
    eventProcessorMetrics.setEvents_process_duration_p99(p99ProcessDuration);

    double lastProcessDuration = lastEventProcessDurationNs_.get() /
        (double) SECOND_IN_NANOS;
    eventProcessorMetrics.setLast_events_process_duration(lastProcessDuration);

    double avgNumberOfEventsReceived1Min =
        metrics_.getMeter(EVENTS_RECEIVED_METRIC).getOneMinuteRate();
    double avgNumberOfEventsReceived5Min =
        metrics_.getMeter(EVENTS_RECEIVED_METRIC).getFiveMinuteRate();
    double avgNumberOfEventsReceived15Min =
        metrics_.getMeter(EVENTS_RECEIVED_METRIC).getFifteenMinuteRate();
    eventProcessorMetrics.setEvents_received_1min_rate(avgNumberOfEventsReceived1Min);
    eventProcessorMetrics.setEvents_received_5min_rate(avgNumberOfEventsReceived5Min);
    eventProcessorMetrics.setEvents_received_15min_rate(avgNumberOfEventsReceived15Min);

    LOG.trace("Events Received: {} Events skipped: {} Avg fetch duration: {} Avg process "
            + "duration: {} Events received rate (1min) : {}",
        eventsReceived, eventsSkipped, avgFetchDuration, avgProcessDuration,
        avgNumberOfEventsReceived1Min);
    return eventProcessorMetrics;
  }

  @Override
  public TEventProcessorMetricsSummaryResponse getEventProcessorSummary() {
    TEventProcessorMetricsSummaryResponse summaryResponse =
        new TEventProcessorMetricsSummaryResponse();
    summaryResponse.setSummary(metrics_.toString());
    if (eventProcessorErrorMsg_ != null) {
      summaryResponse.setError_msg(eventProcessorErrorMsg_);
    }
    return summaryResponse;
  }

  @VisibleForTesting
  Metrics getMetrics() {
    return metrics_;
  }

  /**
   * Process the given list of notification events. Useful for tests which provide a list
   * of events
   * @param currentEventId Current event id on metastore
   * @param events List of NotificationEvents
   * @throws MetastoreNotificationException
   */
  @VisibleForTesting
  protected void processEvents(long currentEventId, List<NotificationEvent> events)
      throws MetastoreNotificationException {
    currentEvent_ = null;
    // update the events received metric before returning
    metrics_.getMeter(EVENTS_RECEIVED_METRIC).mark(events.size());
    if (events.isEmpty()) {
      if (lastSyncedEventId_.get() < currentEventId) {
        // Possible to receive empty list due to event skip list in notification event
        // request. Update the last synced event id with current event id on metastore
        lastSyncedEventId_.set(currentEventId);
        lastSyncedEventTimeSecs_.set(getEventTimeFromHMS(currentEventId));
      }
      return;
    }
    final Timer.Context context =
        metrics_.getTimer(EVENTS_PROCESS_DURATION_METRIC).time();
    Map<MetastoreEvent, Long> eventProcessingTime = new HashMap<>();
    try {
      List<MetastoreEvent> filteredEvents =
          metastoreEventFactory_.getFilteredEvents(events, metrics_);
      if (filteredEvents.isEmpty()) {
        NotificationEvent e = events.get(events.size() - 1);
        lastSyncedEventId_.set(e.getEventId());
        lastSyncedEventTimeSecs_.set(e.getEventTime());
        return;
      }
      for (MetastoreEvent event : filteredEvents) {
        // synchronizing each event processing reduces the scope of the lock so the a
        // potential reset() during event processing is not blocked for longer than
        // necessary
        synchronized (this) {
          if (eventProcessorStatus_ != EventProcessorStatus.ACTIVE) {
            break;
          }
          currentEvent_ = event.metastoreNotificationEvent_;
          String targetName = event.getTargetName();
          String desc = String.format("Processing %s on %s, eventId=%d",
              event.getEventType(), targetName, event.getEventId());
          try (ThreadNameAnnotator tna = new ThreadNameAnnotator(desc)) {
            long startMs = System.currentTimeMillis();
            event.processIfEnabled();
            long elapsedTimeMs = System.currentTimeMillis() - startMs;
            eventProcessingTime.put(event, elapsedTimeMs);
          }
          deleteEventLog_.garbageCollect(event.getEventId());
          lastSyncedEventId_.set(event.getEventId());
          lastSyncedEventTimeSecs_.set(event.getEventTime());
          metrics_.getTimer(AVG_DELAY_IN_CONSUMING_EVENTS).update(
              (System.currentTimeMillis() / 1000) - event.getEventTime(),
                  TimeUnit.SECONDS);
        }
      }
    } catch (CatalogException e) {
      throw new MetastoreNotificationException(String.format(
          "Unable to process event %d of type %s. Event processing will be stopped.",
          currentEvent_.getEventId(), currentEvent_.getEventType()), e);
    } finally {
      long elapsedNs = context.stop();
      lastEventProcessDurationNs_.set(elapsedNs);
      logEventMetrics(eventProcessingTime, elapsedNs);
    }
  }

  private void logEventMetrics(Map<MetastoreEvent, Long> eventProcessingTime,
      long elapsedNs) {
    LOG.info("Time elapsed in processing event batch: {}",
        PrintUtils.printTimeNs(elapsedNs));
    // Only log the metrics when the processing on this batch is slow.
    if (elapsedNs < HdfsTable.LOADING_WARNING_TIME_NS) return;
    // Get the top-10 expensive events
    List<Map.Entry<MetastoreEvent, Long>> eventList =
        new ArrayList<>(eventProcessingTime.entrySet());
    eventList.sort(Map.Entry.<MetastoreEvent, Long>comparingByValue().reversed());
    int num = Math.min(10, eventList.size());
    StringBuilder report = new StringBuilder("Top " + num + " expensive events: ");
    for (Map.Entry<MetastoreEvent, Long> entry : eventList.subList(0, num)) {
      MetastoreEvent event = entry.getKey();
      long durationMs = entry.getValue();
      report.append(String.format("(type=%s, id=%s, target=%s, duration_ms=%d) ",
          event.getEventType(), event.getEventId(), event.getTargetName(), durationMs));
    }
    // Get the top-10 expensive targets
    Map<String, Long> durationPerTable = new HashMap<>();
    for (MetastoreEvent event : eventProcessingTime.keySet()) {
      String targetName = event.getTargetName();
      long durationMs = durationPerTable.getOrDefault(targetName, 0L) +
          eventProcessingTime.get(event);
      durationPerTable.put(targetName, durationMs);
    }
    List<Map.Entry<String, Long>> targetList =
        new ArrayList<>(durationPerTable.entrySet());
    targetList.sort(Map.Entry.<String, Long>comparingByValue().reversed());
    num = Math.min(10, targetList.size());
    report.append("\nTop ").append(num).append(" targets in event processing: ");
    for (Map.Entry<String, Long> entry : targetList.subList(0, num)) {
      String targetName = entry.getKey();
      long durationMs = entry.getValue();
      report.append(String.format("(target=%s, duration_ms=%d) ",
          targetName, durationMs));
    }
    LOG.warn(report.toString());
  }

  /**
   * Updates the current states to the given status.
   */
  private synchronized void updateStatus(EventProcessorStatus toStatus) {
    eventProcessorStatus_ = toStatus;
  }

  private void dumpEventInfoToLog(NotificationEvent event) {
    if (event == null) {
      String error = "Notification event is null";
      LOG.error(error);
      eventProcessorErrorMsg_ += '\n' + error;
      return;
    }
    StringBuilder msg =
        new StringBuilder().append("Event id: ").append(event.getEventId()).append("\n")
            .append("Event Type: ").append(event.getEventType()).append("\n")
            .append("Event time: ").append(event.getEventTime()).append("\n")
            .append("Database name: ").append(event.getDbName()).append("\n");
    if (event.getTableName() != null) {
      msg.append("Table name: ").append(event.getTableName()).append("\n");
    }
    msg.append("Event message: ").append(event.getMessage()).append("\n");
    String msgStr = msg.toString();
    LOG.error(msgStr);
    eventProcessorErrorMsg_ += '\n' + msgStr;
  }

  /**
   * Create a instance of this object if it is not initialized. Currently, this object is
   * a singleton and should only be created during catalogD initialization time, so that
   * the start syncId matches with the catalogD startup time.
   *
   * @param catalogOpExecutor the CatalogOpExecutor instance to which this event
   *     processor belongs.
   * @param startSyncFromId Start event id. Events will be polled starting from this
   *     event id
   * @param eventPollingInterval HMS polling interval in seconds
   * @return this object is already created, or create a new one if it is not yet
   *     instantiated
   */
  public static synchronized ExternalEventsProcessor getInstance(
      CatalogOpExecutor catalogOpExecutor, long startSyncFromId,
      long eventPollingInterval) throws CatalogException {
    if (instance != null) return instance;
    instance =
        new MetastoreEventsProcessor(catalogOpExecutor, startSyncFromId,
            eventPollingInterval);
    return instance;
  }

  @Override
  public MetastoreEventFactory getEventsFactory() {
    return metastoreEventFactory_;
  }

  public static MessageDeserializer getMessageDeserializer() {
    return MESSAGE_DESERIALIZER;
  }

  public static List<String> getEventSkipList() { return EVENT_SKIP_LIST; }
}