refactor KafkaStream initialization