Pulsar IO: KafkaSink handle KeyValue schema

**Is your enhancement request related to a problem? Please describe.**
KafkaSink does not work with KeyValue schema.

**Describe the solution you'd like**
KafkaSink should work with KeyValue schema, I am especially interested in the case of KeyValue

**Describe alternatives you've considered**
Use function to convert from KeyValue to something else

Start at the KafkaSink entry point and trace how message schemas are handled, focusing on KeyValue. Add support for KafkaSink to work with that schema, then verify that the sink accepts and processes KeyValue records; the issue does not name specific files or tests.

Written by the indexing model from the issue text.