ported consumer to kotlin