Kafka消费者没有选择提到的引导服务器

Question

我正在尝试使用SSL实现Kafka消费者，在应用程序中提供所有必需的配置。

当我启动Spring启动Kafka消费者应用程序；消费者试图连接localhost:9092而不是提到Kafka经纪人。

KafkaConfig.java公司

@Bean
    public ConsumerFactory<String, AvroRecord> consumerFactory() throws IOException {
        return new DefaultKafkaConsumerFactory<>(kafkaProps());
    }

    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, AvroRecord>>
    kafkaListenerContainerFactory() throws IOException {
        ConcurrentKafkaListenerContainerFactory<String, AvroRecord> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }

kafkaProps（）正在加载所有与SSL和引导服务器相关的属性。值，我可以在调试模式中看到它。

应用程序.yml

kafka:
  properties:
    basic:
      auth:
        credentials:
          source: USER_INFO
          user: username
          pass: password
    enableAutoRegister: true
    max_count: 100
    max_delay: 5000
    schema:
      registry:
        url: https://schema-registry:8081
        ssl:
          truststore:
            location: <<location>>
            password: pwd
          keystore:
            location: <<location>>
            password: pwd
          key:
            password: pwd
    ssl:
      enabled: true
      protocols: TLSv1.2,TLSv1.1,TLSv1
      truststore:
        type: JKS
        location: <<location>>
        password: pwd
      keystore:
        type: JKS
        location: <<location>>
        password: pwd
      key:
        password: pwd
    security:
      protocol: SSL
    consumer:
      bootstrap-servers: broker1:9092,broker2:9092
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
      max-message-size: 10241024

在应用程序日志中，我得到了下面的日志

18:46:33.964 [main] INFO  o.a.k.c.a.AdminClientConfig requestId=
                    transactionKey= | AdminClientConfig values: 
    bootstrap.servers = [localhost:9092]
    client.dns.lookup = use_all_dns_ips
    client.id = 
    connections.max.idle.ms = 300000


15:53:54.608 [kafka-admin-client-thread | adminclient-1] WARN  o.a.k.c.NetworkClient requestId=
                    transactionKey= | [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.

我找不到它，为什么它连接到本地主机，而不是提到的代理

Answer 1

>

正确的属性是spring.kafka.bootstrap-服务器。您似乎完全缺少Spring前缀。此外，schema.registry.url、ssl.truststore等都被认为是Kafka客户端的单数属性键（字符串），因此（据我所知）不应“嵌套”在YAML对象中

您只尝试在使用者上设置bootstrap属性，而不是AdminClient

在与引导服务器字符串建立初始连接后，您的客户端将始终连接到代理的播发侦听器，因此，如果这是 localhost：9092，将解释 AdminClient 日志输出