- implement kafka reader