changing kafkaStream decoder to be passed in as a method param