Producer Refactor Mk II

examples/AdvancedProducer/Program.cs

@@ -1,4 +1,5 @@
 using System;
+using System.IO;
 using System.Text;
 using System.Collections.Generic;
 using System.Threading.Tasks;
@@ -36,14 +37,8 @@ public static void Main(string[] args)
 
             var config = new Dictionary<string, object> { { "bootstrap.servers", brokerList } };
 
-            using (var producer = new Producer<string, string>(config))
+            using (var producer = new Producer<string, string>(config, new StringSerializer(Encoding.UTF8), new StringSerializer(Encoding.UTF8)))
             {
-                // TODO: work out why explicit cast is needed here.
-                // TODO: remove need to explicitly specify string serializers - assume Utf8StringSerializer in Producer as default.
-                // TODO: allow be be set only in constructor. make readonly.
-                producer.KeySerializer = (ISerializer<string>)new Confluent.Kafka.Serialization.Utf8StringSerializer();
-                producer.ValueSerializer = producer.KeySerializer;
-
                 Console.WriteLine("\n-----------------------------------------------------------------------");
                 Console.WriteLine($"Producer {producer.Name} producing on topic {topicName}.");
                 Console.WriteLine("-----------------------------------------------------------------------");
@@ -68,7 +63,7 @@ public static void Main(string[] args)
                     {
                        text = Console.ReadLine();
                     }
-                    catch
+                    catch (IOException)
                     {
                         // IO exception is thrown when ConsoleCancelEventArgs.Cancel == true.
                         break;
@@ -82,10 +77,10 @@ public static void Main(string[] args)
                     if (index != -1)
                     {
                         key = text.Substring(0, index);
-                        val = text.Substring(index);
+                        val = text.Substring(index + 1);
                     }
 
-                    Task<DeliveryReport> deliveryReport = producer.Produce(topicName, key, val);
+                    Task<DeliveryReport> deliveryReport = producer.ProduceAsync(topicName, key, val);
                     var result = deliveryReport.Result; // synchronously waits for message to be produced.
                     Console.WriteLine($"Partition: {result.Partition}, Offset: {result.Offset}");
                 }

examples/Benchmark/Program.cs

@@ -1,50 +1,56 @@
 using System;
+using System.Linq;
 using System.Collections.Generic;
-using System.Diagnostics;
 using System.Threading;
 using System.Threading.Tasks;
-using Confluent.Kafka.Serialization;
 
 
 namespace Confluent.Kafka.Benchmark
 {
     public class Program
     {
-        public class DeliveryHandler : IDeliveryHandler
+        public class BenchmarkProducer
         {
-            public void SetException(Exception exception)
+            public static void Run(string broker, string topicName, int numberOfMessagesToProduce, int numberOfTests)
             {
-                throw exception;
-            }
-
-            public void SetResult(DeliveryReport deliveryReport)
-            {
-            }
-        }
-
-        public static void Produce(string broker, string topicName, long numMessages)
-        {
-            var deliveryHandler = new DeliveryHandler();
-
-            var config = new Dictionary<string, object> { { "bootstrap.servers", broker } };
-
-            using (var producer = new Producer<Null, byte[]>(config))
-            {
-                // TODO: remove need to explicitly specify this serializer.
-                producer.ValueSerializer = (ISerializer<byte[]>)new ByteArraySerializer();
+                // mirrors the librdkafka performance test example.
+                var config = new Dictionary<string, object>
+                {
+                    { "bootstrap.servers", broker },
+                    { "queue.buffering.max.messages", 500000 },
+                    { "message.send.max.retries", 3 },
+                    { "retry.backoff.ms", 500 },
+                    { "session.timeout.ms", 6000 }
+                };
 
-                Console.WriteLine($"{producer.Name} producing on {topicName}");
-                // TODO: think more about exactly what we want to benchmark.
-                var payload = new byte[] { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 };
-                for (int i = 0; i < numMessages; i++)
+                using (var producer = new Producer(config))
                 {
-                    producer.ProduceWithDeliveryReport(topicName, payload, deliveryHandler);
+                    for (var j=0; j<numberOfTests; ++j)
+                    {
+                        Console.WriteLine($"{producer.Name} producing on {topicName}");
+
+                        byte cnt = 0;
+                        var val = new byte[100].Select(a => ++cnt).ToArray();
+
+                        var startTime = DateTime.Now.Ticks;
+                        var tasks = new Task[numberOfMessagesToProduce];
+                        for (int i = 0; i < numberOfMessagesToProduce; i++)
+                        {
+                            tasks[i] = producer.ProduceAsync(topicName, null, val);
+                        }
+                        Task.WaitAll(tasks);
+                        var duration = DateTime.Now.Ticks - startTime;
+
+                        Console.WriteLine($"Produced {numberOfMessagesToProduce} in {duration/10000.0:F0}ms");
+                        Console.WriteLine($"{numberOfMessagesToProduce / (duration/10000.0):F0} messages/ms");
+                    }
+
+                    Console.WriteLine("Disposing producer");
                 }
-
-                Console.WriteLine("Shutting down");
             }
         }
 
+        // TODO: Update Consumer benchmark for new Consumer when it's written.
         public static async Task<long> Consume(string broker, string topic)
         {
             long n = 0;
@@ -86,24 +92,7 @@ public static void Main(string[] args)
             string brokerList = args[0];
             string topic = args[1];
 
-            long numMessages = 1000000;
-
-            var stopwatch = new Stopwatch();
-
-            // TODO: we really want time from first ack. as it is, this includes producer startup time.
-            stopwatch.Start();
-            Produce(brokerList, topic, numMessages);
-            stopwatch.Stop();
-
-            Console.WriteLine($"Sent {numMessages} messages in {stopwatch.Elapsed}");
-            Console.WriteLine($"{numMessages / stopwatch.Elapsed.TotalSeconds:F0} messages/second");
-
-            stopwatch.Restart();
-            long n = Consume(brokerList, topic).Result;
-            stopwatch.Stop();
-
-            Console.WriteLine($"Received {n} messages in {stopwatch.Elapsed}");
-            Console.WriteLine($"{n / stopwatch.Elapsed.TotalSeconds:F0} messages/second");
+            BenchmarkProducer.Run(brokerList, topic, 5000000, 4);
         }
     }
 }

examples/Misc/Program.cs

@@ -13,7 +13,7 @@ static async Task ListGroups(string brokerList)
         {
             var config = new Dictionary<string, object> { { "bootstrap.servers", brokerList } };
 
-            using (var producer = new Producer<Null, Null>(config))
+            using (var producer = new Producer(config))
             {
                 var groups = await producer.ListGroups(TimeSpan.FromSeconds(10));
                 Console.WriteLine($"Consumer Groups:");
@@ -27,9 +27,7 @@ static async Task ListGroups(string brokerList)
                     {
                         Console.WriteLine($"    {m.MemberId} {m.ClientId} {m.ClientHost}");
                         Console.WriteLine($"    Metadata: {m.MemberMetadata.Length} bytes");
-                        //Console.WriteLine(System.Text.Encoding.UTF8.GetString(m.MemberMetadata));
                         Console.WriteLine($"    Assignment: {m.MemberAssignment.Length} bytes");
-                        //Console.WriteLine(System.Text.Encoding.UTF8.GetString(m.MemberAssignment));
                     }
                 }
             }
@@ -38,7 +36,7 @@ static async Task ListGroups(string brokerList)
         static async Task PrintMetadata(string brokerList)
         {
             var config = new Dictionary<string, object> { { "bootstrap.servers", brokerList } };
-            using (var producer = new Producer<Null, Null>(config))
+            using (var producer = new Producer(config))
             {
                 var meta = await producer.Metadata();
                 Console.WriteLine($"{meta.OriginatingBrokerId} {meta.OriginatingBrokerName}");

examples/SimpleProducer/Program.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Text;
 using System.Collections.Generic;
 using System.Threading.Tasks;
 using Confluent.Kafka.Serialization;
@@ -14,23 +15,22 @@ public static void Main(string[] args)
 
             var config = new Dictionary<string, object> { { "bootstrap.servers", brokerList } };
 
-            using (var producer = new Producer<Null, string>(config))
+            using (var producer = new Producer<Null, string>(config, new NullSerializer(), new StringSerializer(Encoding.UTF8)))
             {
-                // TODO: figure out why the cast below is necessary and how to avoid it.
-                // TODO: There should be no need to specify a serializer for common types like string - I think it should default to the UTF8 serializer.
-                producer.ValueSerializer = (ISerializer<string>)new Confluent.Kafka.Serialization.Utf8StringSerializer();
-
                 Console.WriteLine($"{producer.Name} producing on {topicName}. q to exit.");
 
                 string text;
                 while ((text = Console.ReadLine()) != "q")
                 {
-                    Task<DeliveryReport> deliveryReport = producer.Produce(topicName, text);
+                    Task<DeliveryReport> deliveryReport = producer.ProduceAsync(topicName, null, text);
                     var unused = deliveryReport.ContinueWith(task =>
                     {
                         Console.WriteLine($"Partition: {task.Result.Partition}, Offset: {task.Result.Offset}");
                     });
                 }
+
+                // Tasks are not waited on, it's possible they may still in progress here.
+                producer.Flush();
             }
         }
     }

examples/Wrapped/Program.cs

@@ -0,0 +1,50 @@
+using System.Text;
+using System.Collections.Generic;
+using Confluent.Kafka.Serialization;
+
+namespace Confluent.Kafka.Wrapped
+{
+    /// <summary>
+    ///     An example showing how to wrap a single Producer to produce messages using
+    ///     different serializers.
+    /// </summary>
+    /// <remarks>
+    ///     If you only want to use a single pair of serializers in your application,
+    ///     you should use the Producer&lt;TKey, TValue&gt; constructor instead.
+    /// </remarks>
+    public class Program
+    {
+        public static void Main(string[] args)
+        {
+            var config = new Dictionary<string, object> { { "bootstrap.servers", args[0] } };
+
+            using (var producer = new Producer(config))
+            {
+                // sProducer1 is a lightweight wrapper around a Producer instance that adds
+                // (string, string) serialization. Note that sProducer1 does not need to be
+                // (and cannot be) disposed.
+                var sProducer1 = producer.Wrap<string, string>(new StringSerializer(Encoding.UTF8), new StringSerializer(Encoding.UTF8));
+
+                // sProducer2 is another lightweight wrapper around kafkaProducer that adds
+                // (null, int) serialization. When you do not wish to write any data to a key
+                // or value, the Null type should be used.
+                var sProducer2 = producer.Wrap<Null, int>(new NullSerializer(), new IntSerializer());
+
+                // write (string, string) data to topic "first-topic", statically type checked.
+                sProducer1.ProduceAsync("first-topic", "my-key-value", "my-value");
+
+                // write (null, int) data to topic "second-data". statically type checked, using
+                // the same underlying producer as the producer1.
+                sProducer2.ProduceAsync("second-topic", null, 42);
+
+                // producers are NOT tied to topics. Although it's unusual that you might want to
+                // do so, you can use different serializing producers to write to the same topic.
+                sProducer2.ProduceAsync("first-topic", null, 107);
+
+                // ProducerAsync tasks are not waited on - there is a good chance they are still
+                // in flight.
+                producer.Flush();
+            }
+        }
+    }
+}

examples/Wrapped/project.json

@@ -0,0 +1,25 @@
+{
+    "version": "1.0.0",
+    "authors": ["Confluent Inc"],
+
+    "buildOptions": {
+        "emitEntryPoint": true
+    },
+
+    "dependencies": {
+        "Confluent.Kafka": {
+            "target": "project"
+        }
+    },
+
+    "frameworks": {
+        "netcoreapp1.0": {
+            "dependencies": {
+                "Microsoft.NETCore.App": {
+                    "type": "platform",
+                    "version": "1.0.0"
+                }
+            }
+        }
+    }
+}

src/Confluent.Kafka/DeliveryReport.cs

@@ -0,0 +1,17 @@
+namespace Confluent.Kafka
+{
+    /*
+    TODO: (via ewencp): The equivalent in the Python client fills in more information --
+    the callbacks accept (err, msg) parameters, where the latter is
+    http://docs.confluent.io/3.1.0/clients/confluent-kafka-python/index.html#confluent_kafka.Message
+    Same deal with Go where the DR channel gets one of these:
+    http://docs.confluent.io/3.1.0/clients/confluent-kafka-go/index.html#Message Is this being
+    kept more minimal intentionally?
+    */
+
+    public struct DeliveryReport
+    {
+        public int Partition;
+        public long Offset;
+    }
+}

src/Confluent.Kafka/Handle.cs

@@ -69,6 +69,16 @@ internal void Init(RdKafkaType type, IntPtr config, Config.LogCallback logger)
             callbackTask = StartCallbackTask(callbackCts.Token);
         }
 
