private void processPulsarMsg(Messages<byte[]> messages, long fetchTimeCost) throws Exception {
            for (Message<byte[]> msg : messages) {
                String topicName = msg.getTopicName();
                InLongTopic topic = onlineTopics.get(topicName);
                if (Objects.isNull(topic)) {
                    LOGGER.error("got a message with topic {}, which is not subscribe", topicName);
                    continue;
                }
                // if need seek
                if (msg.getPublishTime() < seeker.getSeekTime()) {
                    seeker.seek();
                    break;
                }
                String offsetKey = getOffset(msg.getMessageId());
                currentConsumer.put(offsetKey, topic, msg.getMessageId());

                // deserialize
                List<InLongMessage> inLongMessages = deserializer
                        .deserialize(context, topic, msg.getProperties(), msg.getData());
                context.addConsumeSuccess(topic, -1, inLongMessages.size(), msg.getData().length,
                        fetchTimeCost);
                int originSize = inLongMessages.size();
                // intercept
                inLongMessages = interceptor.intercept(inLongMessages);
                if (inLongMessages.isEmpty()) {
                    ack(offsetKey);
                    continue;
                }
                int filterSize = originSize - inLongMessages.size();
                context.addConsumeFilter(topic, -1, filterSize);

                List<MessageRecord> msgs = new ArrayList<>();
                msgs.add(new MessageRecord(fetchKey,
                        inLongMessages,
                        offsetKey, System.currentTimeMillis()));
                handleAndCallbackMsg(msgs, topic);
            }
        }
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