high throughput produce

gistfile1.txt

class Program
    {
        static void Main(string[] args)
        {
            var config = new Dictionary<string, object>
            {
                { "bootstrap.servers", "10.200.7.144:9092" }
            };

            var bigNumber = 100;

            var count = 1;
            Action<DeliveryReport<Null, string>> handler = (DeliveryReport<Null, string> dr) => count += 1; // no interlock required.

            using (var producer = new Producer<Null, string>(config, null, new StringSerializer(Encoding.UTF8)))
            {
                var stream = Enumerable.Range(1, bigNumber).Select(i => i.ToString());

                foreach (var msg in stream)
                {
                    while (true) // retry on queue full loop
                    {
                        try
                        {
                            producer.BeginProduce("hello", new Message<Null, string> { Value = msg }, handler);
                            break;
                        }
                        catch (KafkaException e)
                        {
                            if (e.Error.Code == ErrorCode.Local_QueueFull)
                            {
                                Thread.Sleep(100);
                                continue;
                            }
                            
                            // non-recoverable error occured.
                        }
                    }
                }
            }
        }
    }