connect gearpump with Kafka

Will maintain 
```
iterator.setOffset(nextOffsets(partition))
nextOffsets.put(partition, iterator.getOffset())
``` 
in MessageIterator a better idea?

setOffset should only be awared when recovery.