Kafka Topic Resetter Class

gistfile1.txt

package com.altvr.streams.jobs.utils;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

public class TopicOffsetResetter {
	private static final Logger Log = LoggerFactory.getLogger(TopicOffsetResetter.class);

	public static void resetTopicOffsetToBeginning(Properties jobProperties, String topic) {
		Serde<byte[]> byteArraySerde = Serdes.ByteArray();

		Properties consumerProps = new Properties();
		consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, jobProperties.get(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG));
		consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, jobProperties.get(StreamsConfig.APPLICATION_ID_CONFIG));
		consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

		KafkaConsumer<byte[], byte[]> resetConsumer
			= new KafkaConsumer<>(consumerProps, byteArraySerde.deserializer(), byteArraySerde.deserializer());

		resetConsumer.subscribe(Collections.singletonList(topic));

		while (true) {
			resetConsumer.poll(100);
			if (resetConsumer.assignment().size() > 0) break;
		}

		Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();

		if (resetConsumer.assignment().size() != resetConsumer.partitionsFor(topic).size()) {
			Log.error("Did not get assigned all topics -- do you still have a consumer running for this group?");
			resetConsumer.close();
			return;
		}

		for (TopicPartition partition : resetConsumer.assignment()) {
			Log.error("Reset " + partition);
			offsets.put(partition, new OffsetAndMetadata(0));
		}

		Log.error("Committing.");
		resetConsumer.commitSync(offsets);

		Log.error("Done.");
		resetConsumer.close();
	}
}