kafka08 simple consumer firehose

I also changed druid-api package, added a method "public long getLastOffsetCommitTime(); " in "interface Firehose". The  way to call new commit method is hacked by using reflection,  quick and dirty way, need to refactor to add some method to interface. 
Your are correct, when I build this simple consumer firehose, I copy the basic structure  from kafka8 high level firehose. In their factory file , they said  "This is actually not going to do exactly what we want, cause it will be called asynchronously after the persist is complete." So I added a commit method with commit method with offset parameter, that is why the reflections here.  Actually we use the method with offset parameter , so this method never used.  I think I can simply put the offset calculation out of the run method , so nothing will miss. And I don't need the reflection hack. Thanks for the review.