Important: how do I distinguish between fatal errors and ignorable errors?

[winbatch]

I have a 3 node cluster. If I bring one if the nodes down, I get an error callback that one of the node can't be reached. I consider this non fatal since the other nodes are up and viable. Can you either not callback unless all nodes are down, or provide a means of knowing the severity?

[edenhill]

Most of the error reports from rdkafka are informational, there isnt much the application can or should do in case of these errors. But the case with all brokers being down is a good example of an error that the application really should know about, so I'll add two new error/resp codes to signal this:

• All brokers are down
• At least one broker is up

Do you see a use for a third one?:

• All brokers are up

Supplying a severity 'level' to the error_cb is a good idea.

You've highlighted a number of issues with the current API which I will fix for the next SONAME bump.
But since an SONAME bump requires a new package name (librdkafka1 -> librdkafka2) in Debian I want to accumulate as much as possible before doing it.

In the meantime I will try to document some workarounds for the application.

[winbatch]

I'd only want to know real errors (it's an error callback not an
informational callback). So I only want to know if all brokers are down .

On Saturday, January 25, 2014, Magnus Edenhill notifications@github.com
wrote:

Most of the error reports from rdkafka are informational, there isnt much
the application can or should do in case of these errors. But the case with
all brokers being down is a good example of an error that the application
really should know about, so I'll add two new error/resp codes to signal
this:

• All brokers are down
• At least one broker is up

Do you see a use for a third one?:

• All brokers are up

Supplying a severity 'level' to the error_cb is a good idea.

You've highlighted a number of issues with the current API which I will
fix for the next SONAME bump.
But since an SONAME bump requires a new package name (librdkafka1 ->
librdkafka2) in Debian I want to accumulate as much as possible before
doing it.

In the meantime I will try to document some workarounds for the
application.

—
Reply to this email directly or view it on GitHubhttps://github.com//issues/64#issuecomment-33308447
.

[edenhill]

Fair enough, there should probably be an event_cb for the informational stuff.

[winbatch]

Either that or a generic callback where the subscriber indicates the level
of minimal severity requested. (Very similar to how the logging callback
works)

On Saturday, January 25, 2014, Magnus Edenhill notifications@github.com
wrote:

Fair enough, there should probably be an event_cb for the informational
stuff.

—
Reply to this email directly or view it on GitHubhttps://github.com//issues/64#issuecomment-33308540
.

[edenhill]

You can now monitor your error_cb for err==RD_KAFKA_RESP_ERR__ALL_BROKERS_DOWN.
It is triggered when all brokers are in the down state and there has been at least one connection attempt for each one.

Please verify this on your end.

[winbatch]

let me check this with you. Currently any error callback I shut down the
process. Now, I should look for specific error codes and in only specific
cases, I should shutdown?

On Sun, Jan 26, 2014 at 8:08 AM, Magnus Edenhill
notifications@github.comwrote:

You can now monitor your error_cb for
err==RD_KAFKA_RESP_ERR__ALL_BROKERS_DOWN.
It is triggered when all brokers are in the down state and there has been
at least one connection attempt for each one.

Please verify this on your end.

—
Reply to this email directly or view it on GitHubhttps://github.com//issues/64#issuecomment-33316383
.

[edenhill]

Depends on your application's needs of course, but I would say that most errors reported by librdkafka have the potential of being transient/temporary.

That is; a RESOLVE failure could be because of the DNS not being available,
an UNKNOWN_TOPIC because of a cluster split,
and ALL_BROKERS_DOWN because of networking problems,
and so on.

Currently the only error codes signaled through the error_cb are:

• RD_KAFKA_RESP_ERR__ALL_BROKERS_DOWN - all brokers are down
• RD_KAFKA_RESP_ERR__FAIL - generic low level errors (socket failures because of lacking ipv4/ipv6 support)
• RD_KAFKA_RESP_ERR__RESOLVE - failure to resolve the broker address
• RD_KAFKA_RESP_ERR__CRIT_SYS_RESOURCE - failed to create new thread
• RD_KAFKA_RESP_ERR__FS - various file errors in the consumer offset management code
• RD_KAFKA_RESP_ERR__TRANSPORT - failed to connect to single broker, or connection error for single broker
• RD_KAFKA_RESP_ERR__BAD_MSG - received malformed packet from broker (version mismatch?)

