Implement support for keyed messages

Implement support for Keyed message producer. Provides support for two partitioners
* Round robin
* Hashed (default as per kafka clients)