Improved execution log for topics and schemas

src/main/java/com/purbon/kafka/topology/PrincipalManager.java → src/main/java/com/purbon/kafka/topology/AbstractPrincipalManager.java

@@ -1,34 +1,38 @@
 package com.purbon.kafka.topology;
 
-import com.purbon.kafka.topology.actions.accounts.ClearAccounts;
-import com.purbon.kafka.topology.actions.accounts.CreateAccounts;
-import com.purbon.kafka.topology.model.*;
+import com.purbon.kafka.topology.model.Platform;
+import com.purbon.kafka.topology.model.Topic;
+import com.purbon.kafka.topology.model.Topology;
+import com.purbon.kafka.topology.model.User;
 import com.purbon.kafka.topology.model.cluster.ServiceAccount;
 import com.purbon.kafka.topology.serviceAccounts.VoidPrincipalProvider;
 import java.io.IOException;
 import java.io.PrintStream;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
-public class PrincipalManager {
+abstract class AbstractPrincipalManager implements ExecutionPlanUpdater {
 
-  private static final Logger LOGGER = LogManager.getLogger(PrincipalManager.class);
+  private static final Logger LOGGER = LogManager.getLogger(AbstractPrincipalManager.class);
   private final List<String> managedPrefixes;
+  protected PrincipalProvider provider;
+  protected Configuration config;
 
-  private PrincipalProvider provider;
-
-  private Configuration config;
-
-  public PrincipalManager(PrincipalProvider provider, Configuration config) {
+  public AbstractPrincipalManager(PrincipalProvider provider, Configuration config) {
     this.provider = provider;
     this.config = config;
     this.managedPrefixes = config.getServiceAccountManagedPrefixes();
   }
 
-  public void applyCreate(Topology topology, ExecutionPlan plan) throws IOException {
+  @Override
+  public final void updatePlan(ExecutionPlan plan, Map<String, Topology> topologies)
+      throws IOException {
     if (!config.enabledExperimental()) {
       LOGGER.debug("Not running the PrincipalsManager as this is an experimental feature.");
       return;
@@ -38,52 +42,21 @@ public void applyCreate(Topology topology, ExecutionPlan plan) throws IOExceptio
       // This means the management of principals is either not possible or has not been configured
       return;
     }
-
     provider.configure();
 
-    List<String> principals = parseListOfPrincipals(topology);
     Map<String, ServiceAccount> accounts = loadActualClusterStateIfAvailable(plan);
-
-    // build set of principals to be created.
-    Set<ServiceAccount> principalsToBeCreated =
-        principals.stream()
-            .filter(wishPrincipal -> !accounts.containsKey(wishPrincipal))
-            .map(principal -> new ServiceAccount(-1, principal, "Managed by KTB"))
-            .collect(Collectors.toSet());
-
-    if (!principalsToBeCreated.isEmpty()) {
-      plan.add(new CreateAccounts(provider, principalsToBeCreated));
-    }
-  }
-
-  public void applyDelete(Topology topology, ExecutionPlan plan) throws IOException {
-    if (!config.enabledExperimental()) {
-      LOGGER.debug("Not running the PrincipalsManager as this is an experimental feature.");
-      return;
-    }
-    if (provider instanceof VoidPrincipalProvider) {
-      // Do Nothing if the provider is the void one.
-      // This means the management of principals is either not possible or has not been configured
-      return;
-    }
-
-    if (config.isAllowDeletePrincipals()) {
-      provider.configure();
-
+    for (Topology topology : topologies.values()) {
       List<String> principals = parseListOfPrincipals(topology);
-      Map<String, ServiceAccount> accounts = loadActualClusterStateIfAvailable(plan);
-
-      // build list of principals to be deleted.
-      List<ServiceAccount> principalsToBeDeleted =
-          accounts.values().stream()
-              .filter(currentPrincipal -> !principals.contains(currentPrincipal.getName()))
-              .collect(Collectors.toList());
-      if (!principalsToBeDeleted.isEmpty()) {
-        plan.add(new ClearAccounts(provider, principalsToBeDeleted));
-      }
+      doUpdatePlan(plan, topology, principals, accounts);
     }
   }
 
+  protected abstract void doUpdatePlan(
+      ExecutionPlan plan,
+      Topology topology,
+      final List<String> principals,
+      final Map<String, ServiceAccount> accounts);
+
   private Map<String, ServiceAccount> loadActualClusterStateIfAvailable(ExecutionPlan plan)
       throws IOException {
     Set<ServiceAccount> accounts =
@@ -132,7 +105,8 @@ private List<String> parseListOfPrincipals(Topology topology) {
         .collect(Collectors.toList());
   }
 
-  public void printCurrentState(PrintStream out) throws IOException {
+  @Override
+  public final void printCurrentState(PrintStream out) throws IOException {
     out.println("List of Principles: ");
     provider.listServiceAccounts().forEach(out::println);
   }