+        // TODO: Add timout parameter (with default option == block indefinitely) when use rd_kafka_flush.
+        public void Flush()
+        {
+            // TODO: use rd_kafka_flush here instead..
+            while (OutQueueLength > 0)
+            {
+                handle.Poll((IntPtr) 100);
+            }
+        }
+
         public void Dispose()
         {
             Dispose(true);
@@ -83,12 +93,6 @@ protected virtual void Dispose(bool disposing)
 
             if (disposing)
             {
-                // Wait until all outstanding sends have completed.
-                while (OutQueueLength > 0)
-                {
-                    handle.Poll((IntPtr) 100);
-                }
-
                 handle.Dispose();
             }
         }

src/Confluent.Kafka/ISerializingProducer.cs

@@ -0,0 +1,17 @@
+using System.Threading.Tasks;
+using Confluent.Kafka.Serialization;
+
+
+namespace Confluent.Kafka
+{
+    public interface ISerializingProducer<TKey, TValue>
+    {
+        string Name { get; }
+
+        ISerializer<TKey> KeySerializer { get; }
+
+        ISerializer<TValue> ValueSerializer { get; }
+
+        Task<DeliveryReport> ProduceAsync(string topic, TKey key, TValue val, int? partition = null, bool blockIfQueueFull = true);
+    }
+}

