@Override
        public void run() {
            boolean hasPermit;
            while (true) {
                hasPermit = false;
                long fetchTimeCost = -1;
                try {
                    if (context.getConfig().isStopConsume() || stopConsume) {
                        TimeUnit.MILLISECONDS.sleep(50);
                        continue;
                    }

                    if (sleepTime > 0) {
                        TimeUnit.MILLISECONDS.sleep(sleepTime);
                    }

                    context.acquireRequestPermit();
                    hasPermit = true;
                    context.addConsumeTime(null, -1);

                    long startFetchTime = System.currentTimeMillis();
                    Messages<byte[]> pulsarMessages = currentConsumer.batchReceive();
                    fetchTimeCost = System.currentTimeMillis() - startFetchTime;

                    if (null != pulsarMessages && pulsarMessages.size() != 0) {
                        processPulsarMsg(pulsarMessages, fetchTimeCost);
                        sleepTime = 0L;
                    } else {
                        context.addConsumeEmpty(null, -1, fetchTimeCost);
                        emptyFetchTimes++;
                        if (emptyFetchTimes >= context.getConfig().getEmptyPollTimes()) {
                            sleepTime = Math.min((sleepTime += context.getConfig().getEmptyPollSleepStepMs()),
                                    context.getConfig().getMaxEmptyPollSleepMs());
                            emptyFetchTimes = 0;
                        }
                    }
                } catch (Exception e) {
                    context.addConsumeError(null, -1, fetchTimeCost);
                    LOGGER.error("failed to fetch msg ", e);
                } finally {
                    if (hasPermit) {
                        context.releaseRequestPermit();
                    }
                }

                if (closed) {
                    break;
                }
            }
        }
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/PulsarMultiTopicsFetcher.java