Librdkafka: Kafka consumer skips messages

[AlexeyRaga]

Description

I log every single offset I receive, and occasionally I have strange gaps like:

[2018-03-12 04:58:35] Rebalancing, partitions to assign: [PartitionId 44]
[2018-03-12 04:58:36] poll: KafkaResponseError RdKafkaRespErrTimedOut
[2018-03-12 04:58:36] Prefetcher channels has been cleaned
[2018-03-12 04:58:36] crValue: (PartitionId 44,Offset 31101)
[2018-03-12 04:58:36] crValue: (PartitionId 44,Offset 31102)
[2018-03-12 04:58:36] crValue: (PartitionId 44,Offset 31103)
[2018-03-12 04:58:36] crValue: (PartitionId 44,Offset 31104)
[2018-03-12 04:58:36] crValue: (PartitionId 44,Offset 52554)
[2018-03-12 04:58:36] crValue: (PartitionId 44,Offset 52555)
[2018-03-12 04:58:36] crValue: (PartitionId 44,Offset 52556)

Notice that messages with the offsets (31104..52554) are missing, no rebalance happened at that time. No other instances of the job ever handled these offsets. Some of the other partitions also have similar gaps occasionally.

These offsets are definitely in the topic/partition. Compaction is not enabled.

How to reproduce

It is very hard to reproduce, sometimes it happens, sometimes it does not. We are working on understanding how to reproduce it quickly and reliably, so far without much luck.

It is reproducible with master.

Checklist

IMPORTANT: We will close issues where the checklist has not been completed.

Please provide the following information:

• [x] librdkafka version (release number or git tag): b581d0d9df282847f76e8b9e87337161959d39c9
• [x] Apache Kafka version: 0.11
• [x] librdkafka client configuration: enable.auto.commit=false, queued.max.messages.kbytes=10000, log.connection.close=false, auto.offset.reset=earliest
• [x] Operating system: Ubuntu 16.4
• [ ] Provide logs (with debug=.. as necessary) from librdkafka
• [x] Provide broker log excerpts (nothing special about these logs)
• [x] Critical issue (critical since it blocks us very much in production)

[dsturnbull]

This may be related:

