MLHR-1416 

Provide api to track the kafka consumer offset and fix offset reset bug in SimpleKafkaConsumer
How can the offset manager be transient when the user should be able to configure it?