change to latest kafka API