using new kafka spout.