Specific authentication variables are not needed

Schema registry client already have everything for that:
basic.auth.credentials.source=USER_INFO
basic.auth.user.info=fred:letmein

For kafka-gitops, you can provide everything using env variable and
prefix them by SCHEMA_REGISTRY_.

Example:

SCHEMA_REGISTRY_SCHEMA_REGISTRY_SSL_TRUSTSTORE_LOCATION=/ca.jks
SCHEMA_REGISTRY_SCHEMA_REGISTRY_SSL_TRUSTSTORE_PASSWORD=changeme
SCHEMA_REGISTRY_BASIC_AUTH_CREDENTIALS_SOURCE=USER_INFO
SCHEMA_REGISTRY_BASIC_AUTH_USER_INFO=fred:letmein

Author: Jerome Revillard

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

@@ -12,9 +12,6 @@ public class SchemaRegistryConfigLoader {
 
     public static final String SCHEMA_REGISTRY_URL_KEY = "SCHEMA_REGISTRY_URL";
     public static final String SCHEMA_DIRECTORY_KEY = "SCHEMA_DIRECTORY";
-    public static final String SCHEMA_REGISTRY_SASL_JAAS_USERNAME_KEY = "SCHEMA_REGISTRY_SASL_JAAS_USERNAME";
-    public static final String SCHEMA_REGISTRY_SASL_JAAS_PASSWORD_KEY = "SCHEMA_REGISTRY_SASL_JAAS_PASSWORD";
-    public static final String SCHEMA_REGISTRY_SASL_CONFIG_KEY = "SCHEMA_REGISTRY_SASL_CONFIG";
 
     private SchemaRegistryConfigLoader() {}
 
@@ -26,17 +23,11 @@ public static SchemaRegistryConfig load() {
 
     private static void setConfig(SchemaRegistryConfig.Builder builder) {
         Map<String, Object> config = new HashMap<>();
-        AtomicReference<String> username = new AtomicReference<>();
-        AtomicReference<String> password = new AtomicReference<>();
 
         Map<String, String> environment = System.getenv();
 
         environment.forEach((key, value) -> {
-            if (key.equals(SCHEMA_REGISTRY_SASL_JAAS_USERNAME_KEY)) {
-                username.set(value);
-            } else if (key.equals(SCHEMA_REGISTRY_SASL_JAAS_PASSWORD_KEY)) {
-                password.set(value);
-            } else if (key.equals(SCHEMA_REGISTRY_URL_KEY)) {
+            if (key.equals(SCHEMA_REGISTRY_URL_KEY)) {
                 config.put(SCHEMA_REGISTRY_URL_KEY, value);
             } else if (key.equals(SCHEMA_DIRECTORY_KEY)) {
                 config.put(SCHEMA_DIRECTORY_KEY, value);
@@ -47,7 +38,6 @@ private static void setConfig(SchemaRegistryConfig.Builder builder) {
         });
 
         handleDefaultConfig(config);
-        handleAuthentication(username, password, config);
 
         log.info("Schema Registry Config: {}", config);
 
@@ -62,29 +52,8 @@ private static void handleDefaultConfig(Map<String, Object> config) {
             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);
+            log.info("{} not set. Defaulting to current working directory: {}", SCHEMA_DIRECTORY_KEY, CURRENT_WORKING_DIR);
             config.put(SCHEMA_DIRECTORY_KEY, CURRENT_WORKING_DIR);
         }
     }
-
-    private static void handleAuthentication(AtomicReference<String> username, AtomicReference<String> password, Map<String, Object> config) {
-        if (username.get() != null && password.get() != null) {
-            String loginModule = "org.apache.kafka.common.security.plain.PlainLoginModule";
-            String value = String.format("%s required username=\"%s\" password=\"%s\";",
-                    loginModule, escape(username.get()), escape(password.get()));
-            config.put(SCHEMA_REGISTRY_SASL_CONFIG_KEY, value);
-        } else {
-          if(config.get(SCHEMA_REGISTRY_SASL_CONFIG_KEY) == null) {
-              log.info("{} or {} not set. No authentication configured for the Schema Registry",
-                  SCHEMA_REGISTRY_SASL_JAAS_USERNAME_KEY, SCHEMA_REGISTRY_SASL_JAAS_PASSWORD_KEY);
-          }
-        }
-    }
-
-    private static String escape(String value) {
-        if (value != null) {
-            return value.replace("\"", "\\\"");
-        }
-        return null;
-    }
 }

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

@@ -3,11 +3,9 @@
 import java.io.IOException;
 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 org.apache.kafka.common.config.SaslConfigs;
 import com.devshawn.kafka.gitops.config.SchemaRegistryConfig;
 import com.devshawn.kafka.gitops.config.SchemaRegistryConfigLoader;
 import com.devshawn.kafka.gitops.domain.plan.SchemaPlan;
@@ -25,7 +23,6 @@
 import io.confluent.kafka.schemaregistry.client.rest.RestService;
 import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaReference;
 import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException;
-import io.confluent.kafka.schemaregistry.client.security.basicauth.SaslBasicAuthCredentialProvider;
 import io.confluent.kafka.schemaregistry.json.JsonSchemaProvider;
 import io.confluent.kafka.schemaregistry.protobuf.ProtobufSchemaProvider;
 
@@ -182,14 +179,6 @@ public boolean deepEquals(SchemaDetails schemaDetails, SchemaMetadata schemaMeta
 
     public CachedSchemaRegistryClient createSchemaRegistryClient() {
         RestService restService = new RestService(config.getConfig().get(SchemaRegistryConfigLoader.SCHEMA_REGISTRY_URL_KEY).toString());
-        if(config.getConfig().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.getConfig()
-                .get(SchemaRegistryConfigLoader.SCHEMA_REGISTRY_SASL_CONFIG_KEY).toString());
-            saslBasicAuthCredentialProvider.configure(clientConfig);
-            restService.setBasicAuthCredentialProvider(saslBasicAuthCredentialProvider);
-        }
         return new CachedSchemaRegistryClient(restService, 10, config.getConfig());
     }