modify kafkasink to deal with multiple topics