I guess you could treat all but .._TRANSPORT as fatal.

The other error codes are signaled through the delivery report callback (dr_cb) and are message, topic or partition specific.

[winbatch]

Thanks - will use this as a guide. Had a somewhat unrelated question.
Once a successful connection is made to any broker, is the broker list
taken from the original list passed in by the user or is it replaced by the
list retrieved from the broker. For example. If my cluster is comprised
of host1, host2, and host3. If in my initial list I provide to librdkafka
I pass host1, host2, badhost4. Will it 'hold on' to badhost4 or will
replace the list with host1,host2,host3 upon successful connection?

On Sun, Jan 26, 2014 at 8:22 AM, Magnus Edenhill
notifications@github.comwrote:

Depends on your application's needs of course, but I would say that most
errors reported by librdkafka have the potential of being
transient/temporary.

That is; a RESOLVE failure could be because of the DNS not being available,
an UNKNOWN_TOPIC because of a cluster split,
and ALL_BROKERS_DOWN because of networking problems,
and so on.

Currently the only error codes signaled through the error_cb are:

• RD_KAFKA_RESP_ERR__ALL_BROKERS_DOWN - all brokers are down
• RD_KAFKA_RESP_ERR__FAIL - generic low level errors (socket failures
  because of lacking ipv4/ipv6 support)
• RD_KAFKA_RESP_ERR__RESOLVE - failure to resolve the broker address
• RD_KAFKA_RESP_ERR__CRIT_SYS_RESOURCE - failed to create new thread
• RD_KAFKA_RESP_ERR__FS - various file errors in the consumer offset
  management code
• RD_KAFKA_RESP_ERR__TRANSPORT - failed to connect to single broker,
  or connection error for single broker

I guess you could treat all but .._TRANSPORT as fatal.

The other error codes are signaled through the delivery report callback
(dr_cb) and are message, topic or partition specific.

—
Reply to this email directly or view it on GitHubhttps://github.com//issues/64#issuecomment-33316652
.

[edenhill]

The initial list of brokers you specify to librdkafka (either through config "metadata.broker.list" or rd_kafka_brokers_add()) are called the bootstrap brokers: rdkafka will connect to each one and retrieve metadata information containing all brokers and topics in the cluster. The brokers learnt through metadata is ADDED to the list of brokers and rdkafka will connect to them aswell.

In your example that means the final list of brokers in rdkafka would be: host1, host2, badhost4, host3

The bootstrap broker connections are never used for producing or consuming messages, only used for metadata, since they cant be reliably mapped to a specific broker instance.

[winbatch]

Got it. thanks.

On Sun, Jan 26, 2014 at 9:09 AM, Magnus Edenhill
notifications@github.comwrote:

The initial list of brokers you specify to librdkafka (either through
config "metadata.broker.list" or rd_kafka_brokers_add()) are called the
bootstrap brokers: rdkafka will connect to each one and retrieve metadata
information containing all brokers and topics in the cluster. The brokers
learnt through metadata is ADDED to the list of brokers and rdkafka will
connect to them aswell.

In your example that means the final list of brokers in rdkafka would be:
host1, host2, badhost4, host3

The bootstrap broker connections are never used for producing or consuming
messages, only used for metadata, since they cant be reliably mapped to a
specific broker instance.

—
Reply to this email directly or view it on GitHubhttps://github.com//issues/64#issuecomment-33317604
.

[winbatch]

Magnus- on a single broker not being reachable (I intentionally put in wrong port for one broker). I'm getting a -196 (ERR_FAIL). I though that would come as -195 (TRANSPORT)?

[edenhill]

That should now be fixed.
Thanks.

[edenhill]

Also note that I added ..__BAD_MSG to the list of error_cb() error codes in the comment above.

[winbatch]

