Skip to content

Commit f2ac6e7

Browse files
devshawndevshawn
authored
feat: improve performance of plan process (#21)
1 parent 032db0b commit f2ac6e7

File tree

9 files changed

+130
-31
lines changed

9 files changed

+130
-31
lines changed

‎.gitignore

+1
Original file line numberDiff line numberDiff line change
@@ -6,3 +6,4 @@ out/
66
docker/data/
77
state.yaml
88
plan.json
9+
test.py

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

+4
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
import com.devshawn.kafka.gitops.domain.plan.DesiredPlan;
77
import com.devshawn.kafka.gitops.exception.KafkaExecutionException;
88
import com.devshawn.kafka.gitops.exception.MissingConfigurationException;
9+
import com.devshawn.kafka.gitops.exception.PlanIsUpToDateException;
910
import com.devshawn.kafka.gitops.exception.ValidationException;
1011
import com.devshawn.kafka.gitops.exception.WritePlanOutputException;
1112
import com.devshawn.kafka.gitops.service.ParserService;
@@ -34,6 +35,9 @@ public Integer call() {
3435
DesiredPlan desiredPlan = stateManager.plan();
3536
LogUtil.printPlan(desiredPlan, parent.isDeleteDisabled());
3637
return 0;
38+
} catch (PlanIsUpToDateException ex) {
39+
LogUtil.printNoChangesMessage();
40+
return 0;
3741
} catch (MissingConfigurationException ex) {
3842
LogUtil.printGenericError(ex);
3943
} catch (ValidationException ex) {

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

+8
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,8 @@
1+
package com.devshawn.kafka.gitops.exception;
2+
3+
public class PlanIsUpToDateException extends RuntimeException {
4+
5+
public PlanIsUpToDateException() {
6+
super("The current desired state file matches the actual state of the cluster.");
7+
}
8+
}

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

+18-11
Original file line numberDiff line numberDiff line change
@@ -10,21 +10,23 @@
1010
import com.devshawn.kafka.gitops.domain.state.DesiredState;
1111
import com.devshawn.kafka.gitops.domain.state.TopicDetails;
1212
import com.devshawn.kafka.gitops.enums.PlanAction;
13+
import com.devshawn.kafka.gitops.exception.PlanIsUpToDateException;
1314
import com.devshawn.kafka.gitops.exception.ReadPlanInputException;
1415
import com.devshawn.kafka.gitops.exception.WritePlanOutputException;
1516
import com.devshawn.kafka.gitops.service.KafkaService;
16-
import com.devshawn.kafka.gitops.util.LogUtil;
1717
import com.devshawn.kafka.gitops.util.PlanUtil;
1818
import com.fasterxml.jackson.databind.ObjectMapper;
19+
import org.apache.kafka.clients.admin.Config;
1920
import org.apache.kafka.clients.admin.ConfigEntry;
20-
import org.apache.kafka.clients.admin.TopicDescription;
2121
import org.apache.kafka.clients.admin.TopicListing;
2222
import org.apache.kafka.common.acl.AclBinding;
23+
import org.apache.kafka.common.config.ConfigResource;
2324
import org.slf4j.LoggerFactory;
2425

2526
import java.io.FileNotFoundException;
2627
import java.io.FileWriter;
2728
import java.io.IOException;
29+
import java.util.ArrayList;
2830
import java.util.HashMap;
2931
import java.util.List;
3032
import java.util.Map;
@@ -46,21 +48,21 @@ public PlanManager(ManagerConfig managerConfig, KafkaService kafkaService, Objec
4648

4749
public void planTopics(DesiredState desiredState, DesiredPlan.Builder desiredPlan) {
4850
List<TopicListing> topics = kafkaService.getTopics();
51+
List<String> topicNames = topics.stream().map(TopicListing::name).collect(Collectors.toList());
52+
Map<String, List<ConfigEntry>> topicConfigs = fetchTopicConfigurations(topicNames);
4953

5054
desiredState.getTopics().forEach((key, value) -> {
5155
TopicPlan.Builder topicPlan = new TopicPlan.Builder()
5256
.setName(key)
5357
.setTopicDetails(value);
5458

55-
TopicDescription topicDescription = kafkaService.describeTopic(key);
56-
57-
if (topicDescription == null) {
59+
if (!topicNames.contains(key)) {
5860
log.info("[PLAN] Topic {} does not exist; it will be created.", key);
5961
topicPlan.setAction(PlanAction.ADD);
6062
} else {
6163
log.info("[PLAN] Topic {} exists, it will not be created.", key);
6264
topicPlan.setAction(PlanAction.NO_CHANGE);
63-
planTopicConfigurations(key, value, topicPlan);
65+
planTopicConfigurations(key, value, topicConfigs.get(key), topicPlan);
6466
}
6567

6668
desiredPlan.addTopicPlans(topicPlan.build());
@@ -84,10 +86,9 @@ public void planTopics(DesiredState desiredState, DesiredPlan.Builder desiredPla
8486
});
8587
}
8688

87-
private void planTopicConfigurations(String topicName, TopicDetails topicDetails, TopicPlan.Builder topicPlan) {
89+
private void planTopicConfigurations(String topicName, TopicDetails topicDetails, List<ConfigEntry> configs, TopicPlan.Builder topicPlan) {
8890
Map<String, TopicConfigPlan> configPlans = new HashMap<>();
89-
List<ConfigEntry> currentConfigs = kafkaService.describeTopicConfigs(topicName);
90-
List<ConfigEntry> customConfigs = currentConfigs.stream()
91+
List<ConfigEntry> customConfigs = configs.stream()
9192
.filter(it -> it.source() == ConfigEntry.ConfigSource.DYNAMIC_TOPIC_CONFIG)
9293
.collect(Collectors.toList());
9394

@@ -175,8 +176,7 @@ public void planAcls(DesiredState desiredState, DesiredPlan.Builder desiredPlan)
175176
public void validatePlanHasChanges(DesiredPlan desiredPlan, boolean deleteDisabled) {
176177
PlanOverview planOverview = PlanUtil.getOverview(desiredPlan, deleteDisabled);
177178
if (planOverview.getAdd() == 0 && planOverview.getUpdate() == 0 && planOverview.getRemove() == 0) {
178-
LogUtil.printNoChangesMessage();
179-
System.exit(0);
179+
throw new PlanIsUpToDateException();
180180
}
181181
}
182182

@@ -206,4 +206,11 @@ public void writePlanToFile(DesiredPlan desiredPlan) {
206206
}
207207
}
208208
}
209+
210+
private Map<String, List<ConfigEntry>> fetchTopicConfigurations(List<String> topicNames) {
211+
Map<String, List<ConfigEntry>> map = new HashMap<>();
212+
Map<ConfigResource, Config> configs = kafkaService.describeConfigsForTopics(topicNames);
213+
configs.forEach((key, value) -> map.put(key.name(), new ArrayList<ConfigEntry>(value.entries())));
214+
return map;
215+
}
209216
}

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

+20-20
Original file line numberDiff line numberDiff line change
@@ -3,17 +3,30 @@
33
import com.devshawn.kafka.gitops.config.KafkaGitopsConfig;
44
import com.devshawn.kafka.gitops.domain.state.TopicDetails;
55
import com.devshawn.kafka.gitops.exception.KafkaExecutionException;
6-
import org.apache.kafka.clients.admin.*;
6+
import org.apache.kafka.clients.admin.AdminClient;
7+
import org.apache.kafka.clients.admin.AlterConfigOp;
8+
import org.apache.kafka.clients.admin.Config;
9+
import org.apache.kafka.clients.admin.KafkaAdminClient;
10+
import org.apache.kafka.clients.admin.NewTopic;
11+
import org.apache.kafka.clients.admin.TopicListing;
712
import org.apache.kafka.common.KafkaException;
8-
import org.apache.kafka.common.acl.*;
13+
import org.apache.kafka.common.acl.AccessControlEntryFilter;
14+
import org.apache.kafka.common.acl.AclBinding;
15+
import org.apache.kafka.common.acl.AclBindingFilter;
16+
import org.apache.kafka.common.acl.AclOperation;
17+
import org.apache.kafka.common.acl.AclPermissionType;
918
import org.apache.kafka.common.config.ConfigResource;
10-
import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
1119
import org.apache.kafka.common.resource.PatternType;
1220
import org.apache.kafka.common.resource.ResourcePatternFilter;
1321
import org.apache.kafka.common.resource.ResourceType;
1422

15-
import java.util.*;
23+
import java.util.ArrayList;
24+
import java.util.Collection;
25+
import java.util.Collections;
26+
import java.util.List;
27+
import java.util.Map;
1628
import java.util.concurrent.ExecutionException;
29+
import java.util.stream.Collectors;
1730

1831
public class KafkaService {
1932

@@ -76,18 +89,6 @@ public void updateTopicConfig(Map<ConfigResource, Collection<AlterConfigOp>> con
7689
}
7790
}
7891

79-
public TopicDescription describeTopic(String topicName) {
80-
try (final AdminClient adminClient = buildAdminClient()) {
81-
Map<String, TopicDescription> topics = adminClient.describeTopics(Collections.singletonList(topicName)).all().get();
82-
return topics.get(topicName);
83-
} catch (InterruptedException | ExecutionException ex) {
84-
if (!(ex.getCause() instanceof UnknownTopicOrPartitionException)) {
85-
throw new KafkaExecutionException("Error thrown when attempting to describe a Kafka topic", ex.getMessage());
86-
}
87-
}
88-
return null;
89-
}
90-
9192
public List<TopicListing> getTopics() {
9293
try (final AdminClient adminClient = buildAdminClient()) {
9394
Collection<TopicListing> topics = adminClient.listTopics().listings().get();
@@ -97,11 +98,10 @@ public List<TopicListing> getTopics() {
9798
}
9899
}
99100

100-
public List<ConfigEntry> describeTopicConfigs(String topicName) {
101+
public Map<ConfigResource, Config> describeConfigsForTopics(List<String> topicNames) {
101102
try (final AdminClient adminClient = buildAdminClient()) {
102-
ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, topicName);
103-
Config config = adminClient.describeConfigs(Collections.singletonList(resource)).all().get().get(resource);
104-
return new ArrayList<>(config.entries());
103+
List<ConfigResource> resources = topicNames.stream().map(it -> new ConfigResource(ConfigResource.Type.TOPIC, it)).collect(Collectors.toList());
104+
return adminClient.describeConfigs(resources).all().get();
105105
} catch (InterruptedException | ExecutionException ex) {
106106
throw new KafkaExecutionException("Error thrown when attempting to describe a Kafka topic configuration", ex.getMessage());
107107
}

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

+25
Original file line numberDiff line numberDiff line change
@@ -173,4 +173,29 @@ class PlanCommandIntegrationSpec extends Specification {
173173
cleanup:
174174
System.setOut(oldOut)
175175
}
176+
177+
void 'test plan that has no changes'() {
178+
setup:
179+
TestUtils.cleanUpCluster()
180+
TestUtils.seedCluster()
181+
ByteArrayOutputStream out = new ByteArrayOutputStream()
182+
PrintStream oldOut = System.out
183+
System.setOut(new PrintStream(out))
184+
String file = TestUtils.getResourceFilePath("plans/${planName}.yaml")
185+
MainCommand mainCommand = new MainCommand()
186+
CommandLine cmd = new CommandLine(mainCommand)
187+
188+
when:
189+
int exitCode = cmd.execute("-f", file, "plan")
190+
191+
then:
192+
exitCode == 0
193+
out.toString() == TestUtils.getResourceFileContent("plans/no-changes-output.txt")
194+
195+
cleanup:
196+
System.setOut(oldOut)
197+
198+
where:
199+
planName << ["no-changes"]
200+
}
176201
}

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

+14
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
package com.devshawn.kafka.gitops.exception
2+
3+
import spock.lang.Specification
4+
5+
class PlanIsUpToDateExceptionSpec extends Specification {
6+
7+
void 'test PlanIsUpToDateException'() {
8+
when:
9+
PlanIsUpToDateException result = new PlanIsUpToDateException()
10+
11+
then:
12+
result.message == "The current desired state file matches the actual state of the cluster."
13+
}
14+
}

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

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

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

+37
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,37 @@
1+
topics:
2+
delete-topic:
3+
partitions: 1
4+
replication: 1
5+
6+
test-topic:
7+
partitions: 1
8+
replication: 1
9+
10+
topic-with-configs-1:
11+
partitions: 3
12+
replication: 1
13+
configs:
14+
cleanup.policy: compact
15+
segment.bytes: 100000
16+
17+
topic-with-configs-2:
18+
partitions: 6
19+
replication: 1
20+
configs:
21+
retention.ms: 60000
22+
23+
services:
24+
test-service:
25+
principal: "User:test"
26+
type: application
27+
28+
customServiceAcls:
29+
test-service:
30+
read:
31+
name: test-topic
32+
type: TOPIC
33+
pattern: LITERAL
34+
host: "*"
35+
principal: User:test
36+
operation: READ
37+
permission: ALLOW

0 commit comments

Comments
 (0)