feat: add settings and topic blacklist

Author: devshawn

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

@@ -78,6 +78,7 @@ private DesiredPlan generatePlan() {
 
     private void planTopics(DesiredState desiredState, DesiredPlan.Builder desiredPlan) {
         List<TopicListing> topics = kafkaService.getTopics();
+        List<String> prefixesToIgnore = getPrefixedTopicsToIgnore(desiredState);
 
         desiredState.getTopics().forEach((key, value) -> {
             TopicPlan.Builder topicPlan = new TopicPlan.Builder()
@@ -99,6 +100,12 @@ private void planTopics(DesiredState desiredState, DesiredPlan.Builder desiredPl
         });
 
         topics.forEach(currentTopic -> {
+            boolean shouldIgnore = prefixesToIgnore.stream().anyMatch(it -> currentTopic.name().startsWith(it));
+            if (shouldIgnore) {
+                log.info("[PLAN] Ignoring topic {} due to prefix", currentTopic.name());
+                return;
+            }
+
             if (desiredState.getTopics().getOrDefault(currentTopic.name(), null) == null) {
                 TopicPlan topicPlan = new TopicPlan.Builder()
                         .setName(currentTopic.name())
@@ -246,6 +253,14 @@ private void applyAcls(DesiredPlan desiredPlan) {
         });
     }
 
+    private List<String> getPrefixedTopicsToIgnore(DesiredState desiredState) {
+        try {
+            return desiredState.getSettings().get().getTopics().get().getBlacklist().get().getPrefixed();
+        } catch (NoSuchElementException ex) {
+            return Collections.emptyList();
+        }
+    }
+
     private void initializeLogger(boolean verbose) {
         Logger root = (Logger) LoggerFactory.getLogger(Logger.ROOT_LOGGER_NAME);
         Logger kafka = (Logger) LoggerFactory.getLogger("org.apache.kafka");

src/main/java/com/devshawn/kafka/dsf/domain/state/DesiredState.java

@@ -4,11 +4,14 @@
 import org.inferred.freebuilder.FreeBuilder;
 
 import java.util.Map;
+import java.util.Optional;
 
 @FreeBuilder
 @JsonDeserialize(builder = DesiredState.Builder.class)
 public interface DesiredState {
 
+    Optional<Settings> getSettings();
+
     Map<String, TopicDetails> getTopics();
 
     Map<String, AclDetails> getAcls();

src/main/java/com/devshawn/kafka/dsf/domain/state/Settings.java

@@ -0,0 +1,16 @@
+package com.devshawn.kafka.dsf.domain.state;
+
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import org.inferred.freebuilder.FreeBuilder;
+
+import java.util.Optional;
+
+@FreeBuilder
+@JsonDeserialize(builder = Settings.Builder.class)
+public interface Settings {
+
+    Optional<SettingsTopics> getTopics();
+
+    class Builder extends Settings_Builder {
+    }
+}

src/main/java/com/devshawn/kafka/dsf/domain/state/SettingsTopics.java

@@ -0,0 +1,16 @@
+package com.devshawn.kafka.dsf.domain.state;
+
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import org.inferred.freebuilder.FreeBuilder;
+
+import java.util.Optional;
+
+@FreeBuilder
+@JsonDeserialize(builder = SettingsTopics.Builder.class)
+public interface SettingsTopics {
+
+    Optional<SettingsTopicsBlacklist> getBlacklist();
+
+    class Builder extends SettingsTopics_Builder {
+    }
+}

src/main/java/com/devshawn/kafka/dsf/domain/state/SettingsTopicsBlacklist.java

@@ -0,0 +1,16 @@
+package com.devshawn.kafka.dsf.domain.state;
+
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import org.inferred.freebuilder.FreeBuilder;
+
+import java.util.List;
+
+@FreeBuilder
+@JsonDeserialize(builder = SettingsTopicsBlacklist.Builder.class)
+public interface SettingsTopicsBlacklist {
+
+    List<String> getPrefixed();
+
+    class Builder extends SettingsTopicsBlacklist_Builder {
+    }
+}

src/main/java/com/devshawn/kafka/dsf/service/ParserService.java

@@ -10,6 +10,7 @@
 import com.fasterxml.jackson.databind.exc.UnrecognizedPropertyException;
 import com.fasterxml.jackson.databind.exc.ValueInstantiationException;
 import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -30,6 +31,7 @@ public ParserService(File file) {
         this.objectMapper = new ObjectMapper(new YAMLFactory());
         objectMapper.enable(DeserializationFeature.FAIL_ON_READING_DUP_TREE_KEY);
         objectMapper.enable(JsonParser.Feature.STRICT_DUPLICATE_DETECTION);
+        objectMapper.registerModule(new Jdk8Module());
         this.file = file;
     }