Spring-kafka 2.7.4 - not able to connect to dev broker when using @RetryableTopic

[vivekkamineni]

I am using Spring-kafka 2.7.4 and Spring boot: 2.5.3

I'm running into a weird issue...

Here is my config:

@RetryableTopic(attempts = "4", backoff = @Backoff(delay = 300000, maxDelay = 300000 * 4, multiplier = 2), topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE, dltStrategy = DltStrategy.FAIL_ON_ERROR)
	@KafkaListener(topics = "devtopic", groupId = "devgroupid")
	public void consumeFromTopic(TestMessage testMessage){

 }

application.yml:

spring:
  kafka:
    consumer:
      bootstrap-servers: "dev_cluster1.com:9092, dev_cluster2.com:9092, dev_cluster3.com:9092"
      max-poll-records: 50
      key-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer  
      auto-offset-reset: earliest
      properties:
        spring.deserializer.key.delegate.class: org.apache.kafka.common.serialization.StringDeserializer
        spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer
        spring.json.value.default.type: com.test.model.TestMessage
        spring.json.trusted.packages: "*"
        spring.json.use.type.headers: false
        security:
          protocol: SASL_PLAINTEXT
        sasl:
          mechanism: PLAIN
          jaas:
            config: org.apache.kafka.common.security.plain.PlainLoginModule required username="username" password="password";
    listener:
      ack-mode: record
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

When I try running the app pointing to Dev clusters i see the following in the logs saying it is not able to connect to localhost:9092-

	bootstrap.servers = [localhost:9092]
	client.dns.lookup = use_all_dns_ips
	client.id = 
	connections.max.idle.ms = 300000
	default.api.timeout.ms = 60000
	metadata.max.age.ms = 300000
	metric.reporters = []
	metrics.num.samples = 2
	metrics.recording.level = INFO
	metrics.sample.window.ms = 30000
	receive.buffer.bytes = 65536
	reconnect.backoff.max.ms = 1000
	reconnect.backoff.ms = 50
	request.timeout.ms = 30000
	retries = 2147483647
	retry.backoff.ms = 100
	sasl.client.callback.handler.class = null
	sasl.jaas.config = null
	sasl.kerberos.kinit.cmd = /usr/bin/kinit
	sasl.kerberos.min.time.before.relogin = 60000
	sasl.kerberos.service.name = null
	sasl.kerberos.ticket.renew.jitter = 0.05
	sasl.kerberos.ticket.renew.window.factor = 0.8
	sasl.login.callback.handler.class = null
	sasl.login.class = null
	sasl.login.refresh.buffer.seconds = 300
	sasl.login.refresh.min.period.seconds = 60
	sasl.login.refresh.window.factor = 0.8
	sasl.login.refresh.window.jitter = 0.05
	sasl.mechanism = GSSAPI
	security.protocol = PLAINTEXT
	security.providers = null
	send.buffer.bytes = 131072
	socket.connection.setup.timeout.max.ms = 127000
	socket.connection.setup.timeout.ms = 10000
	ssl.cipher.suites = null
	ssl.enabled.protocols = [TLSv1.2]
	ssl.endpoint.identification.algorithm = https
	ssl.engine.factory.class = null
	ssl.key.password = null
	ssl.keymanager.algorithm = SunX509
	ssl.keystore.certificate.chain = null
	ssl.keystore.key = null
	ssl.keystore.location = null
	ssl.keystore.password = null
	ssl.keystore.type = JKS
	ssl.protocol = TLSv1.2
	ssl.provider = null
	ssl.secure.random.implementation = null
	ssl.trustmanager.algorithm = PKIX
	ssl.truststore.certificates = null
	ssl.truststore.location = null
	ssl.truststore.password = null
	ssl.truststore.type = JKS

