title: Kafka very high latency C#
body: <p>I am doing some performance tests on Apache Kafka to compare it with others like RabbitMQ and ActiveMQ. The idea is to use it on a messaging system for agents' communication.</p>
  
  <p>I am testing multiple scenarios (one to one, broadcast and many to one) with different numbers of publishers and subscribers and so different loads. Even in the lowest load scenario of one to one with 10 pairs of agents sending 500 messages with 1ms delay between sends I am experiencing very high latencies (average of ~200ms). And if we go to 100 pairs the numbers rise to ~1500ms. The same thing happens on broadcast and many to one.</p>
  
  <p>I am using Windows with Kafka 2.12-2.5.0 and zookeeper 3.6.1 with C# .Net client Confluent.Kafka 1.4.2. I have already tried some properties like LingerMs = 0 according to some posts I found. I have both Kafka and zookeeper with default settings. </p>
  
  <p>I made a simple test code in which the problem happens: </p>
  
  <pre><code>using System;
  using System.Collections.Generic;
  using System.IO;
  using System.Linq;
  using System.Text;
  using System.Threading;
  using System.Threading.Tasks;
  using Confluent.Kafka;
  
  namespace KafkaSetupAgain
  {
      class Program
      {
          static void Main(string[] args)
          {
              int numberOfMessages = 500;
              int numberOfPublishers = 10;
              int numberOfSubscribers = 10;
              int timeOfRun = 30000;
  
              List&lt;MCVESubscriber&gt; Subscribers = new List&lt;MCVESubscriber&gt;();
              for (int i = 0; i &lt; numberOfSubscribers; i++)
              {
                  MCVESubscriber ZeroMqSubscriber = new MCVESubscriber();
                  new Thread(() =&gt;
                  {
                      ZeroMqSubscriber.read(i.ToString());
                  }).Start();
                  Subscribers.Add(ZeroMqSubscriber);
              }
  
  
              Thread.Sleep(10000);//to make sure all subscribers started
              for (int i = 0; i &lt; numberOfPublishers; i++)
              {
                  MCVEPublisher ZeroMqPublisherBroadcast = new MCVEPublisher();
                  new Thread(() =&gt;
                  {
                      ZeroMqPublisherBroadcast.publish(numberOfMessages, i.ToString());
                  }).Start();
              }
  
              Thread.Sleep(timeOfRun);
              foreach (MCVESubscriber Subscriber in Subscribers)
              {
                  Subscriber.PrintMessages("file.csv");
              }
          }
  
          public class MCVEPublisher
          {
              public void publish(int numberOfMessages, string topic)
              {
                  var config = new ProducerConfig
                  {
                      BootstrapServers = "localhost:9092",
                      LingerMs = 0,
                      Acks = 0,
                  };
                  var producer = new ProducerBuilder&lt;Null, string&gt;(config).Build();
  
                  int success = 0;
                  int failure = 0;
                  Thread.Sleep(3500);
                  for (int i = 0; i &lt; numberOfMessages; i++)
                  {
                      Thread.Sleep(1);
                      long milliseconds = System.Diagnostics.Stopwatch.GetTimestamp() / TimeSpan.TicksPerMillisecond;
  
                      var t = producer.ProduceAsync(topic, new Message&lt;Null, string&gt; { Value = milliseconds.ToString() });
                      t.ContinueWith(task =&gt; {
                          if (task.IsFaulted)
                          {
                              failure++;
                          }
                          else
                          {
                              success++;
                          }
                      });
                  }
  
                  Console.WriteLine("Success: " + success + " Failure:" + failure);
  
              }
          }
  
          public class MCVESubscriber
          {
              private List&lt;string&gt; prints = new List&lt;string&gt;();
  
              public void read(string topic)
              {
                  var config = new ConsumerConfig()
                  {
                      BootstrapServers = "localhost:9092",
                      EnableAutoCommit = false,
                      FetchErrorBackoffMs = 1,
                  };
  
                  var consumerConfig = new ConsumerConfig(config);
                  consumerConfig.GroupId = Guid.NewGuid().ToString();
                  consumerConfig.AutoOffsetReset = AutoOffsetReset.Earliest;
                  consumerConfig.EnableAutoCommit = false;
  
                  using (var consumer = new ConsumerBuilder&lt;Ignore, string&gt;(config).Build())
                  {
                      consumer.Subscribe(new[] { topic });
  
                      while (true)
                      {
                          var consumeResult = consumer.Consume();
  
                          long milliseconds = System.Diagnostics.Stopwatch.GetTimestamp() / TimeSpan.TicksPerMillisecond;
                          prints.Add(consumeResult.Message.Value + ";" + milliseconds.ToString());
                      }
  
                      consumer.Close();
                  }
              }
  
              public void PrintMessages(string path)
              {
                  Console.WriteLine("printing " + prints.Count);
                  File.AppendAllLines(path, prints);
              }
  
          }
      }
  }
  </code></pre>
  
  <p>Does someone what can be the problem? What configs can I change to 
  improve latency?</p>
  
  <p>Thanks,</p>
  
  <p>Davide Costa</p>