How to assign partitions before seek on ConsumerSeekCallback?

user_16691537 0
Viewed 224

I get the following exception

java.lang.IllegalStateException: No current assignment for partition

on

(ConsumerSeekCallback)callback.seek(topic, partition, offset);

From the kafka documentation -

void seek​(java.lang.String topic, int partition, long offset)

Perform a seek operation. When called from ConsumerSeekAware.onPartitionsAssigned(Map, ConsumerSeekCallback) or from ConsumerSeekAware.onIdleContainer(Map, ConsumerSeekCallback) perform the seek immediately on the consumer. When called from elsewhere, queue the seek operation to the consumer. The queued seek will occur after any pending offset commits. The consumer must be currently assigned the specified partition.

What should i do if the partition is not assigned?

spring-kafka apache-kafka
1 Answers

You can capture which topics are assigned to you in ConsumerSeekAware - only perform the seek if you actually have received the topic.

If you extend AbstractConsumerSeekAware, you can call this

    /**
     * Return the callback for the specified topic/partition.
     * @param topicPartition the topic/partition.
     * @return the callback (or null if there is no assignment).
     */
    @Nullable
    protected ConsumerSeekCallback getSeekCallbackFor(TopicPartition topicPartition) {
        return this.callbacks.get(topicPartition);
    }
user_12407630
Related