[2018-03-13 05:39:47] [CONSUME] [thrd:172.32.9.57:9092/1003]: 172.32.9.57:9092/1003: Enqueue 13 message(s) (13 ops) on vodka--atlasdos-submissions-balanced-20171129 [41] fetch queue (qlen 13, v2, last_offset 41051)
[2018-03-13 05:39:47] [CONSUME] [thrd:172.32.9.57:9092/1003]: 172.32.9.57:9092/1003: Enqueue 12 message(s) (12 ops) on vodka--atlasdos-submissions-balanced-20171129 [41] fetch queue (qlen 12, v2, last_offset 41063)
[2018-03-13 05:39:47] [CONSUME] [thrd:172.32.9.57:9092/1003]: 172.32.9.57:9092/1003: Enqueue 11 message(s) (11 ops) on vodka--atlasdos-submissions-balanced-20171129 [41] fetch queue (qlen 11, v2, last_offset 41074)
[2018-03-13 05:39:47] [CONSUME] [thrd:172.32.9.57:9092/1003]: 172.32.9.57:9092/1003: Enqueue 4 message(s) (21475 ops) on vodka--atlasdos-submissions-balanced-20171129 [41] fetch queue (qlen 21475, v2, last_offset 41074)
[2018-03-13 05:39:47] [FETCH] [thrd:172.32.9.57:9092/1003]: 172.32.9.57:9092/1003: Topic vodka--atlasdos-submissions-balanced-20171129 [41] in state active at offset 0 (160481/100000 msgs, 20145/10000 kb queued, opv 2) is not fetchable: queued.min.messages exceeded
[2018-03-13 05:39:47] [FETCHADD] [thrd:172.32.9.57:9092/1003]: 172.32.9.57:9092/1003: Removed vodka--atlasdos-submissions-balanced-20171129 [41] from fetch list (0 entries, opv 2)
[2018-03-13 05:40:43] [OFFSET] [thrd:main]: Topic vodka--atlasdos-submissions-balanced-20171129 [41]: stored offset -1001, committed offset -1001
[2018-03-13 05:40:43] [OFFSET] [thrd:main]: Topic vodka--atlasdos-submissions-balanced-20171129 [41]: setting offset INVALID for commit
[2018-03-13 05:41:44] [OFFSET] [thrd:main]: Topic vodka--atlasdos-submissions-balanced-20171129 [41]: stored offset -1001, committed offset -1001
[2018-03-13 05:41:44] [OFFSET] [thrd:main]: Topic vodka--atlasdos-submissions-balanced-20171129 [41]: setting offset INVALID for commit
[2018-03-13 05:42:45] [OFFSET] [thrd:main]: Topic vodka--atlasdos-submissions-balanced-20171129 [41]: stored offset -1001, committed offset -1001
[2018-03-13 05:42:45] [OFFSET] [thrd:main]: Topic vodka--atlasdos-submissions-balanced-20171129 [41]: setting offset INVALID for commit
[2018-03-13 05:43:47] [OFFSET] [thrd:main]: Topic vodka--atlasdos-submissions-balanced-20171129 [41]: stored offset -1001, committed offset -1001
[2018-03-13 05:43:47] [OFFSET] [thrd:main]: Topic vodka--atlasdos-submissions-balanced-20171129 [41]: setting offset INVALID for commit
[2018-03-13 05:44:46] [OFFSET] [thrd:main]: Topic vodka--atlasdos-submissions-balanced-20171129 [41]: stored offset -1001, committed offset -1001
[2018-03-13 05:44:46] [OFFSET] [thrd:main]: Topic vodka--atlasdos-submissions-balanced-20171129 [41]: setting offset INVALID for commit
[2018-03-13 05:45:48] [OFFSET] [thrd:main]: Topic vodka--atlasdos-submissions-balanced-20171129 [41]: stored offset -1001, committed offset -1001
[2018-03-13 05:45:48] [OFFSET] [thrd:main]: Topic vodka--atlasdos-submissions-balanced-20171129 [41]: setting offset INVALID for commit
[2018-03-13 05:46:49] [OFFSET] [thrd:main]: Topic vodka--atlasdos-submissions-balanced-20171129 [41]: stored offset -1001, committed offset -1001
[2018-03-13 05:46:49] [OFFSET] [thrd:main]: Topic vodka--atlasdos-submissions-balanced-20171129 [41]: setting offset INVALID for commit
[2018-03-13 05:47:51] [OFFSET] [thrd:main]: Topic vodka--atlasdos-submissions-balanced-20171129 [41]: stored offset -1001, committed offset -1001
[2018-03-13 05:47:51] [OFFSET] [thrd:main]: Topic vodka--atlasdos-submissions-balanced-20171129 [41]: setting offset INVALID for commit
[2018-03-13 05:48:51] [OFFSET] [thrd:main]: Topic vodka--atlasdos-submissions-balanced-20171129 [41]: stored offset -1001, committed offset -1001
[2018-03-13 05:48:51] [OFFSET] [thrd:main]: Topic vodka--atlasdos-submissions-balanced-20171129 [41]: setting offset INVALID for commit
[2018-03-13 05:49:52] [OFFSET] [thrd:main]: Topic vodka--atlasdos-submissions-balanced-20171129 [41]: stored offset -1001, committed offset -1001
[2018-03-13 05:49:52] [OFFSET] [thrd:main]: Topic vodka--atlasdos-submissions-balanced-20171129 [41]: setting offset INVALID for commit
[2018-03-13 05:50:55] [OFFSET] [thrd:main]: Topic vodka--atlasdos-submissions-balanced-20171129 [41]: stored offset -1001, committed offset 

[AlexeyRaga]

Some more info on how librdkafka is used in this scenario:

We create a new _messages_ queue (using rd_kafka_queue_new).

Then we redirect messages to that queue using the following algorithm:
When RD_KAFKA_RESP_ERR_ASSIGN_PARTITIONS happens we redirect newly assigned partitions to that our queue (rd_kafka_queue_forward) _before_ calling rd_kafka_assign.
Then we poll from that messages queue when we need messages.

We redirect callbacks to the default consumer queue (rd_kafka_poll_set_consumer)

We do rd_kafka_consumer_poll for callbacks/events asynchronously.
On top we call it (with timeout=0) every time before polling a message, just in case.

We never delete or un-forward these queues, as has been discussed in Gitter, log is here: forward_queues.txt

Is it possible that there is some kind of a race condition?
Like if rd_kafka_queue_forward happens a bunch of messages is going to some other queue? I thought that doing before rd_kafka_assign should prevent that though.

Is there anything wrong in that partitions/queues handling algorithm?

