initial Kafka 0.8 support
initial Kafka 0.8 support