Repository: kafka-ops/kafka-topology-builder
Pull request #194 (2021-01-12 10:53:11)

Add ability to use the name as the end topic name
Whilst it is possible to use a pattern like {{ topic }} when other tooling is being used like ansible there is a collision with {{ }} as it also uses jinja templating.


Patch:
diff --git a/docs/config-values.rst b/docs/config-values.rst
index ccc134a..3290012 100644
--- a/docs/config-values.rst
+++ b/docs/config-values.rst
@@ -43,7 +43,8 @@ If you plan to manage and deploy schemas with KTB, you must define the url to yo
 Topology Builder backend usage and selection
 -----------
 
-It is necessary for the topology builder to keep some state, for example for situations when the tool needs to decide what ACLs to remove.
+It is necessary for the topology builder to keep some state, if it does not retrieve state from the cluster (see also "topology.state.cluster.enabled"),
+for example for situations when the tool needs to decide what ACLs to remove.
 As well this property is important when the tool does not manage all topics in the cluster, so it is important to know it context.
 
 The default implementation is a File, however is possible ot use other systems.
@@ -139,20 +140,55 @@ Retrieve topic management state from local controlled view
 -----------
 
 By default since it's creation KTB has been retrieving the state of topics from the target cluster, this means pulling the actual view directly
-from there (AK cluster) using AdminClient. However, this is not optimal when multiple teams use KTB as a multi tenant tool.
+from there (AK cluster) using AdminClient. To disable this it can be done below.
 
 If you want to manage the current view of topics from the own KTB  cluster state subsystem, you should use this property.
 
 **Property**: *topology.state.topics.cluster.enabled*
 **Default value**: true
 
-This property is fow the time being true as default (backwards compatible), however the local management system for topics will become the
-default in 2.0 and in 3.0 the management using the target cluster will be completely removed.
+This property is for the time being true as default (backwards compatible).
 
 An example to use local topic management state will look like this:
 ::
     topology.state.topics.cluster.enabled=false
 
+
+Retrieve management state from local controlled view
+-----------
+
+KTB for everything apart from topics uses a local state, so that KTB's uses the actual state not its internal state this means pulling the actual view directly,
+we can enable this for everything, topics, acls, service accounts etc. Note this flag supercedes the topology.state.topics.cluster.enabled.
+
+If you want to manage the current view of everything from the own KTB cluster state subsystem, you should use this property.
+
+**Property**: *topology.state.cluster.enabled*
+**Default value**: true
+
+This property is for the time being false as default (backwards compatible).
+
+An example to use actual view management state will look like this:
+::
+    topology.state.cluster.enabled=false
+
+
+Control allowed Topics to be managed by KTB
+-----------
+
+This property is used to control which Topics are allowed to be managed by the KTB, this variable contains a list of allowed prefixes.
+
+**Property**: *topology.topic.managed.prefixes*
+**Default value**: "[]"
+
+An example configuration might look like this:
+::
+    topology.topic.managed.prefixes.0=User:AService
+    topology.topic.managed.prefixes.1=User:BService
+
+If this prefix list is used, only topics that match the prefix will be ever processed, anything else will be ignored.
+This is useful in a shared cluster, to avoid KTB removing/accidentally managing topics managed by other teams with seperate pipelines.
+
+
 Control allowed Service accounts to be managed by KTB
 -----------
 
@@ -166,4 +202,23 @@ An example configuration might look like this:
     topology.service.accounts.managed.prefixes.0=User:AService
     topology.service.accounts.managed.prefixes.1=User:BService
 
-If this prefix list is used, only service accounts that match the prefix will be ever processed, anything else will be ignored.
\ No newline at end of file
+If this prefix list is used, only service accounts that match the prefix will be ever processed, anything else will be ignored.
+This is useful in a shared cluster, to avoid KTB removing/accidentally managing service accounts managed by other teams with seperate pipelines.
+
+Control allowed Group to be managed by KTB
+-----------
+
+Note, currently KTB just manages Group ACLS.
+
+This property is used to control which Group prefixes are allowed to be managed by the KTB, this variable contains a list of allowed prefixes.
+
+**Property**: *topology.group.managed.prefixes*
+**Default value**: "[]"
+
+An example configuration might look like this:
+::
+    topology.group.managed.prefixes.0=NameSpaceA
+    topology.group.managed.prefixes.1=NameSpaceB
+
+If this prefix list is used, only groups that match the prefix will be ever processed, if wildcard it will be managed if the service account is managed by KTB, anything else will be ignored.
+This is useful in a shared cluster, to avoid KTB removing/accidentally managing group acls by other teams with seperate pipelines.
diff --git a/docs/futures/what-acl-management.rst b/docs/futures/what-acl-management.rst
index 542eb06..b4b8933 100644
--- a/docs/futures/what-acl-management.rst
+++ b/docs/futures/what-acl-management.rst
@@ -7,6 +7,18 @@ ACLs
 In the topology descriptor files users can create permissions for different types of applications, Consumers, Producers, Kafka streams apps or Kafka Connectors.
 With this roles, users can easily create the permissions that map directly to their needs.
 
+If desired by organisational purposes e.g. shared cluster a user can decide to filter which ACLS will be managed by prefix, this is done using the managed prefixes config.
+
+Topic acls will be managed if the topic matches
+*topology.topic.managed.prefixes* configuration setting. Check :ref:`config` for details.
+
+Group acls will be managed if the group matches
+*topology.group.managed.prefixes* configuration setting. Check :ref:`config` for details.
+
+All others currently and for the above if a global wildcard, will be managed if principle matches
+*topology.service.accounts.managed.prefixes* configuration setting. Check :ref:`config` for details.
+
+
 Consumers
 ^^^^^^^^^^^
 
diff --git a/docs/futures/what-principal-management.rst b/docs/futures/what-principal-management.rst
index d2cbcb7..0238bb4 100644
--- a/docs/futures/what-principal-management.rst
+++ b/docs/futures/what-principal-management.rst
@@ -29,7 +29,7 @@ If desired by organisational purposes a user can decide to filter witch Service
 Principals Manager Providers
 -----------
 
-As a user of KTB you can select witch providers to use, they could be for example Confluent Cloud or SASL/SCRAM (once `#2 <https://github.com/kafka-ops/kafka-topology-builder/issues/2>`_ is implemented),
+As a user of KTB you can select which providers to use, they could be for example Confluent Cloud or SASL/SCRAM (once `#2 <https://github.com/kafka-ops/kafka-topology-builder/issues/2>`_ is implemented),
 the tool will contact the server and manage the user creation.
 
 In the current version you can only use the Confluent Cloud provider.
diff --git a/docs/futures/what-topic-management.rst b/docs/futures/what-topic-management.rst
index baeb604..94f1008 100644
--- a/docs/futures/what-topic-management.rst
+++ b/docs/futures/what-topic-management.rst
@@ -3,6 +3,9 @@ Managing Topics and their Configuration
 
 The first and foremost important thing we aim to manage is topics, setup the partitions count and replication factor and as well configure their specific characteristics.
 
