Repository: kafka-ops/kafka-topology-builder
Pull request #180 (2020-12-24 12:53:22)

Allow topic state bookkeeping from the cluster backend state system  
* **Please check if the PR fulfills these requirements**
- [x] The commit messages are descriptive
- [x] Tests for the changes have been added (for bug fixes / features)
- [x] Docs have been added / updated (for bug fixes / features)
- [ ] An issue has been created for the pull requests. Some issues might require previous discussion.

* **What kind of change does this PR introduce?** (Bug fix, feature, docs update, ...)



* **What is the current behavior?** (You can also link to an open issue here)

As of now the KTB retrieve the topic status (what has been created) directly from the cluster. There is an option to filter out what the user does not want to consider, however in a more multi tenant scenario, users want to control only what is managed by each of the KTB instances and the number can be bigger.


* **What is the new behavior (if this is a feature change)?**

We will keep the current behaviour as default till the release of 2.0, however, users would have the option to store the managed topics in the status backends. The use of status backends will be the default from 2.0 on.


* **Does this PR introduce a breaking change?** (What changes might users need to make in their application due to this PR?)

Not for now, but will introduce it in 2.0. This behaviour should be noted so it is not forgotten in the 2.0 release.

* **Other information**:


**IMPORTANT**: Please review the CONTRIBUTING.md file for detailed contributing guidelines.
**IMPORTANT**: Your pull request MUST target `master`.

PLEASE REMOVE THIS TEMPLATE BEFORE SUBMITTING


Patch:
diff --git a/docs/config-values.rst b/docs/config-values.rst
index f77cdbd..48d12fd 100644
--- a/docs/config-values.rst
+++ b/docs/config-values.rst
@@ -133,3 +133,21 @@ You can override this behaviour by setting the config below to `false`. And inst
 An example configuration will look like this:
 ::
     topology.connector.allow.topic.create=false
+
+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.
+
+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.
+
+An example to use local topic management state will look like this:
+::
+    topology.state.topics.cluster.enabled=false
\ No newline at end of file
diff --git a/src/main/java/com/purbon/kafka/topology/BackendController.java b/src/main/java/com/purbon/kafka/topology/BackendController.java
index 7c53aed..32e511e 100644
--- a/src/main/java/com/purbon/kafka/topology/BackendController.java
+++ b/src/main/java/com/purbon/kafka/topology/BackendController.java
@@ -23,6 +23,7 @@ public class BackendController {
   private static final String STORE_TYPE = "acls";
 
   private final Backend backend;
+  private Set<String> topics;
   private Set<TopologyAclBinding> bindings;
   private Set<ServiceAccount> serviceAccounts;
 
@@ -34,23 +35,24 @@ public class BackendController {
     this.backend = backend;
     this.bindings = new HashSet<>();
     this.serviceAccounts = new HashSet<>();
+    this.topics = new HashSet<>();
   }
 
-  public void add(List<TopologyAclBinding> bindings) {
+  public void addBindings(List<TopologyAclBinding> bindings) {
     LOGGER.debug(String.format("Adding bindings %s to the backend", bindings));
     this.bindings.addAll(bindings);
   }
 
+  public void addTopics(Set<String> topics) {
+    LOGGER.debug(String.format("Adding topics %s to the backend", topics));
+    this.topics.addAll(topics);
+  }
+
   public void addServiceAccounts(Set<ServiceAccount> serviceAccounts) {
     LOGGER.debug(String.format("Adding Service Accounts %s to the backend", serviceAccounts));
     this.serviceAccounts.addAll(serviceAccounts);
   }
 
-  public void add(TopologyAclBinding binding) {
-    LOGGER.debug(String.format("Adding binding %s to the backend", binding));
-    this.bindings.add(binding);
-  }
-
   public Set<ServiceAccount> getServiceAccounts() {
     return new HashSet<>(serviceAccounts);
   }
@@ -59,6 +61,10 @@ public class BackendController {
     return bindings;
   }
 
+  public Set<String> getTopics() {
+    return topics;
+  }
+
   public void flushAndClose() {
     LOGGER.debug(String.format("Flushing the current state of %s, %s", STORE_TYPE, bindings));
     backend.createOrOpen(Mode.TRUNCATE);
@@ -66,6 +72,8 @@ public class BackendController {
     backend.saveBindings(bindings);
     backend.saveType("ServiceAccounts");
     backend.saveAccounts(serviceAccounts);
+    backend.saveType("Topics");
+    backend.saveTopics(topics);
     backend.close();
   }
 
@@ -74,15 +82,17 @@ public class BackendController {
     backend.createOrOpen();
     bindings.addAll(backend.loadBindings());
     serviceAccounts.addAll(backend.loadServiceAccounts());
+    topics.addAll(backend.loadTopics());
   }
 
   public void reset() {
     LOGGER.debug("Reset the bindings cache");
     bindings.clear();
     serviceAccounts.clear();
+    topics.clear();
   }
 
   public int size() {
-    return bindings.size() + serviceAccounts.size();
+    return bindings.size() + serviceAccounts.size() + topics.size();
   }
 }
diff --git a/src/main/java/com/purbon/kafka/topology/ExecutionPlan.java b/src/main/java/com/purbon/kafka/topology/ExecutionPlan.java
index 85aad09..3de7863 100644
--- a/src/main/java/com/purbon/kafka/topology/ExecutionPlan.java
+++ b/src/main/java/com/purbon/kafka/topology/ExecutionPlan.java
@@ -5,15 +5,18 @@ import com.purbon.kafka.topology.actions.BaseAccountsAction;
 import com.purbon.kafka.topology.actions.access.ClearBindings;
 import com.purbon.kafka.topology.actions.accounts.ClearAccounts;
 import com.purbon.kafka.topology.actions.accounts.CreateAccounts;
+import com.purbon.kafka.topology.actions.topics.DeleteTopics;
+import com.purbon.kafka.topology.actions.topics.SyncTopicAction;
 import com.purbon.kafka.topology.model.cluster.ServiceAccount;
 import com.purbon.kafka.topology.roles.TopologyAclBinding;
+import com.purbon.kafka.topology.utils.StreamUtils;
 import java.io.IOException;
 import java.io.PrintStream;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
-import java.util.stream.Collectors;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
@@ -26,6 +29,7 @@ public class ExecutionPlan {
   private BackendController backendController;
   private Set<TopologyAclBinding> bindings;
   private Set<ServiceAccount> serviceAccounts;
+  private Set<String> topics;
 
   private ExecutionPlan(
       List<Action> plan, PrintStream outputStream, BackendController backendController) {
@@ -33,10 +37,12 @@ public class ExecutionPlan {
     this.outputStream = outputStream;
     this.bindings = new HashSet<>();
     this.serviceAccounts = new HashSet<>();
+    this.topics = new HashSet<>();
     this.backendController = backendController;
     if (backendController.size() > 0) {
       this.bindings.addAll(backendController.getBindings());
       this.serviceAccounts.addAll(backendController.getServiceAccounts());
+      this.topics.addAll(backendController.getTopics());
     }
   }
 
@@ -65,8 +71,9 @@ public class ExecutionPlan {
     }
 
     backendController.reset();
-    backendController.add(new ArrayList<>(bindings));
+    backendController.addBindings(new ArrayList<>(bindings));
     backendController.addServiceAccounts(serviceAccounts);
+    backendController.addTopics(topics);
     backendController.flushAndClose();
   }
 
@@ -76,25 +83,29 @@ public class ExecutionPlan {
       outputStream.println(action);
     } else {
       action.run();
+      if (action instanceof SyncTopicAction) {
+        topics.add(((SyncTopicAction) action).getTopic());
+      } else if (action instanceof DeleteTopics) {
+        List<String> topicsToBeDeleted = ((DeleteTopics) action).getTopicsToBeDeleted();
+        topics =
+            new StreamUtils<>(topics.stream())
+                .filterAsSet(topic -> !topicsToBeDeleted.contains(topic));
+      }
       if (!action.getBindings().isEmpty()) {
         if (action instanceof ClearBindings) {
           bindings =
-              bindings.stream()
-                  .filter(binding -> !action.getBindings().contains(binding))
-                  .collect(Collectors.toSet());
+              new StreamUtils<>(bindings.stream())
+                  .filterAsSet(binding -> !action.getBindings().contains(binding));
         } else {
           bindings.addAll(action.getBindings());
         }
       }
       if (action instanceof BaseAccountsAction) {
         if (action instanceof ClearAccounts) {
-          ClearAccounts clearAccountsAction = (ClearAccounts) action;
+          Collection<ServiceAccount> toDeletePrincipals = ((ClearAccounts) action).getPrincipals();
           serviceAccounts =
-              serviceAccounts.stream()
-                  .filter(
-                      serviceAccount ->
-                          !clearAccountsAction.getPrincipals().contains(serviceAccount))
-                  .collect(Collectors.toSet());
+              new StreamUtils<>(serviceAccounts.stream())
+                  .filterAsSet(sa -> !toDeletePrincipals.contains(sa));
         } else {
           CreateAccounts createAction = (CreateAccounts) action;
           serviceAccounts.addAll(createAction.getPrincipals());
@@ -111,6 +122,10 @@ public class ExecutionPlan {
     return bindings;
   }
 
+  public Set<String> getTopics() {
+    return topics;
+  }
+
   public List<Action> getActions() {
     return plan;
   }
diff --git a/src/main/java/com/purbon/kafka/topology/TopicManager.java b/src/main/java/com/purbon/kafka/topology/TopicManager.java
index f0c8b89..4597bbf 100644
--- a/src/main/java/com/purbon/kafka/topology/TopicManager.java
+++ b/src/main/java/com/purbon/kafka/topology/TopicManager.java
@@ -46,13 +46,8 @@ public class TopicManager {
   }
 
   public void apply(Topology topology, ExecutionPlan plan) throws IOException {
-    // List all topics existing in the cluster, excluding internal topics
-    Set<String> listOfTopics = adminClient.listApplicationTopics();
-    if (listOfTopics.size() > 0)
-      LOGGER.debug(
-          "Full list of topics in the cluster: "
-              + StringUtils.join(new ArrayList<>(listOfTopics), ","));
 
+    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
@@ -86,6 +81,16 @@ public class TopicManager {
     return internalTopicPrefixes.stream().anyMatch(topic::startsWith);
   }
 
+  private Set<String> loadActualClusterStateIfAvailable(ExecutionPlan plan) throws IOException {
+    Set<String> listOfTopics =
+        config.fetchStateFromTheCluster() ? adminClient.listApplicationTopics() : plan.getTopics();
+    if (listOfTopics.size() > 0)
+      LOGGER.debug(
+          "Full list of managed topics in the cluster: "
+              + StringUtils.join(new ArrayList<>(listOfTopics), ","));
+    return listOfTopics;
+  }
+
   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 a327fc1..393f971 100644
--- a/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java
+++ b/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java
@@ -21,21 +21,19 @@ import org.apache.kafka.clients.admin.AdminClientConfig;
 
 public class TopologyBuilderConfig {
 
-  public static final String KAFKA_INTERNAL_TOPIC_PREFIXES = "kafka.internal.topic.prefixes";
-  public static final String ACCESS_CONTROL_IMPLEMENTATION_CLASS =
-      "topology.builder.access.control.class";
+  static final String KAFKA_INTERNAL_TOPIC_PREFIXES = "kafka.internal.topic.prefixes";
+  static final String ACCESS_CONTROL_IMPLEMENTATION_CLASS = "topology.builder.access.control.class";
 
-  public static final String ACCESS_CONTROL_DEFAULT_CLASS =
+  static final String ACCESS_CONTROL_DEFAULT_CLASS =
       "com.purbon.kafka.topology.roles.SimpleAclsProvider";
-  public static final String RBAC_ACCESS_CONTROL_CLASS =
-      "com.purbon.kafka.topology.roles.RBACProvider";
+  static final String RBAC_ACCESS_CONTROL_CLASS = "com.purbon.kafka.topology.roles.RBACProvider";
 
-  public static final String STATE_PROCESSOR_IMPLEMENTATION_CLASS =
+  private static final String STATE_PROCESSOR_IMPLEMENTATION_CLASS =
       "topology.builder.state.processor.class";
 
-  public static final String STATE_PROCESSOR_DEFAULT_CLASS =
+  static final String STATE_PROCESSOR_DEFAULT_CLASS =
       "com.purbon.kafka.topology.backend.FileBackend";
-  public static final String REDIS_STATE_PROCESSOR_CLASS =
+  static final String REDIS_STATE_PROCESSOR_CLASS =
       "com.purbon.kafka.topology.backend.RedisBackend";
 
   static final String REDIS_HOST_CONFIG = "topology.builder.redis.host";
@@ -65,15 +63,18 @@ 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";
-  static final String ALLOW_DELETE_PRINCIPALS = "allow.delete.principals";
+  private static final String ALLOW_DELETE_BINDINGS = "allow.delete.bindings";
+  private static final String ALLOW_DELETE_PRINCIPALS = "allow.delete.principals";
 
-  public static final String CCLOUD_ENV_CONFIG = "ccloud.environment";
+  static final String CCLOUD_ENV_CONFIG = "ccloud.environment";
 
   static final String TOPOLOGY_EXPERIMENTAL_ENABLED_CONFIG = "topology.features.experimental";
   static final String TOPOLOGY_PRINCIPAL_TRANSLATION_ENABLED_CONFIG =
       "topology.translation.principal.enabled";
 
+  public static final String TOPOLOGY_TOPIC_STATE_FROM_CLUSTER =
+      "topology.state.topics.cluster.enabled";
+
   private final Map<String, String> cliParams;
   private Config config;
 
@@ -327,4 +328,8 @@ public class TopologyBuilderConfig {
   public boolean enabledPrincipalTranslation() {
     return config.getBoolean(TOPOLOGY_PRINCIPAL_TRANSLATION_ENABLED_CONFIG);
   }
+
+  public boolean fetchStateFromTheCluster() {
+    return config.getBoolean(TOPOLOGY_TOPIC_STATE_FROM_CLUSTER);
+  }
 }
diff --git a/src/main/java/com/purbon/kafka/topology/actions/topics/DeleteTopics.java b/src/main/java/com/purbon/kafka/topology/actions/topics/DeleteTopics.java
index c5024c7..596e3b3 100644
--- a/src/main/java/com/purbon/kafka/topology/actions/topics/DeleteTopics.java
+++ b/src/main/java/com/purbon/kafka/topology/actions/topics/DeleteTopics.java
@@ -34,4 +34,8 @@ public class DeleteTopics extends BaseAction {
     map.put("topics", topicsToBeDeleted);
     return map;
   }
+
+  public List<String> getTopicsToBeDeleted() {
+    return topicsToBeDeleted;
+  }
 }
diff --git a/src/main/java/com/purbon/kafka/topology/actions/topics/SyncTopicAction.java b/src/main/java/com/purbon/kafka/topology/actions/topics/SyncTopicAction.java
index 563f4fc..746c74b 100644
--- a/src/main/java/com/purbon/kafka/topology/actions/topics/SyncTopicAction.java
+++ b/src/main/java/com/purbon/kafka/topology/actions/topics/SyncTopicAction.java
@@ -35,6 +35,10 @@ public class SyncTopicAction extends BaseAction {
     this.schemaRegistryManager = schemaRegistryManager;
   }
 
+  public String getTopic() {
+    return fullTopicName;
+  }
+
   @Override
   public void run() throws IOException {
     syncTopic(topic, fullTopicName, listOfTopics);
diff --git a/src/main/java/com/purbon/kafka/topology/backend/Backend.java b/src/main/java/com/purbon/kafka/topology/backend/Backend.java
index d8b0584..c51b7c4 100644
--- a/src/main/java/com/purbon/kafka/topology/backend/Backend.java
+++ b/src/main/java/com/purbon/kafka/topology/backend/Backend.java
@@ -16,11 +16,15 @@ public interface Backend {
 
   Set<TopologyAclBinding> loadBindings() throws IOException;
 
+  Set<String> loadTopics() throws IOException;
+
   void saveType(String type);
 
   void saveBindings(Set<TopologyAclBinding> bindings);
 
   void saveAccounts(Set<ServiceAccount> accounts);
 
+  void saveTopics(Set<String> topics);
+
   void close();
 }
diff --git a/src/main/java/com/purbon/kafka/topology/backend/FileBackend.java b/src/main/java/com/purbon/kafka/topology/backend/FileBackend.java
index a62e40d..08d386c 100644
--- a/src/main/java/com/purbon/kafka/topology/backend/FileBackend.java
+++ b/src/main/java/com/purbon/kafka/topology/backend/FileBackend.java
@@ -1,5 +1,6 @@
 package com.purbon.kafka.topology.backend;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.purbon.kafka.topology.BackendController.Mode;
 import com.purbon.kafka.topology.model.cluster.ServiceAccount;
 import com.purbon.kafka.topology.roles.TopologyAclBinding;
@@ -14,9 +15,12 @@ import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.HashSet;
 import java.util.LinkedHashSet;
+import java.util.Objects;
 import java.util.Set;
+import java.util.function.Function;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
+import java.util.stream.Collectors;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
@@ -24,10 +28,13 @@ public class FileBackend implements Backend {
 
   private static final Logger LOGGER = LogManager.getLogger(FileBackend.class);
   public static final String STATE_FILE_NAME = ".cluster-state";
+  static final String SERVICE_ACCOUNTS_TAG = "ServiceAccounts";
+  static final String TOPICS_TAG = "Topics";
+  static final String ACLS_TAG = "acls";
 
   private RandomAccessFile writer;
   private String expression =
-      "^\\'(\\S+)\\',\\s*\\'(\\S+)\\',\\s*\\'(\\S+)\\',\\s*\\'(\\S+)\\',\\s*\\'(\\S+)\\',\\s*\\'(\\S+)\\'";
+      "^\"?\\'(\\S+)\\',\\s*\\'(\\S+)\\',\\s*\\'(\\S+)\\',\\s*\\'(\\S+)\\',\\s*\\'(\\S+)\\',\\s*\\'(\\S+)\\'\"?";
   private Pattern regexp;
 
   public FileBackend() {
@@ -76,7 +83,7 @@ public class FileBackend implements Backend {
         // process service accounts, should break from here.
         break;
       }
-      if (type.equalsIgnoreCase("acls")) {
+      if (type.equalsIgnoreCase(ACLS_TAG)) {
         binding = buildAclBinding(line);
       } else {
         throw new IOException("Binding type ( " + type + " )not supported.");
@@ -87,30 +94,64 @@ public class FileBackend implements Backend {
   }
 
   public Set<ServiceAccount> loadServiceAccounts() throws IOException {
+    return loadItemsFromFile(
+            SERVICE_ACCOUNTS_TAG,
+            line -> {
+              try {
+                return JSON.toObject(line, ServiceAccount.class);
+              } catch (JsonProcessingException e) {
+                LOGGER.error(e);
+                return null;
+              }
+            })
+        .stream()
+        .filter(Objects::nonNull)
+        .map(o -> (ServiceAccount) o)
+        .collect(Collectors.toSet());
+  }
+
+  private boolean foundAControlTag(String line) {
+    return line.equalsIgnoreCase(SERVICE_ACCOUNTS_TAG)
+        || line.equalsIgnoreCase(TOPICS_TAG)
+        || line.equalsIgnoreCase(ACLS_TAG);
+  }
+
+  @Override
+  public Set<String> loadTopics() throws IOException {
+    return loadItemsFromFile(TOPICS_TAG, String::trim).stream()
+        .map(String::valueOf)
+        .collect(Collectors.toSet());
+  }
+
+  private Set<Object> loadItemsFromFile(String tag, Function<String, Object> buildFunction)
+      throws IOException {
     if (writer == null) {
       throw new IOException("state file does not exist");
     }
+    Set<Object> elements = new HashSet<>();
+    BufferedReader in = openLocalStateFile();
+    String line = moveFileToTag(tag, in);
+    if (line != null && line.equalsIgnoreCase(tag)) {
+      while ((line = in.readLine()) != null && !foundAControlTag(line)) {
+        elements.add(buildFunction.apply(line.trim()));
+      }
+    }
+    return elements;
+  }
+
+  private BufferedReader openLocalStateFile() throws IOException {
     Path filePath = Paths.get(STATE_FILE_NAME);
-    Set<ServiceAccount> accounts = new HashSet<>();
-    BufferedReader in = new BufferedReader(new FileReader(filePath.toFile()));
+    return new BufferedReader(new FileReader(filePath.toFile()));
+  }
+
+  private String moveFileToTag(String tag, BufferedReader in) throws IOException {
     String line = null;
     while ((line = in.readLine()) != null) {
-      if (line.equalsIgnoreCase("ServiceAccounts")) {
-        // process service accounts, should break from here.
-        break;
-      }
-    }
-    if (line != null && line.equalsIgnoreCase("ServiceAccounts")) {
-      while ((line = in.readLine()) != null) {
-        ServiceAccount account = (ServiceAccount) JSON.toObject(line, ServiceAccount.class);
-        accounts.add(account);
+      if (line.equalsIgnoreCase(tag)) {
+        break; // process elements, should start from here.
       }
     }
-    return accounts;
-  }
-
-  private TopologyAclBinding buildRBACBinding(String line) {
-    return null;
+    return line;
   }
 
   private TopologyAclBinding buildAclBinding(String line) throws IOException {
@@ -132,42 +173,40 @@ public class FileBackend implements Backend {
   }
 
   public void saveType(String type) {
-    try {
-      writer.writeBytes(type);
-      writer.writeBytes("\n");
-    } catch (IOException e) {
-      LOGGER.error(e);
-    }
+    writeLine(type);
   }
 
   @Override
   public void saveBindings(Set<TopologyAclBinding> bindings) {
-    bindings.stream()
-        .sorted()
-        .forEach(
-            binding -> {
-              try {
-                writer.writeBytes(binding.toString());
-                writer.writeBytes("\n");
-              } catch (IOException e) {
-                LOGGER.error(e);
-              }
-            });
+    bindings.stream().sorted().forEach(b -> writeLine(b.toString()));
   }
 
   @Override
   public void saveAccounts(Set<ServiceAccount> accounts) {
     accounts.forEach(
-        account -> {
+        a -> {
           try {
-            writer.writeBytes(JSON.asString(account));
-            writer.writeBytes("\n");
-          } catch (IOException e) {
+            writeLine(JSON.asString(a));
+          } catch (JsonProcessingException e) {
             LOGGER.error(e);
           }
         });
   }
 
+  @Override
+  public void saveTopics(Set<String> topics) {
+    topics.forEach(this::writeLine);
+  }
+
+  private void writeLine(String line) {
+    try {
+      writer.writeBytes(line);
+      writer.writeBytes("\n");
+    } catch (IOException e) {
+      LOGGER.error(e);
+    }
+  }
+
   @Override
   public void close() {
     try {
diff --git a/src/main/java/com/purbon/kafka/topology/backend/RedisBackend.java b/src/main/java/com/purbon/kafka/topology/backend/RedisBackend.java
index de47b7b..65eed88 100644
--- a/src/main/java/com/purbon/kafka/topology/backend/RedisBackend.java
+++ b/src/main/java/com/purbon/kafka/topology/backend/RedisBackend.java
@@ -47,16 +47,9 @@ public class RedisBackend implements Backend {
     }
   }
 
-  @Override
-  public Set<ServiceAccount> loadServiceAccounts() throws IOException {
-    return new HashSet<>();
-  }
-
   @Override
   public Set<TopologyAclBinding> loadBindings() throws IOException {
-    if (!jedis.isConnected()) {
-      createOrOpen();
-    }
+    connectIfNeed();
 
     Set<TopologyAclBinding> bindings = new HashSet<>();
     String type = jedis.get(KAFKA_TOPOLOGY_BUILDER_TYPE);
@@ -71,6 +64,22 @@ public class RedisBackend implements Backend {
     return bindings;
   }
 
+  private void connectIfNeed() {
+    if (!jedis.isConnected()) {
+      createOrOpen();
+    }
+  }
+
+  @Override
+  public Set<ServiceAccount> loadServiceAccounts() throws IOException {
+    return new HashSet<>();
+  }
+
+  @Override
+  public Set<String> loadTopics() throws IOException {
+    return new HashSet<>();
+  }
+
   @Override
   public void saveType(String type) {
     jedis.set(KAFKA_TOPOLOGY_BUILDER_TYPE, type);
@@ -88,6 +97,9 @@ public class RedisBackend implements Backend {
   @Override
   public void saveAccounts(Set<ServiceAccount> accounts) {}
 
+  @Override
+  public void saveTopics(Set<String> topics) {}
+
   @Override
   public void close() {
     jedis.close();
diff --git a/src/main/java/com/purbon/kafka/topology/utils/JSON.java b/src/main/java/com/purbon/kafka/topology/utils/JSON.java
index ed8098b..848f521 100644
--- a/src/main/java/com/purbon/kafka/topology/utils/JSON.java
+++ b/src/main/java/com/purbon/kafka/topology/utils/JSON.java
@@ -2,12 +2,19 @@ package com.purbon.kafka.topology.utils;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
 import java.util.List;
 import java.util.Map;
 
 public class JSON {
 
-  private static final ObjectMapper mapper = new ObjectMapper();
+  private static final ObjectMapper mapper;
+
+  static {
+    mapper = new ObjectMapper();
+    mapper.registerModule(new Jdk8Module());
+    mapper.findAndRegisterModules();
+  }
 
   public static Map<String, Object> toMap(String jsonString) throws JsonProcessingException {
     return mapper.readValue(jsonString, Map.class);
diff --git a/src/main/java/com/purbon/kafka/topology/utils/StreamUtils.java b/src/main/java/com/purbon/kafka/topology/utils/StreamUtils.java
new file mode 100644
index 0000000..7fd290d
--- /dev/null
+++ b/src/main/java/com/purbon/kafka/topology/utils/StreamUtils.java
@@ -0,0 +1,19 @@
+package com.purbon.kafka.topology.utils;
+
+import java.util.Set;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class StreamUtils<T> {
+
+  private final Stream<T> stream;
+
+  public StreamUtils(Stream<T> stream) {
+    this.stream = stream;
+  }
+
+  public Set<T> filterAsSet(Predicate<T> p) {
+    return stream.filter(p).collect(Collectors.toSet());
+  }
+}
diff --git a/src/main/resources/reference.conf b/src/main/resources/reference.conf
index c0a3603..2b0716f 100644
--- a/src/main/resources/reference.conf
+++ b/src/main/resources/reference.conf
@@ -41,6 +41,11 @@ topology {
       topic.create = true
     }
   }
+  state {
+     topics {
+        cluster.enabled = true
+     }
+  }
 }
 
 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 6c00ac1..72f4ffd 100644
--- a/src/test/java/com/purbon/kafka/topology/AccessControlManagerTest.java
+++ b/src/test/java/com/purbon/kafka/topology/AccessControlManagerTest.java
@@ -68,7 +68,7 @@ public class AccessControlManagerTest {
     TestUtils.deleteStateFile();
     plan = ExecutionPlan.init(backendController, mockPrintStream);
     accessControlManager = new AccessControlManager(aclsProvider, aclsBuilder);
-    doNothing().when(backendController).add(Matchers.anyList());
+    doNothing().when(backendController).addBindings(Matchers.anyList());
     doNothing().when(backendController).flushAndClose();
   }
 
diff --git a/src/test/java/com/purbon/kafka/topology/BackendControllerTest.java b/src/test/java/com/purbon/kafka/topology/BackendControllerTest.java
index 771054f..7d6833e 100644
--- a/src/test/java/com/purbon/kafka/topology/BackendControllerTest.java
+++ b/src/test/java/com/purbon/kafka/topology/BackendControllerTest.java
@@ -6,6 +6,12 @@ import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import com.purbon.kafka.topology.backend.FileBackend;
+import com.purbon.kafka.topology.model.Impl.ProjectImpl;
+import com.purbon.kafka.topology.model.Impl.TopicImpl;
+import com.purbon.kafka.topology.model.Impl.TopologyImpl;
+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.model.cluster.ServiceAccount;
 import com.purbon.kafka.topology.roles.TopologyAclBinding;
 import java.io.IOException;
@@ -44,7 +50,7 @@ public class BackendControllerTest {
         TopologyAclBinding.build(
             ResourceType.CLUSTER.name(), "Topic", "host", "op", "principal", "LITERAL");
 
-    backend.add(binding);
+    backend.addBindings(Collections.singletonList(binding));
 
     assertEquals(1, backend.size());
   }
@@ -60,7 +66,7 @@ public class BackendControllerTest {
 
     ServiceAccount serviceAccount = new ServiceAccount(1, "name", "description");
 
-    backend.add(Collections.singletonList(binding));
+    backend.addBindings(Collections.singletonList(binding));
     backend.addServiceAccounts(Collections.singleton(serviceAccount));
 
     backend.flushAndClose();
@@ -68,4 +74,28 @@ public class BackendControllerTest {
     verify(fileStateProcessor, times(1)).saveBindings(Collections.singleton(binding));
     verify(fileStateProcessor, times(1)).saveAccounts(Collections.singleton(serviceAccount));
   }
+
+  @Test
+  public void testStoreBindingsAndTopics() {
+    BackendController backend = new BackendController(fileStateProcessor);
+
+    Topic topic = new TopicImpl("foo");
+    Project project = new ProjectImpl("project");
+    project.addTopic(topic);
+    Topology topology = new TopologyImpl();
+    topology.setContext("context");
+    topology.addProject(project);
+
+    TopologyAclBinding binding =
+        TopologyAclBinding.build(
+            ResourceType.CLUSTER.name(), "Topic", "host", "op", "principal", "LITERAL");
+
+    backend.addBindings(Collections.singletonList(binding));
+    backend.addTopics(Collections.singleton(topic.getName()));
+
+    backend.flushAndClose();
+
+    verify(fileStateProcessor, times(1)).saveBindings(Collections.singleton(binding));
+    verify(fileStateProcessor, times(1)).saveTopics(Collections.singleton(topic.getName()));
+  }
 }
diff --git a/src/test/java/com/purbon/kafka/topology/ExecutionPlanTest.java b/src/test/java/com/purbon/kafka/topology/ExecutionPlanTest.java
index b52d924..6b83e8d 100644
--- a/src/test/java/com/purbon/kafka/topology/ExecutionPlanTest.java
+++ b/src/test/java/com/purbon/kafka/topology/ExecutionPlanTest.java
@@ -1,13 +1,24 @@
 package com.purbon.kafka.topology;
 
+import static java.util.Collections.singletonList;
 import static org.junit.Assert.assertEquals;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
 import com.purbon.kafka.topology.actions.access.ClearBindings;
 import com.purbon.kafka.topology.actions.access.CreateBindings;
+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.Impl.ProjectImpl;
+import com.purbon.kafka.topology.model.Impl.TopicImpl;
+import com.purbon.kafka.topology.model.Impl.TopologyImpl;
+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.roles.SimpleAclsProvider;
 import com.purbon.kafka.topology.roles.TopologyAclBinding;
+import com.purbon.kafka.topology.schemas.SchemaRegistryManager;
 import com.purbon.kafka.topology.utils.TestUtils;
 import java.io.IOException;
 import java.io.PrintStream;
@@ -24,8 +35,8 @@ import org.mockito.junit.MockitoRule;
 
 public class ExecutionPlanTest {
 
-  ExecutionPlan plan;
-  BackendController backendController;
+  private ExecutionPlan plan;
+  private BackendController backendController;
 
   @Mock PrintStream mockPrintStream;
 
@@ -33,6 +44,10 @@ public class ExecutionPlanTest {
 
   @Rule public MockitoRule mockitoRule = MockitoJUnit.rule();
 
+  @Mock TopologyBuilderAdminClient adminClient;
+
+  @Mock SchemaRegistryManager schemaRegistryManager;
+
   @Before
   public void before() throws IOException {
     TestUtils.deleteStateFile();
@@ -76,7 +91,7 @@ public class ExecutionPlanTest {
     BackendController backendController = new BackendController();
     ExecutionPlan plan = ExecutionPlan.init(backendController, mockPrintStream);
 
-    bindings = new HashSet<>(Arrays.asList(binding2));
+    bindings = new HashSet<>(singletonList(binding2));
     ClearBindings clearBindingsAction = new ClearBindings(aclsProvider, bindings);
 
     plan.add(clearBindingsAction);
@@ -90,4 +105,78 @@ public class ExecutionPlanTest {
     backendController.load();
     assertEquals(1, backendController.size());
   }
+
+  @Test
+  public void addTopicsTest() throws IOException {
+    Topology topology = buildTopologyForTest();
+    Topic topicFoo = topology.getProjects().get(0).getTopics().get(0);
+    Topic topicBar = topology.getProjects().get(0).getTopics().get(1);
+    Set<String> listOfTopics = new HashSet<>();
+
+    SyncTopicAction addTopicAction1 =
+        new SyncTopicAction(
+            adminClient, schemaRegistryManager, topicFoo, topicFoo.toString(), listOfTopics);
+
+    SyncTopicAction addTopicAction2 =
+        new SyncTopicAction(
+            adminClient, schemaRegistryManager, topicBar, topicBar.toString(), listOfTopics);
+
+    plan.add(addTopicAction1);
+    plan.add(addTopicAction2);
+
+    plan.run();
+
+    verify(adminClient, times(1)).createTopic(topicFoo, topicFoo.toString());
+    verify(adminClient, times(1)).createTopic(topicBar, topicBar.toString());
+    assertEquals(2, backendController.size());
+  }
+
+  @Test
+  public void deleteTopicsPreviouslyAddedTest() throws IOException {
+    Topology topology = buildTopologyForTest();
+    Topic topicFoo = topology.getProjects().get(0).getTopics().get(0);
+    Topic topicBar = topology.getProjects().get(0).getTopics().get(1);
+    Set<String> listOfTopics = new HashSet<>();
+
+    SyncTopicAction addTopicAction1 =
+        new SyncTopicAction(
+            adminClient, schemaRegistryManager, topicFoo, topicFoo.toString(), listOfTopics);
+
+    SyncTopicAction addTopicAction2 =
+        new SyncTopicAction(
+            adminClient, schemaRegistryManager, topicBar, topicBar.toString(), listOfTopics);
+
+    plan.add(addTopicAction1);
+    plan.add(addTopicAction2);
+
+    plan.run();
+
+    verify(adminClient, times(1)).createTopic(topicFoo, topicFoo.toString());
+    verify(adminClient, times(1)).createTopic(topicBar, topicBar.toString());
+    assertEquals(2, backendController.size());
+
+    BackendController backendController = new BackendController();
+    ExecutionPlan plan = ExecutionPlan.init(backendController, mockPrintStream);
+
+    DeleteTopics deleteTopicsAction =
+        new DeleteTopics(adminClient, singletonList(topicFoo.toString()));
+    plan.add(deleteTopicsAction);
+
+    plan.run();
+
+    verify(adminClient, times(1)).deleteTopics(singletonList(topicFoo.toString()));
+    assertEquals(1, backendController.size());
+  }
+
+  private Topology buildTopologyForTest() {
+    Topology topology = new TopologyImpl();
+    topology.setContext("context");
+    Project project = new ProjectImpl("project");
+    topology.setProjects(singletonList(project));
+
+    Topic topic = new TopicImpl("foo");
+    Topic topicBar = new TopicImpl("bar");
+    project.setTopics(Arrays.asList(topic, topicBar));
+    return topology;
+  }
 }
diff --git a/src/test/java/com/purbon/kafka/topology/TopicManagerTest.java b/src/test/java/com/purbon/kafka/topology/TopicManagerTest.java
index 71799ad..b258455 100644
--- a/src/test/java/com/purbon/kafka/topology/TopicManagerTest.java
+++ b/src/test/java/com/purbon/kafka/topology/TopicManagerTest.java
@@ -5,6 +5,7 @@ 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 org.mockito.Matchers.any;
 import static org.mockito.Mockito.*;
 
@@ -19,6 +20,8 @@ import com.purbon.kafka.topology.model.Topology;
 import com.purbon.kafka.topology.schemas.SchemaRegistryManager;
 import java.io.IOException;
 import java.io.PrintStream;
+import java.nio.file.Files;
+import java.nio.file.Paths;
 import java.util.*;
 import java.util.stream.Collectors;
 import org.junit.Before;
@@ -34,21 +37,33 @@ public class TopicManagerTest {
 
   @Mock SchemaRegistryManager schemaRegistryManager;
 
-  @Mock BackendController backendController;
+  BackendController backendController;
   ExecutionPlan plan;
+
   @Mock PrintStream outputStream;
 
   @Rule public MockitoRule mockitoRule = MockitoJUnit.rule();
 
   private TopicManager topicManager;
   private HashMap<String, String> cliOps;
+  private Properties props;
+  private TopologyBuilderConfig config;
 
   @Before
   public void setup() throws IOException {
+
+    Files.deleteIfExists(Paths.get(".cluster-state"));
+    backendController = new BackendController();
+
     cliOps = new HashMap<>();
     cliOps.put(BROKERS_OPTION, "");
+    props = new Properties();
+    props.put(TOPOLOGY_TOPIC_STATE_FROM_CLUSTER, "false");
+
     plan = ExecutionPlan.init(backendController, System.out);
-    topicManager = new TopicManager(adminClient, schemaRegistryManager);
+
+    config = new TopologyBuilderConfig(cliOps, props);
+    topicManager = new TopicManager(adminClient, schemaRegistryManager, config);
   }
 
   @Test
@@ -73,23 +88,40 @@ public class TopicManagerTest {
   @Test
   public void topicUpdateTest() throws IOException {
 
+    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");
-    topicB.getConfig().put(NUM_PARTITIONS, "12");
     project.addTopic(topicB);
-    Topology topology = new TopologyImpl();
+
+    topicManager.apply(topology, plan);
+    plan.run();
+
+    verify(adminClient, times(1)).createTopic(topicA, topicA.toString());
+    verify(adminClient, times(1)).createTopic(topicB, topicB.toString());
+
+    ExecutionPlan plan = ExecutionPlan.init(backendController, System.out);
+    TopologyBuilderConfig config = new TopologyBuilderConfig(cliOps, props);
+    TopicManager topicManager = new TopicManager(adminClient, schemaRegistryManager, config);
+
+    topology = new TopologyImpl();
+    project = new ProjectImpl("project");
     topology.addProject(project);
 
-    Set<String> dummyTopicList = new HashSet<>();
-    dummyTopicList.add(topicB.toString());
-    when(adminClient.listApplicationTopics()).thenReturn(dummyTopicList);
+    topicA = new TopicImpl("topicA");
+    project.addTopic(topicA);
+    topicB = new TopicImpl("topicB");
+    topicB.getConfig().put(NUM_PARTITIONS, "12");
+    project.addTopic(topicB);
 
     topicManager.apply(topology, plan);
     plan.run();
 
-    verify(adminClient, times(1)).createTopic(topicA, topicA.toString());
+    verify(adminClient, times(0)).createTopic(topicA, topicA.toString());
+    verify(adminClient, times(0)).createTopic(topicB, topicB.toString());
     verify(adminClient, times(1)).updateTopicConfig(topicB, topicB.toString());
     verify(adminClient, times(1)).getPartitionCount(topicB.toString());
     verify(adminClient, times(1)).updatePartitionCount(topicB, topicB.toString());
diff --git a/src/test/java/com/purbon/kafka/topology/TopologyBuilderAdminClientTest.java b/src/test/java/com/purbon/kafka/topology/TopologyBuilderAdminClientTest.java
index 6df32be..b4c8249 100644
--- a/src/test/java/com/purbon/kafka/topology/TopologyBuilderAdminClientTest.java
+++ b/src/test/java/com/purbon/kafka/topology/TopologyBuilderAdminClientTest.java
@@ -73,7 +73,7 @@ public class TopologyBuilderAdminClientTest {
 
     plan = ExecutionPlan.init(backendController, System.out);
 
-    doNothing().when(backendController).add(Matchers.anyList());
+    doNothing().when(backendController).addBindings(Matchers.anyList());
     doNothing().when(backendController).flushAndClose();
 
     doReturn("foo").when(config).getConfluentCommandTopic();
diff --git a/src/test/java/com/purbon/kafka/topology/backend/FileBackendTest.java b/src/test/java/com/purbon/kafka/topology/backend/FileBackendTest.java
new file mode 100644
index 0000000..700a046
--- /dev/null
+++ b/src/test/java/com/purbon/kafka/topology/backend/FileBackendTest.java
@@ -0,0 +1,79 @@
+package com.purbon.kafka.topology.backend;
+
+import static com.purbon.kafka.topology.backend.FileBackend.ACLS_TAG;
+import static com.purbon.kafka.topology.backend.FileBackend.SERVICE_ACCOUNTS_TAG;
+import static com.purbon.kafka.topology.backend.FileBackend.STATE_FILE_NAME;
+import static com.purbon.kafka.topology.backend.FileBackend.TOPICS_TAG;
+import static java.util.Collections.singletonList;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.purbon.kafka.topology.BackendController.Mode;
+import com.purbon.kafka.topology.model.Impl.ProjectImpl;
+import com.purbon.kafka.topology.model.Impl.TopicImpl;
+import com.purbon.kafka.topology.model.Impl.TopologyImpl;
+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.roles.TopologyAclBinding;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+import org.apache.kafka.common.resource.ResourceType;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class FileBackendTest {
+
+  private FileBackend backend;
+
+  @Before
+  public void setup() {
+    backend = new FileBackend();
+  }
+
+  @After
+  public void after() throws IOException {
+    Files.deleteIfExists(Paths.get(STATE_FILE_NAME));
+  }
+
+  @Test
+  public void testStoreAndLoadBindingsAndTopics() throws IOException {
+    TopologyAclBinding binding =
+        TopologyAclBinding.build(
+            ResourceType.CLUSTER.name(), "Topic", "host", "op", "principal", "LITERAL");
+
+    Topology topology = new TopologyImpl();
+    topology.setContext("context");
+    Project project = new ProjectImpl("project");
+    topology.setProjects(singletonList(project));
+
+    Topic topic = new TopicImpl("foo");
+    Topic topicBar = new TopicImpl("bar");
+    project.setTopics(Arrays.asList(topic, topicBar));
+
+    backend.createOrOpen(Mode.TRUNCATE);
+    backend.saveType(ACLS_TAG);
+    backend.saveBindings(Collections.singleton(binding));
+    backend.saveType(SERVICE_ACCOUNTS_TAG);
+    backend.saveType(TOPICS_TAG);
+    backend.saveTopics(new HashSet<>(Arrays.asList(topic.toString(), topicBar.toString())));
+    backend.close();
+
+    backend = new FileBackend();
+    backend.createOrOpen();
+
+    Set<TopologyAclBinding> bindings = backend.loadBindings();
+    Set<String> topics = backend.loadTopics();
+
+    assertThat(bindings).hasSize(1);
+    assertThat(bindings).contains(binding);
+    assertThat(topics).hasSize(2);
+    assertThat(topics).contains(topic.toString());
+    assertThat(topics).contains(topicBar.toString());
+  }
+}
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 8a107e9..89759e6 100644
--- a/src/test/java/com/purbon/kafka/topology/integration/RBACPRoviderRbacIT.java
+++ b/src/test/java/com/purbon/kafka/topology/integration/RBACPRoviderRbacIT.java
@@ -110,7 +110,7 @@ public class RBACPRoviderRbacIT extends MDSBaseTest {
     plan.run();
 
     // this method is call twice, once for consumers and one for producers
-    verify(cs, times(1)).add(anyList());
+    verify(cs, times(1)).addBindings(anyList());
     verify(cs, times(1)).flushAndClose();
     verifyConsumerAcls(consumers, topicA.toString());
   }
@@ -134,7 +134,7 @@ public class RBACPRoviderRbacIT extends MDSBaseTest {
     plan.run();
 
     // this method is call twice, once for consumers and one for consumers
-    verify(cs, times(1)).add(anyList());
+    verify(cs, times(1)).addBindings(anyList());
     verify(cs, times(1)).flushAndClose();
     verifyProducerAcls(producers, topicA.toString());
   }
@@ -158,7 +158,7 @@ public class RBACPRoviderRbacIT extends MDSBaseTest {
     accessControlManager.apply(topology, plan);
     plan.run();
 
-    verify(cs, times(1)).add(anyList());
+    verify(cs, times(1)).addBindings(anyList());
     verify(cs, times(1)).flushAndClose();
     verifyKStreamsAcls(app);
   }
@@ -181,7 +181,7 @@ public class RBACPRoviderRbacIT extends MDSBaseTest {
     accessControlManager.apply(topology, plan);
     plan.run();
 
-    verify(cs, times(1)).add(anyList());
+    verify(cs, times(1)).addBindings(anyList());
     verify(cs, times(1)).flushAndClose();
     verifyConnectAcls(connector);
   }
@@ -215,7 +215,7 @@ public class RBACPRoviderRbacIT extends MDSBaseTest {
     accessControlManager.apply(topology, plan);
     plan.run();
 
-    verify(cs, times(1)).add(anyList());
+    verify(cs, times(1)).addBindings(anyList());
     verify(cs, times(1)).flushAndClose();
     verifySchemaRegistryAcls(platform);
   }
@@ -241,7 +241,7 @@ public class RBACPRoviderRbacIT extends MDSBaseTest {
     accessControlManager.apply(topology, plan);
     plan.run();
 
-    verify(cs, times(1)).add(anyList());
+    verify(cs, times(1)).addBindings(anyList());
     verify(cs, times(1)).flushAndClose();
     verifyControlCenterAcls(platform);
   }
@@ -266,7 +266,7 @@ public class RBACPRoviderRbacIT extends MDSBaseTest {
     accessControlManager.apply(topology, plan);
     plan.run();
 
-    verify(cs, times(1)).add(anyList());
+    verify(cs, times(1)).addBindings(anyList());
     verify(cs, times(1)).flushAndClose();
 
     verifyKafkaClusterACLs(platform);
@@ -292,7 +292,7 @@ public class RBACPRoviderRbacIT extends MDSBaseTest {
     accessControlManager.apply(topology, plan);
     plan.run();
 
-    verify(cs, times(1)).add(anyList());
+    verify(cs, times(1)).addBindings(anyList());
     verify(cs, times(1)).flushAndClose();
 
     verifyConnectClusterACLs(platform);
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 6c75350..3c90847 100644
--- a/src/test/java/com/purbon/kafka/topology/integration/TopicManagerIT.java
+++ b/src/test/java/com/purbon/kafka/topology/integration/TopicManagerIT.java
@@ -2,7 +2,8 @@ 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 com.purbon.kafka.topology.TopologyBuilderConfig.TOPOLOGY_TOPIC_STATE_FROM_CLUSTER;
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertTrue;
 
 import com.purbon.kafka.topology.BackendController;
@@ -22,6 +23,8 @@ import com.purbon.kafka.topology.schemas.SchemaRegistryManager;
 import io.confluent.kafka.schemaregistry.client.MockSchemaRegistryClient;
 import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
 import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -70,6 +73,8 @@ public class TopicManagerIT {
 
   @Before
   public void before() throws IOException {
+    Files.deleteIfExists(Paths.get(".cluster-state"));
+
     kafkaAdminClient = ContainerTestUtils.getSaslAdminClient(container);
     TopologyBuilderAdminClient adminClient = new TopologyBuilderAdminClient(kafkaAdminClient);
 
@@ -79,6 +84,7 @@ public class TopicManagerIT {
     this.plan = ExecutionPlan.init(new BackendController(), System.out);
 
     Properties props = new Properties();
+    props.put(TOPOLOGY_TOPIC_STATE_FROM_CLUSTER, "false");
 
     HashMap<String, String> cliOps = new HashMap<>();
     cliOps.put(BROKERS_OPTION, "");
@@ -91,11 +97,16 @@ public class TopicManagerIT {
 
   @Test
   public void testTopicCreation() throws ExecutionException, InterruptedException, IOException {
+
+    Topology topology = new TopologyImpl();
+    topology.setContext("testTopicCreation");
+    Project project = new ProjectImpl("project");
+    topology.addProject(project);
+
     HashMap<String, String> config = new HashMap<>();
     config.put(TopicManager.NUM_PARTITIONS, "1");
     config.put(TopicManager.REPLICATION_FACTOR, "1");
 
-    Project project = new ProjectImpl("project");
     Topic topicA = new TopicImpl("topicA", config);
     project.addTopic(topicA);
 
@@ -106,9 +117,6 @@ public class TopicManagerIT {
     Topic topicB = new TopicImpl("topicB", config);
     project.addTopic(topicB);
 
-    Topology topology = new TopologyImpl();
-    topology.addProject(project);
-
     topicManager.apply(topology, plan);
     plan.run();
 
@@ -123,12 +131,13 @@ public class TopicManagerIT {
     config.put("banana", "bar");
 
     Project project = new ProjectImpl("project");
-    Topic topicA = new TopicImpl("topicA", config);
-    project.addTopic(topicA);
-
     Topology topology = new TopologyImpl();
+    topology.setContext("testTopicCreationWithFalseConfig");
     topology.addProject(project);
 
+    Topic topicA = new TopicImpl("topicA", config);
+    project.addTopic(topicA);
+
     topicManager.apply(topology, plan);
     plan.run();
   }
@@ -141,6 +150,7 @@ public class TopicManagerIT {
     config.put(TopicManager.REPLICATION_FACTOR, "1");
 
     Topology topology = new TopologyImpl();
+    topology.setContext("testTopicCreationWithChangedTopology");
     Project project = new ProjectImpl("project");
     topology.addProject(project);
 
@@ -160,7 +170,7 @@ public class TopicManagerIT {
     verifyTopics(Arrays.asList(topicA.toString(), topicB.toString()));
 
     Topology upTopology = new TopologyImpl();
-    upTopology.setContext("foo");
+    upTopology.setContext("testTopicCreationWithChangedTopology");
     Project upProject = new ProjectImpl("bar");
     upTopology.addProject(upProject);
 
@@ -240,16 +250,16 @@ public class TopicManagerIT {
   public void testTopicCreationWithConfig()
       throws ExecutionException, InterruptedException, IOException {
 
+    Topology topology = new TopologyImpl();
+    topology.setContext("testTopicCreationWithConfig-test");
+    Project project = new ProjectImpl("project");
+    topology.addProject(project);
+
     HashMap<String, String> config = buildDummyTopicConfig();
     config.put("retention.bytes", "104857600"); // set the retention.bytes per partition to 100mb
-    Project project = new ProjectImpl("project");
     Topic topicA = new TopicImpl("topicA", config);
     project.addTopic(topicA);
 
-    Topology topology = new TopologyImpl();
-    topology.setContext("testTopicCreationWithConfig-test");
-    topology.addProject(project);
-
     topicManager.apply(topology, plan);
     plan.run();
 
@@ -263,14 +273,14 @@ public class TopicManagerIT {
     config.put("retention.bytes", "104857600"); // set the retention.bytes per partition to 100mb
     config.put("segment.bytes", "104857600");
 
-    Project project = new ProjectImpl("project");
-    Topic topicA = new TopicImpl("topicA", config);
-    project.addTopic(topicA);
-
     Topology topology = new TopologyImpl();
     topology.setContext("testTopicConfigUpdate-test");
+    Project project = new ProjectImpl("project");
     topology.addProject(project);
 
+    Topic topicA = new TopicImpl("topicA", config);
+    project.addTopic(topicA);
+
     topicManager.apply(topology, plan);
     plan.run();
 
@@ -346,8 +356,7 @@ public class TopicManagerIT {
     Set<String> nonInternalTopics =
         topicNames.stream().filter(topic -> !topic.startsWith("_")).collect(Collectors.toSet());
 
-    assertEquals(topicsCount, nonInternalTopics.size());
-
+    assertThat(topicsCount).isLessThanOrEqualTo(nonInternalTopics.size());
     assertTrue("Internal topics not found", isInternal);
   }
 }
diff --git a/src/test/java/com/purbon/kafka/topology/integration/backend/DefaultBackendIT.java b/src/test/java/com/purbon/kafka/topology/integration/backend/DefaultBackendIT.java
index 1a28474..0462471 100644
--- a/src/test/java/com/purbon/kafka/topology/integration/backend/DefaultBackendIT.java
+++ b/src/test/java/com/purbon/kafka/topology/integration/backend/DefaultBackendIT.java
@@ -39,7 +39,7 @@ public class DefaultBackendIT {
 
     Set<ServiceAccount> accounts = new HashSet<>(Arrays.asList(serviceAccount, serviceAccount2));
 
-    backend.add(Collections.singletonList(binding));
+    backend.addBindings(Collections.singletonList(binding));
     backend.addServiceAccounts(accounts);
     backend.flushAndClose();
 
@@ -60,7 +60,7 @@ public class DefaultBackendIT {
         TopologyAclBinding.build(
             ResourceType.CLUSTER.name(), "Topic", "host", "op", "principal", "LITERAL");
 
-    backend.add(Collections.singletonList(binding));
+    backend.addBindings(Collections.singletonList(binding));
     backend.flushAndClose();
 
     // reopen a new connection