Load Kafka connection config from properties file (#68)

Author: gquintana

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

@@ -23,9 +23,13 @@
         description = "Manage Kafka resources with a desired state file.")
 public class MainCommand implements Callable<Integer> {
 
+    @Option(names = {"-c", "--command-config"}, paramLabel = "<file>",
+            description = "Command config properties file.")
+    private File configFile;
+
     @Option(names = {"-f", "--file"}, paramLabel = "<file>",
             description = "Specify the desired state file.", defaultValue = "state.yaml")
-    private File file;
+    private File stateFile;
 
     @Option(names = {"--no-delete"}, description = "Disable the ability to delete resources.")
     private boolean deleteDisabled = false;
@@ -60,8 +64,12 @@ public boolean isVerboseRequested() {
         return verboseRequested;
     }
 
-    public File getFile() {
-        return file;
+    public File getConfigFile() {
+        return configFile;
+    }
+
+    public File getStateFile() {
+        return stateFile;
     }
 
     public boolean isDeleteDisabled() {

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

@@ -2,6 +2,7 @@
 
 import ch.qos.logback.classic.Level;
 import ch.qos.logback.classic.Logger;
+import com.devshawn.kafka.gitops.config.KafkaGitopsConfig;
 import com.devshawn.kafka.gitops.config.KafkaGitopsConfigLoader;
 import com.devshawn.kafka.gitops.config.ManagerConfig;
 import com.devshawn.kafka.gitops.domain.confluent.ServiceAccount;
@@ -60,7 +61,8 @@ public StateManager(ManagerConfig managerConfig, ParserService parserService) {
         initializeLogger(managerConfig.isVerboseRequested());
         this.managerConfig = managerConfig;
         this.objectMapper = initializeObjectMapper();
-        this.kafkaService = new KafkaService(KafkaGitopsConfigLoader.load());
+        KafkaGitopsConfig config = KafkaGitopsConfigLoader.load();
+        this.kafkaService = new KafkaService(config);
         this.parserService = parserService;
         this.roleService = new RoleService();
         this.confluentCloudService = new ConfluentCloudService(objectMapper);

src/main/java/com/devshawn/kafka/gitops/cli/AccountCommand.java

@@ -24,7 +24,7 @@ public class AccountCommand implements Callable<Integer> {
     public Integer call() {
         try {
             System.out.println("Creating service accounts...\n");
-            ParserService parserService = new ParserService(parent.getFile());
+            ParserService parserService = new ParserService(parent.getStateFile());
             StateManager stateManager = new StateManager(generateStateManagerConfig(), parserService);
             stateManager.createServiceAccounts();
             return 0;
@@ -46,7 +46,8 @@ private ManagerConfig generateStateManagerConfig() {
                 .setDeleteDisabled(parent.isDeleteDisabled())
                 .setIncludeUnchangedEnabled(false)
                 .setSkipAclsDisabled(parent.areAclsDisabled())
-                .setStateFile(parent.getFile())
+                .setConfigFile(parent.getConfigFile())
+                .setStateFile(parent.getStateFile())
                 .build();
     }
 }

src/main/java/com/devshawn/kafka/gitops/cli/ApplyCommand.java

@@ -31,7 +31,7 @@ public class ApplyCommand implements Callable<Integer> {
     public Integer call() {
         try {
             System.out.println("Executing apply...\n");
-            ParserService parserService = new ParserService(parent.getFile());
+            ParserService parserService = new ParserService(parent.getStateFile());
             StateManager stateManager = new StateManager(generateStateManagerConfig(), parserService);
             DesiredPlan desiredPlan = stateManager.apply();
             LogUtil.printApplyOverview(PlanUtil.getOverview(desiredPlan, parent.isDeleteDisabled(), parent.areAclsDisabled()));
@@ -55,7 +55,8 @@ private ManagerConfig generateStateManagerConfig() {
                 .setDeleteDisabled(parent.isDeleteDisabled())
                 .setIncludeUnchangedEnabled(false)
                 .setSkipAclsDisabled(parent.areAclsDisabled())
-                .setStateFile(parent.getFile())
+                .setConfigFile(parent.getConfigFile())
+                .setStateFile(parent.getStateFile())
                 .setNullablePlanFile(planFile)
                 .build();
     }

src/main/java/com/devshawn/kafka/gitops/cli/PlanCommand.java

@@ -33,7 +33,7 @@ public class PlanCommand implements Callable<Integer> {
     public Integer call() {
         try {
             System.out.println("Generating execution plan...\n");
-            ParserService parserService = new ParserService(parent.getFile());
+            ParserService parserService = new ParserService(parent.getStateFile());
             StateManager stateManager = new StateManager(generateStateManagerConfig(), parserService);
             DesiredPlan desiredPlan = stateManager.plan();
             LogUtil.printPlan(desiredPlan, parent.isDeleteDisabled(), parent.areAclsDisabled());
@@ -58,7 +58,8 @@ private ManagerConfig generateStateManagerConfig() {
                 .setVerboseRequested(parent.isVerboseRequested())
                 .setDeleteDisabled(parent.isDeleteDisabled())
                 .setIncludeUnchangedEnabled(includeUnchanged)
-                .setStateFile(parent.getFile())
+                .setConfigFile(parent.getConfigFile())
+                .setStateFile(parent.getStateFile())
                 .setSkipAclsDisabled(parent.areAclsDisabled())
                 .setNullablePlanFile(outputFile)
                 .build();

src/main/java/com/devshawn/kafka/gitops/cli/ValidateCommand.java

@@ -21,7 +21,7 @@ public class ValidateCommand implements Callable<Integer> {
     @Override
     public Integer call() {
         try {
-            ParserService parserService = new ParserService(parent.getFile());
+            ParserService parserService = new ParserService(parent.getStateFile());
             StateManager stateManager = new StateManager(generateStateManagerConfig(), parserService);
             stateManager.getAndValidateStateFile();
             LogUtil.printValidationResult("Successfully validated the desired state file.", true);
@@ -38,7 +38,8 @@ private ManagerConfig generateStateManagerConfig() {
                 .setDeleteDisabled(parent.isDeleteDisabled())
                 .setIncludeUnchangedEnabled(false)
                 .setSkipAclsDisabled(parent.areAclsDisabled())
-                .setStateFile(parent.getFile())
+                .setConfigFile(parent.getConfigFile())
+                .setStateFile(parent.getStateFile())
                 .build();
     }
 }

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

@@ -5,21 +5,44 @@
 import org.apache.kafka.common.config.SaslConfigs;
 import org.slf4j.LoggerFactory;
 
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Properties;
 import java.util.concurrent.atomic.AtomicReference;
 
 public class KafkaGitopsConfigLoader {
 
     private static org.slf4j.Logger log = LoggerFactory.getLogger(KafkaGitopsConfigLoader.class);
 
     public static KafkaGitopsConfig load() {
+        return load(null);
+    }
+
+    public static KafkaGitopsConfig load(File configFile) {
         KafkaGitopsConfig.Builder builder = new KafkaGitopsConfig.Builder();
-        setConfig(builder);
+        setConfigFromFile(configFile, builder);
+        setConfigFromEnvironment(builder);
         return builder.build();
     }
 
-    private static void setConfig(KafkaGitopsConfig.Builder builder) {
+    private static void setConfigFromFile(File configFile, KafkaGitopsConfig.Builder builder) {
+        if (configFile == null) {
+            return;
+        }
+        try(InputStream inputStream = new FileInputStream(configFile)) {
+            Properties properties = new Properties();
+            properties.load(inputStream);
+            properties.forEach( (k, v) -> builder.putConfig(k.toString(), v));
+        } catch (IOException ioExc) {
+            log.error("Failed to load config from " + configFile, ioExc);
+        }
+    }
+
+    private static void setConfigFromEnvironment(KafkaGitopsConfig.Builder builder) {
         Map<String, Object> config = new HashMap<>();
         AtomicReference<String> username = new AtomicReference<>();
         AtomicReference<String> password = new AtomicReference<>();

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

@@ -18,6 +18,8 @@ public interface ManagerConfig {
 
     boolean isSkipAclsDisabled();
 
+    File getConfigFile();
+
     File getStateFile();
 
     Optional<File> getPlanFile();

src/test/groovy/com/devshawn/kafka/gitops/config/KafkaGitopsConfigLoaderSpec.groovy

@@ -1,6 +1,6 @@
 package com.devshawn.kafka.gitops.config
 
-
+import org.apache.kafka.clients.CommonClientConfigs
 import org.apache.kafka.common.config.SaslConfigs
 import org.junit.ClassRule
 import org.junit.contrib.java.lang.system.EnvironmentVariables
@@ -45,5 +45,17 @@ class KafkaGitopsConfigLoaderSpec extends Specification {
         config.config.get(SaslConfigs.SASL_JAAS_CONFIG) == "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"te\\\"st\" password=\"te\\\"st-secr\\\"et\";"
     }
 
+    void 'test command config file'() {
+        setup:
+        File configFile = new File(getClass().getResource("/command.properties").toURI())
+
+        when:
+        KafkaGitopsConfig config = KafkaGitopsConfigLoader.load(configFile)
+
+        then:
+        config.config.get(CommonClientConfigs.CLIENT_ID_CONFIG) == "kafka-gitops"
+        config.config.get(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG) == "localhost:9092"
+        config.config.get(SaslConfigs.SASL_MECHANISM) == "PLAIN"
+    }
 
 }

src/test/resources/command.properties

@@ -0,0 +1,2 @@
+bootstrap.servers=commande.9092
+client.id=kafka-gitops