Kafka consumer not picking mentioned Bootstrap servers

user_4242765 0
Viewed 1460

I am trying to implement Kafka consumer with SSL, provide all the required configurations in the application.yml;

When I start the spring boot Kafka consumer application; Consumer is trying to connect the localhost:9092 instead of mentioned Kafka Brokers.

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() is loading all the SSL and bootstrap servers related properties. Values, I can see it in the debug mode.

application.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

In the application logs, I am getting the below log

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.

I am not able to find it, why it is connecting to localhost instead of mentioned brokers

spring-kafka apache-kafka spring-boot
1 Answers

  1. The correct property is spring.kafka.bootstrap-servers. You appear to be missing the spring prefix completely. Also, schema.registry.url, ssl.truststore, etc are all considered singular property keys (strings) to Kafka clients, so (to my knowledge) therefore should not be "nested" in YAML objects

  2. You only tried to set the bootstrap property on the consumer, not the AdminClient

  3. Your client will always connect to advertised.listeners of the broker after making the initial connection to the bootstrap server string, so if that is localhost:9092, would explain the AdminClient log output

user_23086830
Related