rdkafka-backed Producer

[yungchin]

This will depend on the new async Producer coming up in #177.

[yungchin]

@kbourgoin @emmett9001 @rduplain can I bug you guys - and also everyone else who's listening in - for some input on this design choice in particular, before I spend too many hours hammering out the further implementation in the C code?

What this would do is make produce() return a future (either a pykafka.handlers.ResponseFuture or a backported concurrent.futures.Future or something else, if you prefer) and that would just evaluate to None if the message made it to the broker or would raise an appropriate exception from pykafka.exceptions if not (ie if max_retries got exceeded for some messages, and what went wrong). And for self._synchronous=True it could then just evaluate the future as part of produce().

(I'd also volunteer to write the pullreq to make that work the same way in pykafka.Producer if we like it.)

Now the question: I'd like this futures-based design mostly because it is easy to reason about from a user perspective - 1 message in, 1 future out, and you can check it or ignore it, whatever suits - but I'd like your thoughts about the practicality of it in a bigger application. Any and all alternative interface suggestions very welcome.

(I'll note that concurrent.futures can also facilitate a callback interface, if needs be...)

[emmettbutler]

I guess the reason to do this is to take advantage of rdkafka's delivery reports, which makes sense. Implementing these in the Producer should be straightforward, since we're checking for message send success anyway. I believe I'd do it by having OwnedBroker.enqueue create and return a future that it also adds to the enqueued (kv, partition_id) tuple. On the other side, we could fill in the future with whatever values were appropriate based on the fate of the message.

Given that I can picture the implementation already, there's a clear upside (more descriptive return values from produce) and I can't come up with any huge downsides for the average user (who can simply ignore that produce returns anything), I'm +1 to this idea. The question in my mind is whether there is any better way to provide message statuses to the client - a callback interface sounds clumsy, and we already know that a synchronously returned status from produce only gets us half of the way there.

Assuming we agree this makes sense, it should have its own issue number so we can more easily partition the pure python and rdkafka implementations.

[yungchin]

Thanks, yes that's precisely the reason. I was wondering about the same thing - would there be any better way?

I suspect that most users who'd want to evaluate the futures returned (to act on errors) would want to bulk-process them (and with concurrent.futures some facilities for that come provided) in a pattern roughly like so:

# Warning: potential pseudo-code ahead
pending = []
while True:
    for _ in xrange(big_number):
        msg = rawdata_consumer.consume()
        result = do_stuff_with_msg(msg)
        future = nextstage_producer.produce(result)
        pending.append(future)
    timeout = 0  # ie don't wait, just grab what's been delivered
    done, not_done = concurrent.futures.wait(pending, timeout)
    errors = [f.exception() for f in done if f.exception() is not None]
    handle_errors(errors)
    pending = not_done

... and so, one might argue that if that's going to be the pattern, you ought to abstract the pattern into the producer interface, to save everyone the hassle of writing it all the time. On the other hand, it's not a terribly big or complicated pattern, and not baking it into the interface is more flexible, and we can always add convenience-wrappers for this later? So... I guess I'm starting to lean the way of just returning a future (but not a handlers.ResponseFuture, which would lack all the stuff around it).

[yungchin]

Assuming we agree this makes sense, it should have its own issue number [...]

I've put it up at #238 :)

[yungchin]

Quick status update: this now passes all tests that we have for pykafka.Producer, except on pypy. The problems on pypy are probably explained here: http://docs.cython.org/src/userguide/pypy.html#borrowed-references-and-data-pointers - I'm currently keeping borrowed references around in opaque pointers (for the delivery-reporting features). Will hopefully get to fix that later in the week.

The delivery-reporting needs some tests, which I intend to write as part of #238 - but seems to work well enough. That is, save for some test suite additions and bug fixes, this branch is feature complete.

[yungchin]

Another update: delivery-reporting futures now also work fine on pypy (that took a bit of an overhaul actually). That means we can probably declare this branch feature-complete.

The tests still show one minor quirk, which is that librdkafka producer threads aren't cleaned up consistently on pypy, but that should be easily fixed if we define an explicit stop() method like we did for the consumer.

[yungchin]

I'm going to merge this into #176, because there's too much stuff here that I turn out to want to use in the consumer code as well, and cherry-picking and resolving conflicts will become nigh-intractable.

As a final status update here: all the issues mentioned in earlier comments above have been resolved. The one thing left doing, which I'll carry over to #176, is that once #269 has been merged, we should carry the tests over to assert the two producers really have matching interfaces.