Fixes for Kafka consumer

* Don't pass driver's kwargs (which includes path) to KeyedProducer
  constructor.
* Encode topic & key to bytes.

Change-Id: Idec52a421e82660c9f62f97dec29ace5e6dcb09c
Merge "Fixes for Kafka consumer"