Adds GET subject versions in the registry client (#771)

* Fixes return type of get subject versions in schema registry client

* Adds get subject versions method to schema registry client

Author: mikalaisyty

src/Confluent.SchemaRegistry/CachedSchemaRegistryClient.cs

@@ -310,6 +310,13 @@ public Task<List<string>> GetAllSubjectsAsync()
             => restService.GetSubjectsAsync();
 
 
+        /// <summary>
+        ///     Refer to <see cref="Confluent.SchemaRegistry.ISchemaRegistryClient.GetSubjectVersionsAsync(string)" />
+        /// </summary>
+        public async Task<List<int>> GetSubjectVersionsAsync(string subject)
+            => await restService.GetSubjectVersionsAsync(subject).ConfigureAwait(continueOnCapturedContext: false);
+
+
         /// <summary>
         ///     Refer to <see cref="Confluent.SchemaRegistry.ISchemaRegistryClient.IsCompatibleAsync(string, string)" />
         /// </summary>

src/Confluent.SchemaRegistry/ISchemaRegistryClient.cs

@@ -115,6 +115,18 @@ public interface ISchemaRegistryClient : IDisposable
         Task<List<string>> GetAllSubjectsAsync();
 
 
+        /// <summary>
+        ///     Gets a list of versions registered under the specified <paramref name="subject" />.
+        /// </summary>
+        /// <param name="subject">
+        ///     The subject to get versions registered under.
+        /// </param>
+        /// <returns>
+        ///     A list of versions registered under the specified <paramref name="subject" />.
+        /// </returns>
+        Task<List<int>> GetSubjectVersionsAsync(string subject);
+
+
         /// <summary>
         ///     Check if a schema is compatible with latest version registered against a 
         ///     specified subject.

src/Confluent.SchemaRegistry/Rest/IRestService.cs

@@ -36,7 +36,7 @@ internal interface IRestService : IDisposable
         Task<string> GetSchemaAsync(int id);
         Task<Schema> GetSchemaAsync(string subject, int version);
         Task<List<string>> GetSubjectsAsync();
-        Task<List<string>> GetSubjectVersionsAsync(string subject);
+        Task<List<int>> GetSubjectVersionsAsync(string subject);
         Task<int> RegisterSchemaAsync(string subject, string schema);
         Task<Config> SetCompatibilityAsync(string subject, Compatibility compatibility);
         Task<Config> SetGlobalCompatibilityAsync(Compatibility compatibility);

src/Confluent.SchemaRegistry/Rest/RestService.cs

@@ -230,8 +230,8 @@ public async Task<List<string>> GetSubjectsAsync()
             => await RequestListOfAsync<string>("/subjects", HttpMethod.Get)
                         .ConfigureAwait(continueOnCapturedContext: false);
 
-        public async Task<List<string>> GetSubjectVersionsAsync(string subject)
-            => await RequestListOfAsync<string>($"/subjects/{subject}/versions", HttpMethod.Get)
+        public async Task<List<int>> GetSubjectVersionsAsync(string subject)
+            => await RequestListOfAsync<int>($"/subjects/{subject}/versions", HttpMethod.Get)
                         .ConfigureAwait(continueOnCapturedContext: false);
 
         public async Task<Schema> GetSchemaAsync(string subject, int version)

test/Confluent.SchemaRegistry.IntegrationTests/Tests/GetSubjectVersions.cs

@@ -0,0 +1,53 @@
+// Copyright 20 Confluent Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// Refer to LICENSE for more information.
+
+using System;
+using System.Collections.Generic;
+using Xunit;
+
+
+namespace Confluent.SchemaRegistry.IntegrationTests
+{
+    public static partial class Tests
+    {
+        [Theory, MemberData(nameof(SchemaRegistryParameters))]
+        public static void GetSubjectVersions(Config config)
+        {
+            var topicName = Guid.NewGuid().ToString();
+
+            var testSchema1 =
+                "{\"type\":\"record\",\"name\":\"User\",\"namespace\":\"Confluent.Kafka.Examples.AvroSpecific" +
+                "\",\"fields\":[{\"name\":\"name\",\"type\":\"string\"},{\"name\":\"favorite_number\",\"type\":[\"i" +
+                "nt\",\"null\"]},{\"name\":\"favorite_color\",\"type\":[\"string\",\"null\"]}]}";
+
+            var testSchema2 = // compatible with testSchema1
+                "{\"type\":\"record\",\"name\":\"User\",\"namespace\":\"Confluent.Kafka.Examples.AvroSpecific" +
+                "\",\"fields\":[{\"name\":\"name\",\"type\":\"string\"},{\"name\":\"favorite_number\",\"type\":[\"i" +
+                "nt\",\"null\"]},{\"name\":\"favorite_color\",\"type\":[\"string\",\"null\"]}," +
+                "{\"name\":\"favorite_shape\",\"type\":[\"string\",\"null\"], \"default\": \"square\"}]}";
+
+            var sr = new CachedSchemaRegistryClient(new SchemaRegistryConfig { SchemaRegistryUrl = config.Server });
+
+            var subject = sr.ConstructValueSubjectName(topicName);
+            var id1 = sr.RegisterSchemaAsync(subject, testSchema1).Result;
+            var id2 = sr.RegisterSchemaAsync(subject, testSchema2).Result;
+
+            var versions = sr.GetSubjectVersionsAsync(subject).Result;
+
+            Assert.Equal(versions.Count, 2);
+        }
+    }
+}