private void acknowledge(List<JdbcDataPo> dataList) {
        Iterator<JdbcDataPo> iterator = dataList.iterator();
        while (iterator.hasNext()) {
            JdbcDataPo jdbcData = iterator.next();
            try {
                if (jdbcData.getConsumer() != null && jdbcData.getMessageId() != null) {
                    jdbcData.getConsumer().acknowledge(jdbcData.getMessageId());
                }
                iterator.remove();
            } catch (Exception exception) {
                LOG.error("Acknowledge has exception!", exception);
            }
        }
    }
insert

@param msgBody audit data reading from Pulsar or other MessageQueue.
inlong-audit/audit-store/src/main/java/org/apache/inlong/audit/store/service/JdbcService.java