Schema Registry management only if registry url provided

Author: Jerome Revillard

src/main/java/com/devshawn/kafka/gitops/StateManager.java

@@ -97,7 +97,9 @@ private DesiredPlan generatePlan() {
             planManager.planAcls(desiredState, desiredPlan);
         }
         planManager.planTopics(desiredState, desiredPlan);
-        planManager.planSchemas(desiredState, desiredPlan);
+        if (schemaRegistryService.isEnabled()) {
+          planManager.planSchemas(desiredState, desiredPlan);
+        }
         return desiredPlan.build();
     }
 
@@ -113,7 +115,9 @@ public DesiredPlan apply() {
         if (!managerConfig.isSkipAclsDisabled()) {
             applyManager.applyAcls(desiredPlan);
         }
-        applyManager.applySchemas(desiredPlan);
+        if (schemaRegistryService.isEnabled()) {
+          applyManager.applySchemas(desiredPlan);
+        }
 
         return desiredPlan;
     }

src/main/java/com/devshawn/kafka/gitops/config/SchemaRegistryConfigLoader.java

@@ -1,10 +1,8 @@
 package com.devshawn.kafka.gitops.config;
 
-import org.slf4j.LoggerFactory;
-
 import java.util.HashMap;
 import java.util.Map;
-import java.util.concurrent.atomic.AtomicReference;
+import org.slf4j.LoggerFactory;
 
 public class SchemaRegistryConfigLoader {
 
@@ -45,12 +43,7 @@ private static void setConfig(SchemaRegistryConfig.Builder builder) {
     }
 
     private static void handleDefaultConfig(Map<String, Object> config) {
-        final String DEFAULT_URL = "http://localhost:8081";
         final String CURRENT_WORKING_DIR = System.getProperty("user.dir");
-        if (!config.containsKey(SCHEMA_REGISTRY_URL_KEY)) {
-            log.info("{} not set.  Using default value of {}", SCHEMA_REGISTRY_URL_KEY, DEFAULT_URL);
-            config.put(SCHEMA_REGISTRY_URL_KEY, DEFAULT_URL);
-        }
         if (!config.containsKey(SCHEMA_DIRECTORY_KEY)) {
             log.info("{} not set. Defaulting to current working directory: {}", SCHEMA_DIRECTORY_KEY, CURRENT_WORKING_DIR);
             config.put(SCHEMA_DIRECTORY_KEY, CURRENT_WORKING_DIR);

src/main/java/com/devshawn/kafka/gitops/service/SchemaRegistryService.java

@@ -4,7 +4,6 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
-import java.util.Map;
 import java.util.Optional;
 import com.devshawn.kafka.gitops.config.SchemaRegistryConfig;
 import com.devshawn.kafka.gitops.config.SchemaRegistryConfigLoader;
@@ -34,8 +33,8 @@ public SchemaRegistryService(SchemaRegistryConfig config) {
         this.config = config;
     }
 
-    public Map<String, Object> getConfig() {
-      return Collections.unmodifiableMap(config.getConfig());
+    public boolean isEnabled() {
+      return this.config.getConfig().containsKey(SchemaRegistryConfigLoader.SCHEMA_REGISTRY_URL_KEY);
     }
 
     public List<String> getAllSubjects() {

src/test/groovy/com/devshawn/kafka/gitops/ApplyCommandIntegrationSpec.groovy

@@ -64,8 +64,7 @@ class ApplyCommandIntegrationSpec extends Specification {
                 "custom-group-id-connect",
                 "custom-application-id-streams",
                 "custom-storage-topic",
-                "custom-storage-topics",
-                "schema-registry-simple"
+                "custom-storage-topics"
         ]
     }
 

src/test/groovy/com/devshawn/kafka/gitops/PlanCommandIntegrationSpec.groovy

@@ -105,6 +105,7 @@ class PlanCommandIntegrationSpec extends Specification {
 
     void 'test various valid plans with seed - #planName'() {
         setup:
+        TestUtils.cleanUpKafkaCluster()
         TestUtils.seedKafkaCluster()
         String planOutputFile = "/tmp/plan.json"
         String file = TestUtils.getResourceFilePath("plans/${planName}.yaml")
@@ -144,6 +145,7 @@ class PlanCommandIntegrationSpec extends Specification {
 
     void 'test include unchanged flag - #planNam #includeUnchanged'() {
         setup:
+        TestUtils.cleanUpKafkaCluster()
         TestUtils.seedKafkaCluster()
         String planOutputFile = "/tmp/plan.json"
         String file = TestUtils.getResourceFilePath("plans/${planName}.yaml")
@@ -260,6 +262,7 @@ class PlanCommandIntegrationSpec extends Specification {
 
     void 'test plan that has no changes - #includeUnchanged'() {
         setup:
+        TestUtils.cleanUpKafkaCluster()
         TestUtils.seedKafkaCluster()
         ByteArrayOutputStream out = new ByteArrayOutputStream()
         PrintStream oldOut = System.out

src/test/groovy/com/devshawn/kafka/gitops/TestUtils.groovy

@@ -4,7 +4,12 @@ import java.nio.file.Paths
 import org.apache.kafka.clients.CommonClientConfigs
 import org.apache.kafka.clients.admin.AdminClient
 import org.apache.kafka.clients.admin.NewTopic
-import org.apache.kafka.common.acl.*
+import org.apache.kafka.common.acl.AccessControlEntry;
+import org.apache.kafka.common.acl.AccessControlEntryFilter;
+import org.apache.kafka.common.acl.AclBinding;
+import org.apache.kafka.common.acl.AclBindingFilter;
+import org.apache.kafka.common.acl.AclOperation;
+import org.apache.kafka.common.acl.AclPermissionType;
 import org.apache.kafka.common.config.SaslConfigs
 import org.apache.kafka.common.resource.PatternType
 import org.apache.kafka.common.resource.ResourcePattern
@@ -13,19 +18,12 @@ import org.apache.kafka.common.resource.ResourceType
 import com.devshawn.kafka.gitops.config.SchemaRegistryConfigLoader
 import com.devshawn.kafka.gitops.enums.SchemaCompatibility
 import com.devshawn.kafka.gitops.enums.SchemaType
-import com.devshawn.kafka.gitops.exception.ValidationException
 import com.devshawn.kafka.gitops.service.SchemaRegistryService
-import groovy.swing.factory.CollectionFactory
 import io.confluent.kafka.schemaregistry.AbstractSchemaProvider
 import io.confluent.kafka.schemaregistry.ParsedSchema
-import io.confluent.kafka.schemaregistry.SchemaProvider
-import io.confluent.kafka.schemaregistry.avro.AvroSchemaProvider
 import io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient
 import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient
 import io.confluent.kafka.schemaregistry.client.rest.RestService
-import io.confluent.kafka.schemaregistry.client.security.basicauth.SaslBasicAuthCredentialProvider
-import io.confluent.kafka.schemaregistry.json.JsonSchemaProvider
-import io.confluent.kafka.schemaregistry.protobuf.ProtobufSchemaProvider
 import spock.util.concurrent.PollingConditions
 
 class TestUtils {
@@ -196,16 +194,8 @@ class TestUtils {
     }
 
     static CachedSchemaRegistryClient getSchemaRegistryClient() {
-        Map<String, Object> config = SchemaRegistryConfigLoader.load().getConfig();
-        RestService restService = new RestService(config.get(SchemaRegistryConfigLoader.SCHEMA_REGISTRY_URL_KEY).toString())
-        if(config.get(SchemaRegistryConfigLoader.SCHEMA_REGISTRY_SASL_CONFIG_KEY) != null) {
-            SaslBasicAuthCredentialProvider saslBasicAuthCredentialProvider = new SaslBasicAuthCredentialProvider()
-            Map<String, Object> clientConfig = new HashMap<>()
-            clientConfig.put(SaslConfigs.SASL_JAAS_CONFIG, config
-                .get(SchemaRegistryConfigLoader.SCHEMA_REGISTRY_SASL_CONFIG_KEY).toString())
-            saslBasicAuthCredentialProvider.configure(clientConfig)
-            restService.setBasicAuthCredentialProvider(saslBasicAuthCredentialProvider)
-        }
-        return  new CachedSchemaRegistryClient(restService, 10);
+          Map<String, Object> config = SchemaRegistryConfigLoader.load().getConfig();
+          RestService restService = new RestService(config.get(SchemaRegistryConfigLoader.SCHEMA_REGISTRY_URL_KEY).toString())
+        return  new CachedSchemaRegistryClient(restService, 10, config);
     }
 }