implemented  kafka consumer support