Reorganize to use a signle kafka stream