时间戳

首页 > 解决方案 > 如何在@KafkaListeners 中将 groupId 设置为 null

apache-kafka - 如何在@KafkaListeners 中将 groupId 设置为 null

问题描述

关于：这个问题

我正在尝试通过@KafkaListener 阅读压缩主题。我希望每个消费者每次都阅读整个主题。

我无法为每个消费者生成一个唯一的 groupId。所以我想使用一个空组ID。

我尝试将容器和使用者配置为将 groupId 设置为 null，但均未成功。

这是我的容器配置：

 ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
        configurer.configure(factory, kafkaConsumerFactory);
        // Set ackMode to manual and never commit, we are reading from the beginning each time
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        factory.getContainerProperties().setAckOnError(false);
        // Remove groupId, we are consuming all partitions here
        factory.getContainerProperties().setGroupId(null);
        // Enable idle event in order to detect when init phase is over
        factory.getContainerProperties().setIdleEventInterval(1000L);

还试图强制消费者配置：

Map<String, Object> consumerProperties = sprinfKafkaProperties.buildConsumerProperties();
        // Override group id property to force "null"
        consumerProperties.put(ConsumerConfig.GROUP_ID_CONFIG, null);
        ConsumerFactory<Object, Object> kafkaConsumerFactory = new DefaultKafkaConsumerFactory<>(consumerProperties);

当我将容器 groupId 设置为 null 时，将使用具有侦听器 ID 的默认值。

当我强制消费者使用 null groupId 属性时，出现错误：No group.id found in consumer config、container properties 或 @KafkaListener annotation；使用组管理时需要 group.id。

标签： apache-kafkaspring-kafka

解决方案

您不能使用 null group.id。

从卡夫卡文档。

group.id

标识此消费者所属的消费者组的唯一字符串。如果消费者使用 subscribe(topic) 或基于 Kafka 的偏移管理策略使用组管理功能，则需要此属性。

如果你想每次都从头开始阅读，你可以添加一个ConsumerAwareRebalanceListener到容器工厂或者让你的监听器实现ConsumerSeekAware。

在任何一种情况下，当onPartitionsAssigned被调用时，将每个主题/分区搜索到开头。

我无法为每个消费者生成一个唯一的 groupId。

您可以使用 SpEL 表达式来生成 UUID。

编辑

您可以手动分配主题/分区，然后 group.id 可以为空。

@SpringBootApplication
public class So56114299Application {

    public static void main(String[] args) {
        SpringApplication.run(So56114299Application.class, args);
    }

    @Bean
    public NewTopic topic() {
        return new NewTopic("so56114299", 10, (short) 0);
    }

    @KafkaListener(topicPartitions = @TopicPartition(topic = "so56114299",
                          partitions = "#{@finder.partitions('so56114299')}"))
    public void listen(@Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String key, String payload) {
        System.out.println(key + ":" + payload);
    }

    @Bean
    public PartitionFinder finder(ConsumerFactory<String, String> consumerFactory) {
        return new PartitionFinder(consumerFactory);
    }

    public static class PartitionFinder {

        public PartitionFinder(ConsumerFactory<String, String> consumerFactory) {
            this.consumerFactory = consumerFactory;
        }

        private final ConsumerFactory<String, String> consumerFactory;

        public String[] partitions(String topic) {
            try (Consumer<String, String> consumer = consumerFactory.createConsumer()) {
                return consumer.partitionsFor(topic).stream()
                    .map(pi -> "" + pi.partition())
                    .toArray(String[]::new);
            }
        }

    }

}


spring.kafka.consumer.enable-auto-commit=false
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.listener.ack-mode=manual

推荐阅读