[AlexeyRaga]

So far we _suspect_ that the issue has something to do with rebalancing.
When the job doesn't need to rebalance the issue doesn't manifest itself. However, when we start killing consumers (which forces other jobs within the group to rebalance) we start seeing these "jumps".

[AlexeyRaga]

I have found a workaround: in a rebalance callback right after rd_kafka_assign I get committed positions for all of the assigned partitions (or beginning if there are no committed offsets) and rd_kafka_seek to these offsets.

[AlexeyRaga]

No, looks like it is still jumping :(

[edenhill]

In your initial comment you wrote:

no rebalance happened at that time.

but later:

When the job doesn't need to rebalance the issue doesn't manifest itself.

Which is it? :)

[edenhill]

We do rd_kafka_consumer_poll for callbacks/events asynchronously.
On top we call it (with timeout=0) every time before polling a message, just in case.

Are you checking the return value of that latter consumer_poll() call, does it ever return you a message?

[edenhill]

It would be interesting to see this reproduced with debug=cgrp,topic,fetch

[AlexeyRaga]

No, I never never check the result of consumer_poll() under the assumption that it shouldn't return anything because of forwarding.
Hmm, are you saying that there can be a race condition so consumer_poll is called between assign and forward?
I will definitely check that out.

[AlexeyRaga]

I have removed consumer_poll from the place where we poll for messages, now it is only called in one place in an asynchronous loop that handles callbacks.

Here is what we see at the consumer side when printing all the offsets:

[Info] [2018-03-15 23:20:38] Rebalancing, partitions to assign: [PartitionId 48,PartitionId 49,PartitionId 50,PartitionId 51,PartitionId 52,PartitionId 53,PartitionId 54,PartitionId 55,PartitionId 56,PartitionId 57,PartitionId 58,PartitionId 59]
[Info] [2018-03-15 23:20:38] Prefetcher channels has been cleaned
[Info] [2018-03-15 23:20:38] (PartitionId 52,Offset 0)
[Info] [2018-03-15 23:20:38] (PartitionId 52,Offset 1)
[Info] [2018-03-15 23:20:38] (PartitionId 52,Offset 2)
[Info] [2018-03-15 23:20:38] (PartitionId 57,Offset 6376)
[Info] [2018-03-15 23:20:38] (PartitionId 57,Offset 6377)
[Info] [2018-03-15 23:20:38] (PartitionId 57,Offset 6378)
[Info] [2018-03-15 23:20:38] (PartitionId 57,Offset 6379)

Here are the relevant librdkafka logs, the problematic partition is [57]:

