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

Issue:
Make allowDelete optional
As I user I would like to have the option to block `allowDelete` via a config file setting.
In addition it might be an idea to allow/forbid deletion via an environment variable

Pull request description:
Allow granular delete flags as well as ENV override 
fixes #156 #96 

Patch:
diff --git a/.github/workflows/stale.yml b/.github/workflows/stale.yml
index 72357df..60eadd4 100644
--- a/.github/workflows/stale.yml
+++ b/.github/workflows/stale.yml
@@ -1,6 +1,6 @@
 on:
   schedule:
-    - cron: '0 * * * *'
+    - cron: '* 0 * * *'
   push:
     branches:
       - master
diff --git a/docs/handling-delete-in-ktb.rst b/docs/handling-delete-in-ktb.rst
new file mode 100644
index 0000000..0dd59a0
--- /dev/null
+++ b/docs/handling-delete-in-ktb.rst
@@ -0,0 +1,63 @@
+Handling delete in the Kafka Topology Builder
+*******************************
+
+As the reader might already be aware, the Kafka Topology Builder is capable of handling deletes for you.
+This way the project can ensure that the final state of the cluster is consistent with the declarative state in the topology descriptors.
+
+However in some situations having delete enabled might not be what you are looking for.
+For this reason the tool allows you to control it in full granularity.
+
+Delete flag in the CLI
+-----------
+
+The CLI provide you currently with a global DELETE flag, this would allow deletion of all resources controlled by the Kafka Topology builder.
+The user can set this flag (*--allowDelete*) from the CLI as described.
+
+.. code-block:: bash
+
+    $> kafka-topology-builder.sh  --help
+        usage: cli
+          --allowDelete          Permits delete operations for topics and
+                                 configs. (deprecated, to be removed)
+          --brokers <arg>        The Apache Kafka server(s) to connect to.
+          --clientConfig <arg>   The AdminClient configuration file.
+          --dryRun               Print the execution plan without altering
+                                  anything.
+          --help                 Prints usage information.
+          --quiet                Print minimum status update
+          --topology <arg>       Topology config file.
+          --version              Prints useful version information.
+
+*NOTE*: This is a global flag, allowing/deny all delete operations.
+
+By default the value for this variable is false, so the Kafka Topology Builder will **not** delete any resource.
+
+Granular delete flags
+-----------
+
+There could be situations when the reader aim to :
+ * Control delete operations in a more granular way, for example allow delete of bindings (Acls/RBAC) but not topics.
+ * Control the delete via ENV variables. This could be very handy when doing CI/CD integrations and setting variables over the pipeline executions.
+
+This can be done in KTB by using the capabilities provided by the configuration library in use.
+The tool allows the user to set:
+
+* A configuration variable to allow/deny the delete operations.
+* Use an ENV variable as a first priority citizen to handle this operation
+
+*NOTE* ENV variables take priority over other ways to set a config value.
+
+Topics deletion flag
+^^^^^^^^^^^
+The user can control topic deletion by:
+
+- setting the *allow.delete.topics* configuration in the provided file to the tool.
+- set the ENV variable *ALLOW_DELETE_TOPICS* when calling the tool from the CLI.
+
+Bindings deletion flag
+^^^^^^^^^^^
+
+The user can control bindings deletion by:
+
+- setting the *allow.delete.bindings* configuration in the provided file to the tool.
+- set the ENV variable *ALLOW_DELETE_BINDINGS* when calling the tool from the CLI.
\ No newline at end of file
diff --git a/docs/how-to-run-it.rst b/docs/how-to-run-it.rst
index 5ac410c..416f8e8 100644
--- a/docs/how-to-run-it.rst
+++ b/docs/how-to-run-it.rst
@@ -18,7 +18,7 @@ If you are using the CLI tool, you an use the *--help* command to list the diffe
     $> kafka-topology-builder.sh  --help
         usage: cli
           --allowDelete          Permits delete operations for topics and
