public static long lastCommitTimestamp(String groupName, String bootstrapServers) {

    int partition = Math.abs(groupName.hashCode() % 50); // Assume offset topic partition number is 50

    TopicPartition tp = new TopicPartition("__consumer_offsets", partition);

    Properties props = new Properties();

    props.put("bootstrap.servers", bootstrapServers);

    props.put("group.id", "test");

    props.put("enable.auto.commit", "false");

    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

    try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {

    	consumer.assign(Arrays.asList(tp));

    	consumer.poll(Duration.ZERO); // take effect

    	Map<TopicPartition, Long> endOffset = consumer.endOffsets(Collections.singletonList(tp));

    	consumer.seek(tp, endOffset.get(tp) - 1);

    	ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(30));

    	if (records.count() == 0) {

    		throw new IllegalStateException("Consumer group does not commit offsets.");

    	}

    	return records.iterator().next().timestamp();

    }

}