python-kafka, on Windows, returns 'NodeNotReadyError'

Question

I'm able to stream messages from a Kafka topic (running OSX) using the excellent python-kafka package, e.g.:

from kafka import KafkaConsumer

consumer = KafkaConsumer('MyTopic',
                         group_id='alex',
                         bootstrap_servers=['kafka:9092'],
                         auto_offset_reset='largest')

for message in consumer:
    print message.value

A colleague, who's running Windows 7, asked me if he could could stream/filter/alert using the same approach. "No problem", I said. We installed Anaconda Python and Pycharm, and tried to stream some messages.

Unfortunately, a few moments after running the consumer, the following message is displayed:

File "C:\Users\[my_colleague]\AppData\Local\Continuum\Anaconda2\lib\site-packages\kafka\client_async.py", line 598, in connect
    raise Errors.NodeNotReadyError(node_id)
kafka.common.NodeNotReadyError: None

I suspected that the Windows firewall was blocking communications between the consumer and the broker, and so we briefly went and added both Pycharm and python.exe to the list of programs that could traverse the firewall. That didn't help.

At that point, I felt a sense of ownership for the problem and that I should avoid monopolizing his laptop while I troubleshoot. Given the popularity of both Python, Kafka, and Windows, I imagine that we're not the first people to face this issue. Can you see what we're doing wrong?

Answer 1

This is actually an error in the attempt to auto-probe your broker version. It is a non-standard feature that we added to the python client to make it easier to interoperate between broker versions using the same client. My strong recommendation is that once you get past development stages you should pass your broker version explicitly using the api_version parameter. The probed version is logged by kafka-python in case you want to verify, but from comments it sounds like the version here was 0.9. So here,

consumer = KafkaConsumer('MyTopic',
                         api_version=(0, 9),
                         group_id='alex',
                         bootstrap_servers=['kafka:9092'],
                         auto_offset_reset='largest')