Refactored to convert MessageAndMetadata message to String once its consumed from Kafka and pass that around instead of MessageAndMetadata