kafka sink and source for multiple types