Producer error reporting

[emmettbutler]

This pull request provides a mechanism by which the Producer can report errors on a per-message basis. Currently, this is implemented as Futures returned from produce(), though this method has some problems with memory inefficiency and will need to be changed. @yungchin had the idea of instead using a deque of errored messages that client code can consume.

See also #269, #240

[yungchin]

Status update

I just pushed a first shot at replacing the cancelled futures interface with a queue that delivery reports need to be fetched from, see f85e743 - this is still a work in progress, but I just want to solicit some opinions on the usability of this interface, not to be botching it a second time. There are some rough performance numbers at the bottom of this update.

@emmett9001 @kbourgoin @rduplain @amontalenti

Design considerations / questions

• For now, this just adds one new public method Producer.get_delivery_report(self, block, timeout) where the arguments are just the same as those to queue.Queue.get(), and the return value is a (Message, Exception/None) tuple (where None means success). You'd call this once for each call to produce() - see example code with benchmarks below
• I will need to add an init parameter to enable/disable reporting (with reporting disabled it should then run as fast as current master). We may also want other modes of reporting than the currently implemented one, eg "report only failures", or "use one global queue" (rather than the current use of a thread-local queue). I think I'd like to do that sort of switching by just dynamically replacing the get_delivery_report() method based on an option passed to __init__(), rather than creating mixin classes that require bigger overall changes (users would then see different producer types), but I am open to all suggestions and ideas.
• The main reason I started out with a thread-local queue here is that it nicely supports a use model that the futures interface also supported well: if you have something like a multi-threaded web server, with all threads sharing a single producer, but each thread checking its own delivery failures (or each running with sync=True), then you don't want them to de-queue each others reports. On the other hand, I suppose there may be use models where you'd produce() from multiple threads but have a single thread somewhere handling all delivery failures. This would not be possible with the currently proposed code, so maybe we really do need that "use one global queue" option too? (Not that I know of anyone with that particular use model...)

Quick benchmark

This ran on my laptop, with the kafka cluster also on there, so I've kept only one significant decimal in the timings. They ran in ipython with timeit bench_prod(10**5, linger_ms=100). All these were rebased against current master to make them otherwise similar.

• current master at 66adb22 ran in 3.8s with this benchmark:

def bench_prod(n, **kwargs):                                           
    with topic.get_producer(**kwargs) as prod:
        for _ in xrange(n):
            prod.produce(None, partition_key="yow")

• the futures implementation, rebased at 470daf2 ran in 7.3s with

def bench_prod(n, **kwargs):                                           
    with topic.get_producer(**kwargs) as prod:
        concurrent.futures.wait([
            prod.produce(None, partition_key="yow") for _ in xrange(n)])

• the queue implementation at f85e743 ran in 4.5s with

def bench_prod(n, **kwargs):                                           
    with topic.get_producer(**kwargs) as prod:                                   
        for _ in xrange(n):
            prod.produce(None, partition_key="yow") 
        for _ in xrange(n):
            msg, exc = prod.get_delivery_report()
            assert exc is None

Of course master is at a considerable advantage, not doing any reporting, and I think the overhead of the reports-queue implementation is quite reasonable.

[emmettbutler]

I think this function should be factored out, since it's only one line and doesn't make that line significantly more readable than it already is.

[yungchin]

It might grow to a two-liner when we check if reporting is enabled or not, but yeah, it could still be quickly factored out. I had mostly been thinking of what it might all look like if we ended up with a sort of mixin :)

[emmettbutler]

@yungchin this looks great, I'm excited to get it into master. My main thought in response to your design considerations is that I'd prefer to avoid adding switchable modes to this system too eagerly. I think we should figure out which mode is the most likely to be generally useful and present it as the only option. If after that we find that users would like tweaks that support different use cases, we can keep these mode suggestions in mind. (This doesn't include enabling/disabling delivery reporting entirely - I agree that's necessary).

I think we should stick with reporting all deliveries, not just errors. The original impetus for this feature was a user who needed confirmation that each message had been produced, and I think it's fine to risk providing slightly more information than some use cases will require. Making this switchable seems like needless machinery for too small a benefit.

After a bit of thought, I believe it's preferable to use thread local queues. Doing so does force delivery reports to be consumed on the same thread, meaning users can't asynchronously retry their failed messages. On the other hand, a global queue means that we break any situation where a producer is being shared across multiple threads, which I think is something that's reasonable to support. The question boils down to how important we think it is to allow delivery reports to be consumed and retries to happen asynchronously - and my opinion is that it's not extremely important.

[yungchin]

Thanks @emmett9001, cool, that sounds like the best plan really - if nobody comes asking for other modes, we'll have avoided a bunch of code that nobody ends up using. The mode that's here now, with thread-local queues, is probably the most flexible one and would be the one I'd be using.

[yungchin]

@emmett9001 I've updated the docs, and delivery-reporting is now disabled with default settings. When it's disabled, the mini-benchmark posted previously runs .1s faster than current master - presumably that's a little Message.__slots__ boost.

If we wanted to have a "global queue" option later, it would be a very small addition: we could just drop in basically the same class as _DeliveryReportQueue, but not inheriting from threading.local.

Also for later reference: I ran a very quick benchmark with a "report errors only" adaptation, and that runs as fast as the no-reporting option (as you'd expect, if no delivery errors occur). So basically that's 10e5 messages in 3.7s for errors-only, and 4.1/4.5s for report-all (the 4.5s figure includes the calls to get_delivery_report()).