KAFKA-1326 Refactor Sender to support consumer.