Is it possible to reset offsets to a topic for a kafka consumer group in a kafka connector?

Question

My kafka sink connector reads from multiple topics (configured with 10 tasks) and processes upwards of 300 records from all topics. Based on the information held in each record, the connector may perform certain operations.

Here is an example of the key:value pair in a trigger record:

"REPROCESS":"my-topic-1"

Upon reading this record, I would then need to reset the offsets of the topic 'my-topic-1' to 0 in each of its partitions.

I have read in many places that creating a new KafkaConsumer, subscribing to the topic's partitions, then calling the subscribe(...) method is the recommended way. For example,

public class MyTask extends SinkTask {

    @Override
    public void put(Collection<SinkRecord> records) {
        records.forEach(record -> {
        if (record.key().toString().equals("REPROCESS")) {
            reprocessTopicRecords(record);
        } else {
            // do something else
        }
        });
    }
    private void reprocessTopicRecords(SinkRecord record) {
        KafkaConsumer<JsonNode, JsonNode> reprocessorConsumer = 
            new KafkaConsumer<>(reprocessorProps, deserializer, deserializer);
        reprocessorConsumer.subscribe(Arrays.asList(record.value().toString()),
            new ConsumerRebalanceListener() {
                public void onPartitionsRevoked(Collection<TopicPartition> partitions) {}
                public void onPartitionsAssigned(Collection<TopicPartition> partitions) { 
                    // do offset reset here
                }
            }
        );
    }
}

However, the above strategy does not work for my case because: 1. It depends on a group rebalance taking place (does not always happen) 2. 'partitions' passed to the onPartitionsAssigned method are dynamically assigned partitions, meaning these are only a subset to the full set of partitions that will need to have their offset reset. For example, this SinkTask will be assigned only 2 of the 8 partitions that hold the records for 'my-topic-1'.

I've also looked into using assign() but this is not compatible with the distributed consumer model (consumer groups) in the SinkConnector/SinkTask implementation.

I am aware that the kafka command line tool kafka-consumer-groups can do exactly what I want (I think): https://gist.github.com/marwei/cd40657c481f94ebe273ecc16601674b

To summarize, I want to reset the offsets of all partitions for a given topic using Java APIs and let the Sink Connector pick up the offset changes and continue to do what it has been doing (processing records).

Thanks in advance.

Answer 1

I was able to achieve resetting offsets for a kafka connect consumer group by using a series of Confluent's kafka-rest-proxy APIs: https://docs.confluent.io/current/kafka-rest/api.html

This implementation no longer requires the 'trigger record' approach firs described in the original post and is purely Rest API based.

1. Temporarily delete the kafka connector (this deletes the connector's consumers and )

2. Create a consumer instance for the same consumer group ("connect-")

3. Have the instance subscribe to the requested topic you want to reset

4. Do a dummy poll ('subscribe' is evaluated lazily')

5. Reset consumer group topic offsets for specified topic

6. Do a dummy poll ('seek' is evaluated lazily') Commit the current offset state (in the proxy) for the consumer

7. Re-create kafka connector (with same connector name) - after re-balancing, consumers will join the group and read the last committed offset (starting from 0)

8. Delete the temporary consumer instance

If you are able to use the CLI, Steps 2-6 can be replaced with:

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

As for those of you trying to do this in the kafka connector code through native Java APIs, you're out of luck :-(