Repository: Azure/azure-functions-kafka-extension (language: python)

Issue:
Should support external Avro Schema Registry
Should support external Avro schema registry. Currently only fix Avro schemas are supported.


Discussion / hints:
Hi @fbeltrao 
Could you explain more detail of this issue? 
This feature would allow the extension to validate messages against an external schema registry. This is something the Confluent offers to customers to ensure message format consistency.
Instead of having a "hard-coded" schema inside the application, the idea is to use a remote registry.
My company uses this approach so we have a remote repository with all contracts exchange between our systems.
It would be very helpful to see this feature available on the extension.

https://docs.confluent.io/platform/current/schema-registry/schema_registry_onprem_tutorial.html#schema-registry-onprem-tutorial

Fix patch:
diff --git a/samples/dotnet/KafkaFunctionSample/AvroGenericTriggersWithSchemaRegistry.cs b/samples/dotnet/KafkaFunctionSample/AvroGenericTriggersWithSchemaRegistry.cs
new file mode 100644
index 00000000..899851e0
--- /dev/null
+++ b/samples/dotnet/KafkaFunctionSample/AvroGenericTriggersWithSchemaRegistry.cs
@@ -0,0 +1,57 @@
+using System;
+using Microsoft.Extensions.Logging;
+using Newtonsoft.Json;
+using Microsoft.Azure.WebJobs.Extensions.Kafka;
+using Avro.Generic;
+using System.Collections.Generic;
+
+
+namespace KafkaFunctionSample
+{
+    /// <summary>
+    /// Demonstrate using generic avro support and integrating a Confluent schema registry
+    /// In this scenario the schema is retrieved from a registry.
+    /// An instance of GenericRecord will be available in the KafkaEventData.Value property.
+    ///
+    /// This function is disabled by default so it does not interfere with the execution when
+    /// no schema registry is present. To test this, install a schema registry locally
+    /// with <see href="https://docs.confluent.io/platform/current/platform-quickstart.html#cp-quick-start-docker">the tutorial from Confluent</see>
+    /// and uncomment the function name attribute in this file.
+    /// </summary>
+    /// 
+    public class AvroGenericTriggersWithSchemaRegistry
+    {
+        // [FunctionName(nameof(PageViewsSchemaRegistry))]
+        public static void PageViewsSchemaRegistry(
+            [KafkaTrigger("LocalBroker", "pageviews", ConsumerGroup = "azfunc", SchemaRegistryUrl = "localhost:8081")]
+            KafkaEventData<string, GenericRecord>[] kafkaEvents,
+            long[] offsetArray,
+            int[] partitionArray,
+            string[] topicArray,
+            DateTime[] timestampArray,
+            ILogger logger)
+        {
+            for (int i = 0; i < kafkaEvents.Length; i++)
+            {
+                var kafkaEvent = kafkaEvents[i];
+                if (kafkaEvent.Value is GenericRecord genericRecord)
+                {
+                    logger.LogInformation(
+                        $"[{timestampArray[i]}] {topicArray[i]} / {partitionArray[i]} / {offsetArray[i]}: {GenericToJson(genericRecord)}");
+                }
+            }
+        }
+
+        public static string GenericToJson(GenericRecord record)
+        {
+            var props = new Dictionary<string, object>();
+            foreach (var field in record.Schema.Fields)
+            {
+                if (record.TryGetValue(field.Name, out var value))
+                    props[field.Name] = value;
+            }
+
+            return JsonConvert.SerializeObject(props);
+        }
+    }
+}
\ No newline at end of file
diff --git a/samples/dotnet/KafkaFunctionSample/AvroProduceStringTopicFunction.cs b/samples/dotnet/KafkaFunctionSample/AvroProduceStringTopicFunction.cs
new file mode 100644
index 00000000..7c9d3c1d
--- /dev/null
+++ b/samples/dotnet/KafkaFunctionSample/AvroProduceStringTopicFunction.cs
@@ -0,0 +1,75 @@
+using System;
+using System.Threading.Tasks;
+using Avro;
+using Microsoft.AspNetCore.Mvc;
+using Microsoft.Azure.WebJobs;
+using Microsoft.Azure.WebJobs.Extensions.Http;
+using Microsoft.AspNetCore.Http;
+using Microsoft.Extensions.Logging;
+using Microsoft.Azure.WebJobs.Extensions.Kafka;
+using Avro.Generic;
+
+namespace KafkaFunctionSample
+{
+    public static class AvroProduceStringTopicFunction
+    {
+        const string PageViewsSchema = @"{
+  ""type"": ""record"",
+  ""name"": ""pageviews"",
+  ""namespace"": ""ksql"",
+  ""fields"": [
+    {
+      ""name"": ""viewtime"",
+      ""type"": ""long""
+    },
+    {
+      ""name"": ""userid"",
+      ""type"": ""string""
+    },
+    {
+      ""name"": ""pageid"",
+      ""type"": ""string""
+    }
+  ]
+}";
+
+        /// <summary>
+        /// To execute posting an object, execute:
+        /// curl -X POST http://localhost:7071/api/AvroProduceStringTopic
+        /// </summary>
+        /// <param name="req"></param>
+        /// <param name="events"></param>
+        /// <param name="log"></param>
+        /// <returns></returns>
+        [FunctionName(nameof(AvroProduceStringTopic))]
+        public static async Task<IActionResult> AvroProduceStringTopic(
+            [HttpTrigger(AuthorizationLevel.Function, "get", "post", Route = null)]
+            HttpRequest req,
+            [Kafka("LocalBroker", "avroTopic", AvroSchema = PageViewsSchema)]
+            IAsyncCollector<KafkaEventData<GenericRecord>> events,
+            ILogger log)
+        {
+            try
+            {
+                var genericRecord = new GenericRecord((RecordSchema)Schema.Parse(PageViewsSchema));
+                genericRecord.Add("viewtime", 4711L);
+                genericRecord.Add("userid", "User4711");
+                genericRecord.Add("pageid", "Page4711");
+                var kafkaEvent = new KafkaEventData<GenericRecord>()
+                {
+                    Value = genericRecord,
+                };
+
+                await events.AddAsync(kafkaEvent);
+            }
+            catch (Exception ex)
+            {
+                throw new Exception(
+                    "Are you sure the topic 'stringTopic' exists? To create using Confluent Docker quickstart run this command: 'docker-compose exec broker kafka-topics --create --zookeeper zookeeper:2181 --replication-factor 1 --partitions 10 --topic stringTopicTenPartitions'",
+                    ex);
+            }
+
+            return new OkResult();
+        }
+    }
+}
\ No newline at end of file
diff --git a/samples/dotnet/KafkaFunctionSample/AvroProduceStringTopicFunctionWithSchemaRegistry.cs b/samples/dotnet/KafkaFunctionSample/AvroProduceStringTopicFunctionWithSchemaRegistry.cs
new file mode 100644
index 00000000..c971cde3
--- /dev/null
+++ b/samples/dotnet/KafkaFunctionSample/AvroProduceStringTopicFunctionWithSchemaRegistry.cs
@@ -0,0 +1,80 @@
+using System;
+using System.Threading.Tasks;
+using Avro;
+using Microsoft.AspNetCore.Mvc;
+using Microsoft.Azure.WebJobs;
+using Microsoft.Azure.WebJobs.Extensions.Http;
+using Microsoft.AspNetCore.Http;
+using Microsoft.Extensions.Logging;
+using Microsoft.Azure.WebJobs.Extensions.Kafka;
+using Avro.Generic;
+
+namespace KafkaFunctionSample
+{
+    public static class AvroProduceStringTopicFunctionWithSchemaRegistry
+    {
+        const string PageViewsSchema = @"{
+  ""type"": ""record"",
+  ""name"": ""pageviews"",
+  ""namespace"": ""ksql"",
+  ""fields"": [
+    {
+      ""name"": ""viewtime"",
+      ""type"": ""long""
+    },
+    {
+      ""name"": ""userid"",
+      ""type"": ""string""
+    },
+    {
+      ""name"": ""pageid"",
+      ""type"": ""string""
+    }
+  ]
+}";
+
+        /// <summary>
+        /// To execute posting an object, execute:
+        /// curl http://localhost:7071/api/AvroProduceStringTopicSchemaRegistry
+        ///
+        /// To test this, install a schema registry locally with
+        /// <see href="https://docs.confluent.io/platform/current/platform-quickstart.html#cp-quick-start-docker">the tutorial from Confluent</see>.
+        /// You also have to create a topic called "avroTopic" and add a schema via the Confluent web UI.
+        /// The schema that you have to add can be copied from <see cref="PageViewsSchema"/>.
+        /// </summary>
+        /// <param name="req"></param>
+        /// <param name="events"></param>
+        /// <param name="log"></param>
+        /// <returns></returns>
+        [FunctionName(nameof(AvroProduceStringTopicSchemaRegistry))]
+        public static async Task<IActionResult> AvroProduceStringTopicSchemaRegistry(
+            [HttpTrigger(AuthorizationLevel.Function, "get", "post", Route = null)]
+            HttpRequest req,
+            [Kafka("LocalBroker", "avroTopic", SchemaRegistryUrl = "localhost:8081")]
+            IAsyncCollector<KafkaEventData<GenericRecord>> events,
+            ILogger log)
+        {
+            try
+            {
+                var genericRecord = new GenericRecord((RecordSchema)Schema.Parse(PageViewsSchema));
+                genericRecord.Add("viewtime", 4711L);
+                genericRecord.Add("userid", "User4711");
+                genericRecord.Add("pageid", "Page4711");
+                var kafkaEvent = new KafkaEventData<GenericRecord>()
+                {
+                    Value = genericRecord,
+                };
+
+                await events.AddAsync(kafkaEvent);
+            }
+            catch (Exception ex)
+            {
+                throw new Exception(
+                    "Are you sure the topic 'stringTopic' exists? To create using Confluent Docker quickstart run this command: 'docker-compose exec broker kafka-topics --create --zookeeper zookeeper:2181 --replication-factor 1 --partitions 10 --topic stringTopicTenPartitions'",
+                    ex);
+            }
+
+            return new OkResult();
+        }
+    }
+}
\ No newline at end of file
diff --git a/samples/dotnet/KafkaFunctionSample/PageViews.cs b/samples/dotnet/KafkaFunctionSample/PageViews.cs
deleted file mode 100644
index b99dccd0..00000000
--- a/samples/dotnet/KafkaFunctionSample/PageViews.cs
+++ /dev/null
@@ -1,60 +0,0 @@
-using Avro;
-using Avro.Specific;
-using Newtonsoft.Json;
-
-namespace KafkaFunctionSample
-{
-    public class PageViews : ISpecificRecord
-    {
-       public const string SchemaText = @"
-       {
-  ""type"": ""record"",
-  ""name"": ""PageViews"",
-  ""namespace"": ""KafkaFunctionSample"",
-  ""fields"": [
-    {
-      ""name"": ""viewtime"",
-      ""type"": ""long""
-    },
-    {
-      ""name"": ""userid"",
-      ""type"": ""string""
-    },
-    {
-      ""name"": ""pageid"",
-      ""type"": ""string""
-    }
-  ]
-}";
-        public static Schema _SCHEMA = Schema.Parse(SchemaText);
-
-        [JsonIgnore]
-        public virtual Schema Schema => _SCHEMA;
-
-        public long ViewTime { get; set; }
-        public string UserID { get; set; }
-        public string PageID { get; set; }
-
-        public virtual object Get(int fieldPos)
-        {
-            switch (fieldPos)
-            {
-                case 0: return this.ViewTime;
-                case 1: return this.UserID;
-                case 2: return this.PageID;
-                default: throw new AvroRuntimeException("Bad index " + fieldPos + " in Get()");
-            };
-        }
-        public virtual void Put(int fieldPos, object fieldValue)
-        {
-            switch (fieldPos)
-            {
-                case 0: this.ViewTime = (long)fieldValue; break;
-                case 1: this.UserID = (string)fieldValue; break;
-                case 2: this.PageID = (string)fieldValue; break;
-                default: throw new AvroRuntimeException("Bad index " + fieldPos + " in Put()");
-            };
-        }
-
-    }
-}
\ No newline at end of file
diff --git a/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaAttribute.cs b/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaAttribute.cs
index dea4d300..055d20d6 100644
--- a/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaAttribute.cs
+++ b/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaAttribute.cs
@@ -2,8 +2,6 @@
 // Licensed under the MIT License. See License.txt in the project root for license information.
 
 using System;
