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. 
This is will give a guarantee that offset file is not corrupted or you don't happen to shutdown the program when you commit offset .  I use md5 to test if the file content and file name match, if it is not match, then the file is corrupted. I will drop this offset file and start from the offset before this one. (usually I have 5 offset files, last 5 times commit) I will use the correct offset file's commit time to compare with the sink file on disk. If the sink file is generated later than the offset file, then we would not load the sink file.