new sr endpoint client addition (#3899)

Author: gzimmers-cflt

client/src/main/java/io/confluent/kafka/schemaregistry/client/CachedSchemaRegistryClient.java

@@ -22,6 +22,7 @@
 import com.google.common.cache.CacheLoader;
 import com.google.common.cache.LoadingCache;
 import com.google.common.collect.ImmutableMap;
+import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaRegistryServerVersion;
 import io.confluent.kafka.schemaregistry.client.rest.entities.requests.RegisterSchemaRequest;
 import io.confluent.kafka.schemaregistry.client.rest.entities.requests.RegisterSchemaResponse;
 import io.confluent.kafka.schemaregistry.utils.QualifiedSubject;
@@ -990,6 +991,12 @@ public SchemaRegistryDeployment getSchemaRegistryDeployment()
     return restService.getSchemaRegistryDeployment();
   }
 
+  @Override
+  public SchemaRegistryServerVersion getSchemaRegistryServerVersion()
+      throws IOException, RestClientException {
+    return restService.getSchemaRegistryServerVersion();
+  }
+
   @Override
   public Collection<String> getAllSubjects() throws IOException, RestClientException {
     return restService.getAllSubjects();

client/src/main/java/io/confluent/kafka/schemaregistry/client/SchemaRegistryClient.java

@@ -23,6 +23,7 @@
 import io.confluent.kafka.schemaregistry.client.rest.entities.Metadata;
 import io.confluent.kafka.schemaregistry.client.rest.entities.RuleSet;
 import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaRegistryDeployment;
+import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaRegistryServerVersion;
 import io.confluent.kafka.schemaregistry.client.rest.entities.requests.RegisterSchemaResponse;
 import java.io.Closeable;
 import java.io.IOException;
@@ -295,6 +296,11 @@ default SchemaRegistryDeployment getSchemaRegistryDeployment()
     throw new UnsupportedOperationException();
   }
 
+  default SchemaRegistryServerVersion getSchemaRegistryServerVersion()
+      throws IOException, RestClientException {
+    throw new UnsupportedOperationException();
+  }
+
   public Collection<String> getAllSubjects() throws IOException, RestClientException;
 
   default Collection<String> getAllSubjects(boolean lookupDeletedSubject) throws IOException,

client/src/test/java/io/confluent/kafka/schemaregistry/client/CachedSchemaRegistryClientTest.java

@@ -22,6 +22,9 @@
 import io.confluent.kafka.schemaregistry.client.rest.entities.RuleSet;
 import io.confluent.kafka.schemaregistry.client.rest.entities.requests.RegisterSchemaRequest;
 import io.confluent.kafka.schemaregistry.client.rest.entities.requests.RegisterSchemaResponse;
+import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaRegistryDeployment;
+import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaRegistryServerVersion;
+import java.util.ArrayList;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -888,6 +891,44 @@ public void testMissingSchemaCache() throws Exception {
     client.getId(SUBJECT_0, AVRO_SCHEMA_0);
   }
 
+  @Test
+  public void testGetSchemaRegistryDeployment() throws Exception {
+    List<String> deploymentAttributes = new ArrayList<>(Collections.singleton("deploymentScope:opensource"));
+    SchemaRegistryDeployment expectedDeployment = new SchemaRegistryDeployment(deploymentAttributes);
+
+    expect(restService.getSchemaRegistryDeployment())
+        .andReturn(expectedDeployment);
+
+    replay(restService);
+
+    SchemaRegistryDeployment deployment = client.getSchemaRegistryDeployment();
+
+    assertNotNull(deployment);
+    assertEquals(expectedDeployment.getAttributes(), deployment.getAttributes());
+
+    verify(restService);
+  }
+
+  @Test
+  public void testGetSchemaRegistryServerVersion() throws Exception {
+    String version = "7.5.0";
+    String commitId = "abc123def456";
+    SchemaRegistryServerVersion expectedVersion = new SchemaRegistryServerVersion(version, commitId);
+
+    expect(restService.getSchemaRegistryServerVersion())
+        .andReturn(expectedVersion);
+
+    replay(restService);
+
+    SchemaRegistryServerVersion serverVersion = client.getSchemaRegistryServerVersion();
+
+    assertNotNull(serverVersion);
+    assertEquals(version, serverVersion.getVersion());
+    assertEquals(commitId, serverVersion.getCommitId());
+
+    verify(restService);
+  }
+
 
   private static AvroSchema avroSchema(final int i) {
     return new AvroSchema(avroSchemaString(i));

core/src/main/java/io/confluent/kafka/schemaregistry/rest/client/LocalSchemaRegistryClient.java

@@ -29,6 +29,7 @@
 import io.confluent.kafka.schemaregistry.client.rest.entities.Schema;
 import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaReference;
 import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaRegistryDeployment;
+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.ExtendedSchema;
 import io.confluent.kafka.schemaregistry.client.rest.entities.SubjectVersion;
@@ -52,6 +53,7 @@
 import io.confluent.kafka.schemaregistry.rest.VersionId;
 import io.confluent.kafka.schemaregistry.rest.exceptions.Errors;
 import io.confluent.kafka.schemaregistry.rest.exceptions.RestInvalidCompatibilityException;
+import io.confluent.kafka.schemaregistry.utils.AppInfoParser;
 import io.confluent.kafka.schemaregistry.utils.Props;
 import io.confluent.kafka.schemaregistry.rest.exceptions.RestInvalidModeException;
 import io.confluent.kafka.schemaregistry.storage.KafkaSchemaRegistry;
@@ -692,6 +694,12 @@ public SchemaRegistryDeployment getSchemaRegistryDeployment()
     return Props.getSchemaRegistryDeployment(schemaRegistry.properties());
   }
 
