No error for failure to connect to broker

[srinivasvsk]

Somewhat unintentionally i started the spring boot app with KafkaListener without running any broker. It doesnt complain during start up for failure to connect the kafka broker.
Shouldn't this stop server startup like how other message listeners work in spring.
Thanks

[garyrussell]

@kafka only logs such situations at the debug level...

11:22:02.864 [container-kafka-consumer-1] DEBUG o.a.kafka.common.network.Selector - Connection with localhost/127.0.0.1 disconnected
java.net.ConnectException: Connection refused
    at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)

Since the Kafka client manages the connection, not Spring, there's not much we can do.

On the sending side, (every 60 seconds by default) you will see...

11:27:29.742 [main] ERROR o.s.k.s.LoggingProducerListener - Exception thrown when sending a message with key='si.key' and payload='foo0' to topic si.topic:
org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms.

[artembilan]

No, it shouldn't.

Consider the case when your application has many other features and one of them is fallback to some other service when Kafka Broker is down.
When the last one comes back, the Listener Container starts to proceed with message consumption from the topic(s).

Does it make sense for you?

[srinivasvsk]

@garyrussell i see it comes out from the kafka error logs. But if you think of health check on spring boot, it would be great to see something like this on the health endpoint right?

[srinivasvsk]

@artembilan if i remember right you would do for a JMS connection

[artembilan]

on the health endpoint

That would be there when Spring Boot Kafka support will be merged spring-projects/spring-boot#6961

[srinivasvsk]

if its coming it would be great, is there work around mean time

[artembilan]

You can borrow some ideas from here: http://docs.spring.io/spring-boot/docs/1.4.2.RELEASE/reference/htmlsingle/#production-ready-health:

Health information is collected from all HealthIndicator beans defined in your ApplicationContext. Spring Boot includes a number of auto-configured HealthIndicators and you can also write your own.

[mbogoevici]

We can borrow some ideas from the Kafka binder health indicator: https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/blob/master/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java

One particular problem that we've run into is that 'health' is more nuanced in the case of Kafka than for other systems. There is no notion of an generic connection to the cluster - consumers consume specific topics/partititions. A cluster may be healthy with respect to some topics (i.e. there's at least one in-sync replica and there is a leader for each) and unhealthy with respect to others, so it depends of the client's perspective.

One possible criterion to use for deciding on health is based on the state of existing consumers.

[srinivasvsk]

Thanks for the pointer

[srinivasvsk]

Interestingly, you need to a create new KafkaConsumer for health check. Just a thought can the underlying KafkaConsumer instances created in the messagelistener's exposed , so maybe can reuse the same consumer to do something similar to what you do in the kafkabinderhealthindicator

[tkvangorder]

I've run into an interesting side effect of the Kafka communication when the broker is not available.

If you disable the creation of topics and the ability to add partitions in the kafka binder:

spring:
  cloud:
    stream:
      kafka:
        binder:
          autoCreateTopics: false
          autoAddPartitions: false

And then startup your application producer (without the broker running), the application "freezes". I enabled the debug logging in the Kafka selector and very quickly saw it was attempting (and failing) to connect to the broker in a very tight loop.

This presents two problems:

1. There is no feedback that the broker is not available (unless you have debug enabled on the Kafka selector).

2. I am unable to terminate the process from Eclipse. I either have to:

   Run the process via Debug and then have to terminate the kafka producer thread.
   or "kill -9" the process.

[artembilan]

@tkvangorder,

thank you for feddback, but is there some pointer that it is exactly Spring Kafka problem?
Maybe that is how Apache Kafka client is done and would be better to share your experience there?
I'm afraid we can't fix Apache Kafka issues here from Spring Kafka perspective.

Thank you for understanding.

[tkvangorder]

@artembilan

Makes sense, I do think this more a problem with the underlying Kafka client. The use of Spring Cloud Stream + Kafka Binder + "autoCreateTopics = true" works only because of coincidence: It uses the Kafka admin utility in an attempt to create the topic BEFORE the connection to kafka is made. If the zookeeper cluster is not listening, this call fails with some feedback and the application stops.

I will see if I can get some clarification from the Kafka team on how to configure the client such that it fails a little more gracefully.

It also might be possible for the Spring Kafka project to "ping" the server before connecting to it, but admittedly, that seems like a hack.

[artembilan]

Pay attention that producer has sync property to configure: https://docs.spring.io/spring-cloud-stream/docs/Chelsea.SR2/reference/htmlsingle/index.html#_kafka_producer_properties

when you set it into true you will have an exception directly in the calling thread.
In async mode (default) the exception is indeed thrown to the thread in the internal Kafka client executor.

I'm closing this since it isn't related to Spring Kafka and we don't have any control over Kafka client internals.