Async producer retries for failed messages

[vshlapakov]

Another approach to add retries for failed messages (for asynchronous mode only).

1. I've extended ProduceRequest a little bit by adding complete retries count to the fields.
   It's backward compatible, you can miss this param when creating initial ProduceRequest, it's 0 initially.
   (It's not very clean solution in terms of code, but storing retries count per request sounds reasonable.)
2. I've added several retry params for producers also:
   • batch_retry_backoff_ms - backoff timeout to wait before next retry (300 by default);
   • batch_retries_limit - upper limit of retries per request
     (5 by default, None means infinite retries, 0 means no retries).
3. I've used FailedPayloadsException from Improve fault tolerance by handling leadership election and other metadata changes #55 to get failed requests and increment retries for it, or drop it depending on settings.

What do you think about the approach? I'm going to add some additional tests for it asap.

[dpkp]

There are several exceptions that could be raised within the send_produce_request() call chain besides FailedPayloadsError:

NotLeaderForPartitionError -- metadata is likely out of date and should be refreshed before retry
LeaderNotAvailableError -- leader election is underway, backoff + refresh metadata before retry
UnknownTopicOrPartitionError -- broker is neither a leader or a replica, possible that a metadata refresh and retry would fix, but also possible that topic just doesnt exist and auto-topic-creation is not enabled.
ConnectionError -- socket / connection error, backoff + refresh metadata + retry
KafkaUnavailableError -- all broker nodes are offline... backoff
RequestTimedOutError -- broker was unable to get all ACKs specified by the Produce request w/i the required timeout. If the user is ok w/ possible duplicate messages, a retry is fine.
Then also a bunch of message related errors: InvalidMessage, InvalidMessageSize, MessageSizeTooLarge -- which generally should not be retried

[vshlapakov]

Thanks a lot for the clarifications, it seems that we need more smart retrying logic, I'll think about it. So far I've updated the code a little bit to retry all failed requests without exclusions.

[dpkp]

I dont think we can do blanket retries. Many of the errors require resyncing cluster metadata before retry. The error handling logic should handle that.

[vshlapakov]

Sorry, removed/restored the branch by accident, I'm on this issue.

[vshlapakov]

As this PR is directly related with async producer and retries, I also added an ability to limit async queue size (used some commits from #283 & #304 - look like abandoned).

[vshlapakov]

@dpkp I'm still having some odd errors from Travis, but could you check the approach please? I'm not sure if we're fine with having a named tuple as an argument for a producer.

[dpkp]

now that #366 is merged, we can pass fail_on_error=False to client.send_produce_request and get a list of responses/errors. This would allow us to loop through the responses and only retry the reqs that failed. Currently this exception would cause all reqs in the batch to retry, even if only a single req caused the error.

[vshlapakov]

Ok, thanks for pointing me, I missed that PR!

[vshlapakov]

@dpkp Thanks for all your comments, they're extremely helpful, I'll try to fix everything asap.

[vshlapakov]

Rebased and prepared a draft for response.error check.

[dpkp]

awesome! will take a look shortly!

[dpkp]

There are a handful of issues remaining -- need to reenable the test_switch_leader_async test in test/test_failover_integration.py and then fix it along w/ the lint issues failing travis build.

But you've put in a ton of work and I think it is basically there, so I am going to merge and push a few fixes directly. Thanks so much for working on this one!

[vshlapakov]

Cool, thank you @dpkp! If I have some additional time, I'll fix other issues (like additional tests and docstrings) as well in a separate PR.

[dpkp]

Thanks again for your help on this. I put together some edits in #388 (merged yesterday).