private void addWriteNotificationLog(List<NotificationEvent> eventBatch, List<AcidWriteEvent> acidWriteEventList,
                                       Connection dbConn, SQLGenerator sqlGenerator, List<AcidWriteMessage> msgBatch)
          throws MetaException, SQLException {
    LOG.debug("DbNotificationListener: adding write notification log for : {}", eventBatch);
    assert ((dbConn != null) && (sqlGenerator != null));

    int numRows;
    long maxRows = MetastoreConf.getIntVar(conf, ConfVars.JDBC_MAX_BATCH_SIZE);

    try (Statement stmt = dbConn.createStatement()) {
      String st = sqlGenerator.getDbProduct().getPrepareTxnStmt();
      if (st != null) {
        stmt.execute(st);
      }
    } catch (Exception e) {
      LOG.error("Failed to execute query ", e);
      throw new MetaException(e.getMessage());
    }

    ResultSet rs = null;
    String select = sqlGenerator.addForUpdateClause("select \"WNL_ID\", \"WNL_FILES\" from" +
            " \"TXN_WRITE_NOTIFICATION_LOG\" " +
            "where \"WNL_DATABASE\" = ? " +
            "and \"WNL_TABLE\" = ? " + " and (\"WNL_PARTITION\" = ? OR (? IS NULL AND \"WNL_PARTITION\" IS NULL)) " +
            "and \"WNL_TXNID\" = ? ");
    List<Integer> insertList = new ArrayList<>();
    Map<Integer, Pair<Long, String>> updateMap = new HashMap<>();
    try (PreparedStatement pst = dbConn.prepareStatement(select)) {
      for (int i = 0; i < acidWriteEventList.size(); i++) {
        String dbName = acidWriteEventList.get(i).getDatabase();
        String tblName = acidWriteEventList.get(i).getTable();
        String partition = acidWriteEventList.get(i).getPartition();
        Long txnId = acidWriteEventList.get(i).getTxnId();

        LOG.debug("Going to execute query <" + select.replaceAll("\\?", "{}") + ">",
                quoteString(dbName), quoteString(tblName), quoteString(partition));
        pst.setString(1, dbName);
        pst.setString(2, tblName);
        pst.setString(3, partition);
        pst.setString(4, partition);
        pst.setLong(5, txnId);
        rs = pst.executeQuery();
        if (!rs.next()) {
          insertList.add(i);
        } else {
          updateMap.put(i, new Pair<>(rs.getLong(1), rs.getString(2)));
        }
      }
    } catch (Exception e) {
      LOG.error("Failed to execute insert ", e);
      throw new MetaException(e.getMessage());
    } finally {
      close(rs);
    }

    if (insertList.size() != 0) {
      // if rs is empty then no lock is taken and thus it can not cause deadlock.
      long nextNLId = getNextNLId(dbConn, sqlGenerator,
              "org.apache.hadoop.hive.metastore.model.MTxnWriteNotificationLog", insertList.size());

      String insert = "insert into \"TXN_WRITE_NOTIFICATION_LOG\" " +
              "(\"WNL_ID\", \"WNL_TXNID\", \"WNL_WRITEID\", \"WNL_DATABASE\", \"WNL_TABLE\", " +
              "\"WNL_PARTITION\", \"WNL_TABLE_OBJ\", \"WNL_PARTITION_OBJ\", " +
              "\"WNL_FILES\", \"WNL_EVENT_TIME\") VALUES (?,?,?,?,?,?,?,?,?,?)";
      try (PreparedStatement pst = dbConn.prepareStatement(sqlGenerator.addEscapeCharacters(insert))) {
        numRows = 0;
        for (int idx : insertList) {
          String tableObj = msgBatch.get(idx).getTableObjStr();
          String partitionObj = msgBatch.get(idx).getPartitionObjStr();
          String files = ReplChangeManager.joinWithSeparator(msgBatch.get(idx).getFiles());
          String dbName = acidWriteEventList.get(idx).getDatabase();
          String tblName = acidWriteEventList.get(idx).getTable();
          String partition = acidWriteEventList.get(idx).getPartition();
          int currentTime = now();

          pst.setLong(1, nextNLId++);
          pst.setLong(2, acidWriteEventList.get(idx).getTxnId());
          pst.setLong(3, acidWriteEventList.get(idx).getWriteId());
          pst.setString(4, dbName);
          pst.setString(5, tblName);
          pst.setString(6, partition);
          pst.setString(7, tableObj);
          pst.setString(8, partitionObj);
          pst.setString(9, files);
          pst.setInt(10, currentTime);
          LOG.debug("Going to execute insert <" + insert.replaceAll("\\?", "{}") + ">", nextNLId
                  , acidWriteEventList.get(idx).getTxnId(), acidWriteEventList.get(idx).getWriteId()
                  , quoteString(dbName), quoteString(tblName),
                  quoteString(partition), quoteString(tableObj), quoteString(partitionObj), quoteString(files), currentTime);
          pst.addBatch();
          numRows++;
          if (numRows == maxRows) {
            pst.executeBatch();
            numRows = 0;
          }
        }

        if (numRows != 0) {
          pst.executeBatch();
        }
      } catch (Exception e) {
        LOG.error("Failed to execute insert ", e);
        throw new MetaException(e.getMessage());
      }
    }

    if (updateMap.size() != 0) {
      String update = "update \"TXN_WRITE_NOTIFICATION_LOG\" set \"WNL_TABLE_OBJ\" = ? ," +
                " \"WNL_PARTITION_OBJ\" = ? ," +
                " \"WNL_FILES\" = ? ," +
                " \"WNL_EVENT_TIME\" = ?" +
                " where \"WNL_ID\" = ?";
      try (PreparedStatement pst = dbConn.prepareStatement(sqlGenerator.addEscapeCharacters(update))) {
        numRows = 0;
        for (Map.Entry entry : updateMap.entrySet()) {
          int idx = (int) entry.getKey();
          Pair<Long, String> nlIdInfo = (Pair<Long, String>) entry.getValue();
          String tableObj = msgBatch.get(idx).getTableObjStr();
          String partitionObj = msgBatch.get(idx).getPartitionObjStr();
          String files = ReplChangeManager.joinWithSeparator(msgBatch.get(idx).getFiles());
          String existingFiles = nlIdInfo.second;
          long nlId = nlIdInfo.first;
          int currentTime = now();

          if (existingFiles.contains(sqlGenerator.addEscapeCharacters(files))) {
            // If list of files are already present then no need to update it again. This scenario can come in case of
            // retry done to the meta store for the same operation.
            LOG.info("file list " + files + " already present");
            continue;
          }

          files = ReplChangeManager.joinWithSeparator(Lists.newArrayList(files, existingFiles));

          pst.setString(1, tableObj);
          pst.setString(2, partitionObj);
          pst.setString(3, files);
          pst.setInt(4, currentTime);
          pst.setLong(5, nlId);
          LOG.debug("Going to execute update <" + update.replaceAll("\\?", "{}") + ">",
                  quoteString(tableObj), quoteString(partitionObj), quoteString(files), currentTime, nlId);
          pst.addBatch();
          numRows++;
          if (numRows == maxRows) {
            pst.executeBatch();
            numRows = 0;
          }
        }

        if (numRows != 0) {
          pst.executeBatch();
        }
      } catch (Exception e) {
        LOG.error("Failed to execute update ", e);
        throw new MetaException(e.getMessage());
      }
    }
  }
Get the next event ID.

@return The next ID to use for an event.
@throws SQLException if a database access error occurs or this method is
          called on a closed connection
@throws MetaException if the sequence table is not properly initialized
hcatalog/server-extensions/src/main/java/org/apache/hive/hcatalog/listener/DbNotificationListener.java