wenbingshen

> > If we want a topic to support pulsar consumption > > I'm working on supporting Pulsar consumption for messages whose entry format is Kafka recently. After testing the...

> Yeah, you can see the first optional task in #581. Since we tracked the task in StreamNative's private repo, this issue was not updated recently. Good task. Let us...

> is this change supposed to change KOP in a way that we ALWAYS prepend the "magic + pulsar schema version" header in the Kafka payload received on by the...

> @eolivelli The "magic" doesn't need to be prepend in the Kafka payload. The pulsar schema version is enough. Sorry, I think your "magic" should be the meaning of "MARKER".

@BewareMyPower Can we extend the `header` field of `Reocrd` to save the `pulsar schema version` without needing to embed this information in the payload, but this `header` only supports magic=2:...

> IIUC the Header is not available to the Deserializer, but only the payload Yes, It seems that the Deserializer cannot access the Header in the native kafka client.

https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/serialization/Deserializer.java#L52-L61 Header can be obtained from the Deserializer interface.

> that's great. maybe it has been added recently. This new method was introduced from [KIP-336](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=87298242) , so at least `kafka-client-2.1.0` is required to support accessing `Header`.

> Let's talk back to the way by injecting the schema version into the header. It will break the compatibility because after that there is always at least one header...

> For Kafka clients 0.11.0.0 to 2.0, we need to implement `ExtendedSerializer` and `ExtendedDeserializer` introduced from https://issues.apache.org/jira/browse/KAFKA-4208. > It's not easy to solve header compatibility from 0.11.0 to 2.0 with...