kazoo.exceptions.NodeExistsError with PyKafka

[dmitry-saritasa]

apache-storm-1.0.2
hbase-1.2.1
kafka_2.10-0.10.0.0
zookeeper-3.4.9
python 3.5.2
pykafka==2.5.0
kazoo==2.2.1

ops:

zkServer.sh start
$KAFKA_HOME/bin/kafka-server-start.sh $KAFKA_HOME/config/server.properties &!
$KAFKA_HOME/bin/kafka-topics.sh --zookeeper localhost:2181 --create --replication-factor 1 --partitions 4 --topic trivver

$KAFKA_HOME/bin/kafka-topics.sh --zookeeper localhost:2181 --list

    __consumer_offsets
    logons
    metrics
    notifications
    ticks
    trivver

$KAFKA_HOME/bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic trivver

Topic:trivver   PartitionCount:4        ReplicationFactor:1     Configs:
        Topic: trivver  Partition: 0    Leader: 0       Replicas: 0     Isr: 0
        Topic: trivver  Partition: 1    Leader: 0       Replicas: 0     Isr: 0
        Topic: trivver  Partition: 2    Leader: 0       Replicas: 0     Isr: 0
        Topic: trivver  Partition: 3    Leader: 0       Replicas: 0     Isr: 0

zkServer.sh status

ZooKeeper JMX enabled by default
Using config: /home/dmitry/Development/zookeeper/bin/../conf/zoo.cfg
Mode: standalone

java -version 
java version "1.8.0_121"
Java(TM) SE Runtime Environment (build 1.8.0_121-b13)
Java HotSpot(TM) 64-Bit Server VM (build 25.121-b13, mixed mode)

uname -a                         
Linux dmitry.npb.io 4.9.13-200.fc25.x86_64 

Now I run the following python code in two different tmux panels:

ffrom pykafka import KafkaClient, SslConfig
from pykafka.exceptions import ConsumerStoppedException, PartitionOwnedError

client = KafkaClient(hosts="127.0.0.1:9092")
topic = client.topics[b'trivver']

try:
    messages = topic.get_balanced_consumer(
        consumer_group=b'trivver_bigtable',
        zookeeper_connect='localhost:2181')
except (ConsumerStoppedException, PartitionOwnedError):
    print('No connection')
    messages.stop()
    messages.start()

print('Connected')

for message in messages:
    if message is not None:
        print(message.offset, message.value)

first run - no problem second run in a different panel or window or terminal

I'm getting the following traceback:

python consumer.py

Failed to acquire partition <pykafka.partition.Partition at 0x7fb8cb9b3cf8 (id=3)> after 4 retries.
Stopping consumer in response to error
Traceback (most recent call last):
  File "/home/dmitry/Projects/trivver/storm/.venv/storm/lib/python3.5/site-packages/pykafka/balancedconsumer.py", line 633, in _add_partitions
    ephemeral=True
  File "/home/dmitry/Projects/trivver/storm/.venv/storm/lib/python3.5/site-packages/kazoo/client.py", line 834, in create
    sequence=sequence, makepath=makepath).get()
  File "/home/dmitry/Projects/trivver/storm/.venv/storm/lib/python3.5/site-packages/kazoo/handlers/utils.py", line 78, in get
    raise self._exception
  File "/home/dmitry/Projects/trivver/storm/.venv/storm/lib/python3.5/site-packages/kazoo/handlers/utils.py", line 206, in captured_function
    return function(*args, **kwargs)
  File "/home/dmitry/Projects/trivver/storm/.venv/storm/lib/python3.5/site-packages/kazoo/handlers/utils.py", line 224, in captured_function
    value = function(*args, **kwargs)
  File "/home/dmitry/Projects/trivver/storm/.venv/storm/lib/python3.5/site-packages/kazoo/client.py", line 889, in create_completion
    return self.unchroot(result.get())
  File "/home/dmitry/Projects/trivver/storm/.venv/storm/lib/python3.5/site-packages/kazoo/handlers/utils.py", line 72, in get
    raise self._exception
kazoo.exceptions.NodeExistsError: ((), {})

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/home/dmitry/Projects/trivver/storm/.venv/storm/lib/python3.5/site-packages/pykafka/balancedconsumer.py", line 306, in start
    self._rebalance()
  File "/home/dmitry/Projects/trivver/storm/.venv/storm/lib/python3.5/site-packages/pykafka/balancedconsumer.py", line 603, in _rebalance
    self._update_member_assignment()
  File "/home/dmitry/Projects/trivver/storm/.venv/storm/lib/python3.5/site-packages/pykafka/balancedconsumer.py", line 577, in _update_member_assignment
    self._add_partitions(new_partitions - current_zk_parts)
  File "/home/dmitry/Projects/trivver/storm/.venv/storm/lib/python3.5/site-packages/pykafka/balancedconsumer.py", line 636, in _add_partitions
    raise PartitionOwnedError(p)
pykafka.exceptions.PartitionOwnedError
Connected
Traceback (most recent call last):
  File "/home/dmitry/Projects/trivver/storm/.venv/storm/lib/python3.5/site-packages/pykafka/balancedconsumer.py", line 731, in consume
    message = self._consumer.consume(block=block)
AttributeError: 'NoneType' object has no attribute 'consume'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "consumer.py", line 18, in <module>
    for message in messages:
  File "/home/dmitry/Projects/trivver/storm/.venv/storm/lib/python3.5/site-packages/pykafka/balancedconsumer.py", line 745, in __iter__
    message = self.consume(block=True)
  File "/home/dmitry/Projects/trivver/storm/.venv/storm/lib/python3.5/site-packages/pykafka/balancedconsumer.py", line 734, in consume
    raise ConsumerStoppedException
pykafka.exceptions.ConsumerStoppedException

• my virtual environment requirements.txt
• my kafka server.properties
• my zookeeper zoo.cfg

I'm using Fedora 25 / Python 3.5.2

So this is my screenshot: http://pasteboard.co/N6Gai2T8A.jpg It seems that first run of the script and balancedconsumer - gets all partitions managed. The second run doesn't spawn rebalancing for partitions and when I submit simple producer - all messages displayed by a connected consumer.

If however I use a different GROUP name - then both instances of the script continue to run and display exactly the same messages as shown here

[jeffwidman]

This sounds like a pykafka bug rather than a kazoo bug. Can you please report it to them?

As an aside, since you are using Kafka >= 0.9, I strongly recommend switching to using a kafka client that leverages Kafka's built-in consumer group APIs rather than Zookeeper for consumer group coordination. For pykafka, that'd be the ManagedBalancedConsumer rather than the BalancedConsumer that you're using. In kafka-python, that'd be the new KafkaConsumer rather than the old SimpleConsumer.