Consumer Group not working on Kafka 0.8.1

[tavoadolfoleon]

I am using Kafka 0.8.1 with kafka-python tool and I have noticed that the consumer group has no effect on the offset. The consumer keeps reading all the messages form the beginning and when I launch two instances of the same python script the consumers get the same messages, from the beginning to

I found a portion on the code to uncoment when using 0.8.1 but that causes and error when the library is called

"...
consumer = SimpleConsumer(kafka, "my-group", topic, auto_commit=False)
File "build/bdist.linux-x86_64/egg/kafka/consumer.py", line 242, in init
File "build/bdist.linux-x86_64/egg/kafka/consumer.py", line 111, in init
NameError: global name 'OffsetFetchRequest' is not defined
..."

This is the python script:

localconsumer.py

from kafka.client import KafkaClient
from kafka.consumer import SimpleConsumer
from kafka.producer import SimpleProducer, KeyedProducer

kafka = KafkaClient("localhost:9092")
topic = "test"

To consume messages

consumer = SimpleConsumer(kafka, "my-group", topic, auto_commit=False)
for message in consumer:
print message

kafka.close()

[wizzat]

There are two open pull requests which address this problem. The debate is over how to have kafka-python support multiple versions of the kafka server given the fact the server doesn't identify itself. Also, how to keep the code base clean with different versions supported.

On Mar 22, 2014, at 8:00, tavoadolfoleon notifications@github.com wrote:

I am using Kafka 0.8.1 with kafka-python tool and I have noticed that the consumer group has no effect on the offset. The consumer keeps reading all the messages form the beginning and when I launch two instances of the same python script the consumers get the same messages, from the beginning to

I found a portion on the code to uncoment when using 0.8.1 but that causes and error when the library is called

"...
consumer = SimpleConsumer(kafka, "my-group", topic, auto_commit=False)
File "build/bdist.linux-x86_64/egg/kafka/consumer.py", line 242, in init
File "build/bdist.linux-x86_64/egg/kafka/consumer.py", line 111, in init
NameError: global name 'OffsetFetchRequest' is not defined
..."

This is the python script:

##localconsumer.py

from kafka.client import KafkaClient
from kafka.consumer import SimpleConsumer
from fixtures import ZookeeperFixture, KafkaFixture
from kafka.producer import SimpleProducer, KeyedProducer

kafka = KafkaClient("localhost:9092")
topic = "test"

To consume messages

consumer = SimpleConsumer(kafka, "my-group", topic, auto_commit=False)
for message in consumer:

print message

kafka.close()

—
Reply to this email directly or view it on GitHub.

[dpkp]

also -- if you are on the master branch and feel like hacking a bit, you can try including the experimental code in consumer.py:

https://github.com/mumrah/kafka-python/blob/master/kafka/consumer.py#L108-L115

wizzat is correct -- this seems to be blocked on agreeing how to handle multiple server versions

[tavoadolfoleon]

is there anything it can be done to make the consumer group work on this particular version of Kafka 0.8.1, while the debate is going?, any shortcut or a temporary fix to make the consumer group work

would be the simplest thing to indicate the Kafka version manually on a parameter when setting the Kafka connection?

[tavoadolfoleon]

dpkp, I did uncomment the lines you pointed but this error appears when using it:

"...
consumer = SimpleConsumer(kafka, "my-group", topic, auto_commit=False)
File "build/bdist.linux-x86_64/egg/kafka/consumer.py", line 242, in init
File "build/bdist.linux-x86_64/egg/kafka/consumer.py", line 111, in init
NameError: global name 'OffsetFetchRequest' is not defined
..."

[dpkp]

try adding OffsetFetchRequest to the import list on line 11/12 of consumer.py . It's all mostly untested though, so quite possible that it doesn't work.

you might also try rdiomar's fork that implements offsets directly in zookeeper w/o using kafka server apis: https://github.com/rdiomar/kafka-python/tree/zk_offsets .

[dpkp]

also you probably want to set auto_commit=True in SimpleConsumer to tell your consumers to store offsets

[wizzat]

I would recommend using one of the branches which have been offered as a pull request. The easy_kafka branch is what we have in production at Kixeye right now. Before the recent spree of merges I was going to permanently fork the project on 3/29 to get traction back and it uploaded to pypi.

On Mar 22, 2014, at 10:53, Dana Powers notifications@github.com wrote:

also you probably want to set auto_commit=True in SimpleConsumer to tell your consumers to store offsets

—
Reply to this email directly or view it on GitHub.

[geoffrey-young]

is this meant to solve #112 ?

[tavoadolfoleon]

Geoffrey,

I am actually in the process of testing the options, I will be posting the
result checking if anything worked

On Tue, Mar 25, 2014 at 10:41 AM, Geoffrey Young
notifications@github.comwrote:

is this meant to solve #112#112?

Reply to this email directly or view it on GitHubhttps://github.com//issues/147#issuecomment-38572092
.

[tavoadolfoleon]

I did uncomment the lines for Kafka 0.8.1 and include the "import OffsetFetchRequest" at lines 11 and 12 make the errors go away so far.

but still the offset is not set, every time I read it brings all the messages (from beginning) and if I launch two instances of the consumer both show the existing and the new messages

autocommit=true/false doesn't change the behavior

[tavoadolfoleon]

if this is true:

"
... That contract of one message per consumer group only works for the coordinated consumers which are implemented for the JVM only (i.e., Scala and Java clients).

Kafka will eventually have better support for non-JVM clients to handle coordinated consumption at which point we'll implement this functionality.

HTH ...
"

then there is nothing it can be done for the matter

[wizzat]

I've had good luck setting up a consumer per partition. This emulates that behavior fairly well.

-Mark

On Mar 25, 2014, at 12:04, Gustavo A Leon notifications@github.com wrote:

if this is true:

"
... That contract of one message per consumer group only works for the coordinated consumers which are implemented for the JVM only (i.e., Scala and Java clients).

Kafka will eventually have better support for non-JVM clients to handle coordinated consumption at which point we'll implement this functionality.

HTH ...
"

then there is nothing it can be done for the matter

—
Reply to this email directly or view it on GitHub.

[tavoadolfoleon]

when the consumer stop and start again it continues where it was or from beginning? if from beginning how you do it

[tavoadolfoleon]

sorry, I mean, from where it was, how do you do it?

[wizzat]

I'm using this branch: https://github.com/wizzat/kafka-python/tree/easykafka

In particular, each process starts up with a single topic/partition that it listens to. I don't appear to be getting the same data more than once unless I spin up another process listening on the same set of topics.

As to resuming from where it starts, I think it all boils down to this bit (it required some monkeying around).

consumer.py:109:

    if auto_commit:
        for partition in partitions:
            req = OffsetFetchRequest(topic, partition)
            (offset,) = self.client.send_offset_fetch_request(group, [req],
                          callback=get_or_init_offset_callback,
                          fail_on_error=False)
            self.offsets[partition] = offset
    else:
        for partition in partitions:
            self.offsets[partition] = 0