It's interesting- I temporarily changed it to pass only one broker host
even though there are 3. I intentionally pass it the host that is down. I
was surprised to see that I get the callback for all brokers being down
(-187 ) and THEN the callback for the broker being down (-195). I would
have expected the opposite order

On Monday, January 27, 2014, Magnus Edenhill notifications@github.com
wrote:

Also note that I added ..__BAD_MSG to the list of error_cb() error codes
in the comment above.

—
Reply to this email directly or view it on GitHubhttps://github.com//issues/64#issuecomment-33395886
.

[edenhill]

Good point, the order is now reversed.

[winbatch]

I'm not sure this is working after the last update. I pass it a list of hosts (all valid hosts but kafka not running on them). I get a single callback saying that one of the hosts is unreachable, and no error callbacks thereafter. I expected:

• a callback for each host in the list that wasn't reachable
• a callback that all brokers were unreachable

However, if I pass in a single host, and that host is not reachable, I get the 2 callbacks as above.

[winbatch]

Slight update on the above. I do get all the callbacks, but only AFTER I attempt to produce something. As you know, I'd like it to fail immediately, which it seems to do if only a single broker is provided.

Also - when you run rdkafka_example as a consumer - if you pick a partition that has no messages (so it hangs) and then Ctrl-C it - it tells you that all brokers are down...

[edenhill]

rdkafka_example in producer mode is a bad example in this regard because it runs in a single thread that is blocked by fgets(stdin), and thus wont poll error_cb's (et.al) until enter is pressed or program is aborted.

With rdkafka_performance in consume mode I think it works correctly:

./rdkafka_performance -C -t x -p 0 -b localhost:9090,localhost:9093,localhost:9094
% Using random seed 1390965892
%3|1390965892.749|FAIL|rdkafka#consumer-0| localhost:9090/bootstrap: Failed to connect to broker at localhost:9090: Connection refused
%3|1390965892.749|FAIL|rdkafka#consumer-0| localhost:9094/bootstrap: Failed to connect to broker at [localhost]:9094: Connection refused
%3|1390965892.749|FAIL|rdkafka#consumer-0| localhost:9093/bootstrap: Failed to connect to broker at [localhost]:9093: Connection refused
% 0 messages and 0 bytes consumed in 1000ms: 0 msgs/s and 0.00 Mb/s, 0 messages failed, no compression
ERROR CALLBACK: rdkafka#consumer-0: Local: Broker transport failure: localhost:9090/bootstrap: Failed to connect to broker at localhost:9090: Connection refused
ERROR CALLBACK: rdkafka#consumer-0: Local: Broker transport failure: localhost:9094/bootstrap: Failed to connect to broker at [localhost]:9094: Connection refused
ERROR CALLBACK: rdkafka#consumer-0: Local: Broker transport failure: localhost:9093/bootstrap: Failed to connect to broker at [localhost]:9093: Connection refused
ERROR CALLBACK: rdkafka#consumer-0: Local: All broker connections are down: 3/3 brokers are down

[winbatch]

I didn't use example to test, was using my code.

On Tuesday, January 28, 2014, Magnus Edenhill notifications@github.com
wrote:

rdkafka_example in producer mode is a bad example in this regard because
it runs in a single thread that is blocked by fgets(stdin), and thus wont
poll error_cb's (et.al) until enter is pressed or program is aborted.

With rdkafka_performance in consume mode I think it works correctly:

./rdkafka_performance -C -t x -p 0 -b localhost:9090,localhost:9093,localhost:9094
% Using random seed 1390965892
%3|1390965892.749|FAIL|rdkafka#consumer-0| localhost:9090/bootstrap: Failed to connect to broker at localhost:9090: Connection refused
%3|1390965892.749|FAIL|rdkafka#consumer-0| localhost:9094/bootstrap: Failed to connect to broker at [localhost]:9094: Connection refused
%3|1390965892.749|FAIL|rdkafka#consumer-0| localhost:9093/bootstrap: Failed to connect to broker at [localhost]:9093: Connection refused
% 0 messages and 0 bytes consumed in 1000ms: 0 msgs/s and 0.00 Mb/s, 0 messages failed, no compression
ERROR CALLBACK: rdkafka#consumer-0: Local: Broker transport failure: localhost:9090/bootstrap: Failed to connect to broker at localhost:9090: Connection refused
ERROR CALLBACK: rdkafka#consumer-0: Local: Broker transport failure: localhost:9094/bootstrap: Failed to connect to broker at [localhost]:9094: Connection refused
ERROR CALLBACK: rdkafka#consumer-0: Local: Broker transport failure: localhost:9093/bootstrap: Failed to connect to broker at [localhost]:9093: Connection refused
ERROR CALLBACK: rdkafka#consumer-0: Local: All broker connections are down: 3/3 brokers are down

