Added a bunch of log statemtns. Turns out that the kafka daemon wasn't running properly.