Fixed GenericSerializerImpl to work with multiple topics. (#538)

* Fixed GenericSerializerImpl to work with multiple topics.
Added UnitTests

* Revert "Fixed GenericSerializerImpl to work with multiple topics."

This reverts commit b231938.

* GenericSerializerImpl fixed to work with multiple topics

* GenericSerializerImpl - added clearing of local schema cache and trowing and exception in case of schemaId mismatch

* GenericSerializerImpl fixed error message when throwing exception on schema mismatch

Author: NikolayDekov

src/Confluent.Kafka.Avro/GenericSerializerImpl.cs

@@ -111,18 +111,26 @@ public byte[] Serialize(string topic, GenericRecord data)
                 var subjectSchemaPair = new KeyValuePair<string, string>(subject, writerSchemaString);
                 if (!registeredSchemas.Contains(subjectSchemaPair))
                 {
+                    int newSchemaId;
                     // first usage: register/get schema to check compatibility
                     if (autoRegisterSchema)
                     {
-                        schemaIds.Add(
-                            writerSchemaString,
-                            schemaRegistryClient.RegisterSchemaAsync(subject, writerSchemaString).ConfigureAwait(false).GetAwaiter().GetResult());
+                        newSchemaId = schemaRegistryClient.RegisterSchemaAsync(subject, writerSchemaString).ConfigureAwait(false).GetAwaiter().GetResult();
                     }
                     else
                     {
-                        schemaIds.Add(
-                            writerSchemaString,
-                            schemaRegistryClient.GetSchemaIdAsync(subject, writerSchemaString).ConfigureAwait(false).GetAwaiter().GetResult());
+                        newSchemaId = schemaRegistryClient.GetSchemaIdAsync(subject, writerSchemaString).ConfigureAwait(false).GetAwaiter().GetResult();
+                    }
+
+                    if (!schemaIds.ContainsKey(writerSchemaString))
+                    {
+                        schemaIds.Add(writerSchemaString, newSchemaId);
+                    }
+                    else if (schemaIds[writerSchemaString] != newSchemaId)
+                    {
+                        schemaIds.Clear();
+                        registeredSchemas.Clear();
+                        throw new KafkaException(new Error(isKey ? ErrorCode.Local_KeySerialization : ErrorCode.Local_ValueSerialization, $"Duplicate schema registration encountered: Schema ids {schemaIds[writerSchemaString]} and {newSchemaId} are associated with the same schema."));
                     }
 
                     registeredSchemas.Add(subjectSchemaPair);

test/Confluent.Kafka.Avro.IntegrationTests/Tests/ProduceGenericMultipleTopics.cs

@@ -0,0 +1,64 @@
+using System;
+using System.Collections.Generic;
+using Confluent.Kafka.Examples.AvroSpecific;
+using Confluent.Kafka.Serialization;
+using Avro;
+using Avro.Generic;
+using Xunit;
+
+namespace Confluent.Kafka.Avro.IntegrationTests
+{
+    public static partial class Tests
+    {
+        /// <summary>
+        ///     Test that messages produced with the avro serializer can be consumed with the
+        ///     avro deserializer.
+        /// </summary>
+        [Theory, MemberData(nameof(TestParameters))]
+        public static void ProduceGenericMultipleTopics(string bootstrapServers, string schemaRegistryServers)
+        {
+            var s = (RecordSchema)Schema.Parse(
+                @"{
+                    ""namespace"": ""Confluent.Kafka.Examples.AvroSpecific"",
+                    ""type"": ""record"",
+                    ""name"": ""User"",
+                    ""fields"": [
+                        {""name"": ""name"", ""type"": ""string""},
+                        {""name"": ""favorite_number"",  ""type"": [""int"", ""null""]},
+                        {""name"": ""favorite_color"", ""type"": [""string"", ""null""]}
+                    ]
+                  }"
+            );
+
+            var config = new Dictionary<string, object>()
+            {
+                { "bootstrap.servers", bootstrapServers },
+                { "schema.registry.url", schemaRegistryServers }
+            };
+
+            var topic = Guid.NewGuid().ToString();
+            var topic2 = Guid.NewGuid().ToString();
+
+            Message<Null, GenericRecord> dr;
+            Message<Null, GenericRecord> dr2;
+
+            using (var p = new Producer<Null, GenericRecord>(config, null, new AvroSerializer<GenericRecord>()))
+            {
+                var record = new GenericRecord(s);
+                record.Add("name", "my name 2");
+                record.Add("favorite_number", 44);
+                record.Add("favorite_color", null);
+                dr = p.ProduceAsync(topic, null, record).Result;
+                dr2 = p.ProduceAsync(topic2, null, record).Result;
+            }
+
+            Assert.Null(dr.Key);
+            Assert.NotNull(dr.Value);
+
+            Assert.Null(dr2.Key);
+            Assert.NotNull(dr2.Value);
+
+        }
+
+    }
+}