Add properties for producer configuration to Output Binding Attribute definition

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

@@ -49,7 +49,7 @@ public KafkaAttribute()
         /// </summary>
         public Type KeyType { get; set; }
 
-        Type valueType;
+        private Type valueType;
 
         /// <summary>
         /// Gets or sets the Avro data type
@@ -75,13 +75,44 @@ public Type ValueType
         /// </summary>
         public string AvroSchema { get; set; }
 
-        bool IsValidValueType(Type value)
+        private bool IsValidValueType(Type value)
         {
             return
                 typeof(ISpecificRecord).IsAssignableFrom(value) ||
                 typeof(Google.Protobuf.IMessage).IsAssignableFrom(value) ||
                 value == typeof(byte[]) ||
                 value == typeof(string);
         }
+
+        /// <summary>
+        /// Gets or sets the Maximum transmit message size. Default: 1MB
+        /// </summary>
+        public int? MaxMessageBytes { get; set; }
+
+        /// <summary>
+        /// Maximum number of messages batched in one MessageSet. default: 10000
+        /// </summary>
+        public int? BatchSize { get; set; }
+
+        /// <summary>
+        /// When set to `true`, the producer will ensure that messages are successfully produced exactly once and in the original produce order. default: false
+        /// </summary>
+        public bool? EnableIdempotence { get; set; }
+
+        /// <summary>
+        /// Local message timeout. This value is only enforced locally and limits the time a produced message waits for successful delivery. A time of 0 is infinite. This is the maximum time used to deliver a message (including retries). Delivery error occurs when either the retry count or the message timeout are exceeded. default: 300000
+        /// </summary>
+        public int? MessageTimeoutMs { get; set; }
+
+        /// <summary>
+        /// The ack timeout of the producer request in milliseconds. default: 5000
+        /// </summary>
+        public int? RequestTimeoutMs { get; set; }
+
+        /// <summary>
+        /// How many times to retry sending a failing Message. **Note:** default: 2 
+        /// </summary>
+        /// <remarks>Retrying may cause reordering unless <c>EnableIdempotence</c> is set to <c>true</c>.</remarks>
+        public int? MaxRetries { get; set; }
     }
 }

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

@@ -3,7 +3,6 @@
 
 using System;
 using System.Threading;
-using System.Threading.Tasks;
 using Confluent.Kafka;
 using Confluent.SchemaRegistry.Serdes;
 using Microsoft.Extensions.Logging;
@@ -140,7 +139,7 @@ private void DeliveryHandler(DeliveryReport<TKey, TValue> deliveredItem)
             }
             else
             {
-                this.logger.LogError("Failed to delivery message to {topic} / {partition} / {offset}. Error: {error}", deliveredItem.Topic, (int)deliveredItem.Partition, (long)deliveredItem.Offset, deliveredItem.Error.ToString());
+                logger.LogError("Failed to delivery message to {topic} / {partition} / {offset}. Reason: {reason}. Full Error: {error}", deliveredItem.Topic, (int)deliveredItem.Partition, (long)deliveredItem.Offset, deliveredItem.Error.Reason, deliveredItem.Error.ToString());
             }
         }
     }

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

@@ -22,7 +22,7 @@ public class KafkaProducerProvider : IKafkaProducerProvider
         private readonly IConfiguration config;
         private readonly INameResolver nameResolver;
         private readonly ILoggerProvider loggerProvider;
-        ConcurrentDictionary<string, IKafkaProducer> producers = new ConcurrentDictionary<string, IKafkaProducer>();
+        private ConcurrentDictionary<string, IKafkaProducer> producers = new ConcurrentDictionary<string, IKafkaProducer>();
 
         public KafkaProducerProvider(IConfiguration config, INameResolver nameResolver, ILoggerProvider loggerProvider)
         {
@@ -33,8 +33,8 @@ public KafkaProducerProvider(IConfiguration config, INameResolver nameResolver,
 
         public IKafkaProducer Get(KafkaAttribute attribute)
         {
-            var resolvedBrokerList = this.nameResolver.ResolveWholeString(attribute.BrokerList);
-            var brokerListFromConfig = this.config.GetConnectionStringOrSetting(resolvedBrokerList);
+            var resolvedBrokerList = nameResolver.ResolveWholeString(attribute.BrokerList);
+            var brokerListFromConfig = config.GetConnectionStringOrSetting(resolvedBrokerList);
             if (!string.IsNullOrEmpty(brokerListFromConfig))
             {
                 resolvedBrokerList = brokerListFromConfig;
@@ -44,7 +44,7 @@ public IKafkaProducer Get(KafkaAttribute attribute)
             var valueTypeName = attribute.ValueType == null ? string.Empty : attribute.ValueType.AssemblyQualifiedName;
             var producerKey = $"{resolvedBrokerList}:keyTypeName:valueTypeName";
 
-            return producers.GetOrAdd(producerKey, (k) => this.Create(attribute, resolvedBrokerList));
+            return producers.GetOrAdd(producerKey, (k) => Create(attribute, resolvedBrokerList));
         }
 
         private IKafkaProducer Create(KafkaAttribute attribute, string brokerList)
@@ -75,17 +75,19 @@ private IKafkaProducer Create(KafkaAttribute attribute, string brokerList)
 
             return (IKafkaProducer)Activator.CreateInstance(
                 typeof(KafkaProducer<,>).MakeGenericType(keyType, valueType),
-                this.GetProducerConfig(brokerList),
+                GetProducerConfig(attribute, brokerList),
                 avroSchema,
-                this.loggerProvider.CreateLogger(LogCategories.CreateTriggerCategory("Kafka")));
+                loggerProvider.CreateLogger(LogCategories.CreateTriggerCategory("Kafka")));
         }
 
-        private ProducerConfig GetProducerConfig(string brokerList)
+        private ProducerConfig GetProducerConfig(KafkaAttribute attribute, string brokerList) => new ProducerConfig
         {
-            return new ProducerConfig()
-            {
-                BootstrapServers = brokerList,
-            };
-        }
+            BootstrapServers = brokerList,
+            BatchNumMessages = attribute.BatchSize,
+            EnableIdempotence = attribute.EnableIdempotence,
+            MessageSendMaxRetries = attribute.MaxRetries,
+            MessageTimeoutMs = attribute.MessageTimeoutMs,
+            RequestTimeoutMs = attribute.RequestTimeoutMs,
+        };
     }
 }