Less verbose delivery handler and decreased producer queue (#48)

Converted the delivery handler to no longer return the original key, value etc to reduce memory footprint.

Reduced the max queued message size to 101 MB from 1GB when none specified.

builds/csharp/nuget/build_nugets.py

@@ -7,8 +7,8 @@
 from typing import List
 
 version = "0.7.1.0"
-informal_version = "0.7.1.0-dev1"
-nuget_version = "0.7.1.0-dev1"
+informal_version = "0.7.1.0-dev2"
+nuget_version = "0.7.1.0-dev2"
 
 
 def updatecsproj(projfilepath):

src/QuixStreams.Kafka/KafkaProducer.cs

@@ -121,6 +121,11 @@ private void ConfigureProduceDelegate()
         private ProducerConfig GetKafkaProducerConfig(ProducerConfiguration producerConfiguration)
         {
             var config = producerConfiguration.ToProducerConfig();
+            if (string.IsNullOrWhiteSpace(config.DeliveryReportFields))
+            {
+                config.DeliveryReportFields = "none";
+            }
+
             config.Debug = config.Debug;
             if (!string.IsNullOrWhiteSpace(config.Debug))
             {
@@ -492,19 +497,19 @@ private Task SendInternal(KafkaMessage message, ProducerDelegate handler,  Cance
                 return Task.FromCanceled<DeliveryResult<byte[], byte[]>>(cancellationToken);
             }
 
-            var taskSource = new TaskCompletionSource<DeliveryResult<byte[], byte[]>>(TaskCreationOptions.RunContinuationsAsynchronously);
+            var taskSource = new TaskCompletionSource<TopicPartitionOffset>(TaskCreationOptions.RunContinuationsAsynchronously);
 
             void DeliveryHandler(DeliveryReport<byte[], byte[]> report)
             {
                 if (report.Error?.IsError == true)
                 {
                     this.logger.LogTrace("[{0}] {1} {2}", this.configId, report.Error.Code, report.Error.Reason);
                     var wrappedError = new Error(report.Error.Code, $"[{this.configId}] {report.Error.Reason}", report.Error.IsFatal);
-                    taskSource.SetException(new ProduceException<byte[], byte[]>(wrappedError, report));
+                    taskSource.SetException(new KafkaException(wrappedError));
                     return;
                 }
 
-                taskSource.SetResult(report);
+                taskSource.SetResult(report.TopicPartitionOffset);
             }
 
             var success = false;

src/QuixStreams.Kafka/ProducerConfiguration.cs

@@ -36,7 +36,7 @@ public ProducerConfiguration(string brokerList, IDictionary<string, string> prod
 
         /// <summary>
         /// Maximum total message size sum allowed on the queue. This property has higher priority than <see cref="QueueBufferingMaxMessages" />
-        /// default: 1048576
+        /// default: 103424 (101MB)
         /// </summary>
         public int? QueueBufferingMaxKbytes { get; set; }
 
@@ -114,6 +114,11 @@ internal ProducerConfig ToProducerConfig()
                 producerProperties["log_level"] = "0";
             }
 
+            if (!producerProperties.ContainsKey("queue.buffering.max.kbytes") && !QueueBufferingMaxKbytes.HasValue)
+            {
+                QueueBufferingMaxKbytes = 103424;
+            }
+
             var config = new ProducerConfig(producerProperties);
             config.BootstrapServers ??= BrokerList;
             config.QueueBufferingMaxKbytes ??= QueueBufferingMaxKbytes;