+If desired by organisational purposes a user can decide to filter which topics can be managed, this is done using the
+*topology.topics.managed.prefixes* configuration setting. Check :ref:`config` for details.
+
 Topic naming convention
 -----------
 
diff --git a/src/main/java/com/purbon/kafka/topology/AccessControlManager.java b/src/main/java/com/purbon/kafka/topology/AccessControlManager.java
index 6b2829b..7cd35a6 100644
--- a/src/main/java/com/purbon/kafka/topology/AccessControlManager.java
+++ b/src/main/java/com/purbon/kafka/topology/AccessControlManager.java
@@ -24,12 +24,7 @@ import com.purbon.kafka.topology.model.users.platform.ControlCenterInstance;
 import com.purbon.kafka.topology.model.users.platform.SchemaRegistryInstance;
 import com.purbon.kafka.topology.roles.TopologyAclBinding;
 import java.io.PrintStream;
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
+import java.util.*;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -43,6 +38,9 @@ public class AccessControlManager {
   private final TopologyBuilderConfig config;
   private AccessControlProvider controlProvider;
   private BindingsBuilderProvider bindingsBuilder;
+  private final List<String> managedServiceAccountPrefixes;
+  private final List<String> managedTopicPrefixes;
+  private final List<String> managedGroupPrefixes;
 
   public AccessControlManager(
       AccessControlProvider controlProvider, BindingsBuilderProvider builderProvider) {
@@ -56,6 +54,9 @@ public class AccessControlManager {
     this.controlProvider = controlProvider;
     this.bindingsBuilder = builderProvider;
     this.config = config;
+    this.managedServiceAccountPrefixes = config.getServiceAccountManagedPrefixes();
+    this.managedTopicPrefixes = config.getTopicManagedPrefixes();
+    this.managedGroupPrefixes = config.getGroupManagedPrefixes();
   }
 
   /**
@@ -68,7 +69,19 @@ public class AccessControlManager {
   public void apply(final Topology topology, ExecutionPlan plan) {
     List<Action> actions = buildProjectActions(topology);
     actions.addAll(buildPlatformLevelActions(topology));
-    buildUpdateBindingsActions(actions, plan.getBindings()).forEach(plan::add);
+    buildUpdateBindingsActions(actions, loadActualClusterStateIfAvailable(plan)).forEach(plan::add);
+  }
+
+  private Set<TopologyAclBinding> loadActualClusterStateIfAvailable(ExecutionPlan plan) {
+    Set<TopologyAclBinding> bindings =
+        config.fetchStateFromTheCluster() ? providerBindings() : plan.getBindings();
+    return bindings.stream().filter(this::matchesManagedPrefixList).collect(Collectors.toSet());
+  }
+
+  private Set<TopologyAclBinding> providerBindings() {
+    Set<TopologyAclBinding> bindings = new HashSet<>();
+    controlProvider.listAcls().values().forEach(bindings::addAll);
+    return bindings;
   }
 
   /**
@@ -175,9 +188,12 @@ public class AccessControlManager {
     Set<TopologyAclBinding> allFinalBindings =
         actions.stream().flatMap(actionApplyFunction()).collect(Collectors.toSet());
 
-    // Diff of bindings, so we only create what is not already created in the cluster.
     Set<TopologyAclBinding> bindingsToBeCreated =
         allFinalBindings.stream()
+            .filter(Objects::nonNull)
+            // Only create what we manage
+            .filter(this::matchesManagedPrefixList)
+            // Diff of bindings, so we only create what is not already created in the cluster.
             .filter(binding -> !bindings.contains(binding))
             .collect(Collectors.toSet());
 
@@ -201,6 +217,53 @@ public class AccessControlManager {
     return updateActions;
   }
 
+  private boolean matchesManagedPrefixList(TopologyAclBinding topologyAclBinding) {
+    String resourceName = topologyAclBinding.getResourceName();
+    String principle = topologyAclBinding.getPrincipal();
+    // For global wild cards ACL's we manage only if we manage the service account/principle,
+    // regardless.
+    if (resourceName.equals("*")) {
+      return matchesServiceAccountPrefixList(principle);
+    }
+
+    switch (topologyAclBinding.getResourceType()) {
+      case TOPIC:
+        return matchesTopicPrefixList(resourceName);
+      case GROUP:
+        return matchesGroupPrefixList(resourceName);
+      default:
+        return matchesServiceAccountPrefixList(principle);
+    }
+  }
+
+  private boolean matchesTopicPrefixList(String topic) {
+    boolean matches =
+        managedTopicPrefixes.size() == 0
+            || managedTopicPrefixes.stream().anyMatch(topic::startsWith);
+    LOGGER.debug(
+        String.format("Topic %s matches %s with $s", topic, matches, managedTopicPrefixes));
+    return matches;
+  }
+
+  private boolean matchesGroupPrefixList(String group) {
+    boolean matches =
+        managedGroupPrefixes.size() == 0
+            || managedGroupPrefixes.stream().anyMatch(group::startsWith);
+    LOGGER.debug(
+        String.format("Group %s matches %s with $s", group, matches, managedGroupPrefixes));
+    return matches;
+  }
+
+  private boolean matchesServiceAccountPrefixList(String principal) {
+    boolean matches =
+        managedServiceAccountPrefixes.size() == 0
+            || managedServiceAccountPrefixes.stream().anyMatch(principal::startsWith);
+    LOGGER.debug(
+        String.format(
+            "Principal %s matches %s with $s", principal, matches, managedServiceAccountPrefixes));
+    return matches;
+  }
+
   private Function<Action, Stream<TopologyAclBinding>> actionApplyFunction() {
     return action -> {
       try {
diff --git a/src/main/java/com/purbon/kafka/topology/AccessControlProviderFactory.java b/src/main/java/com/purbon/kafka/topology/AccessControlProviderFactory.java
index 7ec401c..681f93f 100644
--- a/src/main/java/com/purbon/kafka/topology/AccessControlProviderFactory.java
+++ b/src/main/java/com/purbon/kafka/topology/AccessControlProviderFactory.java
@@ -1,14 +1,12 @@
 package com.purbon.kafka.topology;
 
-import static com.purbon.kafka.topology.TopologyBuilderConfig.ACCESS_CONTROL_DEFAULT_CLASS;
-import static com.purbon.kafka.topology.TopologyBuilderConfig.MDS_PASSWORD_CONFIG;
-import static com.purbon.kafka.topology.TopologyBuilderConfig.MDS_USER_CONFIG;
-import static com.purbon.kafka.topology.TopologyBuilderConfig.RBAC_ACCESS_CONTROL_CLASS;
+import static com.purbon.kafka.topology.TopologyBuilderConfig.*;
 
 import com.purbon.kafka.topology.api.adminclient.TopologyBuilderAdminClient;
 import com.purbon.kafka.topology.api.ccloud.CCloudCLI;
 import com.purbon.kafka.topology.api.mds.MDSApiClient;
 import com.purbon.kafka.topology.api.mds.MDSApiClientBuilder;
+import com.purbon.kafka.topology.roles.CCloudAclsProvider;
 import com.purbon.kafka.topology.roles.RBACProvider;
 import com.purbon.kafka.topology.roles.SimpleAclsProvider;
 import com.purbon.kafka.topology.roles.acls.AclsBindingsBuilder;
@@ -43,6 +41,11 @@ public class AccessControlProviderFactory {
           Constructor<?> aclsProviderConstructor =
               clazz.getConstructor(TopologyBuilderAdminClient.class);
           return (SimpleAclsProvider) aclsProviderConstructor.newInstance(builderAdminClient);
+        case CONFLUENT_CLOUD_CONTROL_CLASS:
+          Constructor<?> ccloudProviderConstructor =
+              clazz.getConstructor(TopologyBuilderAdminClient.class, TopologyBuilderConfig.class);
+          return (CCloudAclsProvider)
+              ccloudProviderConstructor.newInstance(builderAdminClient, config);
         case RBAC_ACCESS_CONTROL_CLASS:
           Constructor<?> rbacProviderContructor = clazz.getConstructor(MDSApiClient.class);
           MDSApiClient apiClient = apiClientLogIn();
@@ -65,6 +68,8 @@ public class AccessControlProviderFactory {
     try {
       if (accessControlClass.equalsIgnoreCase(ACCESS_CONTROL_DEFAULT_CLASS)) {
         return new AclsBindingsBuilder(config, cCloudUtils);
+      } else if (accessControlClass.equalsIgnoreCase(CONFLUENT_CLOUD_CONTROL_CLASS)) {
+        return new AclsBindingsBuilder(config, cCloudUtils);
       } else if (accessControlClass.equalsIgnoreCase(RBAC_ACCESS_CONTROL_CLASS)) {
         MDSApiClient apiClient = apiClientLogIn();
         apiClient.authenticate();
diff --git a/src/main/java/com/purbon/kafka/topology/KafkaTopologyBuilder.java b/src/main/java/com/purbon/kafka/topology/KafkaTopologyBuilder.java
index 0817ef9..a44d1c4 100644
--- a/src/main/java/com/purbon/kafka/topology/KafkaTopologyBuilder.java
+++ b/src/main/java/com/purbon/kafka/topology/KafkaTopologyBuilder.java
@@ -164,15 +164,23 @@ public class KafkaTopologyBuilder implements AutoCloseable {
             "Running topology builder with TopicManager=[%s], accessControlManager=[%s], dryRun=[%s], isQuite=[%s]",
             topicManager, accessControlManager, config.isDryRun(), config.isQuiet()));
 
+    // Create users should always be first, so user exists when making acl link
+    principalManager.applyCreate(topology, plan);
+
     topicManager.apply(topology, plan);
-    principalManager.apply(topology, plan);
     accessControlManager.apply(topology, plan);
 
+    // Delete users should always be last,
+    // avoids any unlinked acls, e.g. if acl delete or something errors then there is a link still
+    // from the account, and can be re-run or manually fixed more easily
+    principalManager.applyDelete(topology, plan);
+
     plan.run(config.isDryRun());
 
     if (!config.isQuiet() && !config.isDryRun()) {
       topicManager.printCurrentState(System.out);
       accessControlManager.printCurrentState(System.out);
+      principalManager.printCurrentState(System.out);
     }
   }
 
diff --git a/src/main/java/com/purbon/kafka/topology/PrincipalManager.java b/src/main/java/com/purbon/kafka/topology/PrincipalManager.java
index 0561cae..17c0882 100644
--- a/src/main/java/com/purbon/kafka/topology/PrincipalManager.java
+++ b/src/main/java/com/purbon/kafka/topology/PrincipalManager.java
@@ -8,6 +8,7 @@ import com.purbon.kafka.topology.model.User;
 import com.purbon.kafka.topology.model.cluster.ServiceAccount;
 import com.purbon.kafka.topology.serviceAccounts.VoidPrincipalProvider;
 import java.io.IOException;
+import java.io.PrintStream;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -31,22 +32,20 @@ public class PrincipalManager {
     this.managedPrefixes = config.getServiceAccountManagedPrefixes();
   }
 
-  public void apply(Topology topology, ExecutionPlan plan) throws IOException {
+  public void applyCreate(Topology topology, ExecutionPlan plan) throws IOException {
     if (!config.enabledExperimental()) {
       LOGGER.debug("Not running the PrincipalsManager as this is an experimental feature.");
       return;
     }
-    // Do Nothing if the provider is the void one.
-    // This means the management of principals is either not possible or has not been configured
-    List<String> principals = parseListOfPrincipals(topology);
-    if (!(provider instanceof VoidPrincipalProvider)) {
-      provider.configure();
-      managePrincipals(principals, plan);
+    if (provider instanceof VoidPrincipalProvider) {
+      // Do Nothing if the provider is the void one.
+      // This means the management of principals is either not possible or has not been configured
+      return;
     }
-  }
 
-  private void managePrincipals(List<String> principals, ExecutionPlan plan) throws IOException {
+    provider.configure();
 
+    List<String> principals = parseListOfPrincipals(topology);
     Map<String, ServiceAccount> accounts = loadActualClusterStateIfAvailable(plan);
 
     // build list of principals to be created.
@@ -59,9 +58,26 @@ public class PrincipalManager {
     if (!principalsToBeCreated.isEmpty()) {
       plan.add(new CreateAccounts(provider, principalsToBeCreated));
     }
+  }
+
+  public void applyDelete(Topology topology, ExecutionPlan plan) throws IOException {
+    if (!config.enabledExperimental()) {
+      LOGGER.debug("Not running the PrincipalsManager as this is an experimental feature.");
+      return;
+    }
+    if (provider instanceof VoidPrincipalProvider) {
+      // Do Nothing if the provider is the void one.
+      // This means the management of principals is either not possible or has not been configured
+      return;
+    }
 
-    // build list of principals to be deleted.
     if (config.allowDelete() || config.isAllowDeletePrincipals()) {
+      provider.configure();
+
+      List<String> principals = parseListOfPrincipals(topology);
+      Map<String, ServiceAccount> accounts = loadActualClusterStateIfAvailable(plan);
+
+      // build list of principals to be deleted.
       List<ServiceAccount> principalsToBeDeleted =
           accounts.values().stream()
               .filter(currentPrincipal -> !principals.contains(currentPrincipal.getName()))
@@ -111,4 +127,9 @@ public class PrincipalManager {
         .filter(this::matchesPrefixList)
         .collect(Collectors.toList());
   }
+
+  public void printCurrentState(PrintStream out) throws IOException {
+    out.println("List of Principles: ");
+    provider.listServiceAccounts().forEach(out::println);
+  }
 }
diff --git a/src/main/java/com/purbon/kafka/topology/TopicManager.java b/src/main/java/com/purbon/kafka/topology/TopicManager.java
index 4597bbf..bf7d428 100644
--- a/src/main/java/com/purbon/kafka/topology/TopicManager.java
+++ b/src/main/java/com/purbon/kafka/topology/TopicManager.java
@@ -3,16 +3,12 @@ package com.purbon.kafka.topology;
 import com.purbon.kafka.topology.actions.topics.DeleteTopics;
 import com.purbon.kafka.topology.actions.topics.SyncTopicAction;
 import com.purbon.kafka.topology.api.adminclient.TopologyBuilderAdminClient;
-import com.purbon.kafka.topology.model.Project;
 import com.purbon.kafka.topology.model.Topic;
 import com.purbon.kafka.topology.model.Topology;
 import com.purbon.kafka.topology.schemas.SchemaRegistryManager;
 import java.io.IOException;
 import java.io.PrintStream;
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
+import java.util.*;
 import java.util.stream.Collectors;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.logging.log4j.LogManager;
@@ -29,6 +25,7 @@ public class TopicManager {
   private final TopologyBuilderAdminClient adminClient;
   private final TopologyBuilderConfig config;
   private final List<String> internalTopicPrefixes;
+  private final List<String> managedPrefixes;
 
   public TopicManager(
       TopologyBuilderAdminClient adminClient, SchemaRegistryManager schemaRegistryManager) {
@@ -43,31 +40,29 @@ public class TopicManager {
     this.schemaRegistryManager = schemaRegistryManager;
     this.config = config;
     this.internalTopicPrefixes = config.getKafkaInternalTopicPrefixes();
+    this.managedPrefixes = config.getTopicManagedPrefixes();
   }
 
   public void apply(Topology topology, ExecutionPlan plan) throws IOException {
 
     Set<String> listOfTopics = loadActualClusterStateIfAvailable(plan);
-    Set<String> updatedListOfTopics = new HashSet<>();
     // Foreach topic in the topology, sync it's content
     // if topics does not exist already it's created
 
-    for (Project project : topology.getProjects()) {
-      for (Topic topic : project.getTopics()) {
-        String fullTopicName = topic.toString();
-        plan.add(
-            new SyncTopicAction(
-                adminClient, schemaRegistryManager, topic, fullTopicName, listOfTopics));
-        updatedListOfTopics.add(fullTopicName);
-      }
-    }
+    Map<String, Topic> topics = parseMapOfTopics(topology);
+    topics.forEach(
+        (topicName, topic) -> {
+          plan.add(
+              new SyncTopicAction(
+                  adminClient, schemaRegistryManager, topic, topicName, listOfTopics));
+        });
 
     if (config.allowDelete() || config.isAllowDeleteTopics()) {
       // Handle topic delete: Topics in the initial list, but not present anymore after a
       // full topic sync should be deleted
       List<String> topicsToBeDeleted =
           listOfTopics.stream()
-              .filter(topic -> !updatedListOfTopics.contains(topic) && !isAnInternalTopics(topic))
+              .filter(topic -> !topics.containsKey(topic) && !isAnInternalTopics(topic))
               .collect(Collectors.toList());
 
       if (topicsToBeDeleted.size() > 0) {
@@ -77,13 +72,25 @@ public class TopicManager {
     }
   }
 
+  private Map<String, Topic> parseMapOfTopics(Topology topology) {
+    return topology.getProjects().stream()
+        .flatMap(project -> project.getTopics().stream())
+        .filter(this::matchesPrefixList)
+        .collect(Collectors.toMap(Topic::toString, topic -> topic));
+  }
+
   private boolean isAnInternalTopics(String topic) {
     return internalTopicPrefixes.stream().anyMatch(topic::startsWith);
   }
 
   private Set<String> loadActualClusterStateIfAvailable(ExecutionPlan plan) throws IOException {
     Set<String> listOfTopics =
-        config.fetchStateFromTheCluster() ? adminClient.listApplicationTopics() : plan.getTopics();
+        config.fetchTopicStateFromTheCluster()
+            ? adminClient.listApplicationTopics()
+            : plan.getTopics();
+
+    listOfTopics =
+        listOfTopics.stream().filter(this::matchesPrefixList).collect(Collectors.toSet());
     if (listOfTopics.size() > 0)
       LOGGER.debug(
           "Full list of managed topics in the cluster: "
@@ -91,6 +98,17 @@ public class TopicManager {
     return listOfTopics;
   }
 
+  private boolean matchesPrefixList(Topic topic) {
+    return matchesPrefixList(topic.toString());
+  }
+
+  private boolean matchesPrefixList(String topic) {
+    boolean matches =
+        managedPrefixes.size() == 0 || managedPrefixes.stream().anyMatch(topic::startsWith);
+    LOGGER.debug(String.format("Topic %s matches %s with $s", topic, matches, managedPrefixes));
+    return matches;
+  }
+
   void printCurrentState(PrintStream os) throws IOException {
     os.println("List of Topics:");
     adminClient.listTopics().forEach(os::println);
diff --git a/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java b/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java
index 696e97b..dfa9bb8 100644
--- a/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java
+++ b/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java
@@ -26,6 +26,10 @@ public class TopologyBuilderConfig {
 
   static final String ACCESS_CONTROL_DEFAULT_CLASS =
       "com.purbon.kafka.topology.roles.SimpleAclsProvider";
+
+  static final String CONFLUENT_CLOUD_CONTROL_CLASS =
+      "com.purbon.kafka.topology.roles.CCloudAclsProvider";
+
   static final String RBAC_ACCESS_CONTROL_CLASS = "com.purbon.kafka.topology.roles.RBACProvider";
 
   private static final String STATE_PROCESSOR_IMPLEMENTATION_CLASS =
@@ -33,6 +37,7 @@ public class TopologyBuilderConfig {
 
   static final String STATE_PROCESSOR_DEFAULT_CLASS =
       "com.purbon.kafka.topology.backend.FileBackend";
+
   static final String REDIS_STATE_PROCESSOR_CLASS =
       "com.purbon.kafka.topology.backend.RedisBackend";
 
@@ -75,9 +80,15 @@ public class TopologyBuilderConfig {
   public static final String TOPOLOGY_TOPIC_STATE_FROM_CLUSTER =
       "topology.state.topics.cluster.enabled";
 
+  static final String TOPOLOGY_STATE_FROM_CLUSTER = "topology.state.cluster.enabled";
+
   static final String SERVICE_ACCOUNT_MANAGED_PREFIXES =
       "topology.service.accounts.managed.prefixes";
 
+  static final String TOPIC_MANAGED_PREFIXES = "topology.topic.managed.prefixes";
+
+  static final String GROUP_MANAGED_PREFIXES = "topology.group.managed.prefixes";
+
   private final Map<String, String> cliParams;
   private Config config;
 
@@ -241,6 +252,18 @@ public class TopologyBuilderConfig {
         .collect(Collectors.toList());
   }
 
+  public List<String> getTopicManagedPrefixes() {
+    return config.getStringList(TOPIC_MANAGED_PREFIXES).stream()
+        .map(String::trim)
+        .collect(Collectors.toList());
+  }
+
+  public List<String> getGroupManagedPrefixes() {
+    return config.getStringList(GROUP_MANAGED_PREFIXES).stream()
+        .map(String::trim)
+        .collect(Collectors.toList());
+  }
+
   public String getConfluentSchemaRegistryUrl() {
     return config.getString(CONFLUENT_SCHEMA_REGISTRY_URL_CONFIG);
   }
@@ -339,6 +362,10 @@ public class TopologyBuilderConfig {
   }
 
   public boolean fetchStateFromTheCluster() {
-    return config.getBoolean(TOPOLOGY_TOPIC_STATE_FROM_CLUSTER);
+    return config.getBoolean(TOPOLOGY_STATE_FROM_CLUSTER);
+  }
+
+  public boolean fetchTopicStateFromTheCluster() {
+    return fetchStateFromTheCluster() || config.getBoolean(TOPOLOGY_TOPIC_STATE_FROM_CLUSTER);
   }
 }
diff --git a/src/main/java/com/purbon/kafka/topology/model/Impl/TopicImpl.java b/src/main/java/com/purbon/kafka/topology/model/Impl/TopicImpl.java
index 1e3ccea..9ab6874 100644
--- a/src/main/java/com/purbon/kafka/topology/model/Impl/TopicImpl.java
+++ b/src/main/java/com/purbon/kafka/topology/model/Impl/TopicImpl.java
@@ -94,6 +94,7 @@ public class TopicImpl implements Topic, Cloneable {
     this.dataType = dataType;
     this.config = config;
     this.schemas = new ArrayList<>();
+    this.context = new HashMap<>();
     this.replicationFactor = 0;
     this.partitionCount = 0;
     this.appConfig = appConfig;
@@ -133,6 +134,8 @@ public class TopicImpl implements Topic, Cloneable {
     switch (appConfig.getTopicPrefixFormat()) {
       case "default":
         return defaultTopicStructureString(projectPrefix);
+      case "name":
+        return name;
       default:
         return patternBasedTopicNameStructureString();
     }
diff --git a/src/main/java/com/purbon/kafka/topology/roles/CCloudAclsProvider.java b/src/main/java/com/purbon/kafka/topology/roles/CCloudAclsProvider.java
new file mode 100644
index 0000000..8d2aef4
--- /dev/null
+++ b/src/main/java/com/purbon/kafka/topology/roles/CCloudAclsProvider.java
@@ -0,0 +1,160 @@
+package com.purbon.kafka.topology.roles;
+
+import com.purbon.kafka.topology.AccessControlProvider;
+import com.purbon.kafka.topology.TopologyBuilderConfig;
+import com.purbon.kafka.topology.api.adminclient.TopologyBuilderAdminClient;
+import com.purbon.kafka.topology.api.ccloud.CCloudCLI;
+import com.purbon.kafka.topology.model.cluster.ServiceAccount;
+import java.io.IOException;
+import java.util.*;
+import java.util.stream.Collectors;
+import org.apache.kafka.common.acl.AccessControlEntry;
+import org.apache.kafka.common.acl.AclBinding;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+public class CCloudAclsProvider extends SimpleAclsProvider implements AccessControlProvider {
+
+  private static final Logger LOGGER = LogManager.getLogger(CCloudAclsProvider.class);
+
+  private final CCloudCLI cli;
+
+  public CCloudAclsProvider(
+      final TopologyBuilderAdminClient adminClient, final TopologyBuilderConfig config)
+      throws IOException {
+    super(adminClient);
+    this.cli = new CCloudCLI();
+    this.cli.setEnvironment(config.getConfluentCloudEnv());
+  }
+
+  @Override
+  public void createBindings(Set<TopologyAclBinding> bindings) throws IOException {
+    try {
+      Map<String, ServiceAccount> serviceAccounts = cli.serviceAccounts();
+      Set<TopologyAclBinding> ccloudBindings =
+          bindings.stream()
+              .map(b -> convertToConfluentCloudId(serviceAccounts, b))
+              .filter(Objects::nonNull)
+              .collect(Collectors.toSet());
+      super.createBindings(ccloudBindings);
+    } catch (IOException ex) {
+      LOGGER.error(ex);
+      throw ex;
+    }
+  }
+
+  @Override
+  public void clearBindings(Set<TopologyAclBinding> bindings) throws IOException {
+    try {
+      Map<String, ServiceAccount> serviceAccounts = cli.serviceAccounts();
+      Set<TopologyAclBinding> ccloudBindings =
+          bindings.stream()
+              .map(b -> convertToConfluentCloudId(serviceAccounts, b))
+              .filter(Objects::nonNull)
+              .collect(Collectors.toSet());
+      super.clearBindings(ccloudBindings);
+    } catch (IOException ex) {
+      LOGGER.error(ex);
+      throw ex;
+    }
+  }
+
+  @Override
+  public Map<String, List<TopologyAclBinding>> listAcls() {
+    try {
+      Map<Integer, ServiceAccount> serviceAccountsById =
+          cli.serviceAccounts().values().stream()
+              .collect(Collectors.toMap(ServiceAccount::getId, s -> s));
+      Map<String, List<TopologyAclBinding>> map = new HashMap<>();
+      super.listAcls()
+          .forEach(
+              (topic, aclBindings) -> {
+                map.put(
+                    topic,
+                    aclBindings.stream()
+                        .map(
+                            aclBinding ->
+                                convertToServiceAccountName(serviceAccountsById, aclBinding))
+                        .collect(Collectors.toList()));
+              });
+      return map;
+    } catch (IOException e) {
+      return new HashMap<>();
+    }
+  }
+
+  private TopologyAclBinding convertToConfluentCloudId(
+      Map<String, ServiceAccount> serviceAccounts, TopologyAclBinding binding) {
+    if (binding.asAclBinding().isPresent()) {
+      AclBinding aclBinding = convertToConfluentCloudId(serviceAccounts, binding.asAclBinding().get());
+      return aclBinding == null ? null : new TopologyAclBinding(aclBinding);
+    } else {
+      Integer id = getId(serviceAccounts, binding.getPrincipal());
+      return id == null ? null :  getTopologyAclBinding(binding, "User:" + id);
+    }
+  }
+
+  private AclBinding convertToConfluentCloudId(
+      Map<String, ServiceAccount> serviceAccounts, AclBinding aclBinding) {
+    Integer id = getId(serviceAccounts, aclBinding.entry().principal());
+    return id == null ? null : getAclBinding(aclBinding, "User:" + id);
+  }
+
+  private TopologyAclBinding convertToServiceAccountName(
+      Map<Integer, ServiceAccount> serviceAccounts, TopologyAclBinding binding) {
+    if (binding.asAclBinding().isPresent()) {
+      return new TopologyAclBinding(
+          convertToServiceAccountName(serviceAccounts, binding.asAclBinding().get()));
+    } else {
+      String serviceAccountName = getName(serviceAccounts, binding.getPrincipal());
+      return getTopologyAclBinding(binding, serviceAccountName);
+    }
+  }
+
+  private AclBinding convertToServiceAccountName(
+      Map<Integer, ServiceAccount> serviceAccounts, AclBinding aclBinding) {
+    String serviceAccountName = getName(serviceAccounts, aclBinding.entry().principal());
+    return getAclBinding(aclBinding, serviceAccountName);
+  }
+
+  private AclBinding getAclBinding(AclBinding aclBinding, String principle) {
+    AccessControlEntry entry = aclBinding.entry();
+    AccessControlEntry accessControlEntry = new AccessControlEntry(principle, entry.host(), entry.operation(), entry.permissionType());
+    return new AclBinding(aclBinding.pattern(), accessControlEntry);
+  }
+
+  private TopologyAclBinding getTopologyAclBinding(TopologyAclBinding binding, String principle) {
+    return new TopologyAclBinding(
+            binding.getResourceType(),
+            binding.getResourceName(),
+            binding.getHost(),
+            binding.getOperation(),
+            principle,
+            binding.getPattern());
+  }
+
+  private Integer getId(Map<String, ServiceAccount> serviceAccounts, String name) {
+    if (serviceAccounts.containsKey(name)) {
+      return serviceAccounts.get(name).getId();
+    } else {
+      try {
+        return Integer.parseInt(name);
+      } catch (NumberFormatException e) {
+        return null;
+      }
+    }
+  }
+
+  private String getName(Map<Integer, ServiceAccount> serviceAccounts, String id) {
+    try {
+      int cCloudId = Integer.parseInt(id.replace("User:", ""));
+      return getName(serviceAccounts, cCloudId);
+    } catch (NumberFormatException nfe) {
+      return id;
+    }
+  }
+
+  private String getName(Map<Integer, ServiceAccount> serviceAccounts, Integer id) {
+    return serviceAccounts.containsKey(id) ? serviceAccounts.get(id).getName() : id.toString();
+  }
+}
diff --git a/src/main/java/com/purbon/kafka/topology/serviceAccounts/VoidPrincipalProvider.java b/src/main/java/com/purbon/kafka/topology/serviceAccounts/VoidPrincipalProvider.java
index cf4ddbd..0000883 100644
--- a/src/main/java/com/purbon/kafka/topology/serviceAccounts/VoidPrincipalProvider.java
+++ b/src/main/java/com/purbon/kafka/topology/serviceAccounts/VoidPrincipalProvider.java
@@ -3,6 +3,7 @@ package com.purbon.kafka.topology.serviceAccounts;
 import com.purbon.kafka.topology.PrincipalProvider;
 import com.purbon.kafka.topology.model.cluster.ServiceAccount;
 import java.io.IOException;
+import java.util.Collections;
 import java.util.Set;
 
 public class VoidPrincipalProvider implements PrincipalProvider {
@@ -14,7 +15,7 @@ public class VoidPrincipalProvider implements PrincipalProvider {
 
   @Override
   public Set<ServiceAccount> listServiceAccounts() throws IOException {
-    throw new IOException("Not implemented!!");
+    return Collections.emptySet();
   }
 
   @Override
diff --git a/src/main/resources/reference.conf b/src/main/resources/reference.conf
index 1efb73c..1966e49 100644
--- a/src/main/resources/reference.conf
+++ b/src/main/resources/reference.conf
@@ -30,6 +30,7 @@ topology {
       format = "default"
       separator = "."
     }
+    managed.prefixes = []
   }
   project {
     prefix {
@@ -42,6 +43,7 @@ topology {
     }
   }
   state {
+     cluster.enabled = false
      topics {
         cluster.enabled = true
      }
@@ -51,6 +53,9 @@ topology {
       managed.prefixes = []
     }
   }
+  group {
+    managed.prefixes = []
+  }
 }
 
 schema {


Test patch:
diff --git a/src/test/java/com/purbon/kafka/topology/AccessControlManagerTest.java b/src/test/java/com/purbon/kafka/topology/AccessControlManagerTest.java
index 72f4ffd..d60e09f 100644
--- a/src/test/java/com/purbon/kafka/topology/AccessControlManagerTest.java
+++ b/src/test/java/com/purbon/kafka/topology/AccessControlManagerTest.java
@@ -2,12 +2,13 @@ package com.purbon.kafka.topology;
 
 import static com.google.common.collect.Lists.newArrayList;
 import static com.purbon.kafka.topology.BuilderCLI.BROKERS_OPTION;
-import static com.purbon.kafka.topology.TopologyBuilderConfig.OPTIMIZED_ACLS_CONFIG;
+import static com.purbon.kafka.topology.TopologyBuilderConfig.*;
 import static java.util.Arrays.asList;
 import static java.util.Collections.singletonList;
 import static org.hamcrest.CoreMatchers.hasItem;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.collection.IsCollectionWithSize.hasSize;
+import static org.junit.Assert.assertEquals;
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.anyString;
 import static org.mockito.Matchers.eq;
@@ -556,4 +557,142 @@ public class AccessControlManagerTest {
     backendController.reset();
     return backendController;
   }
+
+  @Test
+  public void testToProcessOnlySelectedTopics() throws IOException {
+    Map<String, String> cliOps = new HashMap<>();
+    cliOps.put(BROKERS_OPTION, "");
+
+    Properties props = new Properties();
+    props.put(TOPIC_MANAGED_PREFIXES, Collections.singletonList("NamespaceA"));
+    props.put(TOPIC_PREFIX_FORMAT_CONFIG, "{{topic}}");
+
+    TopologyBuilderConfig config = new TopologyBuilderConfig(cliOps, props);
+    accessControlManager =
+        new AccessControlManager(aclsProvider, new AclsBindingsBuilder(config), config);
+
+    TestTopologyBuilder builder =
+        TestTopologyBuilder.createProject(config)
+            .addTopic("topicA")
+            .addTopic("NamespaceA_topicA")
+            .addConsumer("User:app1");
+
+    accessControlManager.apply(builder.buildTopology(), plan);
+
+    // Check that we only have one action not 2
+    assertEquals(1, plan.getActions().size());
+
+    // Check that the action bindings are for the managed prefix topic, not the non-managed prefix.
+    assertEquals(
+        2,
+        plan.getActions().get(0).getBindings().stream()
+            .filter(
+                b ->
+                    b.getResourceType().equals(ResourceType.TOPIC)
+                        && b.getResourceName().equals("NamespaceA_topicA"))
+            .count());
+    assertEquals(
+        0,
+        plan.getActions().get(0).getBindings().stream()
+            .filter(
+                b ->
+                    b.getResourceType().equals(ResourceType.TOPIC)
+                        && b.getResourceName().equals("topicA"))
+            .count());
+  }
+
+  @Test
+  public void testToProcessOnlySelectedGroupsOrWildcard() throws IOException {
+    Map<String, String> cliOps = new HashMap<>();
+    cliOps.put(BROKERS_OPTION, "");
+
+    Properties props = new Properties();
+    props.put(GROUP_MANAGED_PREFIXES, Collections.singletonList("NamespaceA"));
+
+    TopologyBuilderConfig config = new TopologyBuilderConfig(cliOps, props);
+    accessControlManager =
+        new AccessControlManager(aclsProvider, new AclsBindingsBuilder(config), config);
+
+    TestTopologyBuilder builder =
+        TestTopologyBuilder.createProject(config)
+            .addTopic("topicA")
+            .addConsumer("User:app1", "NamespaceA_ConsumerGroupA")
+            .addConsumer("User:app2", "NamespaceB_ConsumerGroupB")
+            .addConsumer("User:app3", "*");
+
+    accessControlManager.apply(builder.buildTopology(), plan);
+
+    // Check that we only have one action
+    assertEquals(1, plan.getActions().size());
+
+    // Check that the action bindings are for the managed prefix group, not the non-managed prefix.
+    assertEquals(
+        1,
+        plan.getActions().get(0).getBindings().stream()
+            .filter(
+                b ->
+                    b.getResourceType().equals(ResourceType.GROUP)
+                        && b.getResourceName().equals("NamespaceA_ConsumerGroupA"))
+            .count());
+    assertEquals(
+        0,
+        plan.getActions().get(0).getBindings().stream()
+            .filter(
+                b ->
+                    b.getResourceType().equals(ResourceType.GROUP)
+                        && b.getResourceName().equals("NamespaceB_ConsumerGroupB"))
+            .count());
+    assertEquals(
+        1,
+        plan.getActions().get(0).getBindings().stream()
+            .filter(
+                b ->
+                    b.getResourceType().equals(ResourceType.GROUP)
+                        && b.getResourceName().equals("*"))
+            .count());
+  }
+
+  @Test
+  public void testToProcessWildcardGroupOnlySelectedServiceAccounts() throws IOException {
+    Map<String, String> cliOps = new HashMap<>();
+    cliOps.put(BROKERS_OPTION, "");
+
+    Properties props = new Properties();
+    props.put(SERVICE_ACCOUNT_MANAGED_PREFIXES, Collections.singletonList("User:NamespaceA"));
+
+    TopologyBuilderConfig config = new TopologyBuilderConfig(cliOps, props);
+    accessControlManager =
+        new AccessControlManager(aclsProvider, new AclsBindingsBuilder(config), config);
+
+    TestTopologyBuilder builder =
+        TestTopologyBuilder.createProject(config)
+            .addTopic("topicA")
+            .addConsumer("User:NamespaceA_app1", "*")
+            .addConsumer("User:NamespaceB_app2", "*");
+
+    accessControlManager.apply(builder.buildTopology(), plan);
+
+    // Check that we only have one action
+    assertEquals(1, plan.getActions().size());
+
+    // Check that the action bindings are for the managed service group, not the non-managed prefix.
+    assertEquals(
+        1,
+        plan.getActions().get(0).getBindings().stream()
+            .filter(
+                b ->
+                    b.getResourceType().equals(ResourceType.GROUP)
+                        && b.getResourceName().equals("*")
+                        && b.getPrincipal().equals("User:NamespaceA_app1"))
+            .count());
+    assertEquals(
+        0,
+        plan.getActions().get(0).getBindings().stream()
+            .filter(
+                b ->
+                    b.getResourceType().equals(ResourceType.GROUP)
+                        && b.getResourceName().equals("*")
+                        && b.getPrincipal().equals("User:NamespaceB_app2"))
+            .count());
+  }
 }
diff --git a/src/test/java/com/purbon/kafka/topology/PrincipalManagerTest.java b/src/test/java/com/purbon/kafka/topology/PrincipalManagerTest.java
index 403b620..256d97f 100644
--- a/src/test/java/com/purbon/kafka/topology/PrincipalManagerTest.java
+++ b/src/test/java/com/purbon/kafka/topology/PrincipalManagerTest.java
@@ -4,7 +4,7 @@ import static com.purbon.kafka.topology.BuilderCLI.ALLOW_DELETE_OPTION;
 import static com.purbon.kafka.topology.BuilderCLI.BROKERS_OPTION;
 import static com.purbon.kafka.topology.TopologyBuilderConfig.SERVICE_ACCOUNT_MANAGED_PREFIXES;
 import static com.purbon.kafka.topology.TopologyBuilderConfig.TOPOLOGY_EXPERIMENTAL_ENABLED_CONFIG;
-import static com.purbon.kafka.topology.TopologyBuilderConfig.TOPOLOGY_TOPIC_STATE_FROM_CLUSTER;
+import static com.purbon.kafka.topology.TopologyBuilderConfig.TOPOLOGY_STATE_FROM_CLUSTER;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
@@ -74,7 +74,7 @@ public class PrincipalManagerTest {
     props = new Properties();
 
     props.put(TOPOLOGY_EXPERIMENTAL_ENABLED_CONFIG, "true");
-    props.put(TOPOLOGY_TOPIC_STATE_FROM_CLUSTER, "false");
+    props.put(TOPOLOGY_STATE_FROM_CLUSTER, "false");
 
     plan = ExecutionPlan.init(backendController, mockPrintStream);
     config = new TopologyBuilderConfig(cliOps, props);
@@ -93,7 +93,8 @@ public class PrincipalManagerTest {
 
     doNothing().when(provider).configure();
 
-    principalManager.apply(topology, plan);
+    principalManager.applyCreate(topology, plan);
+    principalManager.applyDelete(topology, plan);
 
     Collection<ServiceAccount> accounts =
         Arrays.asList(
@@ -119,7 +120,8 @@ public class PrincipalManagerTest {
 
     doNothing().when(provider).configure();
 
-    principalManager.apply(topology, plan);
+    principalManager.applyCreate(topology, plan);
+    principalManager.applyDelete(topology, plan);
 
     Collection<ServiceAccount> accounts =
         Arrays.asList(
@@ -150,7 +152,8 @@ public class PrincipalManagerTest {
         .when(provider)
         .createServiceAccount(eq("producer"), eq("Managed by KTB"));
 
-    principalManager.apply(topology, plan);
+    principalManager.applyCreate(topology, plan);
+    principalManager.applyDelete(topology, plan);
     plan.run();
     assertThat(plan.getServiceAccounts()).hasSize(2);
 
@@ -162,7 +165,9 @@ public class PrincipalManagerTest {
 
     backendController = new BackendController();
     plan = ExecutionPlan.init(backendController, mockPrintStream);
-    principalManager.apply(topology, plan);
+    principalManager.applyCreate(topology, plan);
+    principalManager.applyDelete(topology, plan);
+    ;
 
     Collection<ServiceAccount> accounts =
         Arrays.asList(new ServiceAccount(124, "producer", "Managed by KTB"));
@@ -186,7 +191,8 @@ public class PrincipalManagerTest {
 
     Topology topology = new TopologyImpl();
 
-    principalManager.apply(topology, mockPlan);
+    principalManager.applyCreate(topology, plan);
+    principalManager.applyDelete(topology, plan);
 
     verify(mockPlan, times(0)).add(any(Action.class));
   }
@@ -216,7 +222,8 @@ public class PrincipalManagerTest {
         .when(provider)
         .createServiceAccount(eq("producer"), eq("Managed by KTB"));
 
-    principalManager.apply(topology, plan);
+    principalManager.applyCreate(topology, plan);
+    principalManager.applyDelete(topology, plan);
     plan.run();
 
     assertThat(plan.getServiceAccounts()).hasSize(1);
diff --git a/src/test/java/com/purbon/kafka/topology/TestTopologyBuilder.java b/src/test/java/com/purbon/kafka/topology/TestTopologyBuilder.java
index 0190161..0edd0e9 100644
--- a/src/test/java/com/purbon/kafka/topology/TestTopologyBuilder.java
+++ b/src/test/java/com/purbon/kafka/topology/TestTopologyBuilder.java
@@ -10,10 +10,12 @@ import com.purbon.kafka.topology.model.users.Consumer;
 import com.purbon.kafka.topology.model.users.Producer;
 import java.util.ArrayList;
 import java.util.HashSet;
+import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 
 public class TestTopologyBuilder {
+  private final TopologyBuilderConfig topologyBuilderConfig;
   private final Topology topology;
   private final Project project;
   private Set<Topic> topics = new HashSet<>();
@@ -21,13 +23,15 @@ public class TestTopologyBuilder {
   private final Set<Producer> producers = new HashSet<>();
 
   public TestTopologyBuilder() {
-    this("ctx", "project");
+    this(new TopologyBuilderConfig(), "ctx", "project");
   }
 
-  public TestTopologyBuilder(String context, String projectName) {
-    topology = new TopologyImpl();
+  public TestTopologyBuilder(
+      TopologyBuilderConfig topologyBuilderConfig, String context, String projectName) {
+    this.topologyBuilderConfig = topologyBuilderConfig;
+    topology = new TopologyImpl(topologyBuilderConfig);
     topology.setContext(context);
-    project = new ProjectImpl(projectName);
+    project = new ProjectImpl(projectName, topologyBuilderConfig);
     topology.addProject(project);
   }
 
@@ -36,7 +40,11 @@ public class TestTopologyBuilder {
   }
 
   public static TestTopologyBuilder createProject(String context, String projectName) {
-    return new TestTopologyBuilder(context, projectName);
+    return new TestTopologyBuilder(new TopologyBuilderConfig(), context, projectName);
+  }
+
+  public static TestTopologyBuilder createProject(TopologyBuilderConfig topologyBuilderConfig) {
+    return new TestTopologyBuilder(topologyBuilderConfig, "ctx", "project");
   }
 
   public Topology buildTopology() {
@@ -47,7 +55,7 @@ public class TestTopologyBuilder {
   }
 
   public TestTopologyBuilder addTopic(String topicName) {
-    addTopic(new TopicImpl(topicName));
+    addTopic(new TopicImpl(topicName, topologyBuilderConfig));
     return this;
   }
 
@@ -70,6 +78,13 @@ public class TestTopologyBuilder {
     return this;
   }
 
+  public TestTopologyBuilder addConsumer(String user, String group) {
+    Consumer consumer = new Consumer(user);
+    consumer.setGroup(Optional.of(group));
+    consumers.add(consumer);
+    return this;
+  }
+
   @SuppressWarnings("UnusedReturnValue")
   public TestTopologyBuilder removeConsumer(String user) {
     consumers.remove(new Consumer(user));
diff --git a/src/test/java/com/purbon/kafka/topology/TopicManagerTest.java b/src/test/java/com/purbon/kafka/topology/TopicManagerTest.java
index b258455..24cfe35 100644
--- a/src/test/java/com/purbon/kafka/topology/TopicManagerTest.java
+++ b/src/test/java/com/purbon/kafka/topology/TopicManagerTest.java
@@ -3,9 +3,7 @@ package com.purbon.kafka.topology;
 import static com.purbon.kafka.topology.BuilderCLI.ALLOW_DELETE_OPTION;
 import static com.purbon.kafka.topology.BuilderCLI.BROKERS_OPTION;
 import static com.purbon.kafka.topology.TopicManager.NUM_PARTITIONS;
-import static com.purbon.kafka.topology.TopologyBuilderConfig.ALLOW_DELETE_TOPICS;
-import static com.purbon.kafka.topology.TopologyBuilderConfig.KAFKA_INTERNAL_TOPIC_PREFIXES;
-import static com.purbon.kafka.topology.TopologyBuilderConfig.TOPOLOGY_TOPIC_STATE_FROM_CLUSTER;
+import static com.purbon.kafka.topology.TopologyBuilderConfig.*;
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.*;
 
@@ -341,4 +339,30 @@ public class TopicManagerTest {
 
     verify(outputStream, times(2)).println(any(Action.class));
   }
+
+  @Test
+  public void testToProcessOnlySelectedTopics() throws IOException {
+
+    props.put(TOPIC_MANAGED_PREFIXES, Collections.singletonList("NamespaceA"));
+    props.put(TOPIC_PREFIX_FORMAT_CONFIG, "{{topic}}");
+    props.put(TOPOLOGY_TOPIC_STATE_FROM_CLUSTER, "true");
+
+    TopologyBuilderConfig config = new TopologyBuilderConfig(cliOps, props);
+    TopicManager topicManager = new TopicManager(adminClient, schemaRegistryManager, config);
+
+    Project project = new ProjectImpl("project");
+    Topic topicA = new TopicImpl("NamespaceA_TopicA", config);
+    project.addTopic(topicA);
+    Topic topicB = new TopicImpl("NamespaceB_TopicB", config);
+    project.addTopic(topicB);
+    Topology topology = new TopologyImpl(config);
+    topology.addProject(project);
+
+    when(adminClient.listApplicationTopics()).thenReturn(new HashSet<>());
+    topicManager.apply(topology, plan);
+    plan.run();
+
+    verify(adminClient, times(1)).createTopic(topicA, topicA.toString());
+    verify(adminClient, times(0)).createTopic(topicB, topicB.toString());
+  }
 }