seek - needs poll

[numenory]

Seek fails unless a poll is done first:

from confluent_kafka import Consumer, KafkaError, TopicPartition
c = Consumer({
    'bootstrap.servers': 'localhost',
    'group.id': 'bob-group',        
    'default.topic.config': {
        'auto.offset.reset': 'smallest',
    }
})

c.subscribe(['topic-test'])
tp = TopicPartition('topic-test', 0, 10)
#msg = c.poll() #shouldn't be necessary think fault in client
c.seek(tp)

Results in:

Traceback (most recent call last):
  File "kafka.confluent.consumer.py", line 29, in <module>
    c.seek(tp)
cimpl.KafkaException: KafkaError{code=_UNKNOWN_PARTITION,val=-190,str="Failed to seek to offset 10: Local: Unknown partition"}

By uncommenting the line "msg = c.poll()", the seek will succeed.

System Info:

Ubuntu 16.04

confluent-kafka-0.11.4
kafka_2.11-1.1.0.tgz
zookeeperd 3.4.10-3 (installed via apt)

[edenhill]

seek() may only be called on partitions that are being actively fetched.
subscribe() is an asynchronous and it typically takes anything from a couple of milliseconds to session.timeout.ms to receive an assignment. So by the time you call seek() there will most likely not be an assignment yet.

Additionally, doing subscribe() and then seek() on a specific partition won't really work since you dont know which partitions will be assigned to you.

If you want to set the start offset for the partitions that are assigned to you you should provide an on_assign callback to subscribe(), it will be called from poll() when your consumer gets a new assignment.
In the on_assign callback you can set the starting offset of each partition by updating the provided TopicPartition list with an .offset for each partition.

Something like this:

    def my_on_assign(consumer, partitions):
        for p in partitions:
             # some starting offset, or use OFFSET_BEGINNING, et, al.
             # the default offset is STORED which means use committed offsets, and if
             # no committed offsets are available use auto.offset.reset config (default latest)
            p.offset = 1234
        # call assign() to start fetching the given partitions.
        consumer.assign(partitions)

     consumer.subscribe(mytopics, on_assign=my_on_assign)

[numenory]

Many thanks and apologies for the report, will close.

[edenhill]

We should probably raise an exception if seek() is called without an active assignment.

[asnare]

I'm looking for a point of clarification here. Normally there are two modes of operation when it comes to partition/consumer assignment:

• .subscribe(…): Automatically handle it in the background by balancing across consumers in the consumer-group;
• .assign(…): Manually decide which partitions to consume from.

In the suggested code sample above, it looks like we start in automatic mode, but then switch to manual mode once the initial assignment has taken place. Is this what's going on, and if so doesn't this prevent the normal automatic reassignments from happening?

[edenhill]

subscribe() joins the consumer group (asynchronously in the background) with a subscription, the subscription is used by the elected group leader to assign a subset of partitions for the subscribed topics to the members in the group.
assign() is controlling the actual fetcher, telling it to start fetching the given partitions.
A high-level KafkaConsumer is thus the combination of subscribe() (to eventually get a partition assignment) and assign() to start consuming the assigned partitions.

Typically an application only needs to call subscribe().
But if the application wants to know its assignment, do any modifications to it (such as changing the starting offsets), it can set up on_assign and/or on_revoke callback that will be triggered as the assignment is assigned or revoked.
To synchronize state between the application and underlying asynchronous client, the application will need to call assign() from these callbacks, to let the underlying client know the application has handled the callback.

[wenkirao]

Please update this wiki with above sample code:

https://github.com/edenhill/librdkafka/wiki/Manually-setting-the-consumer-start-offset

[hrchu]

@wenkirao copied and pasted.

[hrchu]

seek() may only be called on partitions that are being actively fetched.
subscribe() is an asynchronous and it typically takes anything from a couple of milliseconds to session.timeout.ms to receive an assignment. So by the time you call seek() there will most likely not be an assignment yet.

Additionally, doing subscribe() and then seek() on a specific partition won't really work since you dont know which partitions will be assigned to you.

If you want to set the start offset for the partitions that are assigned to you you should provide an on_assign callback to subscribe(), it will be called from poll() when your consumer gets a new assignment.
In the on_assign callback you can set the starting offset of each partition by updating the provided TopicPartition list with an .offset for each partition.

Something like this:

    def my_on_assign(consumer, partitions):
        for p in partitions:
             # some starting offset, or use OFFSET_BEGINNING, et, al.
             # the default offset is STORED which means use committed offsets, and if
             # no committed offsets are available use auto.offset.reset config (default latest)
            p.offset = 1234
        # call assign() to start fetching the given partitions.
        consumer.assign(partitions)

     consumer.subscribe(mytopics, on_assign=my_on_assign)

@edenhill what's the definition of "partitions that are being actively fetched." ? An assigned partition? or a partition which have been read at least one message?