Deadlock when using reset_offset

[zsimpson]

In simpleconsumer I was passing a list of all partitions to reset. Sometimes (often but not always) it would lock up. Adding traces I found that the lock up was at the line that reads:

if partition.fetch_lock.acquire(True):

In the reset_offsets() function.

I changed my code to call the reset_offsets function multiple times, once for each partition. I don't know if this just slows it down enough or what exactly, but it made the deadlock disappear.

My reproduction case is in a somewhat complex environment so I'm unable to provide a simple repro case, but if you'd like me to add traces or other debugging I'd be happy to do so.

[emmettbutler]

Thanks @zsimpson - turns out sharing resources between threads is hard. It might help to see what your debug traces look like when you hit this deadlock.

[zsimpson]

Here's the trace. If you want I'll be happy to add more traces in. I can reproduce it every time right now it seems.

2015-09-04 16:01:05,232 - pykafka.simpleconsumer - INFO - Resetting offsets for 16 partitions
2015-09-04 16:01:05,238 - pykafka.simpleconsumer - INFO - Flushed queue for partition 6
2015-09-04 16:01:05,244 - pykafka.simpleconsumer - INFO - Flushed queue for partition 12
2015-09-04 16:01:05,249 - pykafka.simpleconsumer - INFO - Flushed queue for partition 8
2015-09-04 16:01:05,386 - pykafka.simpleconsumer - DEBUG - Fetched 4054 messages for partition 1
2015-09-04 16:01:05,416 - pykafka.simpleconsumer - DEBUG - Partition 1 queue holds 4054 messages
2015-09-04 16:01:05,422 - pykafka.simpleconsumer - DEBUG - Fetched 4168 messages for partition 3
2015-09-04 16:01:05,450 - pykafka.simpleconsumer - DEBUG - Partition 3 queue holds 4168 messages
2015-09-04 16:01:05,457 - pykafka.simpleconsumer - DEBUG - Fetched 4239 messages for partition 5
2015-09-04 16:01:05,487 - pykafka.simpleconsumer - DEBUG - Partition 5 queue holds 4239 messages
2015-09-04 16:01:05,492 - pykafka.simpleconsumer - DEBUG - Fetched 4205 messages for partition 9
2015-09-04 16:01:05,521 - pykafka.simpleconsumer - DEBUG - Partition 9 queue holds 4205 messages
2015-09-04 16:01:05,528 - pykafka.simpleconsumer - INFO - Resetting offsets in response to OffsetOutOfRangeError
2015-09-04 16:01:05,535 - pykafka.simpleconsumer - INFO - Resetting offsets for 4 partitions
2015-09-04 16:01:05,542 - pykafka.simpleconsumer - INFO - Flushed queue for partition 13
2015-09-04 16:01:05,549 - pykafka.simpleconsumer - INFO - Flushed queue for partition 15
2015-09-04 16:01:05,556 - pykafka.simpleconsumer - INFO - Flushed queue for partition 7
2015-09-04 16:01:05,563 - pykafka.simpleconsumer - INFO - Flushed queue for partition 11
2015-09-04 16:01:05,571 - pykafka.simpleconsumer - DEBUG - Committing offsets for 16 partitions to broker id 1
2015-09-04 16:01:05,584 - pykafka.simpleconsumer - INFO - Resetting offsets in response to OffsetOutOfRangeError
2015-09-04 16:01:05,590 - pykafka.simpleconsumer - INFO - Resetting offsets for 5 partitions
2015-09-04 16:01:05,596 - pykafka.simpleconsumer - INFO - Flushed queue for partition 3
2015-09-04 16:01:05,597 - pykafka.simpleconsumer - INFO - Flushed queue for partition 0
2015-09-04 16:01:05,603 - pykafka.simpleconsumer - INFO - Flushed queue for partition 14
2015-09-04 16:01:05,631 - pykafka.simpleconsumer - INFO - Flushed queue for partition 1
2015-09-04 16:01:05,640 - pykafka.simpleconsumer - INFO - Flushed queue for partition 10
2015-09-04 16:01:05,647 - pykafka.simpleconsumer - INFO - Flushed queue for partition 2
2015-09-04 16:01:05,654 - pykafka.simpleconsumer - INFO - Flushed queue for partition 11
2015-09-04 16:01:05,677 - pykafka.simpleconsumer - INFO - Flushed queue for partition 5
2015-09-04 16:01:05,684 - pykafka.simpleconsumer - INFO - Flushed queue for partition 7
2015-09-04 16:01:05,707 - pykafka.simpleconsumer - INFO - Flushed queue for partition 9

[emmettbutler]

Can you give any information about when you're calling reset_offsets in relation to the rest of the consumer's operation? Any additional traces would also be helpful, since I don't have a ton of context for how this situation is arising.

[zsimpson]

It's the very first thing I do. Essentially I just get the consumer then I call reset_offsets if there's anything in the topic. That if statement was necessary do deal with topics that didn't have any messages on them yet.

    topic = kafka_client.topics[self.topic_id]
    consumer_group_id = 'kayak-'+str(uuid.uuid1())
    consumer = topic.get_simple_consumer( consumer_group_id )

    offsets = list(set([ res.offset[0] for p, res in topic.latest_available_offsets().items() ]))
    if not (len(offsets) == 1 and offsets[0] == -1):
        consumer.reset_offsets( [ (v, -1) for k,v in topic.partitions.items() ] )

I replaced this code with

    topic = kafka_client.topics[self.topic_id]
    consumer_group_id = 'kayak-'+str(uuid.uuid1())
    consumer = topic.get_simple_consumer( consumer_group_id )

    offsets = list(set([ res.offset[0] for p, res in topic.latest_available_offsets().items() ]))
    if not (len(offsets) == 1 and offsets[0] == -1):
        for k,v in topic.partitions.items():
            consumer.reset_offsets( [ (v, -1) ] )

And that seems to bypass the issue. Maybe because it slowed it down? Not really sure.

[ltucker]

The set of "fetch_locks" on owned partitions can be acquired in a variety of different orderings in the reset_offsets(...) call depending on its arguments (and the unreliability of item ordering between different python dicts)

If your call is intersecting with another call to reset_offsets() I believe it can deadlock. Another call to this could be happening asynchronously, for example, in the error handler for an OffsetOutOfRangeError (seen in your trace) and possibly other places. See https://github.com/Parsely/pykafka/blob/master/pykafka/simpleconsumer.py#L211

A rule of thumb here is that resources should be acquired in the same order by anyone who wants more than one. This prevents for example: "thread A holding lock 1 while waiting for lock 2" while "thread B holding lock 2 waiting for lock 1".

This is also likely why calling reset_offsets() one by one does not cause a deadlock (only takes one lock)

I suggest creating a stable sorted ordering for this iteration:
https://github.com/Parsely/pykafka/blob/master/pykafka/simpleconsumer.py#L513

A secondary possibility is that that error handler is being triggered synchronously inside your call to reset_offsets() in which case it may be trying to acquire resources already held by the thread. Re-entrant locks (RLocks) are a possibility if reset_offsets() is supposed to be able to call itself recursively in an error handler. I still suggest stable ordering though.

[emmettbutler]

#265 is a shot at applying consistent ordering to the partitions before locking them. Unfortunately I can't get the tests to run at the moment, but it's a fairly simple fix.

[emmettbutler]

@zsimpson how does #265 affect your deadlock issue?

[rduplain]

Current status: need to replicate error and confirm that this fixes it.