Im trying to consume data from a topic by setting the offset but get assertion error -
from kafka import KafkaConsumer
consumer = KafkaConsumer('foobar1',
bootstrap_servers=['localhost:9092'])
print 'process started'
print consumer.partitions_for_topic('foobar1')
print 'done'
consumer.seek(0,10)
for message in consumer:
print ("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,
message.offset, message.key,
message.value))
print 'process ended'
Error:-
Traceback (most recent call last):
File "/Users/pn/Documents/jobs/ccdn/kafka_consumer_1.py", line 21, in <module>
consumer.seek(0,10)
File "/Users/pn/.virtualenvs/vpsq/lib/python2.7/site-packages/kafka/consumer/group.py", line 549, in seek
assert partition in self._subscription.assigned_partitions(), 'Unassigned partition'
AssertionError: Unassigned partition
Here is an example to solve the problem:
from kafka import KafkaConsumer, TopicPartition
con = KafkaConsumer(bootstrap_servers = my_bootstrapservers)
tp = TopicPartition(my_topic, 0)
con.assign([tp])
con.seek_to_beginning()
con.seek(tp, 1000000)
Reference: kafka consumer seek is not working: AssertionError: Unassigned partition
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With