Confluent-kafka-go: Seek function not working

[martinhynar]

Description

I am trying to make use of Seek function, but I get _Local: Erroneous state_ error when it is called.
I am getting the offset to which I want to seek by calling OffsetsForTimes, but same error is returned also when TopicPartition is constructed manually.

How to reproduce

I have setup described in README here https://github.com/martinhynar/kafka-consumer/tree/seekerror
This is also the code I used when received this error.

Checklist

Please provide the following information:

• [x] confluent-kafka-go and librdkafka version (LibraryVersion()): 721407 0.11.1
• [x] Apache Kafka broker version: confluent-kafka-2.11.noarch 1.0.0-1 (From Confluent repo)
• [x] Client configuration: https://github.com/martinhynar/kafka-consumer/blob/seekerror/kafka-consumer.go#L47
• [x] Operating system: Fedora 27, kernel 4.14.3-300.fc27.x86_64
• [ ] Provide client logs (with "debug": ".." as necessary)
• [ ] Provide broker log excerpts: This is what broker prints when client is executed

[2018-02-01 16:25:11,493] INFO [GroupCoordinator 0]: Preparing to rebalance group kc.local with old generation 20 (__consumer_offsets-39) (kafka.coordinator.group.GroupCoordinator)
[2018-02-01 16:25:11,493] INFO [GroupCoordinator 0]: Stabilized group kc.local generation 21 (__consumer_offsets-39) (kafka.coordinator.group.GroupCoordinator)
[2018-02-01 16:25:11,494] INFO [GroupCoordinator 0]: Assignment received from leader for group kc.local for generation 21 (kafka.coordinator.group.GroupCoordinator)

• [ ] Critical issue: No, it is not

[edenhill]

Seek() may only be used for an already actively fetched partition.
To set the initial starting position you should instead specify the Offset of the TopicPartition in the Assign() call:
https://github.com/edenhill/librdkafka/wiki/Manually-setting-the-consumer-start-offset

[martinhynar]

From https://godoc.org/github.com/confluentinc/confluent-kafka-go/kafka#Consumer.Seek I understood that either Assign (as you are describing) or Subscribe() which is what I have, or I think I have (is Subscribe([]topics, nil) considered self-rebalancing?)

Anyway, changing the code to use Assign makes it work. Thanks

[edenhill]

You can use Seek() with Subscribe() but you need to wait for the assignment to be assigned with Assign(), and without registering a AssignedPartitions handler you won't really know when the implicit Assign() kicks in (or if you call Assign() yourself from the handler).

[martinhynar]

I changed my code to handle also AssignedPartitions and calling Assign() with desired offset when AssignedPartitions event is received. (updated version is tagged here -
https://github.com/martinhynar/kafka-consumer/tree/AssignedPartitions)

The code has one obvious drawback. It rewinds back every time it receives AssignedPartitions event. In case I run this in two instances on multiple partition topic, this will happen on every rebalance. To solve this for one instance, I can have some local state that will hold partitions that were already rewound, but how can I solve this for multi-consumer setup?

---

For the record, if someone else comes across this problem, I would recommend to read introductory description for confluent-kafka-go package itself (https://godoc.org/github.com/confluentinc/confluent-kafka-go/kafka) which I have unfortunately skipped and which contains helpful info.

[edenhill]

There is no generic answer to your question, it depends on your required application logic.
If you don't specify a starting offset for Assign() it will use committed offsets.