%7|1521156039.056|OP|rdkafka#consumer-1| [thrd:main]: lo-werphat--atlasdos-submissions-balanced-20171129 [57] received op FETCH_START (v2) in fetch-state none (opv1)
%7|1521156039.056|FETCH|rdkafka#consumer-1| [thrd:main]: Start fetch for lo-werphat--atlasdos-submissions-balanced-20171129 [57] in state none at offset INVALID (v2)
%7|1521156039.056|PARTSTATE|rdkafka#consumer-1| [thrd:main]: Partition lo-werphat--atlasdos-submissions-balanced-20171129 [57] changed fetch state none -> offset-query
%7|1521156039.056|OFFSET|rdkafka#consumer-1| [thrd:main]: lo-werphat--atlasdos-submissions-balanced-20171129 [57]: offset reset (at offset INVALID) to BEGINNING: no previously committed offset available: Local: No offset stored
%7|1521156039.056|OFFREQ|rdkafka#consumer-1| [thrd:main]: 172.31.10.150:9092/1005: Partition lo-werphat--atlasdos-submissions-balanced-20171129 [57]: querying for logical offset BEGINNING (opv 2)
%7|1521156039.056|OFFSET|rdkafka#consumer-1| [thrd:main]: 172.31.10.150:9092/1005: OffsetRequest (v0, opv 0) for 1 topic(s) and 1 partition(s)
%7|1521156039.056|PARTSTATE|rdkafka#consumer-1| [thrd:main]: Partition lo-werphat--atlasdos-submissions-balanced-20171129 [57] changed fetch state offset-query -> offset-wait
%7|1521156039.057|FETCHDEC|rdkafka#consumer-1| [thrd:172.31.10.150:9092/1005]: Topic lo-werphat--atlasdos-submissions-balanced-20171129 [57]: fetch decide: updating to version 2 (was 0) at offset 0 (was 0)
%7|1521156039.069|FETCH|rdkafka#consumer-1| [thrd:172.31.10.150:9092/1005]: 172.31.10.150:9092/1005: Topic lo-werphat--atlasdos-submissions-balanced-20171129 [57] in state active at offset 0 (12797/100000 msgs, 1660/1000 kb queued, opv 2) is not fetchable: queued.max.messages.kbytes exceeded
%7|1521156039.069|FETCHADD|rdkafka#consumer-1| [thrd:172.31.10.150:9092/1005]: 172.31.10.150:9092/1005: Removed lo-werphat--atlasdos-submissions-balanced-20171129 [57] from fetch list (0 entries, opv 2)
%7|1521156039.075|CONSUME|rdkafka#consumer-1| [thrd:172.31.10.150:9092/1005]: 172.31.10.150:9092/1005: Enqueue 6345 message(s) (6345 ops) on lo-werphat--atlasdos-submissions-balanced-20171129 [57] fetch queue (qlen 6345, v2, last_offset 6344)
%7|1521156039.075|FETCH|rdkafka#consumer-1| [thrd:172.31.10.150:9092/1005]: 172.31.10.150:9092/1005: Topic lo-werphat--atlasdos-submissions-balanced-20171129 [57] in state active at offset 0 (6345/100000 msgs, 822/1000 kb queued, opv 2) is fetchable: 
%7|1521156039.075|FETCHADD|rdkafka#consumer-1| [thrd:172.31.10.150:9092/1005]: 172.31.10.150:9092/1005: Added lo-werphat--atlasdos-submissions-balanced-20171129 [57] to fetch list (2 entries, opv 2)
%7|1521156039.075|FETCH|rdkafka#consumer-1| [thrd:172.31.10.150:9092/1005]: 172.31.10.150:9092/1005: Fetch topic lo-werphat--atlasdos-submissions-balanced-20171129 [57] at offset 6345 (v2)
%7|1521156039.076|CONSUME|rdkafka#consumer-1| [thrd:172.31.10.150:9092/1005]: 172.31.10.150:9092/1005: Enqueue 31 message(s) (31 ops) on lo-werphat--atlasdos-submissions-balanced-20171129 [57] fetch queue (qlen 31, v2, last_offset 6375)
%7|1521156039.076|FETCH|rdkafka#consumer-1| [thrd:172.31.10.150:9092/1005]: 172.31.10.150:9092/1005: Fetch topic lo-werphat--atlasdos-submissions-balanced-20171129 [57] at offset 6376 (v2)
%7|1521156039.077|CONSUME|rdkafka#consumer-1| [thrd:172.31.10.150:9092/1005]: 172.31.10.150:9092/1005: Enqueue 30 message(s) (30 ops) on lo-werphat--atlasdos-submissions-balanced-20171129 [57] fetch queue (qlen 30, v2, last_offset 6405)
%7|1521156039.078|FETCH|rdkafka#consumer-1| [thrd:172.31.10.150:9092/1005]: 172.31.10.150:9092/1005: Fetch topic lo-werphat--atlasdos-submissions-balanced-20171129 [57] at offset 6406 (v2)
%7|1521156039.080|CONSUME|rdkafka#consumer-1| [thrd:172.31.10.150:9092/1005]: 172.31.10.150:9092/1005: Enqueue 1550 message(s) (1550 ops) on lo-werphat--atlasdos-submissions-balanced-20171129 [57] fetch queue (qlen 1550, v2, last_offset 7955)
%7|1521156039.080|FETCH|rdkafka#consumer-1| [thrd:172.31.10.150:9092/1005]: 172.31.10.150:9092/1005: Fetch topic lo-werphat--atlasdos-submissions-balanced-20171129 [57] at offset 7956 (v2)
%7|1521156039.081|CONSUME|rdkafka#consumer-1| [thrd:172.31.10.150:9092/1005]: 172.31.10.150:9092/1005: Enqueue 39 message(s) (39 ops) on lo-werphat--atlasdos-submissions-balanced-20171129 [57] fetch queue (qlen 39, v2, last_offset 7994)
%7|1521156039.081|FETCH|rdkafka#consumer-1| [thrd:172.31.10.150:9092/1005]: 172.31.10.150:9092/1005: Fetch topic lo-werphat--atlasdos-submissions-balanced-20171129 [57] at offset 7995 (v2)

Here is the full log from the instance that handled that partition:
offsets-57.log

