Kafka使用者无法使用Spring Boot 2.2.0.m4连接到除LocalHost:9092以外的代理

Question

我使用Spring Boot 2.2.0.m4和Kafka 2.2.0试图基于https://www.baeldung.com/Spring-Kafka的示例构建一个应用程序。当我为我的主题启用侦听器时，我在使用者上得到以下错误。

[AdminClient Clientid=AdminClient-2]无法建立到节点-1（本地主机/127.0.0.1:9092)的连接。代理可能不可用。

@KafkaListener(topics = "add_app", groupId = "foo")
public void listen(String message) {
    System.out.println("Received Message in group foo: " + message);
}

下面是引用Kafka.BootStrapAddress值的使用者配置类。它被正确地记录。

@Configuration
@Slf4j
public class KafkaConsumerConfig {

    @Value(value = "${kafka.bootstrapAddress}")
    private String bootstrapAddress;

    public ConsumerFactory<String, String> consumerFactory(String groupId) {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        log.info("Created {} using address {}.", this.getClass(), bootstrapAddress);
        return new DefaultKafkaConsumerFactory<>(props);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> fooKafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory("foo"));
        return factory;
    }

Answer 1

解决这个问题的方法相当简单。我只需要将以下内容添加到application.properties文件中。

bootstrap-servers=174.22.22.55:9092

在查看了kafkaproperties.java之后，我发现了这一行：

private List<String> bootstrapServers = new ArrayList<>(Collections.singletonList("localhost:9092"));

private Map<String, Object> buildCommonProperties() {
        Map<String, Object> properties = new HashMap();
        if (this.bootstrapServers != null) {
            properties.put("bootstrap.servers", this.bootstrapServers);
        }

        if (this.clientId != null) {
            properties.put("client.id", this.clientId);
        }

        properties.putAll(this.ssl.buildProperties());
        if (!CollectionUtils.isEmpty(this.properties)) {
            properties.putAll(this.properties);
        }

        return properties;
    }

将containerFactory属性添加到侦听器注释中也可以修复它，并消除对Application.Properties的更改。

@KafkaListener(topics = "add_app", groupId = "foo", containerFactory = "fooKafkaListenerContainerFactory")
public void listen(String message) {
    System.out.println("Received Message in group foo: " + message);
}