Reply to this email directly or view it on GitHubhttps://github.com//issues/64#issuecomment-33553255
.

[winbatch]

And the key is that it should fail without having to produce a message. The
same way it does with a single broker.

On Tuesday, January 28, 2014, Dan Hoffman hoffmandan@gmail.com wrote:

I didn't use example to test, was using my code.

On Tuesday, January 28, 2014, Magnus Edenhill <notifications@github.com<javascript:_e({}, 'cvml', 'notifications@github.com');>>
wrote:

rdkafka_example in producer mode is a bad example in this regard because
it runs in a single thread that is blocked by fgets(stdin), and thus wont
poll error_cb's (et.al) until enter is pressed or program is aborted.

With rdkafka_performance in consume mode I think it works correctly:

./rdkafka_performance -C -t x -p 0 -b localhost:9090,localhost:9093,localhost:9094
% Using random seed 1390965892
%3|1390965892.749|FAIL|rdkafka#consumer-0| localhost:9090/bootstrap: Failed to connect to broker at localhost:9090: Connection refused
%3|1390965892.749|FAIL|rdkafka#consumer-0| localhost:9094/bootstrap: Failed to connect to broker at [localhost]:9094: Connection refused
%3|1390965892.749|FAIL|rdkafka#consumer-0| localhost:9093/bootstrap: Failed to connect to broker at [localhost]:9093: Connection refused
% 0 messages and 0 bytes consumed in 1000ms: 0 msgs/s and 0.00 Mb/s, 0 messages failed, no compression
ERROR CALLBACK: rdkafka#consumer-0: Local: Broker transport failure: localhost:9090/bootstrap: Failed to connect to broker at localhost:9090: Connection refused
ERROR CALLBACK: rdkafka#consumer-0: Local: Broker transport failure: localhost:9094/bootstrap: Failed to connect to broker at [localhost]:9094: Connection refused
ERROR CALLBACK: rdkafka#consumer-0: Local: Broker transport failure: localhost:9093/bootstrap: Failed to connect to broker at [localhost]:9093: Connection refused
ERROR CALLBACK: rdkafka#consumer-0: Local: All broker connections are down: 3/3 brokers are down

Reply to this email directly or view it on GitHubhttps://github.com//issues/64#issuecomment-33553255
.

[edenhill]

I cant reproduce this with rdkafka_example, telling it to not send any messages (-c 0) and to idle (-I):

./rdkafka_performance -P -t x -p 0 -b localhost:9090,localhost:9093,localhost:9094 -c 0 -I
% Using random seed 1390974219
% Sending 0 messages of size 31 bytes
All messages produced, now waiting for 0 deliveries
%3|1390974219.028|FAIL|rdkafka#producer-0| localhost:9090/bootstrap: Failed to connect to broker at localhost:9090: Connection refused
ERROR CALLBACK: rdkafka#producer-0: Local: Broker transport failure: localhost:9090/bootstrap: Failed to connect to broker at localhost:9090: Connection refused
%3|1390974219.028|FAIL|rdkafka#producer-0| localhost:9094/bootstrap: Failed to connect to broker at localhost:9094: Connection refused
% Waiting for 0, 0 messages in outq to be sent. Abort with Ctrl-c
ERROR CALLBACK: rdkafka#producer-0: Local: Broker transport failure: localhost:9094/bootstrap: Failed to connect to broker at localhost:9094: Connection refused
%3|1390974219.028|FAIL|rdkafka#producer-0| localhost:9093/bootstrap: Failed to connect to broker at [localhost]:9093: Connection refused
ERROR CALLBACK: rdkafka#producer-0: Local: Broker transport failure: localhost:9093/bootstrap: Failed to connect to broker at [localhost]:9093: Connection refused
ERROR CALLBACK: rdkafka#producer-0: Local: All broker connections are down: 3/3 brokers are down

