Kafka store abstraction

.gitignore

@@ -23,3 +23,4 @@ target/
 # Log file from schema registry daemon
 logs/
 nohup.out
+/.metadata/

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

@@ -43,8 +43,8 @@ public int getErrorCode() {
   }
 
   @JsonProperty("error_code")
-  public void setErrorCode(int error_code) {
-    this.errorCode = error_code;
+  public void setErrorCode(int errorCode) {
+    this.errorCode = errorCode;
   }
 
   @JsonProperty

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

@@ -51,7 +51,9 @@ public Map<String, Object> getScope() {
 
   public static ServerClusterId of(String kafkaClusterId, String schemaRegistryClusterId) {
     Map<String, Object> clusters = new HashMap<>();
-    clusters.put(KAFKA_CLUSTER, kafkaClusterId);
+    if (kafkaClusterId != null) {
+      clusters.put(KAFKA_CLUSTER, kafkaClusterId);
+    }
     clusters.put(SCHEMA_REGISTRY_CLUSTER, schemaRegistryClusterId);
     Map<String, Object> serverClusterId = new HashMap<>();
     serverClusterId.put("path", Collections.emptyList());

core/src/main/java/io/confluent/kafka/schemaregistry/rest/SchemaRegistryRestApplication.java

@@ -29,6 +29,7 @@
 import io.confluent.kafka.schemaregistry.rest.resources.SubjectVersionsResource;
 import io.confluent.kafka.schemaregistry.rest.resources.SubjectsResource;
 import io.confluent.kafka.schemaregistry.storage.KafkaSchemaRegistry;
+import io.confluent.kafka.schemaregistry.storage.SchemaRegistry;
 import io.confluent.kafka.schemaregistry.storage.serialization.SchemaRegistrySerializer;
 import io.confluent.rest.Application;
 import io.confluent.rest.RestConfigException;
@@ -47,7 +48,7 @@
 public class SchemaRegistryRestApplication extends Application<SchemaRegistryConfig> {
 
   private static final Logger log = LoggerFactory.getLogger(SchemaRegistryRestApplication.class);
-  private KafkaSchemaRegistry schemaRegistry = null;
+  private SchemaRegistry schemaRegistry = null;
   private List<SchemaRegistryResourceExtension> schemaRegistryResourceExtensions = null;
 
   public SchemaRegistryRestApplication(Properties props) throws RestConfigException {
@@ -65,8 +66,8 @@ public SchemaRegistryRestApplication(SchemaRegistryConfig config) {
   }
 
 
-  protected KafkaSchemaRegistry initSchemaRegistry(SchemaRegistryConfig config) {
-    KafkaSchemaRegistry kafkaSchemaRegistry = null;
+  protected SchemaRegistry initSchemaRegistry(SchemaRegistryConfig config) {
+    SchemaRegistry kafkaSchemaRegistry = null;
     try {
       kafkaSchemaRegistry = new KafkaSchemaRegistry(
           config,
@@ -158,6 +159,6 @@ public void onShutdown() {
 
   // for testing purpose only
   public KafkaSchemaRegistry schemaRegistry() {
-    return schemaRegistry;
+    return (KafkaSchemaRegistry)schemaRegistry;
   }
 }

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

@@ -26,7 +26,7 @@
 import io.confluent.kafka.schemaregistry.exceptions.SchemaRegistryStoreException;
 import io.confluent.kafka.schemaregistry.rest.VersionId;
 import io.confluent.kafka.schemaregistry.rest.exceptions.Errors;
-import io.confluent.kafka.schemaregistry.storage.KafkaSchemaRegistry;
+import io.confluent.kafka.schemaregistry.storage.SchemaRegistry;
 import io.confluent.kafka.schemaregistry.utils.QualifiedSubject;
 import io.confluent.rest.annotations.PerformanceMetric;
 import io.swagger.v3.oas.annotations.Operation;
@@ -59,9 +59,9 @@
 public class CompatibilityResource {
 
   private static final Logger log = LoggerFactory.getLogger(CompatibilityResource.class);
-  private final KafkaSchemaRegistry schemaRegistry;
+  private final SchemaRegistry schemaRegistry;
 
-  public CompatibilityResource(KafkaSchemaRegistry schemaRegistry) {
+  public CompatibilityResource(SchemaRegistry schemaRegistry) {
     this.schemaRegistry = schemaRegistry;
   }
 

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

@@ -15,7 +15,7 @@
 
 package io.confluent.kafka.schemaregistry.rest.resources;
 
-import static io.confluent.kafka.schemaregistry.storage.KafkaSchemaRegistry.GLOBAL_RESOURCE_NAME;
+import static io.confluent.kafka.schemaregistry.storage.SchemaRegistry.GLOBAL_RESOURCE_NAME;
 
 import com.google.common.base.CharMatcher;
 import io.confluent.kafka.schemaregistry.CompatibilityLevel;
@@ -28,7 +28,7 @@
 import io.confluent.kafka.schemaregistry.exceptions.UnknownLeaderException;
 import io.confluent.kafka.schemaregistry.rest.exceptions.Errors;
 import io.confluent.kafka.schemaregistry.rest.exceptions.RestInvalidCompatibilityException;
-import io.confluent.kafka.schemaregistry.storage.KafkaSchemaRegistry;
+import io.confluent.kafka.schemaregistry.storage.SchemaRegistry;
 import io.confluent.kafka.schemaregistry.utils.QualifiedSubject;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.Parameter;
@@ -63,11 +63,11 @@
 public class ConfigResource {
 
   private static final Logger log = LoggerFactory.getLogger(ConfigResource.class);
-  private final KafkaSchemaRegistry schemaRegistry;
+  private final SchemaRegistry schemaRegistry;
 
   private final RequestHeaderBuilder requestHeaderBuilder = new RequestHeaderBuilder();
 
-  public ConfigResource(KafkaSchemaRegistry schemaRegistry) {
+  public ConfigResource(SchemaRegistry schemaRegistry) {
     this.schemaRegistry = schemaRegistry;
   }
 

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

@@ -19,7 +19,7 @@
 import io.confluent.kafka.schemaregistry.exceptions.SchemaRegistryException;
 import io.confluent.kafka.schemaregistry.exceptions.SchemaRegistryStoreException;
 import io.confluent.kafka.schemaregistry.rest.exceptions.Errors;
-import io.confluent.kafka.schemaregistry.storage.KafkaSchemaRegistry;
+import io.confluent.kafka.schemaregistry.storage.SchemaRegistry;
 import io.confluent.rest.annotations.PerformanceMetric;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.media.ArraySchema;
@@ -42,9 +42,9 @@
            Versions.JSON, Versions.GENERIC_REQUEST})
 public class ContextsResource {
 
-  private final KafkaSchemaRegistry schemaRegistry;
+  private final SchemaRegistry schemaRegistry;
 
-  public ContextsResource(KafkaSchemaRegistry schemaRegistry) {
+  public ContextsResource(SchemaRegistry schemaRegistry) {
     this.schemaRegistry = schemaRegistry;
   }
 

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

@@ -15,8 +15,7 @@
 
 package io.confluent.kafka.schemaregistry.rest.resources;
 
-import static io.confluent.kafka.schemaregistry.storage.KafkaSchemaRegistry.GLOBAL_RESOURCE_NAME;
-
+import static io.confluent.kafka.schemaregistry.storage.SchemaRegistry.GLOBAL_RESOURCE_NAME;
 import com.google.common.base.CharMatcher;
 import io.confluent.kafka.schemaregistry.client.rest.Versions;
 import io.confluent.kafka.schemaregistry.client.rest.entities.Mode;
@@ -28,7 +27,7 @@
 import io.confluent.kafka.schemaregistry.exceptions.UnknownLeaderException;
 import io.confluent.kafka.schemaregistry.rest.exceptions.Errors;
 import io.confluent.kafka.schemaregistry.rest.exceptions.RestInvalidModeException;
-import io.confluent.kafka.schemaregistry.storage.KafkaSchemaRegistry;
+import io.confluent.kafka.schemaregistry.storage.SchemaRegistry;
 import io.confluent.kafka.schemaregistry.utils.QualifiedSubject;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.Parameter;
@@ -64,11 +63,11 @@
 public class ModeResource {
 
   private static final Logger log = LoggerFactory.getLogger(ModeResource.class);
-  private final KafkaSchemaRegistry schemaRegistry;
+  private final SchemaRegistry schemaRegistry;
 
   private final RequestHeaderBuilder requestHeaderBuilder = new RequestHeaderBuilder();
 
-  public ModeResource(KafkaSchemaRegistry schemaRegistry) {
+  public ModeResource(SchemaRegistry schemaRegistry) {
     this.schemaRegistry = schemaRegistry;
   }
 

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

@@ -22,7 +22,7 @@
 import io.confluent.kafka.schemaregistry.exceptions.SchemaRegistryException;
 import io.confluent.kafka.schemaregistry.exceptions.SchemaRegistryStoreException;
 import io.confluent.kafka.schemaregistry.rest.exceptions.Errors;
-import io.confluent.kafka.schemaregistry.storage.KafkaSchemaRegistry;
+import io.confluent.kafka.schemaregistry.storage.SchemaRegistry;
 import io.confluent.rest.annotations.PerformanceMetric;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.Parameter;
@@ -54,9 +54,9 @@
 public class SchemasResource {
 
   private static final Logger log = LoggerFactory.getLogger(SchemasResource.class);
-  private final KafkaSchemaRegistry schemaRegistry;
+  private final SchemaRegistry schemaRegistry;
 
-  public SchemasResource(KafkaSchemaRegistry schemaRegistry) {
+  public SchemasResource(SchemaRegistry schemaRegistry) {
     this.schemaRegistry = schemaRegistry;
   }
 

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

@@ -18,6 +18,7 @@
 import io.confluent.kafka.schemaregistry.client.rest.Versions;
 import io.confluent.kafka.schemaregistry.client.rest.entities.ServerClusterId;
 import io.confluent.kafka.schemaregistry.storage.KafkaSchemaRegistry;
+import io.confluent.kafka.schemaregistry.storage.SchemaRegistry;
 import io.confluent.rest.annotations.PerformanceMetric;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.responses.ApiResponse;
@@ -39,9 +40,10 @@
 public class ServerMetadataResource {
 
   private static final Logger log = LoggerFactory.getLogger(ServerMetadataResource.class);
-  private final KafkaSchemaRegistry schemaRegistry;
 
-  public ServerMetadataResource(KafkaSchemaRegistry schemaRegistry) {
+  private final SchemaRegistry schemaRegistry;
+
+  public ServerMetadataResource(SchemaRegistry schemaRegistry) {
     this.schemaRegistry = schemaRegistry;
   }
 
@@ -53,8 +55,13 @@ public ServerMetadataResource(KafkaSchemaRegistry schemaRegistry) {
   })
   @PerformanceMetric("metadata.id")
   public ServerClusterId getClusterId() {
-    String kafkaClusterId = schemaRegistry.getKafkaClusterId();
     String schemaRegistryClusterId = schemaRegistry.getGroupId();
+
+    String kafkaClusterId = null;
+    if (schemaRegistry instanceof KafkaSchemaRegistry) {
+      kafkaClusterId = ((KafkaSchemaRegistry)schemaRegistry).getKafkaClusterId();
+    } 
+
     return ServerClusterId.of(kafkaClusterId, schemaRegistryClusterId);
   }
 }

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

@@ -15,7 +15,7 @@
 
 package io.confluent.kafka.schemaregistry.rest.resources;
 
-import static io.confluent.kafka.schemaregistry.storage.KafkaSchemaRegistry.GLOBAL_RESOURCE_NAME;
+import static io.confluent.kafka.schemaregistry.storage.SchemaRegistry.GLOBAL_RESOURCE_NAME;
 
 import com.google.common.base.CharMatcher;
 import io.confluent.kafka.schemaregistry.avro.AvroSchema;
@@ -39,7 +39,7 @@
 import io.confluent.kafka.schemaregistry.exceptions.UnknownLeaderException;
 import io.confluent.kafka.schemaregistry.rest.VersionId;
 import io.confluent.kafka.schemaregistry.rest.exceptions.Errors;
-import io.confluent.kafka.schemaregistry.storage.KafkaSchemaRegistry;
+import io.confluent.kafka.schemaregistry.storage.SchemaRegistry;
 import io.confluent.kafka.schemaregistry.utils.QualifiedSubject;
 import io.confluent.rest.annotations.PerformanceMetric;
 import io.swagger.v3.oas.annotations.Operation;
@@ -79,7 +79,7 @@ public class SubjectVersionsResource {
 
   private static final Logger log = LoggerFactory.getLogger(SubjectVersionsResource.class);
 
-  private final KafkaSchemaRegistry schemaRegistry;
+  private final SchemaRegistry schemaRegistry;
 
   private final RequestHeaderBuilder requestHeaderBuilder = new RequestHeaderBuilder();
 
@@ -88,7 +88,7 @@ public class SubjectVersionsResource {
       + "returns the last registered schema under the specified subject. Note that there may be a "
       + "new latest schema that gets registered right after this request is served.";
 
-  public SubjectVersionsResource(KafkaSchemaRegistry registry) {
+  public SubjectVersionsResource(SchemaRegistry registry) {
     this.schemaRegistry = registry;
   }
 

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

@@ -27,7 +27,7 @@
 import io.confluent.kafka.schemaregistry.exceptions.SchemaRegistryTimeoutException;
 import io.confluent.kafka.schemaregistry.exceptions.SubjectNotSoftDeletedException;
 import io.confluent.kafka.schemaregistry.rest.exceptions.Errors;
-import io.confluent.kafka.schemaregistry.storage.KafkaSchemaRegistry;
+import io.confluent.kafka.schemaregistry.storage.SchemaRegistry;
 import io.confluent.kafka.schemaregistry.utils.QualifiedSubject;
 import io.confluent.rest.annotations.PerformanceMetric;
 import io.swagger.v3.oas.annotations.Operation;
@@ -67,10 +67,10 @@
 public class SubjectsResource {
 
   private static final Logger log = LoggerFactory.getLogger(SubjectsResource.class);
-  private final KafkaSchemaRegistry schemaRegistry;
+  private final SchemaRegistry schemaRegistry;
   private final RequestHeaderBuilder requestHeaderBuilder = new RequestHeaderBuilder();
 
-  public SubjectsResource(KafkaSchemaRegistry schemaRegistry) {
+  public SubjectsResource(SchemaRegistry schemaRegistry) {
     this.schemaRegistry = schemaRegistry;
   }
 

core/src/main/java/io/confluent/kafka/schemaregistry/storage/KafkaSchemaRegistry.java

@@ -104,8 +104,6 @@ public class KafkaSchemaRegistry implements SchemaRegistry, LeaderAwareSchemaReg
    * Schema versions under a particular subject are indexed from MIN_VERSION.
    */
   public static final int MIN_VERSION = 1;
-  // Subject name under which global permissions are stored.
-  public static final String GLOBAL_RESOURCE_NAME = "__GLOBAL";
   public static final int MAX_VERSION = Integer.MAX_VALUE;
   private static final Logger log = LoggerFactory.getLogger(KafkaSchemaRegistry.class);
 
@@ -588,6 +586,7 @@ private boolean isReadOnlyMode(String subject) throws SchemaRegistryStoreExcepti
     return subjectMode == Mode.READONLY || subjectMode == Mode.READONLY_OVERRIDE;
   }
 
+  @Override
   public int registerOrForward(String subject,
                                Schema schema,
                                boolean normalize,
@@ -1711,6 +1710,7 @@ public void setMode(String subject, Mode mode, boolean force)
     }
   }
 
+  @Override
   public void setModeOrForward(String subject, Mode mode, boolean force,
       Map<String, String> headerProperties)
       throws SchemaRegistryStoreException, SchemaRegistryRequestForwardingException,