CODEWITHSUNDEEP
apache-kafkakafka-consumer-api
Patrick Szalapski
Patrick Szalapski

Reputation: 9439

How to advance the consumer offset in a Kafka consumer group to the end?

Is there an easy way--using the Kafka REST API--to advance the consumer offset on all partitions in a consumer group to the end of the partition? Effectively, I want to sometimes skip consuming all remaining messages--for example, if I anticipate re-producing them all.

I know I could retrieve the consumer group, retrieve the partitions, loop through, and seek each one--is there a simpler way?

Upvotes: 3

Views: 8776

Answers (1)

Michael Heil
Michael Heil

Reputation: 18525

According to the documentation on post--consumers-instances-positions-end you can do that for a ConsumerGroup with the following request:

POST /consumers/testgroup/instances/my_consumer/positions/end HTTP/1.1
Host: proxy-instance.kafkaproxy.example.com
Content-Type: application/vnd.kafka.v2+json

{
  "partitions": [
    {
      "topic": "test",
      "partition": 0
    },
    {
      "topic": "test",
      "partition": 1
    }

  ]
}

Still, you need to know in advance the subscribed topics and partitions of the ConsumerGroup in advance.

Edit:

I want to sometimes skip consuming all remaining messages

I see multiple options here but all of them are rather hacky in my point of view and also not using the Kafka Rest API.

Option 1

Change the retention time (retention.ms) of the topics to a small value (like 1), wait a bit to let the LogCleaner delete all messages and change the retention time back to normal. Then produce your new alternative data.

Option 2

Change the ConsumerGroup names of all your consumers to a new ConsumerGroup (configuration group.id) and let the consumers read from the end of the topic by setting auto.offset.reset=latest. Then produce your new alternative data.

Option 3

Similar to my initial answer use the Kafka tool kafka-consumer-groups to manually change the offset of a Consumer Group (e.g. "myConsumer") to the end offset:

bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group myConsumer --topic myTopic --to-latest

Upvotes: 3

Related Questions