Add retry support for producer

[mahendra]

I feel, we must add some retry support for producer.

If client.send_produce_request() fails, because a broker (for a partition) goes down, we can simply do

• conn.close() and
• client.conns.pop(broker)
• try sending the message again, possibly after a retry delay.

This will ensure that the new broker for a partition will come up and we can send the message again.

This can be done after #33 is merged.

[jimjh]

My changes in #55 basically has the client reset its topics_to_broker dictionary when a socket error occurs, so that subsequent requests will trigger _load_metadata_for_topic. This allows the client to discover new leaders.

I support having a retry with good exponential back-off, but I don't know a good place to put it.

[mahendra]

@jimjh We can have it in producer.py/_send_upstream() - with your changes in #55, we can get a list of failed produce requests. We could keep it in a temporary buffer/queue and send it out later.

[dpkp]

#331 and related prs are merged -- async producer should handle retries properly now.