+  @Override
+  public SchemaRegistryServerVersion getSchemaRegistryServerVersion()
+      throws IOException, RestClientException {
+    return new SchemaRegistryServerVersion(AppInfoParser.getVersion(), AppInfoParser.getCommitId());
+  }
+
   @Override
   public void reset() {
   }

core/src/main/java/io/confluent/kafka/schemaregistry/utils/Props.java

@@ -36,7 +36,7 @@ public static SchemaRegistryDeployment getSchemaRegistryDeployment(Map<String, O
       List<?> srDeploymentList = (List<?>) srDeployment;
       // Validate and process each element
       List<String> processedList = srDeploymentList.stream().map(
-          item -> item.toString().trim().toLowerCase()
+          item -> item.toString().trim()
       ).collect(Collectors.toList());
       return new SchemaRegistryDeployment(processedList);
     } else {

core/src/test/java/io/confluent/kafka/schemaregistry/rest/client/LocalSchemaRegistryClientTest.java

@@ -16,31 +16,37 @@
 package io.confluent.kafka.schemaregistry.rest.client;
 
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
 import io.confluent.kafka.schemaregistry.ClusterTestHarness;
 import io.confluent.kafka.schemaregistry.ParsedSchema;
 import io.confluent.kafka.schemaregistry.avro.AvroSchema;
 import io.confluent.kafka.schemaregistry.client.SchemaMetadata;
 import io.confluent.kafka.schemaregistry.client.rest.entities.Config;
 import io.confluent.kafka.schemaregistry.client.rest.entities.Metadata;
 import io.confluent.kafka.schemaregistry.client.rest.entities.Schema;
-import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException;
+import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaRegistryDeployment;
+import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaRegistryServerVersion;
 import io.confluent.kafka.schemaregistry.rest.SchemaRegistryConfig;
 import io.confluent.kafka.schemaregistry.rest.exceptions.RestInvalidSchemaException;
 import io.confluent.kafka.schemaregistry.rest.exceptions.RestOperationNotPermittedException;
 import io.confluent.kafka.schemaregistry.storage.KafkaSchemaRegistry;
 import io.confluent.kafka.schemaregistry.storage.StoreUtils;
 import io.confluent.kafka.schemaregistry.storage.serialization.SchemaRegistrySerializer;
-
-import java.io.IOException;
-import java.util.*;
-
+import io.confluent.kafka.schemaregistry.utils.Props;
 import io.confluent.rest.exceptions.RestNotFoundException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
 import org.junit.Before;
 import org.junit.jupiter.api.Test;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-
-import static org.junit.Assert.*;
-import static org.junit.Assert.assertEquals;
 
 public class LocalSchemaRegistryClientTest extends ClusterTestHarness {
 
@@ -214,4 +220,27 @@ public void testGetByVersion() throws Exception {
         Schema s2 = client.getByVersion(SUBJECT2, 1, false);
         assertEquals(id2, s2.getId().intValue());
     }
+
+    @Test
+  public void testGetSchemaRegistryDeployment() throws Exception {
+    Map<String, Object> props = new HashMap<>();
+    List<String> deploymentAttributes = new ArrayList<String>(Collections.singleton("deploymentScope:opensource"));
+    props.put(Props.PROPERTY_SCHEMA_REGISTRY_DEPLOYMENT_ATTRIBUTES, deploymentAttributes);
+
+    SchemaRegistryDeployment deployment = client.getSchemaRegistryDeployment();
+
+    assertNotNull(deployment);
+    assertEquals(deployment.getAttributes(),
+        new ArrayList<String>()
+    );
+  }
+
+  @Test
+  public void testGetSchemaRegistryServerVersion() throws Exception {
+    SchemaRegistryServerVersion version = client.getSchemaRegistryServerVersion();
+
+    assertNotNull(version);
+    assertNotNull(version.getVersion());
+    assertNotNull(version.getCommitId());
+  }
 }

core/src/test/java/io/confluent/kafka/schemaregistry/utils/PropsTest.java

@@ -182,7 +182,7 @@ public void testGetSchemaRegistryDeploymentWithMixedCaseStrings() {
 
     SchemaRegistryDeployment result = Props.getSchemaRegistryDeployment(props);
     assertNotNull("Should return SchemaRegistryDeployment for mixed case strings", result);
-    List<String> expected = Arrays.asList("confluent", "enterprise", "opensource");
+    List<String> expected = Arrays.asList("CONFLUENT", "Enterprise", "opensource");
     assertEquals("Should convert to lowercase", expected, result.getAttributes());
   }
 
@@ -223,14 +223,14 @@ public void testGetSchemaRegistryDeploymentWithComplexObjects() {
   }
 
   @Test
-  public void testGetSchemaRegistryDeploymentTrimsAndLowercase() {
+  public void testGetSchemaRegistryDeploymentTrims() {
     Map<String, Object> props = new HashMap<>();
     List<String> attributes = Arrays.asList("  CONFLUENT  ", "Enterprise ", " opensource");
     props.put(Props.PROPERTY_SCHEMA_REGISTRY_DEPLOYMENT_ATTRIBUTES, attributes);
 
     SchemaRegistryDeployment result = Props.getSchemaRegistryDeployment(props);
     assertNotNull("Should return SchemaRegistryDeployment", result);
-    List<String> expected = Arrays.asList("confluent", "enterprise", "opensource");
+    List<String> expected = Arrays.asList("CONFLUENT", "Enterprise", "opensource");
     assertEquals("Should trim whitespace and convert to lowercase", expected, result.getAttributes());
   }
 }