Confluent-kafka-go: Why does kafka producer connection not throw an error when the brokers are down

[kafkauser-symc]

Description

Hello - Maybe, I'm missing something obvious here. When I try to connect to kafka which is not running, why doesn't kafka.NewProducer throw and error?

How to reproduce

p, err := kafka.NewProducer(&kafka.ConfigMap{
    "bootstrap.servers": "localhost:9092",
    "default.topic.config": kafka.ConfigMap{
        "acks": "all",
    },
})

   // err is never nil.
if err != nil {
    fmt.Printf("Failed to create producer: %s\n", err)
    os.Exit(1)
}

Checklist

Please provide the following information:

• [ 722431 0.11.5] confluent-kafka-go and librdkafka version (LibraryVersion()):
• [1.0.0 ] Apache Kafka broker version:
• [ ] Client configuration: ConfigMap{...}
• [macOS HighSierra ] Operating system:
• [ ] Provide client logs (with "debug": ".." as necessary)
• [ ] Provide broker log excerpts
• [ ] Critical issue

[bernhard-42]

Same happens to me when I use
p, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": "master1.localdomain"})
which seems to use a default port 9092 instead of
p, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": "master1.localdomain:6667"})

So it looks like a wrong port is also sufficient for the wrong behaviour.

Btw., the first error message I get is when I try to produce the first event:
master1.localdomain:9092/bootstrap: Connect to ipv4#192.168.124.171:9092 failed: Connection refused

Client: macos 10.13.6, librdkafka 0.11.5
Server: linux (HDP 3.0): kafka 1.0.1

[amikar]

Maybe listen to the event channel for the error ?

[kafkauser-symc]

Doesn't the event channel report delivery errors once you start producing messages? Here, I'd like to fail fast and respond to errors.

Here's what the doc says:
Delivery reports are by default emitted on the .Events() channel as *kafka.Message and you should check msg.TopicPartition.Error for nil to find out if the message was succesfully delivered or not.

[aarjan]

It's only possible to know that the connection is not ok only when you get the error delivery message from msg.TopicParition.Err after a default Timeout which is also unknown, since it is retrieved from the C dependency!! :( :(

[edenhill]

You can configure the message timeout with "message.timeout.ms"

[aarjan]

Thanks to @edenhill i got the solution.
You can listen to the kafka.Error from the .Events() channel, which contains the error message with code ErrAllBrokersDown.

case e := <-app.Producer.Events():
                switch ev := e.(type) {
                case *kafka.Message:
                    \\ do things

                case kafka.Error:
                                 fmt.Printf("kafka error: %v,e)