java - 设置并发> 1后spring kafka thorws InstanceAlreadyExistsException异常

Question

我正在使用spring-kafka，如果我不设置ConcurrentKafkaListenerContainerFactory的并发，一切正常，当我将它设置为大于1的数字时，我得到一个异常:

javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=client-3

我的配置:

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> 
    kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new 
    ConcurrentKafkaListenerContainerFactory<String, String>();

    factory.setConcurrency(kafkaConfig.getConcurrency());

    factory.getContainerProperties().setAckMode(AckMode.MANUAL);
    factory.setConsumerFactory(consumerFactory());
    return factory;
}

属性:

kafka.enable-auto-commit=false

kafka.client-id=client-1

kafka.concurrency=2

Answer 1

我打开了一个issue for this on github .目前不支持为每个线程设置不同的 client.id。

作为变通方法，您可以为每个启动一个单独的 KafkaMessageListenerContainer(这是 ConcurrentMessageListenerContainer 在内部执行的操作)。

编辑

虽然不理想，但您可以省略client.id，kafka客户端会为每个生成一个(consumer-1，consumer-2等)