- implement KafkaReader