Kafka Async Producer

[herriojr]

Ok, so I'm trying to integrate with LinkedIn's Camus project and was using kafka-python as the python frontend to talk to Kafka. For Camus, it is expected that there is a single message format per topic (all done in Avro, but that's just a detail). The problem I see with the current API here is if I want to run a Producer in Async mode, I have to have a single producer per topic, and if I run multiple python server processes on a single box to take advantage of all the cores, I'm going to have Process * Topic Producer Daemons per machine and given python doesn't fully respect copy-on-write, that can end up being a lot of memory being used.

It would be much more useful if either the interface allowed processes to share a pipe for writing as well as use a single producer for multiple topics. If I get a little bit of free time, I'll see about adding this in myself. I just wanted to report it as a feature request.

[rdiomar]

Why can't you create multiple async producers for a single topic? It looks to me like the only restriction is that each producer has a single topic. You can have as many producers as you want, each sending to any topic you want, in a single process. It might be nice to have a single producer be able to send to multiple topics, but unless I'm missing something, the current implementation is not lacking any functionality.

In my codebase, I created a global dictionary that maps topic->producer, and wrote a thread-safe get_producer(topic) function. It checks the dict for that topic and returns the producer if it's there. If not, it creates one and adds it.

Let me know if that helps. If you're getting errors with the implementation you want, paste them here and I'll take a look.

[herriojr]

I think I may have mislead you by how I phrased some things in the previous message because I don't want want to create multiple async producers for a single topic. Maybe I misread something in the the Producer code, but what it looks like to me is if it is enabled it to run in async mode, it forks off a process for each Producer created. Also, producers have the topic as part of its members. This means, if I have 50 different topics, I will have 50 different Producers all running in async mode (meaning 50 different processes). Now, if I have an 8 core machine and want to take advantage of all those cores by running more web servers, the total process count will go up to 8 * 50. What I want to take advantage of is Async mode, but I don't want it to re-create a process for a topic when one already exists that can handle that topic. Remember, there's only 1 topic per producer as shown in the following code snippet:

class Producer(object):
    ...
    def __init__(self, client, async=False,
                 req_acks=ACK_AFTER_LOCAL_WRITE,
                 ack_timeout=DEFAULT_ACK_TIMEOUT,
                 batch_send=False,
                 batch_send_every_n=BATCH_SEND_MSG_COUNT,
                 batch_send_every_t=BATCH_SEND_DEFAULT_INTERVAL):

        if batch_send:
            async = True
            assert batch_send_every_n > 0
            assert batch_send_every_t > 0
        else:
            batch_send_every_n = 1
            batch_send_every_t = 3600

        self.client = client
        self.async = async
        self.req_acks = req_acks
        self.ack_timeout = ack_timeout

        if self.async:
            self.queue = Queue()  # Messages are sent through this queue
            self.proc = Process(target=_send_upstream,
                                args=(self.topic,
                                      self.queue,
                                      self.client.copy(),
                                      batch_send_every_t,
                                      batch_send_every_n,
                                      self.req_acks,
                                      self.ack_timeout))

            # Process will die if main thread exits
            self.proc.daemon = True
            self.proc.start()
    ...

What I was proposing, which, correct me if I'm wrong if it already exist, we could actually have multiple processes communicate to a single process for a single topic and start up a single process to handle that topic, or even further, have a single process which can handle a subset of topics. The thing I'm worried about is the total number of Producer daemons running at any given time being (Number of Topics * Number of Services) on a single machine. Maybe I'm completely off-base, but it seems scary that any time a topic is added (which I was using your method of choosing producers based on topic anyways), a new process will be spun up to process the messages and send them out.

Another issue is that Python doesn't really follow copy-on-write as pretty much anytime a variable is accessed it is copied, so if it is already a web server that just wants to send out an async message, it duplicates itself to run the forked producer process.

The other option for me is to create kitchen sink messages, but that ends up bloating the network communication when I really would only use a subset of the protocol at any given time.

[rdiomar]

I see what you mean. The right approach would be to make the producer capable of sending messages to multiple topics, then you only need a single async producer per process. I will gladly take a shot at implementing that.

Not sure I understand the copy-on-write issue. By "web server" do you mean producer? If you're saying that if an async producer is created, a new process will be forked every time you call access it to send messages, that's not true. You can test it by printing the pid in _send_upstream if you like. Or did I misunderstand again?

[herriojr]

The right approach would be to make the producer capable of sending messages to multiple topics, then you only need a single async producer per process.

Yes, I agree with that approach. More or less, that's what I was proposing :P

Not sure I understand the copy-on-write issue.

I'll see if I can find the talk that explains it really well, but python doesn't take advantage of the OS's copy-on-write mechanism which reduces the memory footprint of forked processes compared to the parent process. Python isn't very good at taking advantage of this, so spinning off processes willy-nilly can cause the application to become a memory hog very quickly.

By "web server" do you mean producer?

Pretty much. I was just trying to provide a real world example where this would matter.

If you're saying that if an async producer is created, a new process will be forked every time you call access it to send messages, that's not true.

We both have the same understanding on this.

So to sum things up. It would be very beneficial if a producer supported multiple topics :P

[rdiomar]

Check out #111