refactored kafkasink class