modify kafka sink