-                                 configs.
+                                 configs. (deprecated, to be removed)
           --brokers <arg>        The Apache Kafka server(s) to connect to.
           --clientConfig <arg>   The AdminClient configuration file.
           --dryRun               Print the execution plan without altering
diff --git a/docs/index.rst b/docs/index.rst
index 81c2caf..23577a1 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -61,6 +61,7 @@ Contents
    what-can-you-do
    the-descriptor-files
    how-to-run-it
+   handling-delete-in-ktb
    workflow-setup
    powered-with
    config-values
diff --git a/src/main/java/com/purbon/kafka/topology/AccessControlManager.java b/src/main/java/com/purbon/kafka/topology/AccessControlManager.java
index f87da62..6b2829b 100644
--- a/src/main/java/com/purbon/kafka/topology/AccessControlManager.java
+++ b/src/main/java/com/purbon/kafka/topology/AccessControlManager.java
@@ -186,7 +186,7 @@ public class AccessControlManager {
       updateActions.add(createBindings);
     }
 
-    if (config.allowDelete()) {
+    if (config.allowDelete() || config.isAllowDeleteBindings()) {
       // clear acls that does not appear anymore in the new generated list,
       // but where previously created
       Set<TopologyAclBinding> bindingsToDelete =
diff --git a/src/main/java/com/purbon/kafka/topology/BuilderCLI.java b/src/main/java/com/purbon/kafka/topology/BuilderCLI.java
index 7d28fa8..b6a3d91 100644
--- a/src/main/java/com/purbon/kafka/topology/BuilderCLI.java
+++ b/src/main/java/com/purbon/kafka/topology/BuilderCLI.java
@@ -7,9 +7,13 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import org.apache.commons.cli.*;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 public class BuilderCLI {
 
+  private static final Logger LOGGER = LogManager.getLogger(BuilderCLI.class);
+
   public static final String TOPOLOGY_OPTION = "topology";
   public static final String TOPOLOGY_DESC = "Topology config file.";
 
@@ -24,7 +28,7 @@ public class BuilderCLI {
 
   public static final String ALLOW_DELETE_OPTION = "allowDelete";
   public static final String ALLOW_DELETE_DESC =
-      "Permits delete operations for topics and configs.";
+      "Permits delete operations for topics and configs. (deprecated, to be removed)";
 
   public static final String DRY_RUN_OPTION = "dryRun";
   public static final String DRY_RUN_DESC = "Print the execution plan without altering anything.";
@@ -148,6 +152,13 @@ public class BuilderCLI {
       config.put(BROKERS_OPTION, cmd.getOptionValue(BROKERS_OPTION));
     }
     config.put(ALLOW_DELETE_OPTION, String.valueOf(cmd.hasOption(ALLOW_DELETE_OPTION)));
+    // Add deprecation note for using allow delete CLI option
+    if (cmd.hasOption(ALLOW_DELETE_OPTION)) {
+      LOGGER.info(
+          String.format(
+              "CLI option %s is currently deprecated. Expected to be removed in next major version.",
+              ALLOW_DELETE_OPTION));
+    }
     config.put(DRY_RUN_OPTION, String.valueOf(cmd.hasOption(DRY_RUN_OPTION)));
     config.put(QUIET_OPTION, String.valueOf(cmd.hasOption(QUIET_OPTION)));
     config.put(ADMIN_CLIENT_CONFIG_OPTION, cmd.getOptionValue(ADMIN_CLIENT_CONFIG_OPTION));
diff --git a/src/main/java/com/purbon/kafka/topology/TopicManager.java b/src/main/java/com/purbon/kafka/topology/TopicManager.java
index 56f7fee..f0c8b89 100644
--- a/src/main/java/com/purbon/kafka/topology/TopicManager.java
+++ b/src/main/java/com/purbon/kafka/topology/TopicManager.java
@@ -67,7 +67,7 @@ public class TopicManager {
       }
     }
 
-    if (config.allowDelete()) {
+    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 =
diff --git a/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java b/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java
index 0371066..4f13e2d 100644
--- a/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java
+++ b/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java
@@ -63,6 +63,9 @@ public class TopologyBuilderConfig {
 
   static final String OPTIMIZED_ACLS_CONFIG = "topology.acls.optimized";
 
+  static final String ALLOW_DELETE_TOPICS = "allow.delete.topics";
+  static final String ALLOW_DELETE_BINDINGS = "allow.delete.bindings";
+
   private final Map<String, String> cliParams;
   private Config config;
 
@@ -266,7 +269,7 @@ public class TopologyBuilderConfig {
   }
 
   public boolean allowDelete() {
-    return Boolean.parseBoolean(cliParams.getOrDefault(BuilderCLI.ALLOW_DELETE_OPTION, "true"));
+    return Boolean.parseBoolean(cliParams.getOrDefault(BuilderCLI.ALLOW_DELETE_OPTION, "false"));
   }
 
   public boolean isQuiet() {
@@ -280,4 +283,12 @@ public class TopologyBuilderConfig {
   public FileType getTopologyFileType() {
     return config.getEnum(FileType.class, TOPOLOGY_FILE_TYPE);
   }
+
+  public boolean isAllowDeleteTopics() {
+    return config.getBoolean(ALLOW_DELETE_TOPICS);
+  }
+
+  public boolean isAllowDeleteBindings() {
+    return config.getBoolean(ALLOW_DELETE_BINDINGS);
+  }
 }
diff --git a/src/main/resources/reference.conf b/src/main/resources/reference.conf
index 9e439bc..b9735e9 100644
--- a/src/main/resources/reference.conf
+++ b/src/main/resources/reference.conf
@@ -49,3 +49,12 @@ kafka {
       topic.prefixes = [ "_" ]
    }
 }
+
+allow {
+  delete {
+    topics = false
+    topics = ${?ALLOW_DELETE_TOPICS}
+    bindings = false
+    bindings = ${?ALLOW_DELETE_BINDINGS}
+  }
+}


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 6839f13..70c2ff1 100644
--- a/src/test/java/com/purbon/kafka/topology/AccessControlManagerTest.java
+++ b/src/test/java/com/purbon/kafka/topology/AccessControlManagerTest.java
@@ -1,8 +1,13 @@
 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.TopologyBuilderConfig.ALLOW_DELETE_BINDINGS;
+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.OPTIMIZED_ACLS_CONFIG;
 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;
@@ -131,7 +136,7 @@ public class AccessControlManagerTest {
     Consumer projectConsumer = new Consumer("project-consumer");
     Consumer topicConsumer = new Consumer("topic-consumer");
 
-    List<Consumer> projectConsumers = Collections.singletonList(projectConsumer);
+    List<Consumer> projectConsumers = singletonList(projectConsumer);
 
     Topology topology = new TopologyImpl();
     topology.setContext("testConsumerAclsAtTopicLevel");
@@ -221,7 +226,7 @@ public class AccessControlManagerTest {
     Producer projectProducer = new Producer("project-producer");
     Producer topicProducer = new Producer("topic-producer");
 
-    List<Producer> projectProducers = Collections.singletonList(projectProducer);
+    List<Producer> projectProducers = singletonList(projectProducer);
 
     Topology topology = new TopologyImpl();
     topology.setContext("testProducerAclsAtTopicLevel");
@@ -264,7 +269,7 @@ public class AccessControlManagerTest {
     topics.put(KStream.READ_TOPICS, asList("topicA", "topicB"));
     topics.put(KStream.WRITE_TOPICS, asList("topicC", "topicD"));
     app.setTopics(topics);
-    project.setStreams(Collections.singletonList(app));
+    project.setStreams(singletonList(app));
 
     Topology topology = new TopologyImpl();
     topology.addProject(project);
@@ -298,11 +303,11 @@ public class AccessControlManagerTest {
 
     SchemaRegistryInstance instance = new SchemaRegistryInstance();
     instance.setPrincipal("User:foo");
-    sr.setInstances(Collections.singletonList(instance));
+    sr.setInstances(singletonList(instance));
 
     Map<String, List<User>> rbac = new HashMap<>();
-    rbac.put("SecurityAdmin", Collections.singletonList(new User("User:foo")));
-    rbac.put("ClusterAdmin", Collections.singletonList(new User("User:bar")));
+    rbac.put("SecurityAdmin", singletonList(new User("User:foo")));
+    rbac.put("ClusterAdmin", singletonList(new User("User:bar")));
     sr.setRbac(Optional.of(rbac));
 
     platform.setSchemaRegistry(sr);
@@ -337,7 +342,7 @@ public class AccessControlManagerTest {
     ControlCenterInstance instance = new ControlCenterInstance();
     instance.setPrincipal("User:foo");
     instance.setAppId("appid");
-    c3.setInstances(Collections.singletonList(instance));
+    c3.setInstances(singletonList(instance));
     platform.setControlCenter(c3);
     topology.setPlatform(platform);
 
@@ -359,8 +364,8 @@ public class AccessControlManagerTest {
     Platform platform = new Platform();
     Kafka kafka = new Kafka();
     Map<String, List<User>> rbac = new HashMap<>();
-    rbac.put("Operator", Collections.singletonList(new User("User:foo")));
-    rbac.put("ClusterAdmin", Collections.singletonList(new User("User:bar")));
+    rbac.put("Operator", singletonList(new User("User:foo")));
+    rbac.put("ClusterAdmin", singletonList(new User("User:bar")));
     kafka.setRbac(Optional.of(rbac));
     platform.setKafka(kafka);
     topology.setPlatform(platform);
@@ -384,8 +389,8 @@ public class AccessControlManagerTest {
     Platform platform = new Platform();
     KafkaConnect connect = new KafkaConnect();
     Map<String, List<User>> rbac = new HashMap<>();
-    rbac.put("Operator", Collections.singletonList(new User("User:foo")));
-    rbac.put("ClusterAdmin", Collections.singletonList(new User("User:bar")));
+    rbac.put("Operator", singletonList(new User("User:foo")));
+    rbac.put("ClusterAdmin", singletonList(new User("User:bar")));
     connect.setRbac(Optional.of(rbac));
     platform.setKafkaConnect(connect);
     topology.setPlatform(platform);
@@ -445,7 +450,7 @@ public class AccessControlManagerTest {
 
     List<Consumer> users = asList(new Consumer("User:app1"));
 
-    doReturn(Collections.singletonList(new TopologyAclBinding()))
+    doReturn(singletonList(new TopologyAclBinding()))
         .when(aclsBuilder)
         .buildBindingsForConsumers(users, topicA.toString(), false);
 
@@ -462,8 +467,18 @@ public class AccessControlManagerTest {
     BackendController backendController = new BackendController();
     backendController.load();
     backendController.reset();
+
+    Properties props = new Properties();
+    props.put(KAFKA_INTERNAL_TOPIC_PREFIXES, Arrays.asList("foo.", "_"));
+
+    HashMap<String, String> cliOps = new HashMap<>();
+    cliOps.put(BROKERS_OPTION, "");
+    cliOps.put(ALLOW_DELETE_OPTION, "true");
+
+    TopologyBuilderConfig config = new TopologyBuilderConfig(cliOps, props);
+
     plan = ExecutionPlan.init(backendController, mockPrintStream);
-    accessControlManager = new AccessControlManager(aclsProvider, aclsBuilder);
+    accessControlManager = new AccessControlManager(aclsProvider, aclsBuilder, config);
 
     List<Consumer> consumers = new ArrayList<>();
     consumers.add(new Consumer("User:app1"));
@@ -503,6 +518,122 @@ public class AccessControlManagerTest {
     verify(aclsProvider, times(1)).clearBindings(new HashSet<>(bindingsToDelete));
   }
 
+  @Test
+  public void testBindingsDeleteWithAllOptionsDisabled() throws IOException {
+
+    BackendController backendController = new BackendController();
+    backendController.load();
+    backendController.reset();
+    plan = ExecutionPlan.init(backendController, mockPrintStream);
+
+    Properties props = new Properties();
+    props.put(KAFKA_INTERNAL_TOPIC_PREFIXES, Arrays.asList("foo.", "_"));
+    props.put(ALLOW_DELETE_TOPICS, "false");
+
+    HashMap<String, String> cliOps = new HashMap<>();
+    cliOps.put(BROKERS_OPTION, "");
+    cliOps.put(ALLOW_DELETE_OPTION, "false");
+
+    TopologyBuilderConfig config = new TopologyBuilderConfig(cliOps, props);
+
+    accessControlManager = new AccessControlManager(aclsProvider, aclsBuilder, config);
+
+    List<Consumer> consumers = new ArrayList<>();
+    consumers.add(new Consumer("User:app1"));
+    consumers.add(new Consumer("User:app2"));
+
+    Topic topicA = new TopicImpl("topicA");
+
+    Topology topology = buildTopology(consumers, singletonList(topicA));
+
+    List<TopologyAclBinding> bindings = returnAclsForConsumers(consumers, topicA.getName());
+    doReturn(bindings)
+        .when(aclsBuilder)
+        .buildBindingsForConsumers(any(), eq(topicA.toString()), eq(false));
+
+    accessControlManager.apply(topology, plan);
+    plan.run();
+
+    verify(aclsBuilder, times(1))
+        .buildBindingsForConsumers(refEq(consumers), eq(topicA.toString()), eq(false));
+
+    consumers = new ArrayList<>();
+    consumers.add(new Consumer("User:app1"));
+
+    bindings = returnAclsForConsumers(consumers, topicA.getName());
+    doReturn(bindings)
+        .when(aclsBuilder)
+        .buildBindingsForConsumers(any(), eq(topicA.toString()), eq(false));
+
+    Topology newTopology = buildTopology(consumers, singletonList(topicA));
+
+    accessControlManager.apply(newTopology, plan);
+    plan.run();
+
+    List<TopologyAclBinding> bindingsToDelete =
+        returnAclsForConsumers(singletonList(new Consumer("User:app2")), topicA.getName());
+
+    verify(aclsProvider, times(0)).clearBindings(new HashSet<>(bindingsToDelete));
+  }
+
+  @Test
+  public void testBindingsDeleteWithSpecificConfigEnabled() throws IOException {
+
+    BackendController backendController = new BackendController();
+    backendController.load();
+    backendController.reset();
+    plan = ExecutionPlan.init(backendController, mockPrintStream);
+
+    Properties props = new Properties();
+    props.put(KAFKA_INTERNAL_TOPIC_PREFIXES, Arrays.asList("foo.", "_"));
+    props.put(ALLOW_DELETE_BINDINGS, "true");
+
+    HashMap<String, String> cliOps = new HashMap<>();
+    cliOps.put(BROKERS_OPTION, "");
+    cliOps.put(ALLOW_DELETE_OPTION, "false");
+
+    TopologyBuilderConfig config = new TopologyBuilderConfig(cliOps, props);
+
+    accessControlManager = new AccessControlManager(aclsProvider, aclsBuilder, config);
+
+    List<Consumer> consumers = new ArrayList<>();
+    consumers.add(new Consumer("User:app1"));
+    consumers.add(new Consumer("User:app2"));
+
+    Topic topicA = new TopicImpl("topicA");
+
+    Topology topology = buildTopology(consumers, singletonList(topicA));
+
+    List<TopologyAclBinding> bindings = returnAclsForConsumers(consumers, topicA.getName());
+    doReturn(bindings)
+        .when(aclsBuilder)
+        .buildBindingsForConsumers(any(), eq(topicA.toString()), eq(false));
+
+    accessControlManager.apply(topology, plan);
+    plan.run();
+
+    verify(aclsBuilder, times(1))
+        .buildBindingsForConsumers(refEq(consumers), eq(topicA.toString()), eq(false));
+
+    consumers = new ArrayList<>();
+    consumers.add(new Consumer("User:app1"));
+
+    bindings = returnAclsForConsumers(consumers, topicA.getName());
+    doReturn(bindings)
+        .when(aclsBuilder)
+        .buildBindingsForConsumers(any(), eq(topicA.toString()), eq(false));
+
+    Topology newTopology = buildTopology(consumers, singletonList(topicA));
+
+    accessControlManager.apply(newTopology, plan);
+    plan.run();
+
+    List<TopologyAclBinding> bindingsToDelete =
+        returnAclsForConsumers(singletonList(new Consumer("User:app2")), topicA.getName());
+
+    verify(aclsProvider, times(1)).clearBindings(new HashSet<>(bindingsToDelete));
+  }
+
   private Topology buildTopology(List<Consumer> consumers, List<Topic> topics) {
 
     Project project = new ProjectImpl();
diff --git a/src/test/java/com/purbon/kafka/topology/TopicManagerTest.java b/src/test/java/com/purbon/kafka/topology/TopicManagerTest.java
index 7aa381f..71799ad 100644
--- a/src/test/java/com/purbon/kafka/topology/TopicManagerTest.java
+++ b/src/test/java/com/purbon/kafka/topology/TopicManagerTest.java
@@ -3,6 +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 org.mockito.Matchers.any;
 import static org.mockito.Mockito.*;
@@ -97,6 +98,17 @@ public class TopicManagerTest {
   @Test
   public void topicDeleteTest() throws IOException {
 
+    cliOps = new HashMap<>();
+    cliOps.put(BROKERS_OPTION, "");
+    cliOps.put(ALLOW_DELETE_OPTION, "true");
+
+    Properties props = new Properties();
+
+    TopologyBuilderConfig config = new TopologyBuilderConfig(cliOps, props);
+
+    plan = ExecutionPlan.init(backendController, System.out);
+    topicManager = new TopicManager(adminClient, schemaRegistryManager, config);
+
     // Original Topology
     Topology topology0 = new TopologyImpl();
     Project project0 = new ProjectImpl("project");
@@ -133,6 +145,10 @@ public class TopicManagerTest {
   @Test
   public void topicDeleteWithConfiguredInternalTopicsTest() throws IOException {
 
+    cliOps = new HashMap<>();
+    cliOps.put(BROKERS_OPTION, "");
+    cliOps.put(ALLOW_DELETE_OPTION, "true");
+
     Properties props = new Properties();
     props.put(KAFKA_INTERNAL_TOPIC_PREFIXES, Arrays.asList("foo.", "_"));
 
@@ -201,6 +217,76 @@ public class TopicManagerTest {
     verify(adminClient, times(0)).deleteTopics(Collections.singletonList(topicC));
   }
 
+  @Test
+  public void topicDeleteWithConfiguredNoDeleteOnlyForTopics() throws IOException {
+
+    Properties props = new Properties();
+    props.put(ALLOW_DELETE_TOPICS, "true");
+
+    HashMap<String, String> cliOps = new HashMap<>();
+    cliOps.put(BROKERS_OPTION, "");
+    cliOps.put(ALLOW_DELETE_OPTION, "false");
+
+    TopologyBuilderConfig config = new TopologyBuilderConfig(cliOps, props);
+
+    TopicManager topicManager = new TopicManager(adminClient, schemaRegistryManager, config);
+
+    // Topology after delete action
+    Topology topology = new TopologyImpl();
+    Project project = new ProjectImpl("project");
+    topology.addProject(project);
+
+    Topic topicA = new TopicImpl("topicA");
+    project.addTopic(topicA);
+    Topic topicB = new TopicImpl("topicB");
+    project.addTopic(topicB);
+
+    String topicC = "team.project.topicC";
+
+    Set<String> appTopics = Collections.singleton(topicC);
+    when(adminClient.listApplicationTopics()).thenReturn(appTopics);
+
+    topicManager.apply(topology, plan);
+    plan.run();
+
+    verify(adminClient, times(1)).deleteTopics(Collections.singletonList(topicC));
+  }
+
+  @Test
+  public void topicDeleteWithConfiguredNoDeleteOnlyForTopicsAllDisabled() throws IOException {
+
+    Properties props = new Properties();
+    props.put(ALLOW_DELETE_TOPICS, "false");
+
+    HashMap<String, String> cliOps = new HashMap<>();
+    cliOps.put(BROKERS_OPTION, "");
+    cliOps.put(ALLOW_DELETE_OPTION, "false");
+
+    TopologyBuilderConfig config = new TopologyBuilderConfig(cliOps, props);
+
+    TopicManager topicManager = new TopicManager(adminClient, schemaRegistryManager, config);
+
+    // Topology after delete action
+    Topology topology = new TopologyImpl();
+    Project project = new ProjectImpl("project");
+    topology.addProject(project);
+
+    Topic topicA = new TopicImpl("topicA");
+    project.addTopic(topicA);
+    Topic topicB = new TopicImpl("topicB");
+    project.addTopic(topicB);
+
+    String topicC = "team.project.topicC";
+
+    Set<String> appTopics = Collections.singleton(topicC);
+    when(adminClient.listApplicationTopics()).thenReturn(appTopics);
+
+    topicManager.apply(topology, plan);
+    plan.run();
+
+    verify(adminClient, times(0)).deleteTopics(Collections.singletonList(topicC));
+  }
+
   @Test
   public void dryRunTest() throws IOException {
 
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 9eaf672..89663e8 100644
--- a/src/test/java/com/purbon/kafka/topology/integration/AccessControlManagerIT.java
+++ b/src/test/java/com/purbon/kafka/topology/integration/AccessControlManagerIT.java
@@ -1,5 +1,7 @@
 package com.purbon.kafka.topology.integration;
 
+import static com.purbon.kafka.topology.BuilderCLI.ALLOW_DELETE_OPTION;
+import static com.purbon.kafka.topology.BuilderCLI.BROKERS_OPTION;
 import static org.junit.Assert.assertEquals;
 import static org.mockito.Mockito.when;
 
@@ -86,6 +88,16 @@ public class AccessControlManagerIT {
   @Test
   public void aclsRemoval() throws ExecutionException, InterruptedException, IOException {
 
+    Properties props = new Properties();
+
+    HashMap<String, String> cliOps = new HashMap<>();
+    cliOps.put(BROKERS_OPTION, "");
+    cliOps.put(ALLOW_DELETE_OPTION, "true");
+
+    TopologyBuilderConfig config = new TopologyBuilderConfig(cliOps, props);
+
+    accessControlManager = new AccessControlManager(aclsProvider, bindingsBuilder, config);
+
     // Crate an ACL outside of the control of the state manager.
     List<TopologyAclBinding> bindings =
         bindingsBuilder.buildBindingsForProducers(
@@ -126,6 +138,16 @@ public class AccessControlManagerIT {
   @Test
   public void aclsRemovedTest() throws IOException, ExecutionException, InterruptedException {
 
+    Properties props = new Properties();
+
+    HashMap<String, String> cliOps = new HashMap<>();
+    cliOps.put(BROKERS_OPTION, "");
+    cliOps.put(ALLOW_DELETE_OPTION, "true");
+
+    TopologyBuilderConfig config = new TopologyBuilderConfig(cliOps, props);
+
+    accessControlManager = new AccessControlManager(aclsProvider, bindingsBuilder, config);
+
     TopologyImpl topology = new TopologyImpl();
     topology.setContext("aclsRemovedTest-Integration");
 
diff --git a/src/test/java/com/purbon/kafka/topology/integration/RBACPRoviderRbacIT.java b/src/test/java/com/purbon/kafka/topology/integration/RBACPRoviderRbacIT.java
index 41ce779..8a107e9 100644
--- a/src/test/java/com/purbon/kafka/topology/integration/RBACPRoviderRbacIT.java
+++ b/src/test/java/com/purbon/kafka/topology/integration/RBACPRoviderRbacIT.java
@@ -1,5 +1,7 @@
 package com.purbon.kafka.topology.integration;
 
+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.roles.rbac.RBACPredefinedRoles.DEVELOPER_READ;
 import static com.purbon.kafka.topology.roles.rbac.RBACPredefinedRoles.DEVELOPER_WRITE;
 import static com.purbon.kafka.topology.roles.rbac.RBACPredefinedRoles.RESOURCE_OWNER;
@@ -14,6 +16,7 @@ import static org.mockito.Mockito.verify;
 import com.purbon.kafka.topology.AccessControlManager;
 import com.purbon.kafka.topology.BackendController;
 import com.purbon.kafka.topology.ExecutionPlan;
+import com.purbon.kafka.topology.TopologyBuilderConfig;
 import com.purbon.kafka.topology.api.mds.MDSApiClient;
 import com.purbon.kafka.topology.model.Impl.ProjectImpl;
 import com.purbon.kafka.topology.model.Impl.TopicImpl;
@@ -43,6 +46,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.Test;
 import org.junit.runner.RunWith;
@@ -75,7 +79,16 @@ public class RBACPRoviderRbacIT extends MDSBaseTest {
     plan = ExecutionPlan.init(cs, System.out);
     RBACProvider rbacProvider = new RBACProvider(apiClient);
     RBACBindingsBuilder bindingsBuilder = new RBACBindingsBuilder(apiClient);
-    accessControlManager = new AccessControlManager(rbacProvider, bindingsBuilder);
+
+    Properties props = new Properties();
+
+    HashMap<String, String> cliOps = new HashMap<>();
+    cliOps.put(BROKERS_OPTION, "");
+    cliOps.put(ALLOW_DELETE_OPTION, "true");
+
+    TopologyBuilderConfig config = new TopologyBuilderConfig(cliOps, props);
+
+    accessControlManager = new AccessControlManager(rbacProvider, bindingsBuilder, config);
   }
 
   @Test
diff --git a/src/test/java/com/purbon/kafka/topology/integration/TopicManagerIT.java b/src/test/java/com/purbon/kafka/topology/integration/TopicManagerIT.java
index 0a72cf6..6c75350 100644
--- a/src/test/java/com/purbon/kafka/topology/integration/TopicManagerIT.java
+++ b/src/test/java/com/purbon/kafka/topology/integration/TopicManagerIT.java
@@ -1,11 +1,14 @@
 package com.purbon.kafka.topology.integration;
 
+import static com.purbon.kafka.topology.BuilderCLI.ALLOW_DELETE_OPTION;
+import static com.purbon.kafka.topology.BuilderCLI.BROKERS_OPTION;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
 import com.purbon.kafka.topology.BackendController;
 import com.purbon.kafka.topology.ExecutionPlan;
 import com.purbon.kafka.topology.TopicManager;
+import com.purbon.kafka.topology.TopologyBuilderConfig;
 import com.purbon.kafka.topology.api.adminclient.TopologyBuilderAdminClient;
 import com.purbon.kafka.topology.integration.containerutils.ContainerTestUtils;
 import com.purbon.kafka.topology.integration.containerutils.SaslPlaintextKafkaContainer;
@@ -26,6 +29,7 @@ import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.stream.Collectors;
@@ -73,7 +77,16 @@ public class TopicManagerIT {
     final SchemaRegistryManager schemaRegistryManager =
         new SchemaRegistryManager(schemaRegistryClient, System.getProperty("user.dir"));
     this.plan = ExecutionPlan.init(new BackendController(), System.out);
-    this.topicManager = new TopicManager(adminClient, schemaRegistryManager);
+
+    Properties props = new Properties();
+
+    HashMap<String, String> cliOps = new HashMap<>();
+    cliOps.put(BROKERS_OPTION, "");
+    cliOps.put(ALLOW_DELETE_OPTION, "true");
+
+    TopologyBuilderConfig config = new TopologyBuilderConfig(cliOps, props);
+
+    this.topicManager = new TopicManager(adminClient, schemaRegistryManager, config);
   }
 
   @Test