@VisibleForTesting public static long updateUsingNotificationEvents(RawStore rawStore, long lastEventId)
      throws Exception {
    LOG.debug("updating cache using notification events starting from event id " + lastEventId);
    NotificationEventRequest rqst = new NotificationEventRequest(lastEventId);

    //Add the events which are not related to metadata update
    rqst.addToEventTypeSkipList(MessageBuilder.INSERT_EVENT);
    rqst.addToEventTypeSkipList(MessageBuilder.OPEN_TXN_EVENT);
    rqst.addToEventTypeSkipList(MessageBuilder.COMMIT_TXN_EVENT);
    rqst.addToEventTypeSkipList(MessageBuilder.ABORT_TXN_EVENT);
    rqst.addToEventTypeSkipList(MessageBuilder.ALLOC_WRITE_ID_EVENT);
    rqst.addToEventTypeSkipList(MessageBuilder.ACID_WRITE_EVENT);
    rqst.addToEventTypeSkipList(MessageBuilder.CREATE_FUNCTION_EVENT);
    rqst.addToEventTypeSkipList(MessageBuilder.DROP_FUNCTION_EVENT);
    rqst.addToEventTypeSkipList(MessageBuilder.CREATE_ISCHEMA_EVENT);
    rqst.addToEventTypeSkipList(MessageBuilder.ALTER_ISCHEMA_EVENT);
    rqst.addToEventTypeSkipList(MessageBuilder.DROP_ISCHEMA_EVENT);
    rqst.addToEventTypeSkipList(MessageBuilder.ADD_SCHEMA_VERSION_EVENT);
    rqst.addToEventTypeSkipList(MessageBuilder.ALTER_SCHEMA_VERSION_EVENT);
    rqst.addToEventTypeSkipList(MessageBuilder.DROP_SCHEMA_VERSION_EVENT);

    Deadline.startTimer("getNextNotification");
    NotificationEventResponse resp = rawStore.getNextNotification(rqst);
    Deadline.stopTimer();

    if (resp == null || resp.getEvents() == null) {
      LOG.debug("no events to process");
      return lastEventId;
    }

    List<NotificationEvent> eventList = resp.getEvents();
    LOG.debug("num events to process" + eventList.size());

    for (NotificationEvent event : eventList) {
      long eventId = event.getEventId();
      if (eventId <= lastEventId) {
        LOG.error("Event id is not valid " + lastEventId + " : " + eventId);
        throw new RuntimeException(" event id is not valid " + lastEventId + " : " + eventId);
      }
      lastEventId = eventId;
      String message = event.getMessage();
      LOG.debug("Event to process " + event);
      MessageDeserializer deserializer = MessageFactory.getInstance(event.getMessageFormat()).getDeserializer();
      String catalogName = event.getCatName() == null ? "" : event.getCatName().toLowerCase();
      String dbName = event.getDbName() == null ? "" : event.getDbName().toLowerCase();
      String tableName = event.getTableName() == null ? "" : event.getTableName().toLowerCase();
      if (!shouldCacheTable(catalogName, dbName, tableName)) {
        continue;
      }
      switch (event.getEventType()) {
      case MessageBuilder.ADD_PARTITION_EVENT:
        AddPartitionMessage addPartMessage = deserializer.getAddPartitionMessage(message);
        sharedCache.addPartitionsToCache(catalogName, dbName, tableName, addPartMessage.getPartitionObjs());
        break;
      case MessageBuilder.ALTER_PARTITION_EVENT:
        AlterPartitionMessage alterPartitionMessage = deserializer.getAlterPartitionMessage(message);
        sharedCache
            .alterPartitionInCache(catalogName, dbName, tableName, alterPartitionMessage.getPtnObjBefore().getValues(),
                alterPartitionMessage.getPtnObjAfter());
        //TODO : Use the stat object stored in the alter table message to update the stats in cache.
        updateStatsForAlterPart(rawStore, alterPartitionMessage.getTableObj(), catalogName, dbName, tableName,
            alterPartitionMessage.getPtnObjAfter());
        break;
      case MessageBuilder.ALTER_PARTITIONS_EVENT:
        AlterPartitionsMessage alterPtnsMessage = deserializer.getAlterPartitionsMessage(message);
        List<List<String>> part_vals = new ArrayList<>();
        alterPtnsMessage.getPartitionObjs().forEach(part -> part_vals.add(part.getValues()));
        sharedCache.removePartitionsFromCache(catalogName, dbName, tableName, part_vals);
        break;
      case MessageBuilder.DROP_PARTITION_EVENT:
        DropPartitionMessage dropPartitionMessage = deserializer.getDropPartitionMessage(message);
        for (Map<String, String> partMap : dropPartitionMessage.getPartitions()) {
          sharedCache.removePartitionFromCache(catalogName, dbName, tableName, new ArrayList<>(partMap.values()));
        }
        break;
      case MessageBuilder.CREATE_TABLE_EVENT:
        CreateTableMessage createTableMessage = deserializer.getCreateTableMessage(message);
        sharedCache.addTableToCache(catalogName, dbName, tableName, createTableMessage.getTableObj());
        break;
      case MessageBuilder.ALTER_TABLE_EVENT:
        AlterTableMessage alterTableMessage = deserializer.getAlterTableMessage(message);
        sharedCache.alterTableInCache(catalogName, dbName, tableName, alterTableMessage.getTableObjAfter());
        //TODO : Use the stat object stored in the alter table message to update the stats in cache.
        updateStatsForAlterTable(rawStore, alterTableMessage.getTableObjBefore(), alterTableMessage.getTableObjAfter(),
            catalogName, dbName, tableName);
        break;
      case MessageBuilder.DROP_TABLE_EVENT:
        DropTableMessage dropTableMessage = deserializer.getDropTableMessage(message);
        int batchSize = MetastoreConf.getIntVar(rawStore.getConf(), ConfVars.BATCH_RETRIEVE_OBJECTS_MAX);
        String tableDnsPath = null;
        Path tablePath = new Path(dropTableMessage.getTableObj().getSd().getLocation());
        if (tablePath != null) {
          tableDnsPath = new Warehouse(rawStore.getConf()).getDnsPath(tablePath).toString();
        }

        while (true) {
          Map<String, String> partitionLocations =
              rawStore.getPartitionLocations(catalogName, dbName, tableName, tableDnsPath, batchSize);
          if (partitionLocations == null || partitionLocations.isEmpty()) {
            break;
          }
          sharedCache
              .removePartitionFromCache(catalogName, dbName, tableName, new ArrayList<>(partitionLocations.values()));
        }
        sharedCache.removeTableFromCache(catalogName, dbName, tableName);
        break;
      case MessageBuilder.CREATE_DATABASE_EVENT:
        CreateDatabaseMessage createDatabaseMessage = deserializer.getCreateDatabaseMessage(message);
        sharedCache.addDatabaseToCache(createDatabaseMessage.getDatabaseObject());
        break;
      case MessageBuilder.ALTER_DATABASE_EVENT:
        AlterDatabaseMessage alterDatabaseMessage = deserializer.getAlterDatabaseMessage(message);
        sharedCache.alterDatabaseInCache(catalogName, dbName, alterDatabaseMessage.getDbObjAfter());
        break;
      case MessageBuilder.DROP_DATABASE_EVENT:
        sharedCache.removeDatabaseFromCache(catalogName, dbName);
        break;
      case MessageBuilder.CREATE_CATALOG_EVENT:
      case MessageBuilder.DROP_CATALOG_EVENT:
      case MessageBuilder.ALTER_CATALOG_EVENT:
        // TODO : Need to add cache invalidation for catalog events
        LOG.error("catalog Events are not supported for cache invalidation : " + event.getEventType());
        break;
      case MessageBuilder.UPDATE_TBL_COL_STAT_EVENT:
        UpdateTableColumnStatMessage msg = deserializer.getUpdateTableColumnStatMessage(message);
        sharedCache.alterTableAndStatsInCache(catalogName, dbName, tableName, msg.getWriteId(),
            msg.getColumnStatistics().getStatsObj(), msg.getParameters());
        break;
      case MessageBuilder.DELETE_TBL_COL_STAT_EVENT:
        DeleteTableColumnStatMessage msgDel = deserializer.getDeleteTableColumnStatMessage(message);
        sharedCache.removeTableColStatsFromCache(catalogName, dbName, tableName, msgDel.getColName());
        break;
      case MessageBuilder.UPDATE_PART_COL_STAT_EVENT:
        UpdatePartitionColumnStatMessage msgPartUpdate = deserializer.getUpdatePartitionColumnStatMessage(message);
        sharedCache.alterPartitionAndStatsInCache(catalogName, dbName, tableName, msgPartUpdate.getWriteId(),
            msgPartUpdate.getPartVals(), msgPartUpdate.getParameters(),
            msgPartUpdate.getColumnStatistics().getStatsObj());
        break;
      case MessageBuilder.DELETE_PART_COL_STAT_EVENT:
        DeletePartitionColumnStatMessage msgPart = deserializer.getDeletePartitionColumnStatMessage(message);
        sharedCache.removePartitionColStatsFromCache(catalogName, dbName, tableName, msgPart.getPartValues(),
            msgPart.getColName());
        break;
      case MessageBuilder.ADD_PRIMARYKEY_EVENT:
        AddPrimaryKeyMessage addPrimaryKeyMessage = deserializer.getAddPrimaryKeyMessage(message);
        sharedCache.addPrimaryKeysToCache(catalogName, dbName, tableName, addPrimaryKeyMessage.getPrimaryKeys());
        break;
      case MessageBuilder.ADD_FOREIGNKEY_EVENT:
        AddForeignKeyMessage addForeignKeyMessage = deserializer.getAddForeignKeyMessage(message);
        for (SQLForeignKey fk : addForeignKeyMessage.getForeignKeys()) {
          // This is done because dbName and tblName for Foreign key events are currently set to PK table and db.
          sharedCache.addForeignKeysToCache(catalogName, fk.getFktable_db(), fk.getFktable_name(), Arrays.asList(fk));
        }
        break;
      case MessageBuilder.ADD_NOTNULLCONSTRAINT_EVENT:
        AddNotNullConstraintMessage notNullConstraintMessage = deserializer.getAddNotNullConstraintMessage(message);
        sharedCache.addNotNullConstraintsToCache(catalogName, dbName, tableName,
            notNullConstraintMessage.getNotNullConstraints());
        break;
      case MessageBuilder.ADD_UNIQUECONSTRAINT_EVENT:
        AddUniqueConstraintMessage uniqueConstraintMessage = deserializer.getAddUniqueConstraintMessage(message);
        sharedCache.addUniqueConstraintsToCache(catalogName, dbName, tableName,
            uniqueConstraintMessage.getUniqueConstraints());
        break;
      case MessageBuilder.ADD_DEFAULTCONSTRAINT_EVENT:
        AddDefaultConstraintMessage defaultConstraintMessage = deserializer.getAddDefaultConstraintMessage(message);
        sharedCache.addDefaultConstraintsToCache(catalogName, dbName, tableName,
            defaultConstraintMessage.getDefaultConstraints());
        break;
      case MessageBuilder.ADD_CHECKCONSTRAINT_EVENT:
        AddCheckConstraintMessage checkConstraintMessage = deserializer.getAddCheckConstraintMessage(message);
        sharedCache
            .addCheckConstraintsToCache(catalogName, dbName, tableName, checkConstraintMessage.getCheckConstraints());
        break;
      case MessageBuilder.DROP_CONSTRAINT_EVENT:
        DropConstraintMessage dropConstraintMessage = deserializer.getDropConstraintMessage(message);
        sharedCache.removeConstraintFromCache(catalogName, dbName, tableName, dropConstraintMessage.getConstraint());
        break;
      default:
        LOG.error("Event is not supported for cache invalidation : " + event.getEventType());
      }
    }
    return lastEventId;
  }
Similar to setConf but used from within the tests
This does start the background thread for prewarm and update
@param conf
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/cache/CachedStore.java