public static <T> Map<String, Map<Integer, T>> groupPartitionDataByTopic(Map<TopicPartition, ? extends T> data) {
        Map<String, Map<Integer, T>> dataByTopic = new HashMap<>();
        for (Map.Entry<TopicPartition, ? extends T> entry : data.entrySet()) {
            String topic = entry.getKey().topic();
            int partition = entry.getKey().partition();
            Map<Integer, T> topicData = dataByTopic.computeIfAbsent(topic, t -> new HashMap<>());
            topicData.put(partition, entry.getValue());
        }
        return dataByTopic;
    }
group data by topic

@param data Data to be partitioned
@param <T> Partition data type
@return partitioned data
clients/src/main/java/org/apache/kafka/common/utils/CollectionUtils.java