feat(kafka): add logic to handle protobuf deserialization (#6841)

feat(kafka): add logic to handle protobuf deserialization (#6841) #4801