public void received(Consumer<byte[]> consumer, Message<byte[]> msg) {
                                String body = null;
                                try {
                                    body = new String(msg.getData(), StandardCharsets.UTF_8);
                                    handleMessage(body, consumer, msg.getMessageId());
                                } catch (Exception exception) {
                                    MetricsManager.getInstance().addReceiveFailed(1);
                                    PulsarUtils.acknowledge(consumer, msg.getMessageId());
                                    LOG.error("Invalid audit data: {}", body, exception);
                                }
                            }
Constructor
@param insertServiceList
@param storeConfig
@param mqConfig
inlong-audit/audit-store/src/main/java/org/apache/inlong/audit/store/service/consume/PulsarConsume.java