private void fetchFromKafka() throws Exception {
            context.addConsumeTime(null, -1);

            long startFetchTime = System.currentTimeMillis();
            ConsumerRecords<byte[], byte[]> records = consumer
                    .poll(Duration.ofMillis(context.getConfig().getKafkaFetchWaitMs()));
            long fetchTimeCost = System.currentTimeMillis() - startFetchTime;

            if (null != records && !records.isEmpty()) {
                for (ConsumerRecord<byte[], byte[]> msg : records) {
                    List<MessageRecord> msgs = new ArrayList<>();
                    String topicName = msg.topic();
                    InLongTopic topic = onlineTopics.get(topicName);
                    String offsetKey = getOffset(topicName, msg.partition(), msg.offset());
                    List<InLongMessage> inLongMessages = deserializer
                            .deserialize(context, topic, getMsgHeaders(msg.headers()), msg.value());
                    context.addConsumeSuccess(topic, msg.partition(), inLongMessages.size(), msg.value().length,
                            fetchTimeCost);
                    int originSize = inLongMessages.size();
                    inLongMessages = interceptor.intercept(inLongMessages);
                    if (inLongMessages.isEmpty()) {
                        ack(offsetKey);
                        continue;
                    }
                    int filterSize = originSize - inLongMessages.size();
                    context.addConsumeFilter(topic, msg.partition(), filterSize);
                    msgs.add(new MessageRecord(fetchKey,
                            inLongMessages,
                            offsetKey, System.currentTimeMillis()));
                    handleAndCallbackMsg(msgs, topic, msg.partition());
                }
                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;
                }
            }
        }
put the received msg to onFinished method

@param messageRecords {@link List < MessageRecord >}
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/fetcher/kafka/KafkaMultiTopicsFetcher.java