Debug logging on KafkaTopic write errors

[amajedi]

It would be helpful if the message offset and some additional metadata was logged as a result of write exceptions raised doing writes to kafka topics.

Writes are handled asynchronously in KafkaTopic.flush()

A situation I ran into was when attempting to write a large message, one which exceeded the output topic's message size limit. This caused the write to fail, but there was not enough information logged to debug the problematic message.

I suggest at minimum logging the primary key associated with the output message, in addition to the exception logging already in place.

[moritastic]

+1