Include deleted schemas when getting schemas by subject/version (#16) (#2288)

* Include deleted schemas when getting schemas by subect/version

* Minor fix

Author: rayokota

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

@@ -89,7 +89,7 @@ private async Task CreateSchemaDictUtil(Schema root)
             {
                 foreach (var reference in root.References)
                 {
-                    Schema refSchemaRes = await schemaRegistryClient.GetRegisteredSchemaAsync(reference.Subject, reference.Version);
+                    Schema refSchemaRes = await schemaRegistryClient.GetRegisteredSchemaAsync(reference.Subject, reference.Version, false);
                     await CreateSchemaDictUtil(refSchemaRes);
                 }
             }

src/Confluent.SchemaRegistry/AsyncSerde.cs

@@ -124,7 +124,7 @@ private async Task<IDictionary<string, string>> ResolveReferences(
                 visited.Add(reference.Name);
                 if (!schemas.ContainsKey(reference.Name))
                 {
-                    Schema s = await schemaRegistryClient.GetRegisteredSchemaAsync(reference.Subject, reference.Version)
+                    Schema s = await schemaRegistryClient.GetRegisteredSchemaAsync(reference.Subject, reference.Version, false)
                         .ConfigureAwait(continueOnCapturedContext: false);
                     if (s == null)
                     {
@@ -201,7 +201,7 @@ private async Task<IList<Schema>> GetSchemasBetween(string subject, Schema first
             int version1 = first.Version;
             int version2 = last.Version;
             for (int i = version1 + 1; i < version2; i++) {
-                tasks.Add(schemaRegistryClient.GetRegisteredSchemaAsync(subject, i));
+                tasks.Add(schemaRegistryClient.GetRegisteredSchemaAsync(subject, i, false));
             }
             RegisteredSchema[] schemas = await Task.WhenAll(tasks).ConfigureAwait(continueOnCapturedContext: false);
 

src/Confluent.SchemaRegistry/CachedSchemaRegistryClient.cs

@@ -595,7 +595,7 @@ public async Task<Schema> GetSchemaBySubjectAndIdAsync(string subject, int id, s
 
 
         /// <inheritdoc/>
-        public async Task<RegisteredSchema> GetRegisteredSchemaAsync(string subject, int version)
+        public async Task<RegisteredSchema> GetRegisteredSchemaAsync(string subject, int version, bool ignoreDeletedSchemas = true)
         {
             await cacheMutex.WaitAsync().ConfigureAwait(continueOnCapturedContext: false);
             try

src/Confluent.SchemaRegistry/ISchemaRegistryClient.cs

@@ -194,10 +194,13 @@ public interface ISchemaRegistryClient : IDisposable
         /// <param name="version">
         ///     The version number of schema to get.
         /// </param>
+        /// <param name="ignoreDeletedSchemas">
+        ///     Whether or not to ignore deleted schemas.
+        /// </param>
         /// <returns>
         ///     The schema identified by the specified <paramref name="subject" /> and <paramref name="version" />.
         /// </returns>
-        Task<RegisteredSchema> GetRegisteredSchemaAsync(string subject, int version);
+        Task<RegisteredSchema> GetRegisteredSchemaAsync(string subject, int version, bool ignoreDeletedSchemas = true);
 
 
         /// <summary>

src/Confluent.SchemaRegistry/Rest/IRestService.cs

@@ -34,7 +34,7 @@ internal interface IRestService : IDisposable
         Task<RegisteredSchema> GetLatestWithMetadataAsync(string subject, IDictionary<string, string> metadata, bool ignoreDeletedSchemas);
         Task<Schema> GetSchemaAsync(int id, string format = null);
         Task<Schema> GetSchemaBySubjectAndIdAsync(string subject, int id, string format = null);
-        Task<RegisteredSchema> GetSchemaAsync(string subject, int version);
+        Task<RegisteredSchema> GetSchemaAsync(string subject, int version, bool ignoreDeletedSchemas = true);
         Task<List<string>> GetSubjectsAsync();
         Task<List<int>> GetSubjectVersionsAsync(string subject);
         Task<int> RegisterSchemaAsync(string subject, Schema schema, bool normalize);

src/Confluent.SchemaRegistry/Rest/RestService.cs

@@ -339,9 +339,9 @@ public async Task<List<int>> GetSubjectVersionsAsync(string subject)
             => await RequestListOfAsync<int>($"subjects/{Uri.EscapeDataString(subject)}/versions", HttpMethod.Get)
                 .ConfigureAwait(continueOnCapturedContext: false);
 
-        public async Task<RegisteredSchema> GetSchemaAsync(string subject, int version)
+        public async Task<RegisteredSchema> GetSchemaAsync(string subject, int version, bool ignoreDeletedSchemas = true)
             => SanitizeRegisteredSchema(
-                await RequestAsync<RegisteredSchema>($"subjects/{Uri.EscapeDataString(subject)}/versions/{version}",
+                await RequestAsync<RegisteredSchema>($"subjects/{Uri.EscapeDataString(subject)}/versions/{version}?deleted={!ignoreDeletedSchemas}",
                         HttpMethod.Get)
                     .ConfigureAwait(continueOnCapturedContext: false));
 

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

@@ -61,8 +61,8 @@ public BaseSerializeDeserializeTests()
                         return new Schema(store.Where(x => x.Value == id).First().Key, null, SchemaType.Avro);
                     }
                 });
-            schemaRegistryMock.Setup(x => x.GetRegisteredSchemaAsync(It.IsAny<string>(), It.IsAny<int>())).ReturnsAsync(
-                (string subject, int version) => subjectStore[subject].First(x => x.Version == version)
+            schemaRegistryMock.Setup(x => x.GetRegisteredSchemaAsync(It.IsAny<string>(), It.IsAny<int>(), It.IsAny<bool>())).ReturnsAsync(
+                (string subject, int version, bool ignoreDeletedSchemas) => subjectStore[subject].First(x => x.Version == version)
             );
             schemaRegistryMock.Setup(x => x.GetLatestSchemaAsync(It.IsAny<string>())).ReturnsAsync(
                 (string subject) => subjectStore[subject].Last()