logstash-input-kafka: Kafka logstash input do not continue from where it left off on restart

This is highly undesirable, the reason we are publishing data to kafka is to ensure the consumers can be taken down and can come back up in an asynchronous fashion. Is this a limitation of this plugin? or simply a configuration issue.

Also where does the kafka input save the kafka position per partition ?

I am using the latest version of the plugin with the following kafka config

input {
    kafka {
        zk_connect => "zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181"
        topic_id => "logs"
        consumer_threads => 1
        consumer_restart_on_error => true
        consumer_restart_sleep_ms => 100
        decorate_events => true
    type => "logs"
    }
}