private String getOffset(String topic, int partitionId, long offset) {
            TopicPartition topicPartition = new TopicPartition(topic, partitionId);
            ackOffsetMap.computeIfAbsent(topicPartition, k -> new ConcurrentSkipListMap<>()).put(offset, false);
            return topic + ":" + partitionId + ":" + offset;
        }
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