Are you sure that you call rd_kafka_poll() regularily even when not producing messages?

[winbatch]

define 'regularly' in this context? How many times do I need to call it after setting the broker list? I would assume only once. (Given that this is a callback mechanism, it's weird that I have to call it at all...)

[edenhill]

rd_kafka_poll() provides a way for the application to control which thread a callback is called in.
Without this approach rdkafka could call callbacks at any time in any thread, which will require the application to perform proper locking of its own resources, and it would also impair rdkafka's internal asynchronich design since a user callback could block.

So, an application that has registered at least one callback (error_cb, dr_cb, ..) must call rd_kafka_poll() at regular intervals. What that interval is specifically is up to the application; is it okay for the application to receive errors that are up to 5s old? Then poll it every 5 seconds.
Is the produce message rate in the application so low that it can allow 10s worth of delivery reports for messages to accumulate inside rdkafka? Then poll it every 10s.

But typically it is polled where it makes sense, after a (bunch of) produce() calls, or simply from a main dispatcher loop.
If none of this suits your needs, simply create a specific thread in your application that does nothing but poll rdkafka.

It really depends on what suits the application best.

Also, the application must not put assumptions on when a specific type of error callback is estimated to arrive on error, i.e.:

• Create rdkafka handles and add brokers
• Sleep 5s
• Call rd_kafka_poll() to check if the broker connection failed

This wont work if the host resolving of the brokers takes longer than 5s, or if the connection is slow.

Instead design along these lines:

• Create rdkafka handles and add brokers
• Operate as if Kafka is up and operational - rdkafka abstracts the current state of the cluster for the application so it doesnt need to bother with what brokers are up, which one is leader, if messages needs resending, etc. This means the application can start producing messages right after creating the handle, even if internally no brokers are connected yet.
• Handle errors when they happen, but assume they dont - stay positive!

[winbatch]

Why wouldn't the library simply call the callback only on the threads that
set the callback handler?

On Wednesday, January 29, 2014, Magnus Edenhill notifications@github.com
wrote:

rd_kafka_poll() provides a way for the application to control which
thread a callback is called in.
Without this approach rdkafka could call callbacks at any time in any
thread, which will require the application to perform proper locking of its
own resources, and it would also impair rdkafka's internal asynchronich
design since a user callback could block.

So, an application that has registered at least one callback (error_cb,
dr_cb, ..) must call rd_kafka_poll() at regular intervals. What that
interval is specifically is up to the application; is it okay for the
application to receive errors that are up to 5s old? Then poll it every 5
seconds.
Is the produce message rate in the application so low that it can allow
10s worth of delivery reports for messages to accumulate inside rdkafka?
Then poll it every 10s.

But typically it is polled where it makes sense, after a (bunch of)
produce() calls, or simply from a main dispatcher loop.
If none of this suits your needs, simply create a specific thread in your
application that does nothing but poll rdkafka.

It really depends on what suits the application best.

Also, the application must not put assumptions on when a specific type of
error callback is estimated to arrive on error, i.e.:

• Create rdkafka handles and add brokers
• Sleep 5s
• Call rd_kafka_poll() to check if the broker connection failed

This wont work if the host resolving of the brokers takes longer than 5s,
or if the connection is slow.

Instead design along these lines:

• Create rdkafka handles and add brokers
• Operate as if Kafka is up and operational - rdkafka abstracts the
  current state of the cluster for the application so it doesnt need to
  bother with what brokers are up, which one is leader, if messages needs
  resending, etc. This means the application can start producing messages
  right after creating the handle, even if internally no brokers are
  connected yet.
• Handle errors when they happen, but assume they dont - stay positive!

Reply to this email directly or view it on GitHubhttps://github.com//issues/64#issuecomment-33578833
.

[edenhill]

It is if you call rd_kafka_poll() from that thread :).
rdkafka only operates in its own threads, the only time it runs in the application's threads is when
the application calls one of the rdkafka functions, rd_kafka_poll() being one of them and the only one which will trigger the callbacks.

