allow set/unset group_id at kafka message consumer