1. When we hit an error in producer, we call the callback with a null RecordMetadata. If the callback tries to get the topic/partition, it will hit a NullPointerException. It's probably better to create a RecordMetadata with -1 as the offset.
2. When printing out a Struct, we don't print out the content wrapped in an array properly. So, we will see sth like the following.
[2014-03-09 11:56:24,364] INFO Created 1 requests: [InFlightRequest(expectResponse=true, batches=
3. Need to override the toString() in ProduceResponse.
4. Sender.run(now): It would be good to log metadata request too.
5. Sender.handleDisconnects(): It would be useful to log the correlation id of cancelled inflight requests