changing kafka to use new