Spring-kafka 指定分区消费者，发现消费者不活动

V2EX = way to explore

V2EX 是一个关于分享和探索的地方

现在注册

已注册用户请登录

这是一个创建于 717 天前的主题，其中的信息可能已经有所发展或是发生改变。

问题

问题最近想试试 kafka 的相关操作，在 spring-kafka 3.0.1 版本下。发现当我指定消费者去消费指定分区时，kafka 无法检测到该消费者(消费组内无消费者)，但是却可以正常消费。如果不指定，而是由 kafka 自动平衡消费者的话，就可以检测到

示例代码


// 该代码正常
@KafkaListener(id = "consumer-all", clientIdPrefix = "consumer-all", topics = "topic1", groupId = "mygroup", concurrency = "1")
public void listenAllOne(ConsumerRecord<?, ?> consumerRecord, Acknowledgment ack) {
	System.out.println("这是第一个消费者-----"+consumerRecord.toString());
    ack.acknowledge();
}


// 该消费者可以正常消费，但是无法被 kafka 检测到
@KafkaListener(id="consumer-1", clientIdPrefix = "consumer-1", topicPartitions = {@TopicPartition(topic = "topic2", partitionOffsets = @PartitionOffset(partition = "0,1", initialOffset = "0"))}, groupId = "mygroup1", concurrency = "1")
public void listenTop1(ConsumerRecord<?, ?> consumerRecord, Acknowledgment ack) {
    System.out.println("这是第一个分区消费者-----"+consumerRecord.toString());
    ack.acknowledge();
}

@KafkaListener(id="consumer-2", clientIdPrefix = "consumer-2", topicPartitions = {@TopicPartition(topic = "topic2", partitionOffsets = @PartitionOffset(partition = "2,3", initialOffset = "0"))}, groupId = "mygroup1", concurrency = "1")
public void listenTop2(ConsumerRecord<?, ?> consumerRecord, Acknowledgment ack) {
    System.out.println("这是第二个分区消费者-----"+consumerRecord.toString());
    ack.acknowledge();
}

第 1 条附言 · 2023-01-06 16:41:17 +08:00

CLI查询的方式也是一样的结果，如图