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. 
Ah, ok, yeah, I understand now.  There's some communication that must happen between the Firehose about what state it has persisted and the Plumber about what state it has persisted.  Because there can be a race between the persist of the Sink and the commit back.

I think we can do all of this in the plumber though and keep the Firehose interface from changing.  What do you think about this?

1) The plumber does the persist of the Sink
2) The plumber calls the Runnable from Firehose.commit()
3) The plumber persists one more file indicating that the Sink is good

Then, on startup, the plumber can only load the persisted sinks when both the sink file and the extra file exist.  If the extra file doesn't exist, the node can delete the corrupt segment and move on.  I think there's technically a race with this option as well because you could've persisted the offset and not persisted the file, though...

I know I definitely at least have an issue with the name of the method, `getLastOffsetCommitTime` which is pretty tied to the fact that kafka uses offsets and commits them.  More than that, I'd also prefer not to change the Firehose interface, if possible.  And if it's not possible to change the interface, I'd like to spend some more time thinking about what method signature would lend the most flexibility to Firehose implementations...