Updated to work with the latest kafka-python API changes