[RdKafka] Enable serializers to serialize message keys

[tPl0ch]

Currently the RdKafkaProducer is serializing just the message, which I guess us fine for JSON serialization.

But when you want to use an AvroSerializer that integrates with the Confluent Platform, it is possible to use complex types as keys as well, which in turn means that keys need to be serialized, too.

This change leverages the mutability of the Message by shifting the getKey method call after the call to the Serializer.

[makasim]

Wouldn't it better to add the key dedicated methods to the serializer. Or to introduce key serializer interfaces?

[tPl0ch]

Wouldn't it better to add the key dedicated methods to the serializer?

I would say this is out of question since you'd have to change the stable interface of the SerializerInterface which would break BC.

Or to introduce key serializer interfaces?

That would be a better solution but also way more intrusive (more changes required). I guess I could add the KeySerializerInterface, a default NoopKeySerializer implementation (just returning the unmodified key) and an additional KeySerializerAwareTrait.

Suggestion: Merge this simple & easy to test change now, and implement feature of KeySerializerInterface later.

@makasim What do you think?

[makasim]

Do you have plans to work on a better solution ?

[makasim]

@tPl0ch Could you please share some links on the matter?

[makasim]

What is the key, What role it plays and so on?

[tPl0ch]

@makasim

What is the key, What role it plays and so on?

The key of a Kafka message plays in important role deciding which partition of a topic a message is assigned to (if multiple are available). See this document for more information on partitioning on Kafka topics.

In summary, Kafka only gives ordering guarantees per topic partition. So in order to guarantee order of messages you need to mark related messages (i.e. AggregateId for Domain Events) with a given key.

In our case keys could also be Avro serialized complex types (imagine using serialized ValueObjects as keys), this is why it should be possible to serialize keys separately from the message. You can read about the Confluent platofrm Avro Serialization & Formatters in detail here.

With php-rdkafka partition assignment this is controlled by the constants RD_KAFKA_MSG_PARTITIONER_RANDOM and RD_KAFKA_MSG_PARTITIONER_CONSISTENT, as well by the custom partitioner callback that you can register to override the default Kafka hash alogrithm (which is CRC32 currently).

Do you have plans to work on a better solution?

Yes, I would provide a PR this evening prototyping the solution with an additional KeySerializer which would not rely on message mutability.

Also for future major versions of this library the decision to have mutable message implementations should IMO be revisited.
In return of pthreads for PHP 7.2 and the more & more important role of parallel and async programming having immutable data structures would drastically help with sharing contexts between threads.