updating code to accommodate the latest changes in kafka 0.8