Fix gateway with Pulsar

[cbornet]

No description provided.

[eolivelli]

Lgtm

The only thing that I am not sure is that on the Producer we enforce a String schema on the topic.
As the topic is in theory schemaless we should keep BYTES. But we can address it in a follow up work

[cbornet]

You're right. It should be BYTES.
The other concern I have is whether the read MessageId contains the info for all partitions.

[eolivelli]

We have to add test cases for partitioned topics. I guess we have to manually keep track of each partition

[cbornet]

The only thing that I am not sure is that on the Producer we enforce a String schema on the topic.
As the topic is in theory schemaless we should keep BYTES

We can do like Kafka and wait for the first message to infer the schema type based on the Java type. I'll do it.