Trying to replicate a message sent through Kafka that has headers leads to an error

[lbilger]

If I use this library (I tried functions/config/EventHubCopy but I believe it's a general problem) to replicate an Event Hub where messages come in through Kafka and those messages have headers, I get the following error message:

2021-10-06T08:19:47.923 [Error] Executed 'Functions.copy-all' (Failed, Id=4c649116-0fff-4013-b104-d6f79885b280, Duration=2ms)Serialization operation failed due to unsupported type System.Byte[].

This is probably due to the different encoding of headers / application properties in Kafka vs. AMQP, but still it should be possible to copy the message unchanged without any serialization / deserialization required.

There has already been some discussion on this with @jsquire in my StackOverflow question

Steps to reproduce

• Create an Event Hubs Namespace and an Event Hub
• Create a Shared Access Policy with write access
• Write a message to the Event Hub using kcat, e.g.

  kcat -b "${event_hubs_namespace}.servicebus.windows.net:9093" -X security.protocol=sasl_ssl -X sasl.mechanisms=PLAIN -X sasl.username='$ConnectionString' -X sasl.password="${connection_string}" -t "${event_hub_name}" -H headerkey=headervalue -k key -P <<< value
  

• follow the instructions in this repository to replicate the Event Hub to another one

[jsquire]

@clemensv: Do we have any guidance for transforming an event published with Kafka into a form that the native Event Hubs SDKs would produce? Would this potentially be a corner case to consider for other replication tools as well?

[jeff-lauterbach-by]

I just hit this same issue and stumbled on this issue by coincidence. Is this due to an issue in the replication libary this repo provides or a gap with the Event Hubs SDK? Would I be better off leveraging a more native kafka solution?

[lbilger]

I think it's a problem with the SDK, I got the same error when I tried to create my own Azure Function to copy the data. I think the reason is the different way headers are serialized in AMQP vs. Kafka, as explained in the Microsoft docs.
Still, if the SDK can hand a header in as a Byte[], it would be nice if it could also write it out this way.

[jsquire]

This is something that the replication tasks would be responsible for addressing. Depending on where the exception is occurring, replication may be able to work around it by accessing the raw AMQP message or may need to fallback to using a Kafka-aware client for these scenarios.

The Event Hubs SDK is AMQP-based and does not support other protocols or formats, by design. Events are expected to conform to AMQP message specification and transformed with that constraint. It does allow access to the raw AMQP message and data can be exchanged in that format without working directly with the model attributes, which bypasses some of the transforms.

Any chance that you're willing to share a full stack trace for the exception and/or offer more detail about what behavior you're observing in the Function? That would help to understand what work arounds may be possible to help get you folks unblocked in the interim.