Both kafka_franz and kafka inputs emit the kafka_key metadata as string, but bytes should be used instead

[mihaitodor]

Run either the kafka_franz or the kafka output to produce 2 messages. For the first one, set key: "" and for the second one, do not set this field.

Now, try running this config:

input:
  kafka:
    ...

output:
  stdout: {}
  processors:
    - mapping:
        root = metadata("kafka_key").format_json()

Note that for both messages the output is "" and I can't think of any way to distinguish between them. However, for the second message, we should output null instead.

As far as I can tell, we allow implicit conversion from bytes to string, so maybe we can just get rid of the string cast here and here since we now have structured metadata...

UPDATE: Another issue is that the key interpolation for the outputs can only produce strings, so we don't have a way right now to emit null instead of "" when using interpolation.