feat: improve performance of plan process (#21)

Author: devshawn

.gitignore

@@ -6,3 +6,4 @@ out/
 docker/data/
 state.yaml
 plan.json
+test.py

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

@@ -6,6 +6,7 @@
 import com.devshawn.kafka.gitops.domain.plan.DesiredPlan;
 import com.devshawn.kafka.gitops.exception.KafkaExecutionException;
 import com.devshawn.kafka.gitops.exception.MissingConfigurationException;
+import com.devshawn.kafka.gitops.exception.PlanIsUpToDateException;
 import com.devshawn.kafka.gitops.exception.ValidationException;
 import com.devshawn.kafka.gitops.exception.WritePlanOutputException;
 import com.devshawn.kafka.gitops.service.ParserService;
@@ -34,6 +35,9 @@ public Integer call() {
             DesiredPlan desiredPlan = stateManager.plan();
             LogUtil.printPlan(desiredPlan, parent.isDeleteDisabled());
             return 0;
+        } catch (PlanIsUpToDateException ex) {
+            LogUtil.printNoChangesMessage();
+            return 0;
         } catch (MissingConfigurationException ex) {
             LogUtil.printGenericError(ex);
         } catch (ValidationException ex) {

src/main/java/com/devshawn/kafka/gitops/exception/PlanIsUpToDateException.java

@@ -0,0 +1,8 @@
+package com.devshawn.kafka.gitops.exception;
+
+public class PlanIsUpToDateException extends RuntimeException {
+
+    public PlanIsUpToDateException() {
+        super("The current desired state file matches the actual state of the cluster.");
+    }
+}

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

@@ -10,21 +10,23 @@
 import com.devshawn.kafka.gitops.domain.state.DesiredState;
 import com.devshawn.kafka.gitops.domain.state.TopicDetails;
 import com.devshawn.kafka.gitops.enums.PlanAction;
+import com.devshawn.kafka.gitops.exception.PlanIsUpToDateException;
 import com.devshawn.kafka.gitops.exception.ReadPlanInputException;
 import com.devshawn.kafka.gitops.exception.WritePlanOutputException;
 import com.devshawn.kafka.gitops.service.KafkaService;
-import com.devshawn.kafka.gitops.util.LogUtil;
 import com.devshawn.kafka.gitops.util.PlanUtil;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.kafka.clients.admin.Config;
 import org.apache.kafka.clients.admin.ConfigEntry;
-import org.apache.kafka.clients.admin.TopicDescription;
 import org.apache.kafka.clients.admin.TopicListing;
 import org.apache.kafka.common.acl.AclBinding;
+import org.apache.kafka.common.config.ConfigResource;
 import org.slf4j.LoggerFactory;
 
 import java.io.FileNotFoundException;
 import java.io.FileWriter;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -46,21 +48,21 @@ public PlanManager(ManagerConfig managerConfig, KafkaService kafkaService, Objec
 
     public void planTopics(DesiredState desiredState, DesiredPlan.Builder desiredPlan) {
         List<TopicListing> topics = kafkaService.getTopics();
+        List<String> topicNames = topics.stream().map(TopicListing::name).collect(Collectors.toList());
+        Map<String, List<ConfigEntry>> topicConfigs = fetchTopicConfigurations(topicNames);
 
         desiredState.getTopics().forEach((key, value) -> {
             TopicPlan.Builder topicPlan = new TopicPlan.Builder()
                     .setName(key)
                     .setTopicDetails(value);
 
-            TopicDescription topicDescription = kafkaService.describeTopic(key);
-
-            if (topicDescription == null) {
+            if (!topicNames.contains(key)) {
                 log.info("[PLAN] Topic {} does not exist; it will be created.", key);
                 topicPlan.setAction(PlanAction.ADD);
             } else {
                 log.info("[PLAN] Topic {} exists, it will not be created.", key);
                 topicPlan.setAction(PlanAction.NO_CHANGE);
-                planTopicConfigurations(key, value, topicPlan);
+                planTopicConfigurations(key, value, topicConfigs.get(key), topicPlan);
             }
 
             desiredPlan.addTopicPlans(topicPlan.build());
@@ -84,10 +86,9 @@ public void planTopics(DesiredState desiredState, DesiredPlan.Builder desiredPla
         });
     }
 
-    private void planTopicConfigurations(String topicName, TopicDetails topicDetails, TopicPlan.Builder topicPlan) {
+    private void planTopicConfigurations(String topicName, TopicDetails topicDetails, List<ConfigEntry> configs, TopicPlan.Builder topicPlan) {
         Map<String, TopicConfigPlan> configPlans = new HashMap<>();
-        List<ConfigEntry> currentConfigs = kafkaService.describeTopicConfigs(topicName);
-        List<ConfigEntry> customConfigs = currentConfigs.stream()
+        List<ConfigEntry> customConfigs = configs.stream()
                 .filter(it -> it.source() == ConfigEntry.ConfigSource.DYNAMIC_TOPIC_CONFIG)
                 .collect(Collectors.toList());
 
@@ -175,8 +176,7 @@ public void planAcls(DesiredState desiredState, DesiredPlan.Builder desiredPlan)
     public void validatePlanHasChanges(DesiredPlan desiredPlan, boolean deleteDisabled) {
         PlanOverview planOverview = PlanUtil.getOverview(desiredPlan, deleteDisabled);
         if (planOverview.getAdd() == 0 && planOverview.getUpdate() == 0 && planOverview.getRemove() == 0) {
-            LogUtil.printNoChangesMessage();
-            System.exit(0);
+            throw new PlanIsUpToDateException();
         }
     }
 
@@ -206,4 +206,11 @@ public void writePlanToFile(DesiredPlan desiredPlan) {
             }
         }
     }
+
+    private Map<String, List<ConfigEntry>> fetchTopicConfigurations(List<String> topicNames) {
+        Map<String, List<ConfigEntry>> map = new HashMap<>();
+        Map<ConfigResource, Config> configs = kafkaService.describeConfigsForTopics(topicNames);
+        configs.forEach((key, value) -> map.put(key.name(), new ArrayList<ConfigEntry>(value.entries())));
+        return map;
+    }
 }

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

@@ -3,17 +3,30 @@
 import com.devshawn.kafka.gitops.config.KafkaGitopsConfig;
 import com.devshawn.kafka.gitops.domain.state.TopicDetails;
 import com.devshawn.kafka.gitops.exception.KafkaExecutionException;
-import org.apache.kafka.clients.admin.*;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.AlterConfigOp;
+import org.apache.kafka.clients.admin.Config;
+import org.apache.kafka.clients.admin.KafkaAdminClient;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.admin.TopicListing;
 import org.apache.kafka.common.KafkaException;
-import org.apache.kafka.common.acl.*;
+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.ConfigResource;
-import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
 import org.apache.kafka.common.resource.PatternType;
 import org.apache.kafka.common.resource.ResourcePatternFilter;
 import org.apache.kafka.common.resource.ResourceType;
 
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
 import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
 
 public class KafkaService {
 
@@ -76,18 +89,6 @@ public void updateTopicConfig(Map<ConfigResource, Collection<AlterConfigOp>> con
         }
     }
 
-    public TopicDescription describeTopic(String topicName) {
-        try (final AdminClient adminClient = buildAdminClient()) {
-            Map<String, TopicDescription> topics = adminClient.describeTopics(Collections.singletonList(topicName)).all().get();
-            return topics.get(topicName);
-        } catch (InterruptedException | ExecutionException ex) {
-            if (!(ex.getCause() instanceof UnknownTopicOrPartitionException)) {
-                throw new KafkaExecutionException("Error thrown when attempting to describe a Kafka topic", ex.getMessage());
-            }
-        }
-        return null;
-    }
-
     public List<TopicListing> getTopics() {
         try (final AdminClient adminClient = buildAdminClient()) {
             Collection<TopicListing> topics = adminClient.listTopics().listings().get();
@@ -97,11 +98,10 @@ public List<TopicListing> getTopics() {
         }
     }
 
-    public List<ConfigEntry> describeTopicConfigs(String topicName) {
+    public Map<ConfigResource, Config> describeConfigsForTopics(List<String> topicNames) {
         try (final AdminClient adminClient = buildAdminClient()) {
-            ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, topicName);
-            Config config = adminClient.describeConfigs(Collections.singletonList(resource)).all().get().get(resource);
-            return new ArrayList<>(config.entries());
+            List<ConfigResource> resources = topicNames.stream().map(it -> new ConfigResource(ConfigResource.Type.TOPIC, it)).collect(Collectors.toList());
+            return adminClient.describeConfigs(resources).all().get();
         } catch (InterruptedException | ExecutionException ex) {
             throw new KafkaExecutionException("Error thrown when attempting to describe a Kafka topic configuration", ex.getMessage());
         }

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

@@ -173,4 +173,29 @@ class PlanCommandIntegrationSpec extends Specification {
         cleanup:
         System.setOut(oldOut)
     }
+
+    void 'test plan that has no changes'() {
+        setup:
+        TestUtils.cleanUpCluster()
+        TestUtils.seedCluster()
+        ByteArrayOutputStream out = new ByteArrayOutputStream()
+        PrintStream oldOut = System.out
+        System.setOut(new PrintStream(out))
+        String file = TestUtils.getResourceFilePath("plans/${planName}.yaml")
+        MainCommand mainCommand = new MainCommand()
+        CommandLine cmd = new CommandLine(mainCommand)
+
+        when:
+        int exitCode = cmd.execute("-f", file, "plan")
+
+        then:
+        exitCode == 0
+        out.toString() == TestUtils.getResourceFileContent("plans/no-changes-output.txt")
+
+        cleanup:
+        System.setOut(oldOut)
+
+        where:
+        planName << ["no-changes"]
+    }
 }

src/test/groovy/com/devshawn/kafka/gitops/exception/PlanIsUpToDateExceptionSpec.groovy

@@ -0,0 +1,14 @@
+package com.devshawn.kafka.gitops.exception
+
+import spock.lang.Specification
+
+class PlanIsUpToDateExceptionSpec extends Specification {
+
+    void 'test PlanIsUpToDateException'() {
+        when:
+        PlanIsUpToDateException result = new PlanIsUpToDateException()
+
+        then:
+        result.message == "The current desired state file matches the actual state of the cluster."
+    }
+}

src/test/resources/plans/no-changes-output.txt

@@ -0,0 +1,3 @@
+Generating execution plan...
+
+[SUCCESS] There are no necessary changes; the actual state matches the desired state.

src/test/resources/plans/no-changes.yaml

@@ -0,0 +1,37 @@
+topics:
+  delete-topic:
+    partitions: 1
+    replication: 1
+
+  test-topic:
+    partitions: 1
+    replication: 1
+
+  topic-with-configs-1:
+    partitions: 3
+    replication: 1
+    configs:
+      cleanup.policy: compact
+      segment.bytes: 100000
+
+  topic-with-configs-2:
+    partitions: 6
+    replication: 1
+    configs:
+      retention.ms: 60000
+
+services:
+  test-service:
+    principal: "User:test"
+    type: application
+
+customServiceAcls:
+  test-service:
+    read:
+      name: test-topic
+      type: TOPIC
+      pattern: LITERAL
+      host: "*"
+      principal: User:test
+      operation: READ
+      permission: ALLOW