Producer futures

[yungchin]

This closes #238.

Error responses from kafka previously only appeared in log output. With the changes here, produce() returns a concurrent.futures.Future, through which any errors that persist after max_retries can be communicated to user code.

For sync=True, we simply evaluate the Future immediately, so that kafka error responses are raised directly from produce().

This touches a few more lines than a minimal implementation would have done, I hope that's ok. I felt that slapping yet another item (ie the delivery_future) onto message_partition_tup was just going to get too unwieldly, so instead of message_partition_tup we're now moving instances of Message around. The second structural change is in _send_request(), where to_retry previously was an iterable of messages, but given that we were now going to add exceptions to that, and the error responses always apply to a whole MessageSet, to_retry is now an iterable of MessageSet, Exception.

[yungchin]

@rduplain @emmett9001 @kbourgoin I've just updated the description above. Would be grateful if you can check that this makes sense in terms of interface - and implementation of course.

[kbourgoin]

Related to #238.

[kbourgoin]

Connected to #238.

[yungchin]

@emmett9001 can I bug you to review yet another one? :)

[emmettbutler]

[nitpick] No need for this linebreak, I think. I've been using 90 characters as a length limit, since github windows hold at least that without scrolling left to right.

[yungchin]

Thanks, yep, that's it! There are also some convenience functions in the futures module, with which in your example you could split the list up in completed and pending futures, and throw the pending futures back on the list for the next round (otherwise checking future.exception() will block until the future is completed). I should probably steal that example and stick it somewhere in the docs, actually.

[emmettbutler]

Yes, I think it's a good idea to include a code sample like this right above your paragraph in the README on how to use the futures.

[yungchin]

Ok, I was poking around a bit with sys.getsizeof and the __slots__ change seems to reduce Message overhead by 10x (Message.__dict__ was over 1kB on py2 at least). However.... Future and its members weigh in at about twice that. The BDFL giveth, and the BDFL taketh away, or something like that.

[emmettbutler]

Oh wow. In that case we may want a way to turn off futures with a kwarg or something. We should keep that in mind if memory usage ever becomes a problem, but I don't think we necessarily need to add it now.

[yungchin]

With the changes I made as part of f4f6df1, if the user ignores the future, it's gone immediately (the message only holds a weakref to it). So you'd still have the overhead of creating and destroying one, but you wouldn't have one lying around for every message in your outbound-queue.

(The weakref is luckily not in possession of a __dict__, so it's a lot smaller)