@Override
    public void insert(AuditData msgBody, Consumer<byte[]> consumer, MessageId messageId) {
        if (!isAuditDataValid(msgBody)) {
            MetricsManager.getInstance().addInvalidData();
            PulsarUtils.acknowledge(consumer, messageId);
            LOG.error("Invalid audit data: {} ", msgBody);
            return;
        }
        JdbcDataPo data = new JdbcDataPo();
        data.setConsumer(consumer);
        data.setMessageId(messageId);
        data.setIp(msgBody.getIp());
        data.setThreadId(msgBody.getThreadId());
        data.setDockerId(msgBody.getDockerId());
        data.setPacketId(msgBody.getPacketId());
        data.setSdkTs(new Timestamp(msgBody.getSdkTs()));
        data.setLogTs(new Timestamp(msgBody.getLogTs()));
        data.setAuditId(msgBody.getAuditId());
        data.setAuditTag(msgBody.getAuditTag());
        data.setAuditVersion(msgBody.getAuditVersion());
        data.setCount(msgBody.getCount());
        data.setDelay(msgBody.getDelay());
        data.setInLongGroupId(msgBody.getInlongGroupId());
        data.setInLongStreamId(msgBody.getInlongStreamId());
        data.setSize(msgBody.getSize());
        data.setUpdateTime(new Timestamp(System.currentTimeMillis()));
        try {
            receiveQueue.offer(data, Long.MAX_VALUE, TimeUnit.MILLISECONDS);
        } catch (InterruptedException exception) {
            LOG.error("Insert data has InterruptedException ", 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