I think if we put `override def sinkFields = Fields("key", "value")` in VersionedKeyValSource it should work.