- implement kafkareader