RequestTimedOutError when polling with all partitions paused

[hburrows]

When I pause all my partitions and call poll I get a RequestTimedOutError after 40 seconds (presumably due to request.timeout.ms). This is then shortly followed by an Auto offset commit failed for group XYZ... error. I'm pausing all my partitions so I can perform a long "batch" operation and continue to poll the topic for heartbeat purposes so Kafka doesn't think I'm dead but NOT actually fetch any messages. Once the batch operation is done I'll resume the partitions. I'm essentially trying to implement a background heartbeat by interleaving a "no-op" poll into the batch operation on a regular interval (every 2 secs)

Why am I doing this? I want to load process state from a compacted topic during process startup. I also want to use Kafka's automatic partition assignment. I use a consistent hash key for all my topics so the data in my partitions aligns from topic to topic. When a process starts it connects to a topic and gets its partition assignments. I then load state from the compacted topic by manually assigning the partitions and using the partition numbers corresponding with what was automatically assigned from the main input topic. I'll read messages from the compacted topic up to a predetermined offset and call poll on the main topic on a regular 2 sec. interval. This seems like it should work. However, after 40 secs I get RequestTimedOutError, closing connection, etc. errors.

Should this work? Why would kafka-python be trying to commit the offset for a paused partition? What would be causing the RequestTimedOutError error? Is there an alternative way that I can accomplish what I describe above (I'm more than happy to try an alternative approach).

[dpkp]

I'm not sure what would cause that -- definitely worth investigation.

On the general design, I've actually taken very similar approaches but I generally do not use dynamic partition assignment to make the partition <-> partition joins easier to manage with separate processes.

[hburrows]

@dpkp Thanks for the quick response. If it would help I can post a example that reproduces the problem. It's essentially....

consumer = KafkaConsumer(...)
consumer.pause(*list(consumer.assignment()))
while performing_batch_operations:
  consumer.poll()   # gets called every 2 seconds
consumer.resume(*list(consumer.assignment()))

After 40 seconds looping in the while it'll error with RequestTimedOutError

[dpkp]

how do you manage assignments changing during poll? dont you also need to worry about pausing any new partitions that you've been assigned?

[hburrows]

If the assignments change during the batch load (or poll) we'll start from scratch reloading the state from the compacted topic again. We will only start over if we're assigned new partitions we weren't previously assigned; otherwise we just discard the data for the partitions we're no longer assigned.

This isn't necessarily ideal... and in theory could result in thrashing. But we're finding the partition assignments to be stable unless a process dies or Kafka assumes it's dead - and in either case we want to fix it so it doesn't continue happening.

[hburrows]

Little more data on this issue.... was thinking that maybe auto commit of offsets might have something to do with this so I disabled that (enable_auto_commit=False) in favor of manually committing after processing the batch of messages returned from poll. The auto offset commit setting doesn't change behavior.

One other thing I failed to mention is that I'm using version 1.3.1

[hburrows]

I think we've figured this out. Not sure if you'd consider it a bug or not? We were performing long running "batch" operations inside on_partitions_assigned of our ConsumerRebalanceListener subclass. The first call to on_partitions_assigned happens on the first poll and apparently before the HeartbeatTask is scheduled to run; I don't believe the heartbeat task gets scheduled until after on_partitions_assigned returns . Thus... as we poll no heartbeats are ever sent.

Our solution is to remove any long running "batch" operations from the ConsumerRebalanceListener listeners and run them from within the main poll loop. The ConsumerRebalanceListener listeners now set a flag that the main poll loop inspects to signal a "batch" operation. This allows any internal "housekeeping" that is associated with a rebalance to occur before attempting to run the "batch" job. To keep the heartbeat active during a long running batch we 1) pause all partitions and then 2) regularly call poll with a timeout_ms of 100 (a timeout_ms of 0 or a value that is too small doesn't result in a heartbeat request ???). We can also abort batch jobs so if a rebalance happens during a batch it's immediately aborted and restarted. Generally seems to be working.

How kafka-python decides to send a heartbeat is a mystery and very complicated. There should be a predictable and transparent way to get a heartbeat request to be transmitted. The heartbeat is so integral to using automatic partition assignment you need to understand how to make it happen predictably.

[dpkp]

Glad to hear you found the error. Heartbeat management is definitely convoluted. I've copied the initial java client design, but they have since modified their approach to now use a separate background thread to manage heartbeats. It's probably worth investigating a similar switch in kafka-python.