Retrieve version for writer schema when getting migrations (#2358)

* Retrieve version for writer schema when getting migrations

Fixes #2356

* Minor cleanup

Author: rayokota

src/Confluent.SchemaRegistry.Serdes.Avro/GenericDeserializerImpl.cs

@@ -72,7 +72,7 @@ public async Task<GenericRecord> Deserialize(string topic, Headers headers, byte
                 }
 
                 string subject = GetSubjectName(topic, isKey, null);
-                Schema latestSchema = null;
+                RegisteredSchema latestSchema = null;
                 if (subject != null)
                 {
                     latestSchema = await GetReaderSchema(subject)

src/Confluent.SchemaRegistry.Serdes.Avro/SpecificDeserializerImpl.cs

@@ -126,7 +126,7 @@ public async Task<T> Deserialize(string topic, Headers headers, byte[] array, bo
                 }
 
                 string subject = GetSubjectName(topic, isKey, null);
-                Schema latestSchema = null;
+                RegisteredSchema latestSchema = null;
                 if (subject != null)
                 {
                     latestSchema = await GetReaderSchema(subject)

src/Confluent.SchemaRegistry.Serdes.Json/JsonDeserializer.cs

@@ -175,7 +175,7 @@ public override async Task<T> DeserializeAsync(ReadOnlyMemory<byte> data, bool i
             bool isKey = context.Component == MessageComponentType.Key;
             string topic = context.Topic;
             string subject = GetSubjectName(topic, isKey, null);
-            Schema latestSchema = null;
+            RegisteredSchema latestSchema = null;
             if (subject != null)
             {
                 latestSchema = await GetReaderSchema(subject)

src/Confluent.SchemaRegistry.Serdes.Protobuf/ProtobufDeserializer.cs

@@ -128,7 +128,7 @@ public override async Task<T> DeserializeAsync(ReadOnlyMemory<byte> data, bool i
             // Currently Protobuf does not support migration rules because of lack of support for DynamicMessage
             // See https://github.com/protocolbuffers/protobuf/issues/658
             /*
-            Schema latestSchema = await SerdeUtils.GetReaderSchema(schemaRegistryClient, subject, useLatestWithMetadata, useLatestVersion)
+            RegisteredSchema latestSchema = await SerdeUtils.GetReaderSchema(schemaRegistryClient, subject, useLatestWithMetadata, useLatestVersion)
                 .ConfigureAwait(continueOnCapturedContext: false);
             */
 

src/Confluent.SchemaRegistry/AsyncSerde.cs

@@ -166,11 +166,14 @@ await ResolveReferences(s, schemas, visited)
             return schemas;
         }
 
-        protected async Task<IList<Migration>> GetMigrations(string subject, Schema writerSchema, Schema readerSchema)
+        protected async Task<IList<Migration>> GetMigrations(string subject, Schema writer, RegisteredSchema readerSchema)
         {
+            var writerSchema = await schemaRegistryClient.LookupSchemaAsync(subject, writer, false, false)
+                .ConfigureAwait(continueOnCapturedContext: false);
+
             RuleMode migrationMode;
-            Schema first;
-            Schema last;
+            RegisteredSchema first;
+            RegisteredSchema last;
             IList<Migration> migrations = new List<Migration>();
             if (writerSchema.Version < readerSchema.Version)
             {
@@ -217,7 +220,7 @@ protected async Task<IList<Migration>> GetMigrations(string subject, Schema writ
             return migrations;
         }
 
-        private async Task<IList<Schema>> GetSchemasBetween(string subject, Schema first, Schema last)
+        private async Task<IList<Schema>> GetSchemasBetween(string subject, RegisteredSchema first, RegisteredSchema last)
         {
             if (last.Version - first.Version <= 1)
             {

test/Confluent.SchemaRegistry.Serdes.UnitTests/BaseSerializeDeserialize.cs

@@ -45,7 +45,15 @@ public BaseSerializeDeserializeTests()
             var schemaRegistryMock = new Mock<ISchemaRegistryClient>();
             schemaRegistryMock.Setup(x => x.ConstructValueSubjectName(testTopic, It.IsAny<string>())).Returns($"{testTopic}-value");
             schemaRegistryMock.Setup(x => x.RegisterSchemaAsync(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<bool>())).ReturnsAsync(
-                (string topic, string schema, bool normalize) => store.TryGetValue(schema, out int id) ? id : store[schema] = store.Count + 1
+                (string subject, string schema, bool normalize) => store.TryGetValue(schema, out int id) ? id : store[schema] = store.Count + 1
+            );
+            schemaRegistryMock.Setup(x => x.LookupSchemaAsync(It.IsAny<string>(), It.IsAny<Schema>(), It.IsAny<bool>(), It.IsAny<bool>())).ReturnsAsync(
+                (string subject, Schema schema, bool ignoreDeleted, bool normalize) =>
+                {
+                    return subjectStore[subject].First(x =>
+                        x.SchemaString == schema.SchemaString
+                    );
+                }
             );
             schemaRegistryMock.Setup(x => x.GetSchemaBySubjectAndIdAsync(It.IsAny<string>(), It.IsAny<int>(), It.IsAny<string>())).ReturnsAsync(
                 (string subject, int id, string format) =>