private void handleAndCallbackMsg(List<MessageRecord> messageRecords) {
            long start = System.currentTimeMillis();
            try {
                context.addCallBack(topic, -1);
                context.getConfig().getCallback().onFinishedBatch(messageRecords);
                context.addCallBackSuccess(topic, -1, messageRecords.size(),
                        System.currentTimeMillis() - start);
            } catch (Exception e) {
                context.addCallBackFail(topic, -1, messageRecords.size(),
                        System.currentTimeMillis() - start);
                LOGGER.error("failed to callback", e);
            }
        }
put the received msg to onFinished method

@param messageRecords {@link List}
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/fetcher/pulsar/PulsarSingleTopicFetcher.java