Merge branch '7.1.x' into 7.2.x by anshul-goyal

client/src/main/java/io/confluent/kafka/schemaregistry/client/rest/RestService.java

@@ -25,6 +25,7 @@
 import io.confluent.kafka.schemaregistry.client.rest.entities.Config;
 import io.confluent.kafka.schemaregistry.client.rest.entities.ErrorMessage;
 import io.confluent.kafka.schemaregistry.client.rest.entities.Schema;
+import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaRegistryServerVersion;
 import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaString;
 import io.confluent.kafka.schemaregistry.client.rest.entities.ServerClusterId;
 import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaReference;
@@ -143,6 +144,9 @@ public class RestService implements Configurable {
   private static final TypeReference<ServerClusterId> GET_CLUSTER_ID_RESPONSE_TYPE =
       new TypeReference<ServerClusterId>() {
       };
+  private static final TypeReference<SchemaRegistryServerVersion> GET_SR_VERSION_RESPONSE_TYPE =
+          new TypeReference<SchemaRegistryServerVersion>() {
+          };
 
   private static final int HTTP_CONNECT_TIMEOUT_MS = 60000;
   private static final int HTTP_READ_TIMEOUT_MS = 60000;
@@ -1171,6 +1175,12 @@ public ServerClusterId getClusterId(Map<String, String> requestProperties)
                         requestProperties, GET_CLUSTER_ID_RESPONSE_TYPE);
   }
 
+  public SchemaRegistryServerVersion getSchemaRegistryServerVersion()
+          throws IOException, RestClientException {
+    return httpRequest("/v1/metadata/version", "GET", null,
+            DEFAULT_REQUEST_PROPERTIES, GET_SR_VERSION_RESPONSE_TYPE);
+  }
+
   private static List<String> parseBaseUrl(String baseUrl) {
     List<String> baseUrls = Arrays.asList(baseUrl.split("\\s*,\\s*"));
     if (baseUrls.isEmpty()) {

client/src/main/java/io/confluent/kafka/schemaregistry/client/rest/entities/SchemaRegistryServerVersion.java

@@ -0,0 +1,76 @@
+/*
+ * Copyright 2022 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (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.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.kafka.schemaregistry.client.rest.entities;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.Objects;
+
+@JsonInclude(JsonInclude.Include.NON_EMPTY)
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class SchemaRegistryServerVersion {
+
+  private String version;
+  private String commitId;
+
+  @JsonCreator
+  public SchemaRegistryServerVersion(@JsonProperty("version") String version,
+                                     @JsonProperty("commitId") String commitId) {
+    this.version = version;
+    this.commitId = commitId;
+  }
+
+  @JsonProperty("version")
+  public String getVersion() {
+    return version;
+  }
+
+  @JsonProperty("version")
+  public void setVersion(String version) {
+    this.version = version;
+  }
+
+  @JsonProperty("commitId")
+  public String getCommitId() {
+    return commitId;
+  }
+
+  @JsonProperty("commitId")
+  public void setCommitId(String commitId) {
+    this.commitId = commitId;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    SchemaRegistryServerVersion srVersion = (SchemaRegistryServerVersion) o;
+    return Objects.equals(version, srVersion.version)
+            && Objects.equals(commitId, srVersion.commitId);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(version, commitId);
+  }
+}

core/src/main/java/io/confluent/kafka/schemaregistry/rest/resources/ServerMetadataResource.java

@@ -16,8 +16,10 @@
 package io.confluent.kafka.schemaregistry.rest.resources;
 
 import io.confluent.kafka.schemaregistry.client.rest.Versions;
+import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaRegistryServerVersion;
 import io.confluent.kafka.schemaregistry.client.rest.entities.ServerClusterId;
 import io.confluent.kafka.schemaregistry.storage.KafkaSchemaRegistry;
+import io.confluent.kafka.schemaregistry.utils.AppInfoParser;
 import io.confluent.rest.annotations.PerformanceMetric;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.responses.ApiResponse;
@@ -57,4 +59,14 @@ public ServerClusterId getClusterId() {
     String schemaRegistryClusterId = schemaRegistry.getGroupId();
     return ServerClusterId.of(kafkaClusterId, schemaRegistryClusterId);
   }
+
+  @GET
+  @Path("/version")
+  @Operation(summary = "Get Schema Registry server version", responses = {
+      @ApiResponse(responseCode = "500",
+                      description = "Error code 50001 -- Error in the backend data store\n")
+  })
+  public SchemaRegistryServerVersion getSchemaRegistryVersion() {
+    return new SchemaRegistryServerVersion(AppInfoParser.getVersion(), AppInfoParser.getCommitId());
+  }
 }

core/src/test/java/io/confluent/kafka/schemaregistry/rest/RestApiTest.java

@@ -21,6 +21,7 @@
 import io.confluent.kafka.schemaregistry.avro.AvroUtils;
 import io.confluent.kafka.schemaregistry.client.rest.RestService;
 import io.confluent.kafka.schemaregistry.client.rest.entities.Schema;
+import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaRegistryServerVersion;
 import io.confluent.kafka.schemaregistry.client.rest.entities.ServerClusterId;
 import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaReference;
 import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaString;
@@ -30,6 +31,7 @@
 import io.confluent.kafka.schemaregistry.rest.exceptions.Errors;
 import io.confluent.kafka.schemaregistry.rest.exceptions.RestInvalidSubjectException;
 import io.confluent.kafka.schemaregistry.rest.exceptions.RestInvalidVersionException;
+import io.confluent.kafka.schemaregistry.utils.AppInfoParser;
 import io.confluent.kafka.schemaregistry.utils.TestUtils;
 
 import org.apache.avro.Schema.Parser;
@@ -1796,6 +1798,13 @@ public void testGetClusterId() throws Exception {
     }
   }
 
+  @Test
+  public void testGetSchemaRegistryServerVersion() throws Exception {
+      SchemaRegistryServerVersion srVersion = restApp.restClient.getSchemaRegistryServerVersion();
+      assertEquals(AppInfoParser.getVersion(), srVersion.getVersion());
+      assertEquals(AppInfoParser.getCommitId(), srVersion.getCommitId());
+  }
+
   @Test
   public void testHttpResponseHeaders() throws Exception {
     String baseUrl = restApp.restClient.getBaseUrls().current();