-using Avro.Specific;
-using Confluent.Kafka;
 using Microsoft.Azure.WebJobs.Description;
 
 namespace Microsoft.Azure.WebJobs.Extensions.Kafka
@@ -145,5 +143,20 @@ public KafkaAttribute()
         /// being sent to cluster. Larger value allows more batching results in high throughput.
         /// </summary>
         public int LingerMs { get; set; } = 5;
+
+        /// <summary>
+        /// URL for the Avro Schema Registry
+        /// </summary>
+        public string SchemaRegistryUrl { get; set; }
+
+        /// <summary>
+        /// Username for the Avro Schema Registry
+        /// </summary>
+        public string SchemaRegistryUsername { get; set; }
+
+        /// <summary>
+        /// Password for the Avro Schema Registry
+        /// </summary>
+        public string SchemaRegistryPassword { get; set; }
     }
 }
\ No newline at end of file
diff --git a/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaAttributeBinding.cs b/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaAttributeBinding.cs
index dd729bda..4e4299d9 100644
--- a/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaAttributeBinding.cs
+++ b/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaAttributeBinding.cs
@@ -24,7 +24,7 @@ internal class KafkaAttributeBinding : IBinding
         private readonly INameResolver nameResolver;
 
         public KafkaAttributeBinding(
-            string parameterName, 
+            string parameterName,
             KafkaAttribute attribute,
             IKafkaProducerFactory kafkaProducerFactory,
             IArgumentBinding<KafkaProducerEntity> argumentBinding,
diff --git a/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaAttributeBindingProvider.cs b/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaAttributeBindingProvider.cs
index 044a1f7b..f1d00716 100644
--- a/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaAttributeBindingProvider.cs
+++ b/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaAttributeBindingProvider.cs
@@ -1,7 +1,6 @@
 ﻿// Copyright (c) .NET Foundation. All rights reserved.
 // Licensed under the MIT License. See License.txt in the project root for license information.
 
-using Confluent.Kafka;
 using Microsoft.Azure.WebJobs.Host.Bindings;
 using Microsoft.Extensions.Configuration;
 using System;
@@ -44,7 +43,6 @@ public Task<IBinding> TryCreateAsync(BindingProviderContext context)
             {
                 return Task.FromResult<IBinding>(null);
             }
-            
 
             var argumentBinding = InnerProvider.TryCreate(parameter);
             var keyAndValueTypes = SerializationHelper.GetKeyAndValueTypes(attribute.AvroSchema, parameter.ParameterType, typeof(string));
diff --git a/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaProducerEntity.cs b/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaProducerEntity.cs
index dfc028dc..e161ddd3 100644
--- a/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaProducerEntity.cs
+++ b/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaProducerEntity.cs
@@ -1,14 +1,10 @@
 ﻿// Copyright (c) .NET Foundation. All rights reserved.
 // Licensed under the MIT License. See License.txt in the project root for license information.
 
-using Confluent.Kafka;
 using System;
 using System.Collections;
-using System.Collections.Generic;
-using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
-using System.Xml;
 
 namespace Microsoft.Azure.WebJobs.Extensions.Kafka
 {
diff --git a/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaProducerFactory.cs b/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaProducerFactory.cs
index 0f4ef14b..e93a7b4f 100644
--- a/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaProducerFactory.cs
+++ b/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaProducerFactory.cs
@@ -3,16 +3,10 @@
 
 using System;
 using System.Collections.Concurrent;
-using System.Reflection;
 using System.Text;
-using Avro.Generic;
-using Avro.Specific;
 using Confluent.Kafka;
-using Microsoft.Azure.WebJobs.Host;
-using Microsoft.Azure.WebJobs.Logging;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.Logging;
-using Microsoft.Extensions.Options;
 
 namespace Microsoft.Azure.WebJobs.Extensions.Kafka
 {
@@ -88,7 +82,12 @@ private IKafkaProducer Create(Handle producerBaseHandle, KafkaProducerEntity ent
             var valueType = entity.ValueType ?? typeof(byte[]);
             var keyType = entity.KeyType ?? typeof(Null);
 
-            var valueSerializer = SerializationHelper.ResolveValueSerializer(valueType, entity.AvroSchema);
+            var avroSchema = this.config.ResolveSecureSetting(nameResolver, entity.AvroSchema);
+            var schemaRegistryUrl = this.config.ResolveSecureSetting(nameResolver, entity.Attribute.SchemaRegistryUrl);
+            var schemaRegistryUsername = this.config.ResolveSecureSetting(nameResolver, entity.Attribute.SchemaRegistryUsername);
+            var schemaRegistryPassword = this.config.ResolveSecureSetting(nameResolver, entity.Attribute.SchemaRegistryPassword);
+
+            var valueSerializer = SerializationHelper.ResolveValueSerializer(valueType, avroSchema, schemaRegistryUrl, schemaRegistryUsername, schemaRegistryPassword);
 
             return (IKafkaProducer)Activator.CreateInstance(
                 typeof(KafkaProducer<,>).MakeGenericType(keyType, valueType),
diff --git a/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Serialization/SerializationHelper.cs b/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Serialization/SerializationHelper.cs
index 089b6b0f..ed2dcc97 100644
--- a/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Serialization/SerializationHelper.cs
+++ b/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Serialization/SerializationHelper.cs
@@ -2,6 +2,7 @@
 // Licensed under the MIT License. See License.txt in the project root for license information.
 
 using System;
+using System.Collections.Generic;
 using System.Linq;
 using System.Reflection;
 using Avro.Generic;
@@ -16,34 +17,25 @@ namespace Microsoft.Azure.WebJobs.Extensions.Kafka
 {
     internal static class SerializationHelper
     {
-        internal static object ResolveValueDeserializer(Type valueType, string specifiedAvroSchema)
+        internal static object ResolveValueDeserializer(Type valueType, string specifiedAvroSchema, string schemaRegistryUrl, string schemaRegistryUsername, string schemaRegistryPassword)
         {
-            if (typeof(Google.Protobuf.IMessage).IsAssignableFrom(valueType))
+            if (typeof(IMessage).IsAssignableFrom(valueType))
             {
                 return Activator.CreateInstance(typeof(ProtobufDeserializer<>).MakeGenericType(valueType));
             }
 
             var isSpecificRecord = typeof(ISpecificRecord).IsAssignableFrom(valueType);
-            if (isSpecificRecord || typeof(GenericRecord).IsAssignableFrom(valueType))
+            if (!isSpecificRecord && !typeof(GenericRecord).IsAssignableFrom(valueType))
             {
-                if (string.IsNullOrWhiteSpace(specifiedAvroSchema))
-                {
-                    if (!isSpecificRecord)
-                    {
-                        throw new ArgumentNullException(nameof(specifiedAvroSchema), $@"parameter is required when creating an generic avro deserializer");
-                    }
+                return null;
+            }
 
-                    specifiedAvroSchema = ((ISpecificRecord)Activator.CreateInstance(valueType)).Schema.ToString();
-                }
+            var schemaRegistry = CreateSchemaRegistry(valueType, specifiedAvroSchema, schemaRegistryUrl, schemaRegistryUsername, schemaRegistryPassword, isSpecificRecord);
 
-                var methodInfo = typeof(SerializationHelper).GetMethod(nameof(CreateAvroDeserializer), BindingFlags.Static | BindingFlags.NonPublic);
-                var genericMethod = methodInfo.MakeGenericMethod(valueType);
+            var methodInfo = typeof(SerializationHelper).GetMethod(nameof(CreateAvroDeserializer), BindingFlags.Static | BindingFlags.NonPublic);
+            var genericMethod = methodInfo.MakeGenericMethod(valueType);
 
-                var schemaRegistry = new LocalSchemaRegistry(specifiedAvroSchema);
-                return genericMethod.Invoke(null, new object[] { schemaRegistry });
-            }
-                       
-            return null;
+            return genericMethod.Invoke(null, new object[] { schemaRegistry });
         }
 
         private static IDeserializer<TValue> CreateAvroDeserializer<TValue>(ISchemaRegistryClient schemaRegistry)
@@ -51,32 +43,47 @@ private static IDeserializer<TValue> CreateAvroDeserializer<TValue>(ISchemaRegis
             return new AvroDeserializer<TValue>(schemaRegistry).AsSyncOverAsync();
         }
 
-        internal static object ResolveValueSerializer(Type valueType, string specifiedAvroSchema)
+        internal static object ResolveValueSerializer(Type valueType, string specifiedAvroSchema, string schemaRegistryUrl, string schemaRegistryUsername, string schemaRegistryPassword)
         {
-            if (typeof(Google.Protobuf.IMessage).IsAssignableFrom(valueType))
+            if (typeof(IMessage).IsAssignableFrom(valueType))
             {
                 return Activator.CreateInstance(typeof(ProtobufSerializer<>).MakeGenericType(valueType));
             }
 
             var isSpecificRecord = typeof(ISpecificRecord).IsAssignableFrom(valueType);
-            if (isSpecificRecord || typeof(GenericRecord).IsAssignableFrom(valueType))
+            if (!isSpecificRecord && !typeof(GenericRecord).IsAssignableFrom(valueType))
             {
-                if (string.IsNullOrWhiteSpace(specifiedAvroSchema))
-                {
-                    if (!isSpecificRecord)
-                    {
-                        throw new ArgumentNullException(nameof(specifiedAvroSchema), $@"parameter is required when creating an generic avro serializer");
-                    }
+                return null;
+            }
 
-                    specifiedAvroSchema = ((ISpecificRecord)Activator.CreateInstance(valueType)).Schema.ToString();
-                }
+            var schemaRegistry = CreateSchemaRegistry(valueType, specifiedAvroSchema, schemaRegistryUrl, schemaRegistryUsername, schemaRegistryPassword, isSpecificRecord);
 
-                var schemaRegistry = new LocalSchemaRegistry(specifiedAvroSchema);
-                var serializer = Activator.CreateInstance(typeof(AvroSerializer<>).MakeGenericType(valueType), schemaRegistry, null /* config */);
-                return typeof(SyncOverAsyncSerializerExtensionMethods).GetMethod("AsSyncOverAsync").MakeGenericMethod(valueType).Invoke(null, new object[] { serializer });
+            var serializer = Activator.CreateInstance(typeof(AvroSerializer<>).MakeGenericType(valueType), schemaRegistry, null /* config */);
+            return typeof(SyncOverAsyncSerializerExtensionMethods).GetMethod("AsSyncOverAsync").MakeGenericMethod(valueType).Invoke(null, new object[] { serializer });
+        }
+
+        private static ISchemaRegistryClient CreateSchemaRegistry(Type valueType, string specifiedAvroSchema, string schemaRegistryUrl, string schemaRegistryUsername, string schemaRegistryPassword, bool isSpecificRecord)
+        {
+            if (string.IsNullOrWhiteSpace(specifiedAvroSchema) && isSpecificRecord)
+            {
+                specifiedAvroSchema = ((ISpecificRecord)Activator.CreateInstance(valueType)).Schema.ToString();
             }
 
-            return null;
+            if (!string.IsNullOrWhiteSpace(specifiedAvroSchema))
+            {
+                return new LocalSchemaRegistry(specifiedAvroSchema);
+            }
+            if (schemaRegistryUrl != null)
+            {
+                var schemaRegistryConfig = new List<KeyValuePair<string, string>>();
+                schemaRegistryConfig.Add(new KeyValuePair<string, string>("schema.registry.url", schemaRegistryUrl));
+                if (schemaRegistryUsername != null && schemaRegistryPassword != null) {
+                    var authString = schemaRegistryUsername + ":" + schemaRegistryPassword;
+                    schemaRegistryConfig.Add(new KeyValuePair<string, string>("schema.registry.basic.auth.user.info", authString));
+                }
+                return new CachedSchemaRegistryClient(schemaRegistryConfig.ToArray());
+            }
+            throw new ArgumentNullException(nameof(specifiedAvroSchema), $@"parameter is required when creating an generic avro serializer");
         }
 
         internal class GetKeyAndValueTypesResult
@@ -102,7 +109,7 @@ internal static GetKeyAndValueTypesResult GetKeyAndValueTypes(string avroSchemaF
             while (valueType.HasElementType && valueType.GetElementType() != typeof(byte))
             {
                 valueType = valueType.GetElementType();
-            }            
+            }
 
             if (!valueType.IsPrimitive)
             {
diff --git a/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Trigger/KafkaTriggerAttribute.cs b/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Trigger/KafkaTriggerAttribute.cs
index 1837bf01..9981767b 100644
--- a/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Trigger/KafkaTriggerAttribute.cs
+++ b/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Trigger/KafkaTriggerAttribute.cs
@@ -2,7 +2,6 @@
 // Licensed under the MIT License. See License.txt in the project root for license information.
 
 using Avro.Specific;
-using Confluent.Kafka;
 using Microsoft.Azure.WebJobs.Description;
 using System;
 
@@ -42,7 +41,7 @@ public KafkaTriggerAttribute(string brokerList, string topic)
         /// Gets or sets the consumer group
         /// </summary>
         public string ConsumerGroup { get; set; }
-        
+
 
         /// <summary>
         /// Gets or sets the Avro schema.
@@ -114,6 +113,21 @@ public KafkaTriggerAttribute(string brokerList, string topic)
         /// </summary>
         public long LagThreshold { get => lagThreshold.GetValueOrDefault(1000L); set => lagThreshold = value; }
 
+        /// <summary>
+        /// URL for the Avro Schema Registry
+        /// </summary>
+        public string SchemaRegistryUrl { get; set; }
+
+        /// <summary>
+        /// Username for the Avro Schema Registry
+        /// </summary>
+        public string SchemaRegistryUsername { get; set; }
+
+        /// <summary>
+        /// Password for the Avro Schema Registry
+        /// </summary>
+        public string SchemaRegistryPassword { get; set; }
+
         bool IsValidValueType(Type value)
         {
             return
diff --git a/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Trigger/KafkaTriggerAttributeBindingProvider.cs b/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Trigger/KafkaTriggerAttributeBindingProvider.cs
index e3f8e510..dbd23244 100644
--- a/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Trigger/KafkaTriggerAttributeBindingProvider.cs
+++ b/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Trigger/KafkaTriggerAttributeBindingProvider.cs
@@ -6,7 +6,6 @@
 using System.Threading.Tasks;
 using Confluent.Kafka;
 using Microsoft.Azure.WebJobs.Extensions.Kafka.Trigger;
-using Microsoft.Azure.WebJobs.Host;
 using Microsoft.Azure.WebJobs.Host.Bindings;
 using Microsoft.Azure.WebJobs.Host.Listeners;
 using Microsoft.Azure.WebJobs.Host.Triggers;
@@ -37,7 +36,7 @@ public KafkaTriggerAttributeBindingProvider(
             this.nameResolver = nameResolver;
             this.options = options;
             this.logger = loggerFactory.CreateLogger(LogCategories.CreateTriggerCategory("Kafka"));
-        }   
+        }
 
         public Task<ITriggerBinding> TryCreateAsync(TriggerBindingProviderContext context)
         {
@@ -48,11 +47,13 @@ public Task<ITriggerBinding> TryCreateAsync(TriggerBindingProviderContext contex
                 return Task.FromResult<ITriggerBinding>(null);
             }
 
-            var consumerConfig = CreateConsumerConfiguration(attribute);
-
             var keyAndValueTypes = SerializationHelper.GetKeyAndValueTypes(attribute.AvroSchema, parameter.ParameterType, typeof(string));
-            var valueDeserializer = SerializationHelper.ResolveValueDeserializer(keyAndValueTypes.ValueType, keyAndValueTypes.AvroSchema);            
+            var schemaRegistryUrl = this.config.ResolveSecureSetting(nameResolver, attribute.SchemaRegistryUrl);
+            var schemaRegistryUsername = this.config.ResolveSecureSetting(nameResolver, attribute.SchemaRegistryUsername);
+            var schemaRegistryPassword = this.config.ResolveSecureSetting(nameResolver, attribute.SchemaRegistryPassword);
+            var valueDeserializer = SerializationHelper.ResolveValueDeserializer(keyAndValueTypes.ValueType, keyAndValueTypes.AvroSchema, schemaRegistryUrl, schemaRegistryUsername, schemaRegistryPassword);
 
+            var consumerConfig = CreateConsumerConfiguration(attribute);
             var binding = CreateBindingStrategyFor(keyAndValueTypes.KeyType ?? typeof(Ignore), keyAndValueTypes.ValueType, keyAndValueTypes.RequiresKey, valueDeserializer, parameter, consumerConfig);
             return Task.FromResult<ITriggerBinding>(new KafkaTriggerBindingWrapper(binding));
         }
@@ -77,7 +78,7 @@ Task<IListener> listenerCreator(ListenerFactoryContext factoryContext, bool sing
                     valueDeserializer,
                     this.logger,
                     factoryContext.Descriptor.Id);
-                
+
                 return Task.FromResult<IListener>(listener);
             }
 
@@ -95,7 +96,7 @@ private KafkaListenerConfiguration CreateConsumerConfiguration(KafkaTriggerAttri
                 LagThreshold = attribute.LagThreshold
             };
 
-            if (attribute.AuthenticationMode != BrokerAuthenticationMode.NotSet || 
+            if (attribute.AuthenticationMode != BrokerAuthenticationMode.NotSet ||
                 attribute.Protocol != BrokerProtocol.NotSet)
             {
                 consumerConfig.SaslPassword = this.config.ResolveSecureSetting(nameResolver, attribute.Password);


Test patch:
diff --git a/test/Microsoft.Azure.WebJobs.Extensions.Kafka.EndToEndTests/Constants.cs b/test/Microsoft.Azure.WebJobs.Extensions.Kafka.EndToEndTests/Constants.cs
index e2e16a9c..bc25df2b 100644
--- a/test/Microsoft.Azure.WebJobs.Extensions.Kafka.EndToEndTests/Constants.cs
+++ b/test/Microsoft.Azure.WebJobs.Extensions.Kafka.EndToEndTests/Constants.cs
@@ -9,6 +9,8 @@ internal static class Constants
         internal const string StringTopicWithLongKeyAndTenPartitionsName = "stringTopicWithLongKeyTenPartitions";
         internal const string MyAvroRecordTopicName = "myAvroRecordTopic";
         internal const string MyProtobufTopicName = "myProtobufTopic";
+        internal const string SchemaRegistryTopicName = "schemaRegistryTopic";
         internal const string ConsumerGroupID = "e2e_tests";
+        internal const string SchemaRegistryUrl = "localhost:8081";
     }
 }
diff --git a/test/Microsoft.Azure.WebJobs.Extensions.Kafka.EndToEndTests/KafkaEndToEndTests.cs b/test/Microsoft.Azure.WebJobs.Extensions.Kafka.EndToEndTests/KafkaEndToEndTests.cs
index b1f5eb6a..299f4902 100644
--- a/test/Microsoft.Azure.WebJobs.Extensions.Kafka.EndToEndTests/KafkaEndToEndTests.cs
+++ b/test/Microsoft.Azure.WebJobs.Extensions.Kafka.EndToEndTests/KafkaEndToEndTests.cs
@@ -1044,5 +1044,41 @@ await TestHelpers.Await(() =>
                 await Task.Delay(1500);
             }
         }
+
+        // <summary>
+        // This test expects a schema registry running at localhost:8081.
+        // <see cref="Constants.SchemaRegistryUrl"/>
+        // </summary>
+        [Fact]
+        public async Task SingleItem_With_Schema_Registry()
+        {
+            const int producedMessagesCount = 1;
+            
+            var loggerProvider1 = CreateTestLoggerProvider();
+
+            using (var host = await StartHostAsync(
+                       new[] { typeof(SingleItem_With_Schema_Registry), typeof(KafkaOutputFunctions) },
+                       loggerProvider1))
+            {
+                var jobHost = host.GetJobHost();
+
+                await jobHost.CallAsync(
+                    GetStaticMethod(typeof(KafkaOutputFunctions),
+                        nameof(KafkaOutputFunctions.Produce_AsyncCollector_PageView_With_String_Key)),
+                    new { topic = Constants.SchemaRegistryTopicName });
+
+                await TestHelpers.Await(() =>
+                {
+                    var foundCount = loggerProvider1.GetAllUserLogMessages().Count(p => p.FormattedMessage != null);
+                    return foundCount == producedMessagesCount;
+                });
+
+                // Give time for the commit to be saved
+                await Task.Delay(1500);
+
+                await host.StopAsync();
+            }
+
+        }
     }
 }
diff --git a/test/Microsoft.Azure.WebJobs.Extensions.Kafka.EndToEndTests/KafkaOutputFunctions.cs b/test/Microsoft.Azure.WebJobs.Extensions.Kafka.EndToEndTests/KafkaOutputFunctions.cs
index f2f0fd17..3a6d52ae 100644
--- a/test/Microsoft.Azure.WebJobs.Extensions.Kafka.EndToEndTests/KafkaOutputFunctions.cs
+++ b/test/Microsoft.Azure.WebJobs.Extensions.Kafka.EndToEndTests/KafkaOutputFunctions.cs
@@ -5,6 +5,8 @@
 using System.Collections.Generic;
 using System.Text;
 using System.Threading.Tasks;
+using Avro;
+using Avro.Generic;
 
 namespace Microsoft.Azure.WebJobs.Extensions.Kafka.EndToEndTests
 {
@@ -82,7 +84,7 @@ public static KafkaEventData<string, string>[] Produce_Return_Parameter_KafkaEve
             return list.ToArray();
         }
 
-        
+
         [return: Kafka("LocalBroker", Constants.StringTopicWithTenPartitionsName)]
         public static string[] Produce_Return_Parameter_Raw_String_Array(
             string topic,
@@ -182,7 +184,7 @@ public static async Task Produce_AsyncCollector_Raw_SpecificAvro(
         {
 
             foreach (var c in content)
-            {                
+            {
                 await output.AddAsync(new MyAvroRecord()
                 {
                     ID = c,
@@ -245,5 +247,46 @@ public static async Task Produce_AsyncCollector_Protobuf_With_String_Key(
                 i++;
             }
         }
+
+        public static async Task Produce_AsyncCollector_PageView_With_String_Key(
+            string topic,
+            [Kafka(BrokerList = "LocalBroker", SchemaRegistryUrl = Constants.SchemaRegistryUrl)]
+            IAsyncCollector<KafkaEventData<string, GenericRecord>> output
+        )
+        {
+            const string pageViewsSchema = @"{
+  ""type"": ""record"",
+  ""name"": ""PageViews"",
+  ""namespace"": ""ksql"",
+  ""fields"": [
+    {
+      ""name"": ""ViewTime"",
+      ""type"": ""long""
+    },
+    {
+      ""name"": ""UserID"",
+      ""type"": ""string""
+    },
+    {
+      ""name"": ""PageID"",
+      ""type"": ""string""
+    }
+  ]
+}";
+
+            var record = new GenericRecord((RecordSchema)Schema.Parse(pageViewsSchema));
+            record.Add("UserID", "4711");
+            record.Add("PageID", "4712");
+            record.Add("ViewTime", 4713L);
+
+            var message = new KafkaEventData<string, GenericRecord>()
+            {
+                Key = "key",
+                Topic = topic,
+                Value = record
+            };
+
+            await output.AddAsync(message);
+        }
     }
 }
\ No newline at end of file
diff --git a/test/Microsoft.Azure.WebJobs.Extensions.Kafka.EndToEndTests/TriggerFunctions.cs b/test/Microsoft.Azure.WebJobs.Extensions.Kafka.EndToEndTests/TriggerFunctions.cs
index 79794ff5..f22487e4 100644
--- a/test/Microsoft.Azure.WebJobs.Extensions.Kafka.EndToEndTests/TriggerFunctions.cs
+++ b/test/Microsoft.Azure.WebJobs.Extensions.Kafka.EndToEndTests/TriggerFunctions.cs
@@ -25,7 +25,8 @@ public static void Trigger(
     }
 
     internal static class MultiItem_KafkaEventData_String_With_Ignore_Key_Trigger
-    {        public static void Trigger(
+    {
+        public static void Trigger(
                [KafkaTrigger("LocalBroker", Constants.StringTopicWithTenPartitionsName, ConsumerGroup = Constants.ConsumerGroupID)] KafkaEventData<Ignore, string>[] kafkaEvents,
                ILogger log)
         {
@@ -70,7 +71,7 @@ public static void Trigger(
             }
         }
     }
-    
+
 
     internal static class MultiItem_String_With_Long_Key_Trigger
     {
@@ -284,8 +285,8 @@ public static void Trigger(
             throw new Exception("unhandled error");
         }
     }
-    
-    internal static class SingleEventTrigger_With_Activity 
+
+    internal static class SingleEventTrigger_With_Activity
     {
         public static void Trigger(
             [KafkaTrigger("LocalBroker", Constants.StringTopicWithOnePartitionName, ConsumerGroup = Constants.ConsumerGroupID)] KafkaEventData<string> kafkaEvent,
@@ -318,4 +319,14 @@ public static void Trigger(
             }
         }
     }
+
+    internal static class SingleItem_With_Schema_Registry
+    {
+        public static void Trigger(
+            [KafkaTrigger("LocalBroker", Constants.SchemaRegistryTopicName, ConsumerGroup = Constants.ConsumerGroupID, SchemaRegistryUrl = Constants.SchemaRegistryUrl)] KafkaEventData<string, GenericRecord> kafkaEvent,
+            ILogger log)
+        {
+            log.LogInformation(kafkaEvent.Value.ToString());
+        }
+    }
 }
diff --git a/test/Microsoft.Azure.WebJobs.Extensions.Kafka.EndToEndTests/kafka-singlenode-compose.yaml b/test/Microsoft.Azure.WebJobs.Extensions.Kafka.EndToEndTests/kafka-singlenode-compose.yaml
index e5df7666..1cad45a6 100644
--- a/test/Microsoft.Azure.WebJobs.Extensions.Kafka.EndToEndTests/kafka-singlenode-compose.yaml
+++ b/test/Microsoft.Azure.WebJobs.Extensions.Kafka.EndToEndTests/kafka-singlenode-compose.yaml
@@ -36,4 +36,16 @@ services:
       KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
       KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
       KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
-      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
\ No newline at end of file
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+  schema-registry:
+    image: confluentinc/cp-schema-registry
+    depends_on:
+      - zookeeper
+      - kafka
+    ports:
+      - 8081:8081
+    environment:
+      SCHEMA_REGISTRY_LISTENERS: "http://0.0.0.0:8081"
+      SCHEMA_REGISTRY_HOST_NAME: "localhost"
+      SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: zookeeper:2181
+      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: 'kafka:29092'
diff --git a/test/Microsoft.Azure.WebJobs.Extensions.Kafka.UnitTests/KafkaProducerFactoryTest.cs b/test/Microsoft.Azure.WebJobs.Extensions.Kafka.UnitTests/KafkaProducerFactoryTest.cs
index 2c299bde..a912d106 100644
--- a/test/Microsoft.Azure.WebJobs.Extensions.Kafka.UnitTests/KafkaProducerFactoryTest.cs
+++ b/test/Microsoft.Azure.WebJobs.Extensions.Kafka.UnitTests/KafkaProducerFactoryTest.cs
@@ -9,7 +9,6 @@
 using Avro.Generic;
 using Confluent.Kafka;
 using Confluent.Kafka.SyncOverAsync;
-using Confluent.SchemaRegistry.Serdes;
 
 using Microsoft.Azure.WebJobs.Extensions.Kafka.UnitTests.Helpers;
 using Microsoft.Extensions.Configuration;
@@ -119,6 +118,28 @@ public void When_Avro_Schema_Is_Provided_Should_Create_GenericRecord_Listener()
             Assert.IsType<SyncOverAsyncSerializer<GenericRecord>>(typedProducer.ValueSerializer);
         }
 
+        [Fact]
+        public void When_Schema_Registry_Is_Provided_Should_Create_GenericRecord_Listener()
+        {
+            var attribute = new KafkaAttribute("brokers:9092", "myTopic");
+
+            attribute.SchemaRegistryUrl = "localhost";
+
+            var entity = new KafkaProducerEntity()
+            {
+                Attribute = attribute,
+                ValueType = typeof(GenericRecord),
+            };
+
+            var factory = new KafkaProducerFactory(emptyConfiguration, new DefaultNameResolver(emptyConfiguration), NullLoggerFactory.Instance);
+            var producer = factory.Create(entity);
+
+            Assert.NotNull(producer);
+            Assert.IsType<KafkaProducer<Null, GenericRecord>>(producer);
+            var typedProducer = (KafkaProducer<Null, GenericRecord>)producer;
+            Assert.NotNull(typedProducer.ValueSerializer);
+            Assert.IsType<SyncOverAsyncSerializer<GenericRecord>>(typedProducer.ValueSerializer);
+        }
 
         [Fact]
         public void When_Value_Type_Is_Specific_Record_Should_Create_SpecificRecord_Listener()
@@ -183,7 +204,7 @@ public void GetProducerConfig_When_No_Auth_Defined_Should_Not_Contain_Auth_Setti
 
             var factory = new KafkaProducerFactory(emptyConfiguration, new DefaultNameResolver(emptyConfiguration), NullLoggerFactory.Instance);
             var config = factory.GetProducerConfig(entity);
-            Assert.Equal(0, config.Count(x=>x.Key.StartsWith("sasl.")));
+            Assert.Equal(0, config.Count(x => x.Key.StartsWith("sasl.")));
             Assert.Null(config.SaslMechanism);
             Assert.Null(config.SaslPassword);
             Assert.Null(config.SaslUsername);
@@ -285,7 +306,7 @@ public void GetProducerConfig_When_Ssl_Locations_Resolve_InAzure_Should_Contain_
             AzureEnvironment.SetEnvironmentVariable(AzureFunctionsFileHelper.AzureHomeEnvVarName, currentFolder);
 
             var sslCertificate = this.CreateFile(Path.Combine(currentFolder, AzureFunctionsFileHelper.AzureDefaultFunctionPathPart1, AzureFunctionsFileHelper.AzureDefaultFunctionPathPart2, "sslCertificate.pfx"));
-            var sslCa = this.CreateFile(Path.Combine(currentFolder, AzureFunctionsFileHelper.AzureDefaultFunctionPathPart1, AzureFunctionsFileHelper.AzureDefaultFunctionPathPart2, "sslCa.pem")); 
+            var sslCa = this.CreateFile(Path.Combine(currentFolder, AzureFunctionsFileHelper.AzureDefaultFunctionPathPart1, AzureFunctionsFileHelper.AzureDefaultFunctionPathPart2, "sslCa.pem"));
             var sslKeyLocation = this.CreateFile(Path.Combine(currentFolder, AzureFunctionsFileHelper.AzureDefaultFunctionPathPart1, AzureFunctionsFileHelper.AzureDefaultFunctionPathPart2, "sslKey.key"));
 
             var attribute = new KafkaAttribute("brokers:9092", "myTopic")
diff --git a/test/Microsoft.Azure.WebJobs.Extensions.Kafka.UnitTests/KafkaTriggerAttributeBindingProviderTest.cs b/test/Microsoft.Azure.WebJobs.Extensions.Kafka.UnitTests/KafkaTriggerAttributeBindingProviderTest.cs
index 15377c4e..1c79edc5 100644
--- a/test/Microsoft.Azure.WebJobs.Extensions.Kafka.UnitTests/KafkaTriggerAttributeBindingProviderTest.cs
+++ b/test/Microsoft.Azure.WebJobs.Extensions.Kafka.UnitTests/KafkaTriggerAttributeBindingProviderTest.cs
@@ -4,7 +4,6 @@
 using Avro.Generic;
 using Confluent.Kafka;
 using Confluent.Kafka.SyncOverAsync;
-using Confluent.SchemaRegistry.Serdes;
 using Microsoft.Azure.WebJobs.Host.Executors;
 using Microsoft.Azure.WebJobs.Host.Listeners;
 using Microsoft.Azure.WebJobs.Host.Protocols;
@@ -74,6 +73,7 @@ static void GenericAvro_Fn([KafkaTrigger("brokers:9092", "myTopic", AvroSchema =
         static void GenericAvroWithoutKey_Fn([KafkaTrigger("brokers:9092", "myTopic", AvroSchema = "fake")] KafkaEventData<GenericRecord> genericRecord) { }
         static void GenericAvro_WithLongKey_Fn([KafkaTrigger("brokers:9092", "myTopic", AvroSchema = "fake")] KafkaEventData<long, GenericRecord> genericRecord) { }
         static void GenericAvro_WithStringKey_Fn([KafkaTrigger("brokers:9092", "myTopic", AvroSchema = "fake")] KafkaEventData<string, GenericRecord> genericRecord) { }
+        static void GenericWithSchemaRegistry_Fn([KafkaTrigger("brokers:9092", "myTopic", SchemaRegistryUrl = "localhost:8081")] KafkaEventData<GenericRecord> genericRecord) { }
 
         static void RawSpecificAvro_Fn([KafkaTrigger("brokers:9092", "myTopic")] MyAvroRecord myAvroRecord) { }
         static void SpecificAvro_Fn([KafkaTrigger("brokers:9092", "myTopic")] KafkaEventData<Null, MyAvroRecord> myAvroRecord) { }
@@ -140,9 +140,9 @@ public async Task When_No_Type_Is_Set_Should_Create_ByteArray_Listener(string fu
                 config,
                 Options.Create(new KafkaOptions()),
                 new KafkaEventDataConvertManager(NullLogger.Instance),
-                new DefaultNameResolver(config),                
+                new DefaultNameResolver(config),
                 NullLoggerFactory.Instance);
-            
+
             var parameterInfo = new TriggerBindingProviderContext(this.GetParameterInfo(functionName), default);
 
             var triggerBinding = await bindingProvider.TryCreateAsync(parameterInfo);
@@ -193,6 +193,7 @@ public async Task When_String_Value_Type_Is_Set_Should_Create_String_Listener(st
         [InlineData(nameof(GenericAvro_Fn), typeof(Null))]
         [InlineData(nameof(GenericAvroWithoutKey_Fn), typeof(string))]
         [InlineData(nameof(RawGenericAvro_Fn), typeof(string))]
+        [InlineData(nameof(GenericWithSchemaRegistry_Fn), typeof(string))]
         public async Task When_Avro_Schema_Is_Provided_Should_Create_GenericRecord_Listener(string functionName, Type expectedKeyType)
         {
             var attribute = new KafkaTriggerAttribute("brokers:9092", "myTopic")
@@ -223,8 +224,8 @@ public async Task When_Avro_Schema_Is_Provided_Should_Create_GenericRecord_Liste
 
 
             Assert.NotNull(listener);
-            AssertIsCorrectKafkaListener(listener, expectedKeyType, typeof(GenericRecord), typeof(SyncOverAsyncDeserializer<GenericRecord>));            
-        }       
+            AssertIsCorrectKafkaListener(listener, expectedKeyType, typeof(GenericRecord), typeof(SyncOverAsyncDeserializer<GenericRecord>));
+        }
 
         [Theory]
         [InlineData(nameof(SpecificAvro_Fn), typeof(Null))]