[AlexeyRaga]

We managed to reproduce the same issue without any queue forwarding.
The way to reproduce it is to have a job running and then scale it up and down (in terms of the number of consumers in a group).
At some point the issue pops up.

[edenhill]

It is weird in that debug log that it clearly enqueues the first 6345 messages, and then a couple of milliseconds they are gone, with no indication of a flush barrier or anything.

Could you share relevant parts of your rebalance callback, where you call assign, where you set up queue forwarding, and what your consumer loop looks like?

[edenhill]

Also, a full debug run debug=all might be warranted at this point if it is easy to reproduce (preferably without queue forwarding to make things simpler)

[AlexeyRaga]

I'll try to explain my rebalance callback code, etc. here. I'll try to get debug=all at some point during the weekend, although without forwarding is actually more difficult to get because it would require to change some code.

Here is how the rebalance callback looks like (a bit simplified):

   :: (KafkaConsumer -> RebalanceEvent -> IO ()) -- an action to call
   -> KafkaConsumer -- Kafka handle (wrapped)
   -> KafkaError -- Kafka error (wrapped)
   -> [TopicPartition] -- a list of topic/partitions that is provided by librdkafka 
   -> IO () -- return no value
rebalanceCallback action consumer error ps =
  case error of
    KafkaResponseError RdKafkaRespErrAssignPartitions -> do
        msgQueue <- getMessageQueue consumer
        forM_ ps (\tp -> redirectPartitionQueue consumer (tpTopicName tp) (tpPartition tp) msgQueue)
        action consumer (RebalanceBeforeAssign ps)
        void $ assign consumer ps
        action consumer (RebalanceAssign ps)
    KafkaResponseError RdKafkaRespErrRevokePartitions -> do
        action consumer (RebalanceBeforeRevoke ps)
        void $ assign consumer []
        action consumer (RebalanceRevoke ps)

Here we have an action that we call during the rebalance callback. We usually use it for logging. We call it before and after assignment (providing different arguments).

Before even calling assign we forward partitions queues to msgQueue.
redirectPartitionQueue is responsible for doing it, and it is also responsible for making sure that destroy_* is eventually called on the obtained partition queue.

In this example I forward partitions _before_ calling assign. I thought that it may prevent some nasty race conditions at the point when partitions are already assigned but not yet forwarded.
I have tried doing it the other way around and it didn't help.

The consumer loop looks like:

runConsumerLoop :: KafkaConsumer -> Timeout -> IO ()
runConsumerLoop consumer ct = forkIO go
  where
    go = do
      void $ rdKafkaConsumerPoll consumer
      go

Here we run a recursive loop (go) in a separate (green) thread (using forkIO). It simply polls the consumer and discards the result.
I think at some point I tried to check the result and crash the program if it is not null, which never happened, but I don't clearly remember.

[edenhill]

'll try to get debug=all at some point during the weekend, although without forwarding is actually more difficult to get because it would require to change some code.

with the forwarding code is okay.

[AlexeyRaga]

Here is librdkafka log with debug=all. instance-3.log.gz

Relevant partition ID is 13.

Here is the relevant consumer log:

[Info] [2018-03-16 12:25:14] Instantiating Schema Registry
[Info] [2018-03-16 12:25:14] Creating Kafka Consumer
[Info] [2018-03-16 12:25:14] Creating Kafka Producer
[Info] [2018-03-16 12:25:14] Spawning prefetcher
[Info] [2018-03-16 12:25:14] Processing submissions
[Info] [2018-03-16 12:25:14] Rebalancing, partitions to assign: [PartitionId 12,PartitionId 13,PartitionId 14,PartitionId 15,PartitionId 16,PartitionId 17,PartitionId 18,PartitionId 19,PartitionId 20,PartitionId 21,PartitionId 22,PartitionId 23]
[Info] [2018-03-16 12:25:15] Prefetcher channels has been cleaned
[Info] [2018-03-16 12:25:15] (PartitionId 12,Offset 0)
[Info] [2018-03-16 12:25:15] (PartitionId 12,Offset 1)
[Info] [2018-03-16 12:25:15] (PartitionId 13,Offset 6620)
[Info] [2018-03-16 12:25:16] (PartitionId 13,Offset 6621)
[Info] [2018-03-16 12:25:16] (PartitionId 13,Offset 6622)

[dsturnbull]

I am seemingly having success with this by changing our rebalance callback to use the given rd_kafka_topic_partition_list_t * when calling rd_kafka_assign().

