Modified to use new kafkaOffset setting; added code that can reset the offset with ZooKeeper