Kafka自定义分区-每个分区用户数据自定义的使用者分配器

Question

我正在实现一个自定义消费者的主题/分区分配在Kafka。为此，我将重写AbstractPartitionAssignor抽象类，该类又实现ConsumerPartitionAssignor接口。

作为自定义赋值器的一部分，我希望发送一个关于消费者订阅的每个主题的每个分区的单个（浮动）信息。

我知道可以通过重写ConsumerPartitionAssignor接口的默认方法ByteBuffer subscriptionUserData(Set topics) 向赋值器发送自定义数据。

但是，问题是，从上面的方法签名中，我无法获得为使用者注册的每个主题分配给带下划线使用者的分区列表。

public static final class Subscription {
    private final List<String> topics;
    private final ByteBuffer userData;
    private final List<TopicPartition> ownedPartitions;
    private Optional<String> groupInstanceId;
    .....

谢谢你。

Answer 1

要解决上述问题，只需使用赋值器回调函数onassignment，如下所示

...
private List<TopicPartition> memberAssignment = null;
     @Override
        public void onAssignment(Assignment assignment, ConsumerGroupMetadata metadata) {
            memberAssignment = assignment.partitions();
            this.generation = metadata.generationId();
        }

并在函数subscriptionUserData中使用MemberAssignment（或Assignor类中的任何位置）获取当前分配给每个使用者的分区列表。

public ByteBuffer subscriptionUserData(Set<String> topics)