restructure kafka