KafkaSink, added shutdown to KafkaSource