2021-08-26 15:18:29.055  INFO 74336 --- [           main] o.a.kafka.common.utils.AppInfoParser     : Kafka version: 2.7.1
2021-08-26 15:18:29.056  INFO 74336 --- [           main] o.a.kafka.common.utils.AppInfoParser     : Kafka commitId: 61dbce85d0d41457
2021-08-26 15:18:29.056  INFO 74336 --- [           main] o.a.kafka.common.utils.AppInfoParser     : Kafka startTimeMs: 1630005509054
2021-08-26 15:18:29.078  WARN 74336 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.
2021-08-26 15:18:29.184  WARN 74336 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.
2021-08-26 15:18:29.290  WARN 74336 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.
2021-08-26 15:18:29.600  WARN 74336 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.
2021-08-26 15:18:30.011  WARN 74336 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.
2021-08-26 15:18:30.923  WARN 74336 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.
2021-08-26 15:18:32.099  INFO 74336 --- [on(2)-127.0.0.1] inMXBeanRegistrar$SpringApplicationAdmin : Application shutdown requested.
2021-08-26 15:18:32.140  WARN 74336 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.
2021-08-26 15:18:32.212  INFO 74336 --- [on(3)-127.0.0.1] inMXBeanRegistrar$SpringApplicationAdmin : Application shutdown requested.
2021-08-26 15:18:32.341  INFO 74336 --- [on(4)-127.0.0.1] inMXBeanRegistrar$SpringApplicationAdmin : Application shutdown requested.
2021-08-26 15:18:32.461  INFO 74336 --- [on(5)-127.0.0.1] inMXBeanRegistrar$SpringApplicationAdmin : Application shutdown requested.
2021-08-26 15:18:32.612  INFO 74336 --- [on(6)-127.0.0.1] inMXBeanRegistrar$SpringApplicationAdmin : Application shutdown requested.
2021-08-26 15:18:32.858  INFO 74336 --- [on(7)-127.0.0.1] inMXBeanRegistrar$SpringApplicationAdmin : Application shutdown requested.
2021-08-26 15:18:33.262  WARN 74336 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.
2021-08-26 15:18:34.179  WARN 74336 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.
2021-08-26 15:18:35.312  WARN 74336 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.
2021-08-26 15:18:36.434  WARN 74336 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.

However if my Kafka is running locally, somehow it is able to fetch records from the Dev broker..

The app fetches records from Dev if I comment out @RetryableTopic

Can you please let me know if I am missing anything in the configuration..

[garyrussell]

That makes no sense at all; it's hard for us to tell what's wrong without much more info.

Are you saying it works without the @RetryableTopic?

[vivekkamineni]

hi Gary,

Thanks for responding.

yes.

//@RetryableTopic(attempts = "4", backoff = @Backoff(delay = 300000, maxDelay = 300000 * 4, multiplier = 2), topicSuffixingStrategy //= TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE, dltStrategy = DltStrategy.FAIL_ON_ERROR)
	@KafkaListener(topics = "devtopic", groupId = "devgroupid")
	public void consumeFromTopic(TestMessage testMessage){

 }

I just commented out @RetryableTopic like above and kept everything else same, it is able to fetch records from Dev.
If i put back @RetryableTopic, it gives me the below logs:

2021-08-26 15:18:29.078  WARN 74336 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.
2021-08-26 15:18:29.184  WARN 74336 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.
2021-08-26 15:18:29.290  WARN 74336 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.
2021-08-26 15:18:29.600  WARN 74336 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.
2021-08-26 15:18:30.011  WARN 74336 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.
2021-08-26 15:18:30.923  WARN 74336 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.

Everything works well when i point to local kafka

[garyrussell]

I can't reproduce it; it works fine for me; something must be wrong with your configuration.

Can you provide a Minimal, Complete, Reproducible Example that exhibits this behavior?

[vivekkamineni]

demo.zip

@garyrussell -- i attached a simple project which tries to connect to our dev kafka (changed the names) along with the logs

Kafka should not be running locally to reproduce this.

Please review and let me know of any configuration changes, i am trying to connect to Dev instead of local Kafka

[garyrussell]

You are setting the bootstrap-servers for consumers only; the retry topic uses the KafkaAdmin to create the retry topics and it has no bootstrap servers specified.

Use spring.kafka.bootstrap-servers which will then be applied to consumers, producers, and the admin.

The separate properties for each is, say, for when you want to consume from one cluster and produce to a different one.

[vivekkamineni]

Thanks @garyrussell for the guidance! "spring.kafka.bootstrap-servers" resolved my issue. Here is my final yml:

//For kafka Admin - common properties
  kafka:
    bootstrap-servers: "dev_cluster1.com:9092, dev_cluster2.com:9092, dev_cluster3.com:9092"
    properties:
      spring.deserializer.key.delegate.class: org.apache.kafka.common.serialization.StringDeserializer
      spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer
      spring.json.value.default.type: com.test.model.TestMessage
      spring.json.trusted.packages: "*"
      spring.json.use.type.headers: false
      security:
        protocol: SASL_PLAINTEXT
      sasl:
        mechanism: PLAIN
        jaas:
          config: org.apache.kafka.common.security.plain.PlainLoginModule required username="username" password="password";
    listener:
      ack-mode: record   

    
// All my consumers should apply the below configuration overriding the defaults.    
    consumer:
      max-poll-records: 50
      auto-offset-reset: earliest
      key-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer      
      


// All my producers should apply the below configuration  
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer