Repository: kafka-ops/kafka-topology-builder (java)

Issue:
Reduce number of ACLs created by moving to prefixed ACLs
Given the limitations on the total number of ACLs in Confluent Cloud 
(currently 1,000 for basic and standard, and 10,000 for dedicated cluster)
and when use centralized ACLs (currently 1,000 per cluster), we should strive to optimize the
total number of ACLs being created.

- Currently, we create one literal ACL for each consumer/producer and each topic in a project for the necessary operations (READ/WRITE and DESCRIBE).
This could be replaced with a single prefixed ACL for each consumer/producer and operation combination.

- READ entries for consumer groups be prefixed instead of providing wildcard access by default.

Pull request description:
Add an option of grouping ACLs for consumers and producers 
This function is selected via configuration option.

tasks:

- [x] Grouping of ACLs for traditional ACLs
- [x] Grouping of ACLs for traditional for RBAC
- [x] Update documentation 

closes #80

Patch:
diff --git a/docs/config-values.rst b/docs/config-values.rst
index c2e94d1..57ba729 100644
--- a/docs/config-values.rst
+++ b/docs/config-values.rst
@@ -67,6 +67,18 @@ As a user you can customize:
 - **Property**: *topology.project.prefix.format*, to set the project level name format, it should be a subset of the previous one.
 - **Property**: *topology.topic.prefix.separator*, to select a custom separator between attributes.
 
+Optimised number of ACLs and RBAC bindings
+-----------
+
+This property is used to reduce the number of ACLs, or RBAC bindings, created. In the normal operational mode, the KTB, will create direct pair of bindings for each user and topic.
+However for some organisations, it might be enough, to create an optimised list by using prefixed bindings.
+
+**Property**: *topology.acls.optimized*
+**Default value**: "false"
+
+An example configuration might look like this:
+::
+    topology.acls.optimized=true
 
 Internal topics prefixes
 -----------
diff --git a/docs/futures/what-acl-management.rst b/docs/futures/what-acl-management.rst
index e43c56f..6ee13a1 100644
--- a/docs/futures/what-acl-management.rst
+++ b/docs/futures/what-acl-management.rst
@@ -40,6 +40,31 @@ Consumer definition with principal "User:App0" and without an specific consumer
 
 Consumer definition with principal "User:App0" and consumer group name "foo".
 
+In the default mode the KTB will create dedicated ACL for each user and topic pair. For organisations that aim not to have dedicated pair or rules the KTB offer the option
+to optimise the number of ACLs using prefixed rules.
+
+The optimised ACLs/RBAC can be enabled using the *topology.acls.optimized* configuration property.
+
+Producers
+^^^^^^^^^^^
+As a user of KTB you can configure the required set of producers for your application.
+
+Producers have a principal.
+
+.. code-block:: YAML
+
+  ---
+    context: "context"
+    source: "source"
+    projects:
+      - name: "foo"
+        producers:
+          - principal: "User:App0"
+
+In the default mode the KTB will create dedicated ACL for each user and topic pair. For organisations that aim not to have dedicated pair or rules the KTB offer the option
+to optimise the number of ACLs using prefixed rules.
+
+The optimised ACLs/RBAC can be enabled using the *topology.acls.optimized* configuration property.
 
 Connectors
 ^^^^^^^^^^^
diff --git a/src/main/conf/topology-builder.properties b/src/main/conf/topology-builder.properties
index 671d62b..4e6d702 100644
--- a/src/main/conf/topology-builder.properties
+++ b/src/main/conf/topology-builder.properties
@@ -27,6 +27,9 @@
 ## Topology files type, the default value is YAML
 # topology.file.type=JSON
 
+## Use optimised access control bindings
+# topology.acls.optimized=true
+
 ## SASL connection configuration
 #sasl.mechanism=PLAIN
 #sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
diff --git a/src/main/java/com/purbon/kafka/topology/AccessControlManager.java b/src/main/java/com/purbon/kafka/topology/AccessControlManager.java
index 2cc79ce..a328b28 100644
--- a/src/main/java/com/purbon/kafka/topology/AccessControlManager.java
+++ b/src/main/java/com/purbon/kafka/topology/AccessControlManager.java
@@ -7,16 +7,8 @@ import static com.purbon.kafka.topology.model.Component.SCHEMA_REGISTRY;
 import com.purbon.kafka.topology.actions.Action;
 import com.purbon.kafka.topology.actions.access.ClearBindings;
 import com.purbon.kafka.topology.actions.access.CreateBindings;
-import com.purbon.kafka.topology.actions.access.builders.BuildBindingsForConsumer;
-import com.purbon.kafka.topology.actions.access.builders.BuildBindingsForControlCenter;
-import com.purbon.kafka.topology.actions.access.builders.BuildBindingsForKConnect;
-import com.purbon.kafka.topology.actions.access.builders.BuildBindingsForKStreams;
-import com.purbon.kafka.topology.actions.access.builders.BuildBindingsForProducer;
-import com.purbon.kafka.topology.actions.access.builders.BuildBindingsForSchemaRegistry;
-import com.purbon.kafka.topology.actions.access.builders.rbac.BuildBindingsForConnectorAuthorization;
-import com.purbon.kafka.topology.actions.access.builders.rbac.BuildBindingsForSchemaAuthorization;
-import com.purbon.kafka.topology.actions.access.builders.rbac.BuildClusterLevelBinding;
-import com.purbon.kafka.topology.actions.access.builders.rbac.BuildPredefinedBinding;
+import com.purbon.kafka.topology.actions.access.builders.*;
+import com.purbon.kafka.topology.actions.access.builders.rbac.*;
 import com.purbon.kafka.topology.model.Component;
 import com.purbon.kafka.topology.model.DynamicUser;
 import com.purbon.kafka.topology.model.Platform;
