title: Python librdkafka producer perform against the native Apache Kafka Producer
body: <p>I am testing Apache Kafka Producer with native java implementation against Python's confluent-kafka to see which has the maximum throughput.</p>
  
  <p>I am deploying a Kafka cluster with 3 Kafka brokers and 3 zookeeper instances using docker-compose. My docker compose file: <a href="https://paste.fedoraproject.org/paste/bn7rr2~YRuIihZ06O3Q6vw/raw" rel="nofollow noreferrer">https://paste.fedoraproject.org/paste/bn7rr2~YRuIihZ06O3Q6vw/raw</a></p>
  
  <p>It's a very simple code with mostly default options for Python confluent-kafka and some config changes in java producer to match that of confluent-kafka.</p>
  
  <p>Python Code:</p>
  
  <pre class="lang-py prettyprint-override"><code>from confluent_kafka import Producer
  
  producer = Producer({'bootstrap.servers': 'kafka-1:19092,kafka-2:29092,kafka-3:39092', 'linger.ms': 300, "max.in.flight.requests.per.connection": 1000000, "queue.buffering.max.kbytes": 1048576, "message.max.bytes": 1000000,
      'default.topic.config': {'acks': "all"}})
  
  ss = '0123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357'
  
  def f():
      import time
      start = time.time()
      for i in xrange(1000000):
          try:
              producer.produce('test-topic', ss)
          except Exception:
              producer.poll(1)
              try:
                  producer.produce('test-topic', ss)
              except Exception:
                  producer.flush(30)
                  producer.produce('test-topic', ss)
          producer.poll(0)
      producer.flush(30)
      print(time.time() - start)
  
  
  if __name__ == '__main__':
      f()
  </code></pre>
  
  <p>Java implementation. Configuration same as config in librdkafka. Changed the linger.ms and callback as suggested by Edenhill.</p>
  
  <pre class="lang-java prettyprint-override"><code>package com.amit.kafka;
  
  import org.apache.kafka.clients.producer.*;
  import org.apache.kafka.common.serialization.LongSerializer;
  import org.apache.kafka.common.serialization.StringSerializer;
  
  import java.nio.charset.Charset;
  import java.util.Properties;
  import java.util.concurrent.TimeUnit;
  
  public class KafkaProducerExampleAsync {
  
      private final static String TOPIC = "test-topic";
      private final static String BOOTSTRAP_SERVERS = "kafka-1:19092,kafka-2:29092,kafka-3:39092";
  
      private static Producer&lt;String, String&gt; createProducer() {
          int bufferMemory = 67108864;
          int batchSizeBytes = 1000000;
          String acks = "all";
  
          Properties props = new Properties();
  
          props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
          props.put(ProducerConfig.CLIENT_ID_CONFIG, "KafkaExampleProducer");
          props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class.getName());
          props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
          props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSizeBytes);
          props.put(ProducerConfig.LINGER_MS_CONFIG, 100);
          props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
          props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1000000);
          props.put(ProducerConfig.ACKS_CONFIG, acks);
  
          return new KafkaProducer&lt;&gt;(props);
      }
  
      static void runProducer(final int sendMessageCount) throws InterruptedException {
          final Producer&lt;String, String&gt; producer = createProducer();
          final String msg = "0123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357";
  
          final ProducerRecord&lt;String, String&gt; record = new ProducerRecord&lt;&gt;(TOPIC, msg);
          final long[] new_time = new long[1];
  
          try {
              for (long index = 0; index &lt; sendMessageCount; index++) {
                  producer.send(record, new Callback() {
                      public void onCompletion(RecordMetadata metadata, Exception e) {
                          // This if-else is to only start timing this when first message reach kafka
                          if(e != null) {
                             e.printStackTrace();
                          } else {
                              if (new_time[0] == 0) {
                                  new_time[0] = System.currentTimeMillis();
                              }
                          }
                      }
                  });
              }
          } finally {
              // producer.flush();
              producer.close();
              System.out.printf("Total time %d ms\n", System.currentTimeMillis() - new_time[0]);
          }
      }
  
      public static void main(String... args) throws Exception {
          if (args.length == 0) {
              runProducer(1000000);
          } else {
              runProducer(Integer.parseInt(args[0]));
          }
      }
  }
  </code></pre>
  
  <hr>
  
  <h2>Benchmark results(Edited after making changes recommended by Edenhill)</h2>
  
  <p><strong>Acks = 0</strong>, <strong>Messages: 1000000</strong></p>
  
  <p>Java: 12.066</p>
  
  <p>Python: 9.608 seconds</p>
  
  <p><strong>Acks: all</strong>, <strong>Messages: 1000000</strong></p>
  
  <p>Java: <s>45.763</s> 11.917 seconds</p>
  
  <p>Python: 14.3029 seconds</p>
  
  <hr>
  
  <p>Java implementation is performing same as Python implementation even after making all the changes that I could think of and the ones suggested by Edenhill in the comment below.</p>
  
  <p>There are various benchmarks about the performance of Kafka in Python but I couldn't find any comparing librdkafka or python Kafka against Apache Kafka.</p>
  
  <p>I have two questions:</p>
  
  <ol>
  <li><p>Is this test enough to come to the conclusion that with default config's and message of size 1Kb librdkafka is faster?</p></li>
  <li><p>Does anyone have experience or a source(blog, doc etc) benchmarking librdkafka against confluent-kafka?</p></li>
  </ol>