Changes to support kafka 2.8