running topology, reading from kafka