Kafka producer/consumer with 0.8.2-beta

- producer/consumer module splitted
- KafkaSink with java kafka producer without any duplicate queue
Why do we need the retention value to get the initial value of hash? Cannot it just be `new AtomicLong(new Random().nextInt(1000))`