Kafka 0.8

Make some changes to work with Kafka 0.8.