Confluent-kafka-go: Logger integration

[jimrobinson]

I ran a test to see what happened when my application encountered a broker that was shut down. Instead of getting back an errors from kafka.NewProducer or from setting a (*kafka.Message).TopicPartition.Error on message submission, what I was seeing was that the library appears to be writing directly to stderr. An example:

%3|1490436188.086|ERROR|kpush#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv6#[::1]:9092 failed: Connection refused
%3|1490436188.086|ERROR|kpush#producer-1| [thrd:localhost:9092/bootstrap]: 1/1 brokers are down

Is there some way to control this behavior? I didn't see any options for controlling the logging output in

https://github.com/confluentinc/confluent-kafka-go
http://docs.confluent.io/3.2.0/clients/confluent-kafka-go/index.html

[edenhill]

Broker connection errors are considered temporary and the client will try to recover automatically, that's why these errors are not directly propagated as per-message errors but rather global errors (or logs).
The not-yet-produced messages will remain in internal queues until the relevant broker(s) comes up again or the messages time out (message.timeout.ms, def 5 minutes).

Regarding the logging output: it is not currently possible but it is on the roadmap.
Do you have a preference for how you would like to receive log messages? A Log event on the .Events channel? A separate channel? something else?

[jimrobinson]

Hi,

I think a separate "notification" log channel would be great. Otherwise possibly allow someone the option of the librdkafka library just calling the standard logger in https://golang.org/pkg/log/, thereby letting them handle sending the log to a file or whatever they want via that package's options.

[tchap]

Hi, I vote for a separate channel to receive a structured as well-defined errors, no strings please.

[eliaslevy]

May also want to give the option to just configure the log callback, which can be done in librdkafka via rd_kafka_conf_set_log_cb. The current lack of control over log output is unworkable in a number of environments.

[edenhill]

That is somewhat problematic since the librdkafka log_cb may be called from internal librdkafka threads, it might be safer to emit log messages as Events and let the application poll/consume them.
This is what we did in the Python client.

[eliaslevy]

I wonder about possible issues with that approach, as it would delay the logging of messages, which may be lot in the case of some fatal failure in the meantime. But it would certainly be an improvement on the current state.

[tchap]

A separate channel perhaps? It would have to enabled (by default disabled) using a special config key I guess so that the producer/consumer does not deadlock when the channel is not being received upon.

[eliaslevy]

That could work. A separate go routine can consume the log messages.

[scraly]

Is it some news about this point?
In a golang app using confluent-kafka-go, how to handle with connection to broker & librdkafka err logs?
Thx

[edenhill]

This is on our roadmap, we're aiming to add proper logging support later this year.
PRs are of course welcome before that :)
Corresponding Python client's logger, start here:
https://github.com/confluentinc/confluent-kafka-python/blob/master/confluent_kafka/src/confluent_kafka.c#L1669

[onorua]

I believe it is implemented in v0.11.6:

    for e := range p.Producer.Events() {
        switch ev := e.(type) {
        case *kafka.Message:
                        // reply message ....
        case kafka.Error:
                         log.Errorf("%v", ev)
            if ev.Code() == kafka.ErrAllBrokersDown {
                // do what is required when brokers are down
            }
        }

I believe this issue can be closed

[edenhill]

v0.11.6 added proper Error reporting, but there is still no support for specifying an alternate logger. All logs currently go to stderr.

We're scheduling this for early 2019, unless someone from the community wants to help out.

This boils down to enabling librdkafka log queue and modifying the event loop in event.go to translate librdkafka log events to the Go log framework.

[PChou]

I implement this recently, and use both separate goroutine and channel: https://github.com/confluentinc/confluent-kafka-go/pull/295

And i think this feature is quite useful for online issue debug.

[rigelbm]

Merged in #421.