private void handleAndCallbackMsg(MessageRecord messageRecord) {
            long start = System.currentTimeMillis();
            try {
                context.addCallBack(topic, -1);
                context.getConfig().getCallback().onFinishedBatch(Collections.singletonList(messageRecord));
                context.addCallBackSuccess(topic, -1, 1,
                        System.currentTimeMillis() - start);
            } catch (Exception e) {
                context.addCallBackFail(topic, -1, 1,
                        System.currentTimeMillis() - start);
                LOG.error("failed to callback {}", e.getMessage(), e);
            }
        }
put the received msg to onFinished method

@param messageRecord {@link MessageRecord}
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/fetcher/tube/TubeSingleTopicFetcher.java