src/Confluent.Kafka/Impl/LibRdKafka.cs

@@ -341,14 +341,14 @@ internal static ErrorCode committed(IntPtr rk, IntPtr partitions, IntPtr timeout
         internal static ErrorCode position(IntPtr rk, IntPtr partitions)
             => _position(rk, partitions);
 
-        private static Func<IntPtr, int, IntPtr, byte[], UIntPtr, byte[], UIntPtr,
+        private static Func<IntPtr, int, IntPtr, IntPtr, UIntPtr, IntPtr, UIntPtr,
                 IntPtr, IntPtr> _produce;
         internal static IntPtr produce(
                 IntPtr rkt,
                 int partition,
                 IntPtr msgflags,
-                byte[] val, UIntPtr len,
-                byte[] key, UIntPtr keylen,
+                IntPtr val, UIntPtr len,
+                IntPtr key, UIntPtr keylen,
                 IntPtr msg_opaque)
             => _produce(rkt, partition, msgflags, val, len, key, keylen, msg_opaque);
 
@@ -607,8 +607,8 @@ internal static extern IntPtr rd_kafka_produce(
                     IntPtr rkt,
                     int partition,
                     IntPtr msgflags,
-                    byte[] val, UIntPtr len,
-                    byte[] key, UIntPtr keylen,
+                    IntPtr val, UIntPtr len,
+                    IntPtr key, UIntPtr keylen,
                     IntPtr msg_opaque);
 
             [DllImport(DllName, CallingConvention = CallingConvention.Cdecl)]
@@ -644,5 +644,6 @@ internal static extern IntPtr rd_kafka_brokers_add(IntPtr rk,
             [DllImport(DllName, CallingConvention = CallingConvention.Cdecl)]
             internal static extern IntPtr rd_kafka_wait_destroyed(IntPtr timeout_ms);
         }
+
     }
 }