Is there a way to reset offsets of a Kafka consumer group through an API?

This is a solution via CLI:

List the topics to which the group is subscribed:

kafka-consumer-groups --bootstrap-server <kafkahost:port> --group <group_id> --describe

Note the values under "CURRENT-OFFSET" and "LOG-END-OFFSET". "CURRENT-OFFSET" is the offset where this consumer group is currently at in each of the partitions.

Reset the consumer offset for a topic (preview):

kafka-consumer-groups --bootstrap-server <kafkahost:port> --group <group_id> --topic <topic_name> --reset-offsets --to-earliest

This will print the expected result of the reset, but not actually run it.

Reset the consumer offset for a topic (execute):

kafka-consumer-groups --bootstrap-server <kafkahost:port> --group <group_id> --topic <topic_name> --reset-offsets --to-earliest --execute

This will execute the reset and reset the consumer group offset for the specified topic back to 0.

Repeat 1 to check if the reset is successful

Spring for Apache Kafka provides some convenience mechanisms for performing seeks, either during application initialization, or at any time thereafter.

The simplest is to have your listener extend AbstractConsumerSeekAware or implement ConsumerSeekAware.

To reset an offset, the code has to get the consumer by group id, get started, and stopped. A client has to be created (while consumer is stopped). Once the client has been created, the code has to connect, reset the offset, and then disconnect (as posted below). Once it has reset the consumer group has to be restarted again.

The code below below resets the offset. The key is to get the consumer instance to be able to stop and start the consumers because it cannot reset if a consumer is running. We have this running from the command line which is nice. Hopefully this helps.

return kafka.getConsumer( args.groupId )
    .then(consumerInstance => {
        consumer = consumerInstance;
        return consumer.run();
    }).then(()=> {
        return consumer.stop();
    }).then(() => {
         // reset kafka client offset
    })).then(()=> {
        return consumer.run();
    }).

// reset kafka client offset
const { groupId, topic } = args;

const admin = createKafkaClient().admin();

return admin.connect()
    .then(() =>
        admin.resetOffsets({
            groupId,
            topic,
            earliest: true,
    })
    .then(() => admin.disconnect();