@@ -86,24 +78,11 @@ public class AccessControlManager {
     List<Action> actions = new ArrayList<>();
 
     for (Project project : topology.getProjects()) {
-      project
-          .getTopics()
-          .forEach(
-              topic -> {
-                final String fullTopicName = topic.toString();
-                if (!project.getConsumers().isEmpty()) {
-                  Action action =
-                      new BuildBindingsForConsumer(
-                          bindingsBuilder, project.getConsumers(), fullTopicName);
-                  actions.add(action);
-                }
-                if (!project.getProducers().isEmpty()) {
-                  Action action =
-                      new BuildBindingsForProducer(
-                          bindingsBuilder, project.getProducers(), fullTopicName);
-                  actions.add(action);
-                }
-              });
+      if (config.shouldOptimizeAcls()) {
+        actions.addAll(buildOptimizeConsumerAndProducerAcls(project));
+      } else {
+        actions.addAll(buildDetailedConsumerAndProducerAcls(project));
+      }
       // Setup global Kafka Stream Access control lists
       String topicPrefix = project.namePrefix();
       for (KStream app : project.getStreams()) {
@@ -129,6 +108,40 @@ public class AccessControlManager {
     return actions;
   }
 
+  private List<Action> buildOptimizeConsumerAndProducerAcls(Project project) {
+    List<Action> actions = new ArrayList<>();
+    actions.add(
+        new BuildBindingsForConsumer(
+            bindingsBuilder, project.getConsumers(), project.namePrefix(), true));
+    actions.add(
+        new BuildBindingsForProducer(
+            bindingsBuilder, project.getProducers(), project.namePrefix(), true));
+    return actions;
+  }
+
+  private List<Action> buildDetailedConsumerAndProducerAcls(Project project) {
+    List<Action> actions = new ArrayList<>();
+    project
+        .getTopics()
+        .forEach(
+            topic -> {
+              final String fullTopicName = topic.toString();
+              if (!project.getConsumers().isEmpty()) {
+                Action action =
+                    new BuildBindingsForConsumer(
+                        bindingsBuilder, project.getConsumers(), fullTopicName, false);
+                actions.add(action);
+              }
+              if (!project.getProducers().isEmpty()) {
+                Action action =
+                    new BuildBindingsForProducer(
+                        bindingsBuilder, project.getProducers(), fullTopicName, false);
+                actions.add(action);
+              }
+            });
+    return actions;
+  }
+
   /**
    * Build a list of actions required to create or delete necessary bindings
    *
diff --git a/src/main/java/com/purbon/kafka/topology/BindingsBuilderProvider.java b/src/main/java/com/purbon/kafka/topology/BindingsBuilderProvider.java
index 97638a5..2dc1c58 100644
--- a/src/main/java/com/purbon/kafka/topology/BindingsBuilderProvider.java
+++ b/src/main/java/com/purbon/kafka/topology/BindingsBuilderProvider.java
@@ -19,9 +19,11 @@ public interface BindingsBuilderProvider {
   List<TopologyAclBinding> buildBindingsForStreamsApp(
       String principal, String topicPrefix, List<String> readTopics, List<String> writeTopics);
 
-  List<TopologyAclBinding> buildBindingsForConsumers(Collection<Consumer> consumers, String topic);
+  List<TopologyAclBinding> buildBindingsForConsumers(
+      Collection<Consumer> consumers, String resource, boolean prefixed);
 
-  List<TopologyAclBinding> buildBindingsForProducers(Collection<Producer> producers, String topic);
+  List<TopologyAclBinding> buildBindingsForProducers(
+      Collection<Producer> principals, String resource, boolean prefixed);
 
   default TopologyAclBinding setPredefinedRole(
       String principal, String predefinedRole, String topicPrefix) {
diff --git a/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java b/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java
index a8f700a..89901b6 100644
--- a/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java
+++ b/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java
@@ -62,6 +62,8 @@ public class TopologyBuilderConfig {
 
   static final String TOPOLOGY_FILE_TYPE = "topology.file.type";
 
+  static final String OPTIMIZED_ACLS_CONFIG = "topology.acls.optimized";
+
   private final Map<String, String> cliParams;
   private Config config;
 
@@ -257,6 +259,10 @@ public class TopologyBuilderConfig {
     return config.getString(TOPIC_PREFIX_SEPARATOR_CONFIG);
   }
 
+  public Boolean shouldOptimizeAcls() {
+    return config.getBoolean(OPTIMIZED_ACLS_CONFIG);
+  }
+
   public List<String> getTopologyValidations() {
     List<String> classes = config.getStringList(TOPOLOGY_VALIDATIONS_CONFIG);
     return classes.stream().map(String::trim).collect(Collectors.toList());
diff --git a/src/main/java/com/purbon/kafka/topology/actions/access/builders/BuildBindingsForConsumer.java b/src/main/java/com/purbon/kafka/topology/actions/access/builders/BuildBindingsForConsumer.java
index e97ec7f..95402de 100644
--- a/src/main/java/com/purbon/kafka/topology/actions/access/builders/BuildBindingsForConsumer.java
+++ b/src/main/java/com/purbon/kafka/topology/actions/access/builders/BuildBindingsForConsumer.java
@@ -6,34 +6,36 @@ import com.purbon.kafka.topology.model.users.Consumer;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.stream.Collectors;
 
 public class BuildBindingsForConsumer extends BaseAccessControlAction {
 
   private final String fullTopicName;
   private final List<Consumer> consumers;
   private final BindingsBuilderProvider builderProvider;
+  private boolean prefixed;
 
   public BuildBindingsForConsumer(
-      BindingsBuilderProvider builderProvider, List<Consumer> consumers, String fullTopicName) {
+      BindingsBuilderProvider builderProvider,
+      List<Consumer> consumers,
+      String fullTopicName,
+      boolean prefixed) {
     super();
     this.consumers = consumers;
     this.fullTopicName = fullTopicName;
     this.builderProvider = builderProvider;
+    this.prefixed = prefixed;
   }
 
   @Override
   protected void execute() {
-    bindings = builderProvider.buildBindingsForConsumers(consumers, fullTopicName);
+    bindings = builderProvider.buildBindingsForConsumers(consumers, fullTopicName, prefixed);
   }
 
   @Override
   protected Map<String, Object> props() {
-    List<String> principals =
-        consumers.stream().map(c -> c.getPrincipal()).collect(Collectors.toList());
     Map<String, Object> map = new HashMap<>();
     map.put("Operation", getClass().getName());
-    map.put("Principals", principals);
+    map.put("Principals", consumers);
     map.put("Topic", fullTopicName);
     return map;
   }
diff --git a/src/main/java/com/purbon/kafka/topology/actions/access/builders/BuildBindingsForProducer.java b/src/main/java/com/purbon/kafka/topology/actions/access/builders/BuildBindingsForProducer.java
index 8e9b78f..62839b7 100644
--- a/src/main/java/com/purbon/kafka/topology/actions/access/builders/BuildBindingsForProducer.java
+++ b/src/main/java/com/purbon/kafka/topology/actions/access/builders/BuildBindingsForProducer.java
@@ -3,38 +3,39 @@ package com.purbon.kafka.topology.actions.access.builders;
 import com.purbon.kafka.topology.BindingsBuilderProvider;
 import com.purbon.kafka.topology.actions.BaseAccessControlAction;
 import com.purbon.kafka.topology.model.users.Producer;
-import java.io.IOException;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.stream.Collectors;
 
 public class BuildBindingsForProducer extends BaseAccessControlAction {
 
   private final BindingsBuilderProvider builderProvider;
   private final List<Producer> producers;
   private final String fullTopicName;
+  private final boolean prefixed;
 
   public BuildBindingsForProducer(
-      BindingsBuilderProvider builderProvider, List<Producer> producers, String fullTopicName) {
+      BindingsBuilderProvider builderProvider,
+      List<Producer> producers,
+      String fullTopicName,
+      boolean prefixed) {
     super();
     this.builderProvider = builderProvider;
     this.producers = producers;
     this.fullTopicName = fullTopicName;
+    this.prefixed = prefixed;
   }
 
   @Override
-  protected void execute() throws IOException {
-    bindings = builderProvider.buildBindingsForProducers(producers, fullTopicName);
+  protected void execute() {
+    bindings = builderProvider.buildBindingsForProducers(producers, fullTopicName, prefixed);
   }
 
   @Override
   protected Map<String, Object> props() {
-    List<String> principals =
-        producers.stream().map(p -> p.getPrincipal()).collect(Collectors.toList());
     Map<String, Object> map = new HashMap<>();
     map.put("Operation", getClass().getName());
-    map.put("Principals", principals);
+    map.put("Principals", producers);
     map.put("Topic", fullTopicName);
     return map;
   }
diff --git a/src/main/java/com/purbon/kafka/topology/roles/acls/AclsBindingsBuilder.java b/src/main/java/com/purbon/kafka/topology/roles/acls/AclsBindingsBuilder.java
index ed8b2e2..aa657fc 100644
--- a/src/main/java/com/purbon/kafka/topology/roles/acls/AclsBindingsBuilder.java
+++ b/src/main/java/com/purbon/kafka/topology/roles/acls/AclsBindingsBuilder.java
@@ -88,14 +88,16 @@ public class AclsBindingsBuilder implements BindingsBuilderProvider {
 
   @Override
   public List<TopologyAclBinding> buildBindingsForConsumers(
-      Collection<Consumer> consumers, String topic) {
-    return toList(consumers.stream().flatMap(consumer -> consumerAclsStream(consumer, topic)));
+      Collection<Consumer> consumers, String resource, boolean prefixed) {
+    return toList(
+        consumers.stream().flatMap(consumer -> consumerAclsStream(consumer, resource, prefixed)));
   }
 
   @Override
   public List<TopologyAclBinding> buildBindingsForProducers(
-      Collection<Producer> producers, String topic) {
-    return toList(producers.stream().flatMap(p -> producerAclsStream(p, topic)));
+      Collection<Producer> producers, String resource, boolean prefixed) {
+    return toList(
+        producers.stream().flatMap(producer -> producerAclsStream(producer, resource, prefixed)));
   }
 
   @Override
@@ -113,16 +115,15 @@ public class AclsBindingsBuilder implements BindingsBuilderProvider {
     return bindingStream.map(TopologyAclBinding::new).collect(Collectors.toList());
   }
 
-  private Stream<AclBinding> producerAclsStream(Producer producer, String topic) {
+  private Stream<AclBinding> producerAclsStream(Producer producer, String topic, boolean prefixed) {
+    PatternType patternType = prefixed ? PatternType.PREFIXED : PatternType.LITERAL;
 
     List<AclBinding> bindings = new ArrayList<>();
 
     bindings.addAll(
         Arrays.asList(
-            buildTopicLevelAcl(
-                producer.getPrincipal(), topic, PatternType.LITERAL, AclOperation.DESCRIBE),
-            buildTopicLevelAcl(
-                producer.getPrincipal(), topic, PatternType.LITERAL, AclOperation.WRITE)));
+            buildTopicLevelAcl(producer.getPrincipal(), topic, patternType, AclOperation.DESCRIBE),
+            buildTopicLevelAcl(producer.getPrincipal(), topic, patternType, AclOperation.WRITE)));
 
     producer
         .getTransactionId()
@@ -154,16 +155,13 @@ public class AclsBindingsBuilder implements BindingsBuilderProvider {
     return bindings.stream();
   }
 
-  private Stream<AclBinding> consumerAclsStream(Consumer consumer, String topic) {
+  private Stream<AclBinding> consumerAclsStream(Consumer consumer, String topic, boolean prefixed) {
+    PatternType patternType = prefixed ? PatternType.PREFIXED : PatternType.LITERAL;
     return Stream.of(
-        buildTopicLevelAcl(
-            consumer.getPrincipal(), topic, PatternType.LITERAL, AclOperation.DESCRIBE),
-        buildTopicLevelAcl(consumer.getPrincipal(), topic, PatternType.LITERAL, AclOperation.READ),
+        buildTopicLevelAcl(consumer.getPrincipal(), topic, patternType, AclOperation.DESCRIBE),
+        buildTopicLevelAcl(consumer.getPrincipal(), topic, patternType, AclOperation.READ),
         buildGroupLevelAcl(
-            consumer.getPrincipal(),
-            consumer.groupString(),
-            PatternType.LITERAL,
-            AclOperation.READ));
+            consumer.getPrincipal(), consumer.groupString(), patternType, AclOperation.READ));
   }
 
   private Stream<AclBinding> streamsAppStream(
diff --git a/src/main/java/com/purbon/kafka/topology/roles/rbac/RBACBindingsBuilder.java b/src/main/java/com/purbon/kafka/topology/roles/rbac/RBACBindingsBuilder.java
index 9c2fb5f..98daf95 100644
--- a/src/main/java/com/purbon/kafka/topology/roles/rbac/RBACBindingsBuilder.java
+++ b/src/main/java/com/purbon/kafka/topology/roles/rbac/RBACBindingsBuilder.java
@@ -119,21 +119,21 @@ public class RBACBindingsBuilder implements BindingsBuilderProvider {
 
   @Override
   public List<TopologyAclBinding> buildBindingsForConsumers(
-      Collection<Consumer> consumers, String topic) {
+      Collection<Consumer> consumers, String resource, boolean prefixed) {
+    String patternType = prefixed ? PREFIX : LITERAL;
     List<TopologyAclBinding> bindings = new ArrayList<>();
     consumers.forEach(
         consumer -> {
           TopologyAclBinding binding =
-              apiClient.bind(consumer.getPrincipal(), DEVELOPER_READ, topic, LITERAL);
+              apiClient.bind(consumer.getPrincipal(), DEVELOPER_READ, resource, patternType);
           bindings.add(binding);
-          String pattern = consumer.groupString().equals("*") ? PREFIX : LITERAL;
           binding =
               apiClient.bind(
                   consumer.getPrincipal(),
                   RESOURCE_OWNER,
                   consumer.groupString(),
                   "Group",
-                  pattern);
+                  LITERAL);
           bindings.add(binding);
         });
     return bindings;
@@ -141,12 +141,13 @@ public class RBACBindingsBuilder implements BindingsBuilderProvider {
 
   @Override
   public List<TopologyAclBinding> buildBindingsForProducers(
-      Collection<Producer> producers, String topic) {
+      Collection<Producer> producers, String resource, boolean prefixed) {
+    String patternType = prefixed ? PREFIX : LITERAL;
     List<TopologyAclBinding> bindings = new ArrayList<>();
     producers.forEach(
         producer -> {
           TopologyAclBinding binding =
-              apiClient.bind(producer.getPrincipal(), DEVELOPER_WRITE, topic, LITERAL);
+              apiClient.bind(producer.getPrincipal(), DEVELOPER_WRITE, resource, patternType);
           bindings.add(binding);
         });
     return bindings;
diff --git a/src/main/resources/reference.conf b/src/main/resources/reference.conf
index be1b6b0..9e439bc 100644
--- a/src/main/resources/reference.conf
+++ b/src/main/resources/reference.conf
@@ -1,4 +1,7 @@
 topology {
+  acls {
+    optimized = false
+  }
   file {
     type = "YAML"
   }


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 c2523fe..444c9e2 100644
--- a/src/test/java/com/purbon/kafka/topology/AccessControlManagerTest.java
+++ b/src/test/java/com/purbon/kafka/topology/AccessControlManagerTest.java
@@ -1,5 +1,7 @@
 package com.purbon.kafka.topology;
 
+import static com.purbon.kafka.topology.BuilderCLI.BROKERS_OPTION;
+import static com.purbon.kafka.topology.TopologyBuilderConfig.OPTIMIZED_ACLS_CONFIG;
 import static java.util.Arrays.asList;
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.anyString;
@@ -42,6 +44,7 @@ import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Properties;
 import java.util.stream.Collectors;
 import org.apache.kafka.common.acl.AclBinding;
 import org.apache.kafka.common.acl.AclOperation;
@@ -97,9 +100,42 @@ public class AccessControlManagerTest {
 
     doReturn(new ArrayList<TopologyAclBinding>())
         .when(aclsBuilder)
-        .buildBindingsForConsumers(users, topicA.toString());
+        .buildBindingsForConsumers(users, topicA.toString(), false);
     accessControlManager.apply(topology, plan);
-    verify(aclsBuilder, times(1)).buildBindingsForConsumers(eq(users), eq(topicA.toString()));
+    verify(aclsBuilder, times(1))
+        .buildBindingsForConsumers(eq(users), eq(topicA.toString()), eq(false));
+  }
+
+  @Test
+  public void newConsumerOptimisedACLsCreation() {
+
+    HashMap<String, String> cliOps = new HashMap<>();
+    cliOps.put(BROKERS_OPTION, "");
+    Properties props = new Properties();
+    props.put(OPTIMIZED_ACLS_CONFIG, true);
+
+    TopologyBuilderConfig config = new TopologyBuilderConfig(cliOps, props);
+    accessControlManager = new AccessControlManager(aclsProvider, aclsBuilder, config);
+
+    List<Consumer> consumers = new ArrayList<>();
+    consumers.add(new Consumer("User:app1"));
+    Project project = new ProjectImpl();
+    project.setConsumers(consumers);
+
+    Topic topicA = new TopicImpl("topicA");
+    project.addTopic(topicA);
+
+    Topology topology = new TopologyImpl();
+    topology.addProject(project);
+
+    List<Consumer> users = asList(new Consumer("User:app1"));
+
+    doReturn(new ArrayList<TopologyAclBinding>())
+        .when(aclsBuilder)
+        .buildBindingsForConsumers(users, project.namePrefix(), true);
+    accessControlManager.apply(topology, plan);
+    verify(aclsBuilder, times(1))
+        .buildBindingsForConsumers(eq(users), eq(project.namePrefix()), eq(true));
   }
 
   @Test
@@ -118,9 +154,40 @@ public class AccessControlManagerTest {
 
     doReturn(new ArrayList<TopologyAclBinding>())
         .when(aclsBuilder)
-        .buildBindingsForProducers(producers, topicA.toString());
+        .buildBindingsForProducers(producers, topicA.toString(), false);
+
+    accessControlManager.apply(topology, plan);
+    verify(aclsBuilder, times(1))
+        .buildBindingsForProducers(eq(producers), eq(topicA.toString()), eq(false));
+  }
+
+  @Test
+  public void newProducerOptimizedACLsCreation() {
+    HashMap<String, String> cliOps = new HashMap<>();
+    cliOps.put(BROKERS_OPTION, "");
+    Properties props = new Properties();
+    props.put(OPTIMIZED_ACLS_CONFIG, true);
+
+    TopologyBuilderConfig config = new TopologyBuilderConfig(cliOps, props);
+    accessControlManager = new AccessControlManager(aclsProvider, aclsBuilder, config);
+
+    List<Producer> producers = new ArrayList<>();
+    producers.add(new Producer("User:app1"));
+    Project project = new ProjectImpl();
+    project.setProducers(producers);
+
+    Topic topicA = new TopicImpl("topicA");
+    project.addTopic(topicA);
+
+    Topology topology = new TopologyImpl();
+    topology.addProject(project);
+
+    doReturn(new ArrayList<TopologyAclBinding>())
+        .when(aclsBuilder)
+        .buildBindingsForProducers(producers, project.namePrefix(), true);
     accessControlManager.apply(topology, plan);
-    verify(aclsBuilder, times(1)).buildBindingsForProducers(eq(producers), eq(topicA.toString()));
+    verify(aclsBuilder, times(1))
+        .buildBindingsForProducers(eq(producers), eq(project.namePrefix()), eq(true));
   }
 
   @Test
@@ -317,7 +384,7 @@ public class AccessControlManagerTest {
 
     doReturn(new ArrayList<TopologyAclBinding>())
         .when(aclsBuilder)
-        .buildBindingsForConsumers(users, topicA.toString());
+        .buildBindingsForConsumers(users, topicA.toString(), false);
     accessControlManager.apply(topology, plan);
 
     plan.run(true);
@@ -343,12 +410,14 @@ public class AccessControlManagerTest {
     Topology topology = buildTopology(consumers, asList(topicA));
 
     List<TopologyAclBinding> bindings = returnAclsForConsumers(consumers, topicA.getName());
-    doReturn(bindings).when(aclsBuilder).buildBindingsForConsumers(any(), eq(topicA.toString()));
+    doReturn(bindings)
+        .when(aclsBuilder)
+        .buildBindingsForConsumers(any(), eq(topicA.toString()), eq(false));
 
     accessControlManager.apply(topology, plan);
     plan.run();
 
-    verify(aclsBuilder, times(1)).buildBindingsForConsumers(consumers, topicA.toString());
+    verify(aclsBuilder, times(1)).buildBindingsForConsumers(consumers, topicA.toString(), false);
 
     consumers = new ArrayList<>();
     consumers.add(new Consumer("User:app1"));
@@ -356,7 +425,7 @@ public class AccessControlManagerTest {
     bindings = returnAclsForConsumers(consumers, topicA.getName());
     doReturn(bindings)
         .when(aclsBuilder)
-        .buildBindingsForConsumers(eq(consumers), eq(topicA.toString()));
+        .buildBindingsForConsumers(eq(consumers), eq(topicA.toString()), eq(false));
 
     Topology newTopology = buildTopology(consumers, asList(topicA));
 
diff --git a/src/test/java/com/purbon/kafka/topology/AclsBindingsBuilderTest.java b/src/test/java/com/purbon/kafka/topology/AclsBindingsBuilderTest.java
index cd22517..cdab754 100644
--- a/src/test/java/com/purbon/kafka/topology/AclsBindingsBuilderTest.java
+++ b/src/test/java/com/purbon/kafka/topology/AclsBindingsBuilderTest.java
@@ -41,7 +41,7 @@ public class AclsBindingsBuilderTest {
     Consumer consumer = new Consumer("User:foo");
 
     List<TopologyAclBinding> aclBindings =
-        builder.buildBindingsForConsumers(Collections.singleton(consumer), "bar");
+        builder.buildBindingsForConsumers(Collections.singleton(consumer), "bar", false);
     assertThat(aclBindings.size()).isEqualTo(3);
     assertThat(aclBindings)
         .contains(buildTopicLevelAcl("User:foo", "bar", PatternType.LITERAL, AclOperation.READ));
@@ -56,7 +56,7 @@ public class AclsBindingsBuilderTest {
   public void testProducerAclsBuilder() {
     Producer producer = new Producer("User:foo");
     List<TopologyAclBinding> aclBindings =
-        builder.buildBindingsForProducers(Collections.singleton(producer), "bar");
+        builder.buildBindingsForProducers(Collections.singleton(producer), "bar", false);
     assertThat(aclBindings.size()).isEqualTo(2);
     assertThat(aclBindings)
         .contains(buildTopicLevelAcl("User:foo", "bar", PatternType.LITERAL, AclOperation.WRITE));
@@ -70,7 +70,7 @@ public class AclsBindingsBuilderTest {
     Producer producer = new Producer("User:foo");
     producer.setTransactionId(Optional.of("1234"));
     List<TopologyAclBinding> aclBindings =
-        builder.buildBindingsForProducers(Collections.singleton(producer), "bar");
+        builder.buildBindingsForProducers(Collections.singleton(producer), "bar", false);
     assertThat(aclBindings.size()).isEqualTo(5);
 
     assertThat(aclBindings)
@@ -104,7 +104,7 @@ public class AclsBindingsBuilderTest {
     Producer producer = new Producer("User:foo");
     producer.setIdempotence(Optional.of(true));
     List<TopologyAclBinding> aclBindings =
-        builder.buildBindingsForProducers(Collections.singleton(producer), "bar");
+        builder.buildBindingsForProducers(Collections.singleton(producer), "bar", false);
     assertThat(aclBindings.size()).isEqualTo(3);
 
     assertThat(aclBindings)
diff --git a/src/test/java/com/purbon/kafka/topology/RbacProviderTest.java b/src/test/java/com/purbon/kafka/topology/RbacProviderTest.java
index 92d7315..584ca40 100644
--- a/src/test/java/com/purbon/kafka/topology/RbacProviderTest.java
+++ b/src/test/java/com/purbon/kafka/topology/RbacProviderTest.java
@@ -1,6 +1,9 @@
 package com.purbon.kafka.topology;
 
+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.roles.rbac.RBACBindingsBuilder.LITERAL;
+import static com.purbon.kafka.topology.roles.rbac.RBACBindingsBuilder.PREFIX;
 import static com.purbon.kafka.topology.roles.rbac.RBACPredefinedRoles.DEVELOPER_READ;
 import static com.purbon.kafka.topology.roles.rbac.RBACPredefinedRoles.DEVELOPER_WRITE;
 import static org.mockito.Matchers.any;
@@ -40,6 +43,7 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Properties;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
@@ -58,6 +62,8 @@ public class RbacProviderTest {
   @Rule public MockitoRule mockitoRule = MockitoJUnit.rule();
 
   private AccessControlManager accessControlManager;
+  private RBACProvider aclsProvider;
+  private RBACBindingsBuilder bindingsBuilder;
 
   @Before
   public void setup() {
@@ -65,13 +71,13 @@ public class RbacProviderTest {
     apiClient.setSchemaRegistryClusterID("sr");
     apiClient.setKafkaClusterId("ak");
 
-    RBACProvider aclsProvider = new RBACProvider(apiClient);
-    RBACBindingsBuilder bindingsBuilder = new RBACBindingsBuilder(apiClient);
+    aclsProvider = new RBACProvider(apiClient);
+    bindingsBuilder = new RBACBindingsBuilder(apiClient);
     accessControlManager = new AccessControlManager(aclsProvider, bindingsBuilder);
   }
 
   @Test
-  public void newConsumerACLsCreation() throws IOException {
+  public void newConsumerACLsCreation() {
 
     List<Consumer> consumers = new ArrayList<>();
     consumers.add(new Consumer("User:app1"));
@@ -95,7 +101,39 @@ public class RbacProviderTest {
   }
 
   @Test
-  public void newProducerACLsCreation() throws IOException {
+  public void newConsumerOptimisedACLsCreation() {
+
+    HashMap<String, String> cliOps = new HashMap<>();
+    cliOps.put(BROKERS_OPTION, "");
+    Properties props = new Properties();
+    props.put(OPTIMIZED_ACLS_CONFIG, true);
+
+    TopologyBuilderConfig config = new TopologyBuilderConfig(cliOps, props);
+    accessControlManager = new AccessControlManager(aclsProvider, bindingsBuilder, config);
+
+    List<Consumer> consumers = new ArrayList<>();
+    consumers.add(new Consumer("User:app1"));
+    Project project = new ProjectImpl();
+    project.setConsumers(consumers);
+
+    Topic topicA = new TopicImpl("topicA");
+    project.addTopic(topicA);
+
+    Topology topology = new TopologyImpl();
+    topology.addProject(project);
+
+    doReturn(new TopologyAclBinding())
+        .when(apiClient)
+        .bind("User:app1", DEVELOPER_READ, project.namePrefix(), PREFIX);
+
+    accessControlManager.apply(topology, plan);
+
+    verify(apiClient, times(1))
+        .bind(eq("User:app1"), eq(DEVELOPER_READ), eq(project.namePrefix()), eq(PREFIX));
+  }
+
+  @Test
+  public void newProducerACLsCreation() {
 
     List<Producer> producers = new ArrayList<>();
     producers.add(new Producer("User:app1"));
@@ -119,7 +157,38 @@ public class RbacProviderTest {
   }
 
   @Test
-  public void newKafkaStreamsAppACLsCreation() throws IOException {
+  public void newProducerOptimizedACLsCreation() {
+
+    HashMap<String, String> cliOps = new HashMap<>();
+    cliOps.put(BROKERS_OPTION, "");
+    Properties props = new Properties();
+    props.put(OPTIMIZED_ACLS_CONFIG, true);
+
+    TopologyBuilderConfig config = new TopologyBuilderConfig(cliOps, props);
+    accessControlManager = new AccessControlManager(aclsProvider, bindingsBuilder, config);
+    List<Producer> producers = new ArrayList<>();
+    producers.add(new Producer("User:app1"));
+    Project project = new ProjectImpl();
+    project.setProducers(producers);
+
+    Topic topicA = new TopicImpl("topicA");
+    project.addTopic(topicA);
+
+    Topology topology = new TopologyImpl();
+    topology.addProject(project);
+
+    doReturn(new TopologyAclBinding())
+        .when(apiClient)
+        .bind("User:app1", DEVELOPER_WRITE, project.namePrefix(), PREFIX);
+
+    accessControlManager.apply(topology, plan);
+
+    verify(apiClient, times(1))
+        .bind(eq("User:app1"), eq(DEVELOPER_WRITE), eq(project.namePrefix()), eq(PREFIX));
+  }
+
+  @Test
+  public void newKafkaStreamsAppACLsCreation() {
 
     Project project = new ProjectImpl();
 
diff --git a/src/test/java/com/purbon/kafka/topology/integration/AccessControlManagerIT.java b/src/test/java/com/purbon/kafka/topology/integration/AccessControlManagerIT.java
index e9b2e42..fd9c2bc 100644
--- a/src/test/java/com/purbon/kafka/topology/integration/AccessControlManagerIT.java
+++ b/src/test/java/com/purbon/kafka/topology/integration/AccessControlManagerIT.java
@@ -88,7 +88,7 @@ public class AccessControlManagerIT {
     // Crate an ACL outside of the control of the state manager.
     List<TopologyAclBinding> bindings =
         bindingsBuilder.buildBindingsForProducers(
-            Collections.singleton(new Producer("User:foo")), "bar");
+            Collections.singleton(new Producer("User:foo")), "bar", false);
     aclsProvider.createBindings(new HashSet<>(bindings));
 
     List<Consumer> consumers = new ArrayList<>();