[winbatch]

Ok... To me though if I always have to make a call to fire a callback
that's no better than a get status method with a return code- I've list the
async nature that the callback should provide. For now I'll just call
poll more often ;)

On Wednesday, January 29, 2014, Magnus Edenhill notifications@github.com
wrote:

It is if you call rd_kafka_poll() from that thread :).
rdkafka only operates in its own threads, the only time it runs in the
application's threads is when
the application calls one of the rdkafka functions, rd_kafka_poll() being
one of them and the only one which will trigger the callbacks.

Reply to this email directly or view it on GitHubhttps://github.com//issues/64#issuecomment-33579325
.

[winbatch]

I've 'lost' (damn autocorrect)

On Wednesday, January 29, 2014, Dan Hoffman hoffmandan@gmail.com wrote:

Ok... To me though if I always have to make a call to fire a callback
that's no better than a get status method with a return code- I've list the
async nature that the callback should provide. For now I'll just call
poll more often ;)

On Wednesday, January 29, 2014, Magnus Edenhill <notifications@github.com<javascript:_e({}, 'cvml', 'notifications@github.com');>>
wrote:

It is if you call rd_kafka_poll() from that thread :).
rdkafka only operates in its own threads, the only time it runs in the
application's threads is when
the application calls one of the rdkafka functions, rd_kafka_poll() being
one of them and the only one which will trigger the callbacks.

Reply to this email directly or view it on GitHubhttps://github.com//issues/64#issuecomment-33579325
.

[edenhill]

Its impossible for rdkafka to spontaneously call a callback in an application thread since rdkafka does not run any code in those threads unless the application calls an rdkafka function.

Internally there's a queue of ops (error, dr, ..) that the rdkafka threads enqueue when things happen.
rd_kafka_poll() simply chews ops off this queue and calls the registered callbacks for each one.
It could just as well be an interface like:

while ((op = rd_kafka_get_next_op(rk))) {
  if (op->type == error)
    handle error;
  else if (op->type == dr)
    handle delivery report:
  ...

But I like the simple and isolated callback way.

[winbatch]

Any way for me to get the queue count or otherwise get callbacks for all
items on the queue with a single poll?

On Wednesday, January 29, 2014, Magnus Edenhill notifications@github.com
wrote:

Its impossible for rdkafka to spontaneously call a callback in an
application thread since rdkafka does not run any code in those threads
unless the application calls an rdkafka function.

Internally there's a queue of ops (error, dr, ..) that the rdkafka threads
enqueue when things happen.
rd_kafka_poll() simply chews ops off this queue and calls the registered
callbacks for each one.
It could just as well be an interface like:

while ((op = rd_kafka_get_next_op(rk))) {
if (op->type == error)
handle error;
else if (op->type == dr)
handle delivery report:
...

But I like the simple and isolated callback way.

Reply to this email directly or view it on GitHubhttps://github.com//issues/64#issuecomment-33579944
.

[edenhill]

rd_kafka_poll() will serve all ops on the queue in one go.
I.e., if there are 5 errors and 190 delivery reports you will get 5+190 callbacks in one rd_kafka_poll() call.

[winbatch]

Ok. Then in my case it must be as you stated- that my poll call is
happening before all host failures have been detected/added to the queue.
It may be that because you are using localhost that it happened quick
enough.

On Wednesday, January 29, 2014, Magnus Edenhill notifications@github.com
wrote:

rd_kafka_poll() will serve all ops on the queue in one go.
I.e., if there are 5 errors and 190 delivery reports you will get 5+190
callbacks in one rd_kafka_poll() call.

Reply to this email directly or view it on GitHubhttps://github.com//issues/64#issuecomment-33581165
.

[edenhill]

Yeah, makes sense. Keep polling! :)

[edenhill]

I guess this issue has been resolved, right?
Reopen if not.