Using kafka_headers in value expressions

[jcrean]

Using the Kafka source in a pipeline and trying to figure out if its possible to use Kafka headers to do things like conditional routing or dynamic targeting of OpenSearch indices.

For example, something like:

routes:
  - source_app1: "getMetadata(\"kafka_headers/source\") == \"app1\""

Since Kafka header values are byte arrays and expressions only support basic types, this throws:

ExpressionCoercionException: Unsupported type for value [B@abcdef

Is there a way to use the kafka_headers that get added to the event metadata? Some way to force it to parse the byte array as a string?

[kkondaka]

@jcrean would it work for you if Data Prepper has a way to convert "kafka_headers/source" from byte array to string?