added a kafka consume (not sure if works)