Add EnableDeliveryReports to KafkaAttribute

Author: gliljas

README.md

@@ -322,6 +322,7 @@ The settings exposed here are targeted to more advanced users that want to custo
 |LibkafkaDebug|debug|Both
 |MetadataMaxAgeMs|metadata.max.age.ms|Both
 |SocketKeepaliveEnable|socket.keepalive.enable|Both
+|EnableDeliveryReports|Feature of Confluent.Kafka|Output
 
 **NOTE:** `MetadataMaxAgeMs` default is `180000` `SocketKeepaliveEnable` default is `true` otherwise, the default value is the same as the [Configuration properties](https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md). The reason of the default settings, refer to this [issue](https://github.com/Azure/azure-functions-kafka-extension/issues/187).
 

src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaAttribute.cs

@@ -15,6 +15,9 @@ namespace Microsoft.Azure.WebJobs.Extensions.Kafka
     [Binding]
     public sealed class KafkaAttribute : Attribute
     {
+        private bool? enableDeliveryReports;
+        private int? maxMessageBytes;
+
         /// <summary>
         /// Initialize a new instance of the <see cref="KafkaAttribute"/>
         /// </summary>
@@ -54,7 +57,7 @@ public KafkaAttribute()
         /// <summary>
         /// Gets or sets the Maximum transmit message size. Default: 1MB
         /// </summary>
-        public int? MaxMessageBytes { get; set; }
+        public int MaxMessageBytes { get => maxMessageBytes.GetValueOrDefault(1000000); set => maxMessageBytes = value; }
 
         /// <summary>
         /// Maximum number of messages batched in one MessageSet. default: 10000
@@ -139,5 +142,12 @@ public KafkaAttribute()
         /// ssl.key.password in librdkafka
         /// </summary>
         public string SslKeyPassword { get; set; }
+
+        /// <summary>
+        /// Specifies whether to enable notification of delivery reports. Typically you should
+        /// set this parameter to true. Set it to false for "fire and forget" semantics and
+        /// a small boost in performance. default: true importance: low
+        /// </summary>
+        public bool EnableDeliveryReports { get => enableDeliveryReports.GetValueOrDefault(true); set => enableDeliveryReports = value; }
     }
 }

src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaProducerFactory.cs

@@ -118,7 +118,9 @@ public ProducerConfig GetProducerConfig(KafkaProducerEntity entity)
             {
                 BootstrapServers = this.config.ResolveSecureSetting(nameResolver, entity.Attribute.BrokerList),
                 BatchNumMessages = entity.Attribute.BatchSize,
+                EnableDeliveryReports = entity.Attribute.EnableDeliveryReports,
                 EnableIdempotence = entity.Attribute.EnableIdempotence,
+                MessageMaxBytes = entity.Attribute.MaxMessageBytes,
                 MessageSendMaxRetries = entity.Attribute.MaxRetries,
                 MessageTimeoutMs = entity.Attribute.MessageTimeoutMs,
                 RequestTimeoutMs = entity.Attribute.RequestTimeoutMs,

test/Microsoft.Azure.WebJobs.Extensions.Kafka.UnitTests/KafkaProducerFactoryTest.cs

@@ -304,5 +304,37 @@ public void GetProducerConfig_When_Ssl_Locations_Resolve_InAzure_Should_Contain_
             Assert.Equal(sslCa.FullName, config.SslCaLocation);
             Assert.Equal(sslKeyLocation.FullName, config.SslKeyLocation);
         }
+
+        [Fact]
+        public void GetProducerConfig_Copies_Properties_From_Attribute()
+        {
+            var attribute = new KafkaAttribute("brokers:9092", "myTopic")
+            {
+                EnableDeliveryReports = false,
+                BatchSize = 123,
+                EnableIdempotence = true,
+                MaxMessageBytes = 234,
+                MaxRetries = 345,
+                MessageTimeoutMs = 456,
+                RequestTimeoutMs = 567
+            };
+
+            var entity = new KafkaProducerEntity
+            {
+                Attribute = attribute
+            };
+
+            var factory = new KafkaProducerFactory(emptyConfiguration, new DefaultNameResolver(emptyConfiguration), NullLoggerProvider.Instance);
+            var config = factory.GetProducerConfig(entity);
+
+            Assert.Equal(attribute.EnableDeliveryReports, config.EnableDeliveryReports);
+            Assert.Equal(attribute.BatchSize, config.BatchNumMessages);
+            Assert.Equal(attribute.EnableIdempotence, config.EnableIdempotence);
+            Assert.Equal(attribute.MaxMessageBytes, config.MessageMaxBytes);
+            Assert.Equal(attribute.MaxRetries, config.MessageSendMaxRetries);
+            Assert.Equal(attribute.MessageTimeoutMs, config.MessageTimeoutMs);
+            Assert.Equal(attribute.RequestTimeoutMs, config.RequestTimeoutMs);
+            Assert.Equal(attribute.EnableDeliveryReports, config.EnableDeliveryReports);
+        }
     }
 }