Before:

1. callback function receives rd_kafka_topic_partition_list_t *partitions, we extract topic, partition, offset into some other data structure
2. create a new list with rd_kafka_topic_partition_list_new()
3. use rd_kafka_topic_partition_list_add() for each t, p, o in that other data structure
4. pass this new rd_kafka_topic_partition_list_t * to rd_kafka_assign().

After:

1. callback function receives rd_kafka_topic_partition_list_t *partitions
2. sends the same pointer to rd_kafka_assign().

Edit: this doesn't actually do anything.

[AlexeyRaga]

We have tested the fix @dsturnbull and it seems to have no affect on the situation.
Skips are still there after this fix.

[dsturnbull]

Probably what led me to believe that the rd_kafka_topic_partition_list_t * change worked was testing it with mega logging enabled. As soon as logging is cut back, I see this issue again.

[AlexeyRaga]

I clarify what @dsturnbull is saying a bit: when we subscribe to a log callback and enable enough librdkafka logs then we see no skipping issue. This makes us believe that there is some race condition somewhere.
The issue is reproducible when log callback is not set up (no matter if librdkafka logs are enabled or not).

[ataz]

Any progress on this? @AlexeyRaga @edenhill ? We are having the same issue with skipped messages, and we've also seen some correlation to when rebalancing occurs...

[AlexeyRaga]

@ataz oh, at least I am not alone here :)
How do you use librdkafka? I am using it through Haskell so my examples are less relevant to @edenhill because Haskell runtime is highly concurrent and is harder to debug, unfortunately.

I have found a VERY ugly workaround: After calling rd_kafka_queue_forward for all the partitions that are being assigned, but before calling assign I do sleep for 1 second. Then I don't see jumps. At least I cannot reproduce them locally. I haven't been seriously testing in production yet, but I have done lots of testing locally simulating rebalancing, scaling up and down, and I can't reproduce jumps with this sleep.

The relevant code is here:
https://github.com/haskell-works/hw-kafka-client/blob/master/src/Kafka/Consumer/Callbacks.hs#L89

My conclusion, for now, is that:
1) This ugly fix can be a workaround, but it is unsound because I cannot reason about it: I cannot say that it fixes jumps completely for all cases
2) I believe that there is an issue somewhere in librdkafka that is related to multithreading and forwarding. Looks like in some unfortunate cases fetched messages could be "forwarded" to nowhere. But I have no idea what really happens.

Please try this workaround, if it works for you then great.
But for us this issue is HUGE and we have started working on not relying on Kafka when possible, despite convenience.

[ataz]

Thanks @AlexeyRaga for the detailed response! We are using Python 3.6 with confluent-kafka-python 0.11.4 and librdkafka 0.11.4. We are currently working on reproducing the issue (we've added a watchdog to log when a skip happens), and will report back here when we have more information. The only thing we know so far is that the skips are related to when we have major rebalancing (most common example is if a consumer ends up in a crashback-loop which then causes constant rebalancing). Would be happy to hear your thoughts @edenhill if there's been any progress on this yet?

[realapak]

Hi all, I'm facing this situation too.

Any news regarding this?

[renatomefi]

We also have the same issue, we use the PHP binder https://github.com/arnaud-lb/php-rdkafka, we also use alpine 3.10 which ships librdkafka 1.0.1.

We do also think the issue is related to the rebalances!

Maybe a lot more people have the same issue but they might not be doing deep checks like this.

I'd love to see this issue raising in priority, @edenhill what can we do to help? Do you think some of the comments can help on creating a reproducible scenario? I'm willing to invest time to help, but I'd like your input before I go on a dead-end or such. Thanks!

[edenhill]

If this can be reproduced on a recent version of librdkafka (>v1.0.1), please provide debug logs with debug=cgrp or better debug=cgrp,fetch,broker,topic (noisy).

[TomGoBravo]

PSA: I thought I was having the same problem with python confluent_kafka version ('1.1.0', 16842752) and libversion ('1.1.0', 16843007) but tracked down my problem to ignoring the message returned by Consumer.poll(). Watch out for that if you use a pattern such as

while keep_going:
  for msg in consumer.consume(20000):
    if msg.error() is not None: ...
    .... do stuff with msg ....
  # Don't forget to run callbacks
  consumer.poll(timeout=0)  # And DON'T FORGET TO HANDLE THIS MESSAGE! :-(...