feat: Implement --skip-topics to skip any topic-related actions

README.md

@@ -39,12 +39,13 @@ Documentation on how to install and use this tool can be found on our [documenta
 Run `kafka-gitops` to view the help output.
 
 ```bash
-Usage: kafka-gitops [-hvV] [--no-delete] [--skip-acls] [-f=<file>] [COMMAND]
+Usage: kafka-gitops [-hvV] [--no-delete] [--skip-acls] [--skip-topics] [-f=<file>] [COMMAND]
 Manage Kafka resources with a desired state file.
   -f, --file=<file>   Specify the desired state file.
   -h, --help          Display this help message.
       --no-delete     Disable the ability to delete resources.
       --skip-acls     Do not take ACLs into account during plans or applies.
+      --skip-topics   Do not take topics into account during plans or applies.
   -v, --verbose       Show more detail during execution.
   -V, --version       Print the current version of this tool.
 Commands:

docs/quick-start.md

@@ -99,6 +99,12 @@ If running against a Kafka cluster with no authorizer configured or if you simpl
 kafka-gitops --skip-acls plan
 ```
 
+Conversely, if you're only interested in managing ACLs you can ignore changes to topics completely. This can be done by running:
+
+```bash
+kafka-gitops --skip-topics plan
+```
+
 ## Apply
 
 To execute a plan against the cluster, we use the apply command.

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

@@ -37,6 +37,9 @@ public class MainCommand implements Callable<Integer> {
     @Option(names = {"--skip-acls"}, description = "Do not take ACLs into account during plans or applies.")
     private boolean skipAcls = false;
 
+    @Option(names = {"--skip-topics"}, description = "Do not take topics into account during plans or applies.")
+    private boolean skipTopics = false;
+
     @Option(names = {"-h", "--help"}, usageHelp = true, description = "Display this help message.")
     private boolean helpRequested = false;
 
@@ -80,6 +83,10 @@ public boolean areAclsDisabled() {
         return skipAcls;
     }
 
+    public boolean areTopicsDisabled() {
+        return skipTopics;
+    }
+
     public static void main(String[] args) {
         int exitCode = new CommandLine(new MainCommand()).execute(args);
         System.exit(exitCode);

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

@@ -81,7 +81,10 @@ public DesiredStateFile getAndValidateStateFile() {
     public DesiredPlan plan() {
         DesiredPlan desiredPlan = generatePlan();
         planManager.writePlanToFile(desiredPlan);
-        planManager.validatePlanHasChanges(desiredPlan, managerConfig.isDeleteDisabled(), managerConfig.isSkipAclsDisabled());
+        planManager.validatePlanHasChanges(desiredPlan,
+                managerConfig.isDeleteDisabled(),
+                managerConfig.isSkipAclsDisabled(),
+                managerConfig.isSkipTopicsDisabled());
         return desiredPlan;
     }
 
@@ -91,7 +94,11 @@ private DesiredPlan generatePlan() {
         if (!managerConfig.isSkipAclsDisabled()) {
             planManager.planAcls(desiredState, desiredPlan);
         }
-        planManager.planTopics(desiredState, desiredPlan);
+
+        if (!managerConfig.isSkipTopicsDisabled()) {
+            planManager.planTopics(desiredState, desiredPlan);
+        }
+
         return desiredPlan.build();
     }
 
@@ -101,9 +108,13 @@ public DesiredPlan apply() {
             desiredPlan = generatePlan();
         }
 
-        planManager.validatePlanHasChanges(desiredPlan, managerConfig.isDeleteDisabled(), managerConfig.isSkipAclsDisabled());
+        planManager.validatePlanHasChanges(desiredPlan, managerConfig.isDeleteDisabled(),
+                managerConfig.isSkipAclsDisabled(), managerConfig.isSkipTopicsDisabled());
+
+        if (!managerConfig.isSkipTopicsDisabled()) {
+            applyManager.applyTopics(desiredPlan);
+        }
 
-        applyManager.applyTopics(desiredPlan);
         if (!managerConfig.isSkipAclsDisabled()) {
             applyManager.applyAcls(desiredPlan);
         }

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

@@ -46,6 +46,7 @@ private ManagerConfig generateStateManagerConfig() {
                 .setDeleteDisabled(parent.isDeleteDisabled())
                 .setIncludeUnchangedEnabled(false)
                 .setSkipAclsDisabled(parent.areAclsDisabled())
+                .setSkipTopicsDisabled(parent.areTopicsDisabled())
                 .setNullableConfigFile(parent.getConfigFile())
                 .setStateFile(parent.getStateFile())
                 .build();

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

@@ -34,7 +34,10 @@ public Integer call() {
             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()));
+            LogUtil.printApplyOverview(PlanUtil.getOverview(desiredPlan,
+                    parent.isDeleteDisabled(),
+                    parent.areAclsDisabled(),
+                    parent.areTopicsDisabled()));
             return 0;
         } catch (PlanIsUpToDateException ex) {
             LogUtil.printNoChangesMessage();
@@ -55,6 +58,7 @@ private ManagerConfig generateStateManagerConfig() {
                 .setDeleteDisabled(parent.isDeleteDisabled())
                 .setIncludeUnchangedEnabled(false)
                 .setSkipAclsDisabled(parent.areAclsDisabled())
+                .setSkipTopicsDisabled(parent.areTopicsDisabled())
                 .setNullableConfigFile(parent.getConfigFile())
                 .setStateFile(parent.getStateFile())
                 .setNullablePlanFile(planFile)

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

@@ -36,7 +36,7 @@ public Integer call() {
             ParserService parserService = new ParserService(parent.getStateFile());
             StateManager stateManager = new StateManager(generateStateManagerConfig(), parserService);
             DesiredPlan desiredPlan = stateManager.plan();
-            LogUtil.printPlan(desiredPlan, parent.isDeleteDisabled(), parent.areAclsDisabled());
+            LogUtil.printPlan(desiredPlan, parent.isDeleteDisabled(), parent.areAclsDisabled(), parent.areTopicsDisabled());
             return 0;
         } catch (PlanIsUpToDateException ex) {
             LogUtil.printNoChangesMessage();
@@ -61,6 +61,7 @@ private ManagerConfig generateStateManagerConfig() {
                 .setNullableConfigFile(parent.getConfigFile())
                 .setStateFile(parent.getStateFile())
                 .setSkipAclsDisabled(parent.areAclsDisabled())
+                .setSkipTopicsDisabled(parent.areTopicsDisabled())
                 .setNullablePlanFile(outputFile)
                 .build();
     }

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

@@ -38,6 +38,7 @@ private ManagerConfig generateStateManagerConfig() {
                 .setDeleteDisabled(parent.isDeleteDisabled())
                 .setIncludeUnchangedEnabled(false)
                 .setSkipAclsDisabled(parent.areAclsDisabled())
+                .setSkipTopicsDisabled(parent.areTopicsDisabled())
                 .setNullableConfigFile(parent.getConfigFile())
                 .setStateFile(parent.getStateFile())
                 .build();

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

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

src/main/java/com/devshawn/kafka/gitops/manager/PlanManager.java

@@ -171,8 +171,11 @@ public void planAcls(DesiredState desiredState, DesiredPlan.Builder desiredPlan)
         });
     }
 
-    public void validatePlanHasChanges(DesiredPlan desiredPlan, boolean deleteDisabled, boolean skipAclsDisabled) {
-        PlanOverview planOverview = PlanUtil.getOverview(desiredPlan, deleteDisabled, skipAclsDisabled);
+    public void validatePlanHasChanges(DesiredPlan desiredPlan,
+                                       boolean deleteDisabled,
+                                       boolean skipAclsDisabled,
+                                       boolean skipTopicsDisabled) {
+        PlanOverview planOverview = PlanUtil.getOverview(desiredPlan, deleteDisabled, skipAclsDisabled, skipTopicsDisabled);
         if (planOverview.getAdd() == 0 && planOverview.getUpdate() == 0 && planOverview.getRemove() == 0) {
             throw new PlanIsUpToDateException();
         }

src/main/java/com/devshawn/kafka/gitops/util/LogUtil.java

@@ -16,8 +16,11 @@ public final class LogUtil {
     private LogUtil() {
     }
 
-    public static void printPlan(DesiredPlan desiredPlan, boolean deleteDisabled, boolean skipAclsDisabled) {
-        PlanOverview planOverview = PlanUtil.getOverview(desiredPlan, deleteDisabled, skipAclsDisabled);
+    public static void printPlan(DesiredPlan desiredPlan,
+                                 boolean deleteDisabled,
+                                 boolean skipAclsDisabled,
+                                 boolean skipTopicsDisabled) {
+        PlanOverview planOverview = PlanUtil.getOverview(desiredPlan, deleteDisabled, skipAclsDisabled, skipTopicsDisabled);
 
         printLegend(planOverview);
 
@@ -27,7 +30,7 @@ public static void printPlan(DesiredPlan desiredPlan, boolean deleteDisabled, bo
         printAclOverview(desiredPlan, deleteDisabled);
         desiredPlan.getAclPlans().forEach(LogUtil::printAclPlan);
 
-        printOverview(desiredPlan, deleteDisabled, skipAclsDisabled);
+        printOverview(desiredPlan, deleteDisabled, skipAclsDisabled, skipTopicsDisabled);
     }
 
     public static void printValidationResult(String message, boolean success) {
@@ -132,8 +135,11 @@ public static void printPostApply() {
      * Helpers
      */
 
-    private static void printOverview(DesiredPlan desiredPlan, boolean deleteDisabled, boolean skipAclsDisabled) {
-        PlanOverview planOverview = PlanUtil.getOverview(desiredPlan, deleteDisabled, skipAclsDisabled);
+    private static void printOverview(DesiredPlan desiredPlan,
+                                      boolean deleteDisabled,
+                                      boolean skipAclsDisabled,
+                                      boolean skipTopicsDisabled) {
+        PlanOverview planOverview = PlanUtil.getOverview(desiredPlan, deleteDisabled, skipAclsDisabled, skipTopicsDisabled);
         System.out.printf("%s: %s, %s, %s.%n", bold("Plan"), toCreate(planOverview.getAdd()),
                 toUpdate(planOverview.getUpdate()), toDelete(planOverview.getRemove()));
     }

src/main/java/com/devshawn/kafka/gitops/util/PlanUtil.java

@@ -9,10 +9,16 @@
 
 public class PlanUtil {
 
-    public static PlanOverview getOverview(DesiredPlan desiredPlan, boolean deleteDisabled, boolean skipAclsDisabled) {
+    public static PlanOverview getOverview(DesiredPlan desiredPlan,
+                                           boolean deleteDisabled,
+                                           boolean skipAclsDisabled,
+                                           boolean skipTopicsDisabled) {
         EnumMap<PlanAction, Long> map = getPlanActionMap();
-        desiredPlan.getTopicPlans().forEach(it -> addToMap(map, it.getAction(), deleteDisabled));
-        if(!skipAclsDisabled) {
+        if (!skipTopicsDisabled) {
+            desiredPlan.getTopicPlans().forEach(it -> addToMap(map, it.getAction(), deleteDisabled));
+        }
+
+        if (!skipAclsDisabled) {
             desiredPlan.getAclPlans().forEach(it -> addToMap(map, it.getAction(), deleteDisabled));
         }
         return buildPlanOverview(map);

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

@@ -87,6 +87,31 @@ class ApplyCommandIntegrationSpec extends Specification {
         ]
     }
 
+    void 'test skip-topics flag'() {
+        setup:
+        ByteArrayOutputStream out = new ByteArrayOutputStream()
+        PrintStream oldOut = System.out
+        System.setOut(new PrintStream(out))
+        String file = TestUtils.getResourceFilePath("plans/${planFile}-plan.json")
+        MainCommand mainCommand = new MainCommand()
+        CommandLine cmd = new CommandLine(mainCommand)
+
+        when:
+        int exitCode = cmd.execute("-f", file, "--skip-topics", "apply", "-p", file)
+
+        then:
+        exitCode == 0
+        out.toString() == TestUtils.getResourceFileContent("plans/${planFile}-apply-output.txt")
+
+        cleanup:
+        System.setOut(oldOut)
+
+        where:
+        planFile << [
+                "skip-topics-apply"
+        ]
+    }
+
     void 'test various valid applies with seed - #planFile #deleteDisabled'() {
         setup:
         TestUtils.seedCluster()

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

@@ -97,6 +97,32 @@ class PlanCommandIntegrationSpec extends Specification {
         ]
     }
 
+    void 'test skip-topics flag'() {
+        setup:
+        String planOutputFile = "/tmp/plan.json"
+        String file = TestUtils.getResourceFilePath("plans/${planName}.yaml")
+        MainCommand mainCommand = new MainCommand()
+        CommandLine cmd = new CommandLine(mainCommand)
+
+        when:
+        int exitCode = cmd.execute("-f", file, "--skip-topics", "plan", "-o", planOutputFile)
+
+        then:
+        exitCode == 0
+
+        when:
+        String actualPlan = TestUtils.getFileContent(planOutputFile)
+        String expectedPlan = TestUtils.getResourceFileContent("plans/${planName}-plan.json")
+
+        then:
+        JSONAssert.assertEquals(expectedPlan, actualPlan, true)
+
+        where:
+        planName << [
+                "skip-topics"
+        ]
+    }
+
     void 'test various valid plans with seed - #planName'() {
         setup:
         TestUtils.cleanUpCluster()

src/test/resources/plans/skip-topics-apply-apply-output.txt

@@ -0,0 +1,73 @@
+Executing apply...
+
+Applying: [CREATE]
+
++ [ACL] test-service-0
+	 + resource_name: another.topic.0
+	 + resource_type: TOPIC
+	 + resource_pattern: LITERAL
+	 + resource_principal: User:test
+	 + host: *
+	 + operation: WRITE
+	 + permission: ALLOW
+
+
+Successfully applied.
+
+Applying: [CREATE]
+
++ [ACL] test-service-1
+	 + resource_name: MY_TOPIC
+	 + resource_type: TOPIC
+	 + resource_pattern: LITERAL
+	 + resource_principal: User:test
+	 + host: *
+	 + operation: READ
+	 + permission: ALLOW
+
+
+Successfully applied.
+
+Applying: [CREATE]
+
++ [ACL] test-service-2
+	 + resource_name: test-service
+	 + resource_type: GROUP
+	 + resource_pattern: LITERAL
+	 + resource_principal: User:test
+	 + host: *
+	 + operation: READ
+	 + permission: ALLOW
+
+
+Successfully applied.
+
+Applying: [CREATE]
+
++ [ACL] my-other-service-0
+	 + resource_name: another.topic.0
+	 + resource_type: TOPIC
+	 + resource_pattern: LITERAL
+	 + resource_principal: User:test
+	 + host: *
+	 + operation: READ
+	 + permission: ALLOW
+
+
+Successfully applied.
+
+Applying: [CREATE]
+
++ [ACL] my-other-service-1
+	 + resource_name: my-other-service
+	 + resource_type: GROUP
+	 + resource_pattern: LITERAL
+	 + resource_principal: User:test
+	 + host: *
+	 + operation: READ
+	 + permission: ALLOW
+
+
+Successfully applied.
+
+[SUCCESS] Apply complete! Resources: 5 created, 0 updated, 0 deleted.