如何从自定义PartitionAssignor实现中获取机架id或用户信息

import org.apache.kafka.clients.consumer.ConsumerPartitionAssignor;
import org.apache.kafka.clients.consumer.ConsumerPartitionAssignor.Assignment;
import org.apache.kafka.clients.consumer.ConsumerPartitionAssignor.GroupSubscription;
import org.apache.kafka.clients.consumer.ConsumerPartitionAssignor.Subscription;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.utils.CircularIterator;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

public class CustomPartitionAssignor implements ConsumerPartitionAssignor {
    @Override
    public Map<String, Assignment> assign(Cluster metadata, GroupSubscription groupSubscription) {
        Map<String, Assignment> assignments = new HashMap<>();

        // 获取消费者订阅的主题和分区信息
        Subscription subscription = groupSubscription.subscription();
        Set<String> topics = subscription.topics();
        Map<String, List<TopicPartition>> topicPartitions = new HashMap<>();

        // 根据主题获取分区信息
        for (String topic : topics) {
            List<TopicPartition> partitions = metadata.partitionsForTopic(topic);
            topicPartitions.put(topic, partitions);
        }

        // 获取消费者的元数据信息
        Map<String, byte[]> userData = subscription.userData();

        // 根据机架id或用户信息进行分区分配
        for (String memberId : groupSubscription.members()) {
            List<TopicPartition> assignedPartitions = new ArrayList<>();

            // 根据memberId获取机架id或用户信息
            byte[] memberData = userData.get(memberId);
            String rackId = getRackId(memberData); // 获取机架id
            String userInfo = getUserInfo(memberData); // 获取用户信息

            // 根据机架id或用户信息进行分区分配
            for (String topic : topics) {
                List<TopicPartition> partitions = topicPartitions.get(topic);
                for (TopicPartition partition : partitions) {
                    // 根据机架id或用户信息进行分区分配策略
                    if (shouldAssignToPartition(partition, rackId, userInfo)) {
                        assignedPartitions.add(partition);
                    }
                }
            }

            assignments.put(memberId, new Assignment(new ArrayList<>(assignedPartitions)));
        }

        return assignments;
    }

    // 根据业务逻辑实现机架id的获取
    private String getRackId(byte[] memberData) {
        // 实现获取机架id的逻辑
        return "rack1";
    }

    // 根据业务逻辑实现用户信息的获取
    private String getUserInfo(byte[] memberData) {
        // 实现获取用户信息的逻辑
        return "user1";
    }

    // 根据业务逻辑实现分区分配策略
    private boolean shouldAssignToPartition(TopicPartition partition, String rackId, String userInfo) {
        // 实现分区分配策略的逻辑
        // 根据机架id或用户信息判断是否分配给该消费者
        return true;
    }
}