Repository: kafka-ops/kafka-topology-builder
Pull request #127 (2020-11-12 14:08:29)

Fix ACL removal bug
fixes important error reported at #125 

purbon: @akselh this PR should fix the problem you detected, I apologize for the inconvenience. 

Patch:
diff --git a/src/main/java/com/purbon/kafka/topology/AccessControlManager.java b/src/main/java/com/purbon/kafka/topology/AccessControlManager.java
index 084e1a7..25a1ae5 100644
--- a/src/main/java/com/purbon/kafka/topology/AccessControlManager.java
+++ b/src/main/java/com/purbon/kafka/topology/AccessControlManager.java
@@ -181,8 +181,10 @@ public class AccessControlManager {
             .filter(binding -> !bindings.contains(binding))
             .collect(Collectors.toSet());
 
-    CreateBindings createBindings = new CreateBindings(controlProvider, bindingsToBeCreated);
-    updateActions.add(createBindings);
+    if (!bindingsToBeCreated.isEmpty()) {
+      CreateBindings createBindings = new CreateBindings(controlProvider, bindingsToBeCreated);
+      updateActions.add(createBindings);
+    }
 
     if (config.allowDelete()) {
       // clear acls that does not appear anymore in the new generated list,
@@ -191,9 +193,10 @@ public class AccessControlManager {
           bindings.stream()
               .filter(binding -> !allFinalBindings.contains(binding))
               .collect(Collectors.toSet());
-
-      ClearBindings clearBindings = new ClearBindings(controlProvider, bindingsToDelete);
-      updateActions.add(clearBindings);
+      if (!bindingsToDelete.isEmpty()) {
+        ClearBindings clearBindings = new ClearBindings(controlProvider, bindingsToDelete);
+        updateActions.add(clearBindings);
+      }
     }
     return updateActions;
   }
diff --git a/src/main/java/com/purbon/kafka/topology/BackendController.java b/src/main/java/com/purbon/kafka/topology/BackendController.java
index 7ab5bd7..fcd13d7 100644
--- a/src/main/java/com/purbon/kafka/topology/BackendController.java
+++ b/src/main/java/com/purbon/kafka/topology/BackendController.java
@@ -12,6 +12,11 @@ import org.apache.logging.log4j.Logger;
 
 public class BackendController {
 
+  public enum Mode {
+    TRUNCATE,
+    APPEND
+  }
+
   private static final Logger LOGGER = LogManager.getLogger(BackendController.class);
 
   private static final String STORE_TYPE = "acls";
@@ -44,7 +49,7 @@ public class BackendController {
 
   public void flushAndClose() {
     LOGGER.debug(String.format("Flushing the current state of %s, %s", STORE_TYPE, bindings));
-    backend.createOrOpen();
+    backend.createOrOpen(Mode.TRUNCATE);
     backend.saveType(STORE_TYPE);
     backend.saveBindings(bindings);
     backend.close();
diff --git a/src/main/java/com/purbon/kafka/topology/ExecutionPlan.java b/src/main/java/com/purbon/kafka/topology/ExecutionPlan.java
index 0535a4d..dc1aab8 100644
--- a/src/main/java/com/purbon/kafka/topology/ExecutionPlan.java
+++ b/src/main/java/com/purbon/kafka/topology/ExecutionPlan.java
@@ -28,6 +28,9 @@ public class ExecutionPlan {
     this.outputStream = outputStream;
     this.bindings = new HashSet<>();
     this.backendController = backendController;
+    if (backendController.size() > 0) {
+      this.bindings.addAll(backendController.getBindings());
+    }
   }
 
   public void add(Action action) {
@@ -70,7 +73,7 @@ public class ExecutionPlan {
         if (action instanceof ClearBindings) {
           bindings =
               bindings.stream()
-                  .filter(binding -> action.getBindings().contains(binding))
+                  .filter(binding -> !action.getBindings().contains(binding))
                   .collect(Collectors.toSet());
         } else {
           bindings.addAll(action.getBindings());
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 2dc3b91..18dd956 100644
--- a/src/main/java/com/purbon/kafka/topology/backend/Backend.java
+++ b/src/main/java/com/purbon/kafka/topology/backend/Backend.java
@@ -1,5 +1,6 @@
 package com.purbon.kafka.topology.backend;
 
+import com.purbon.kafka.topology.BackendController;
 import com.purbon.kafka.topology.roles.TopologyAclBinding;
 import java.io.IOException;
 import java.net.URI;
@@ -9,6 +10,8 @@ public interface Backend {
 
   void createOrOpen();
 
+  void createOrOpen(BackendController.Mode mode);
+
   Set<TopologyAclBinding> load() throws IOException;
 
   Set<TopologyAclBinding> load(URI uri) throws IOException;
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 dbce655..00f54aa 100644
--- a/src/main/java/com/purbon/kafka/topology/backend/FileBackend.java
+++ b/src/main/java/com/purbon/kafka/topology/backend/FileBackend.java
@@ -1,12 +1,12 @@
 package com.purbon.kafka.topology.backend;
 
+import com.purbon.kafka.topology.BackendController.Mode;
 import com.purbon.kafka.topology.roles.TopologyAclBinding;
 import java.io.BufferedReader;
 import java.io.File;
 import java.io.FileReader;
-import java.io.FileWriter;
 import java.io.IOException;
-import java.io.Writer;
+import java.io.RandomAccessFile;
 import java.net.URI;
 import java.nio.file.Path;
 import java.nio.file.Paths;
@@ -21,7 +21,7 @@ public class FileBackend implements Backend {
 
   private static final Logger LOGGER = LogManager.getLogger(FileBackend.class);
 
-  private Writer writer;
+  private RandomAccessFile writer;
   private String expression =
       "^\\'(\\S+)\\',\\s*\\'(\\S+)\\',\\s*\\'(\\S+)\\',\\s*\\'(\\S+)\\',\\s*\\'(\\S+)\\',\\s*\\'(\\S+)\\'";
   private Pattern regexp;
@@ -33,10 +33,22 @@ public class FileBackend implements Backend {
 
   @Override
   public void createOrOpen() {
+    createOrOpen(Mode.APPEND);
+  }
+
+  @Override
+  public void createOrOpen(Mode mode) {
     try {
-      writer = new FileWriter(filename());
+      writer = new RandomAccessFile(filename(), "rw");
+      Path path = Paths.get(filename());
+      if (path.toFile().exists()) {
+        writer.seek(0);
+        if (mode.equals(Mode.TRUNCATE)) {
+          writer.getChannel().truncate(0);
+        }
+      }
     } catch (IOException e) {
-      e.printStackTrace();
+      LOGGER.error(e);
     }
   }
 
@@ -92,8 +104,8 @@ public class FileBackend implements Backend {
 
   public void saveType(String type) {
     try {
-      writer.write(type);
-      writer.write("\n");
+      writer.writeBytes(type);
+      writer.writeBytes("\n");
     } catch (IOException e) {
       LOGGER.error(e);
     }
@@ -104,8 +116,8 @@ public class FileBackend implements Backend {
     bindings.forEach(
         binding -> {
           try {
-            writer.write(binding.toString());
-            writer.write("\n");
+            writer.writeBytes(binding.toString());
+            writer.writeBytes("\n");
           } catch (IOException e) {
             LOGGER.error(e);
           }
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 d97c922..61e1e93 100644
--- a/src/main/java/com/purbon/kafka/topology/backend/RedisBackend.java
+++ b/src/main/java/com/purbon/kafka/topology/backend/RedisBackend.java
@@ -1,5 +1,6 @@
 package com.purbon.kafka.topology.backend;
 
+import com.purbon.kafka.topology.BackendController.Mode;
 import com.purbon.kafka.topology.roles.TopologyAclBinding;
 import java.io.IOException;
 import java.net.URI;
@@ -14,6 +15,7 @@ import redis.clients.jedis.Jedis;
 public class RedisBackend implements Backend {
 
   private static final Logger LOGGER = LogManager.getLogger(RedisBackend.class);
+
   static final String KAFKA_TOPOLOGY_BUILDER_BINDINGS = "kafka.topology.builder.bindings";
   static final String KAFKA_TOPOLOGY_BUILDER_TYPE = "kafka.topology.builder.type";
 
@@ -33,7 +35,16 @@ public class RedisBackend implements Backend {
 
   @Override
   public void createOrOpen() {
+    createOrOpen(Mode.APPEND);
+  }
+
+  @Override
+  public void createOrOpen(Mode mode) {
     jedis.connect();
+    if (mode.equals(Mode.TRUNCATE)) {
+      jedis.del(KAFKA_TOPOLOGY_BUILDER_TYPE);
+      jedis.del(KAFKA_TOPOLOGY_BUILDER_BINDINGS);
+    }
   }
 
   @Override


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 fea1576..98da7aa 100644
--- a/src/test/java/com/purbon/kafka/topology/AccessControlManagerTest.java
+++ b/src/test/java/com/purbon/kafka/topology/AccessControlManagerTest.java
@@ -41,6 +41,8 @@ import com.purbon.kafka.topology.roles.TopologyAclBinding;
 import com.purbon.kafka.topology.roles.acls.AclsBindingsBuilder;
 import java.io.IOException;
 import java.io.PrintStream;
+import java.nio.file.Files;
+import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -82,6 +84,7 @@ public class AccessControlManagerTest {
 
   @Before
   public void setup() throws IOException {
+    Files.deleteIfExists(Paths.get(".cluster-state"));
     plan = ExecutionPlan.init(backendController, mockPrintStream);
     accessControlManager = new AccessControlManager(aclsProvider, aclsBuilder);
     doNothing().when(backendController).add(Matchers.anyList());
@@ -450,28 +453,30 @@ public class AccessControlManagerTest {
   public void testDryRunMode() throws IOException {
 
     plan = ExecutionPlan.init(backendController, mockPrintStream);
+    Topology topology = new TopologyImpl();
+    topology.setContext("foo");
 
     List<Consumer> consumers = new ArrayList<>();
     consumers.add(new Consumer("User:app1"));
-    Project project = new ProjectImpl();
+
+    Project project = new ProjectImpl("project");
     project.setConsumers(consumers);
+    topology.addProject(project);
 
     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>())
+    doReturn(Collections.singletonList(new TopologyAclBinding()))
         .when(aclsBuilder)
         .buildBindingsForConsumers(users, topicA.toString(), false);
+
     accessControlManager.apply(topology, plan);
 
     plan.run(true);
 
-    verify(mockPrintStream, times(2)).println(any(Action.class));
+    verify(mockPrintStream, times(1)).println(any(Action.class));
   }
 
   @Test
diff --git a/src/test/java/com/purbon/kafka/topology/ExecutionPlanTest.java b/src/test/java/com/purbon/kafka/topology/ExecutionPlanTest.java
new file mode 100644
index 0000000..643d45b
--- /dev/null
+++ b/src/test/java/com/purbon/kafka/topology/ExecutionPlanTest.java
@@ -0,0 +1,97 @@
+package com.purbon.kafka.topology;
+
+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.roles.SimpleAclsProvider;
+import com.purbon.kafka.topology.roles.TopologyAclBinding;
+import java.io.IOException;
+import java.io.PrintStream;
+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.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnit;
+import org.mockito.junit.MockitoRule;
+
+public class ExecutionPlanTest {
+
+  ExecutionPlan plan;
+  BackendController backendController;
+
+  @Mock
+  PrintStream mockPrintStream;
+
+  @Mock
+  SimpleAclsProvider aclsProvider;
+
+  @Rule
+  public MockitoRule mockitoRule = MockitoJUnit.rule();
+
+
+  @Before
+  public void before() throws IOException {
+    Files.deleteIfExists(Paths.get(".cluster-state"));
+    backendController = new BackendController();
+    plan = ExecutionPlan.init(backendController, mockPrintStream);
+  }
+
+  @Test
+  public void addBindingsTest() throws IOException {
+    TopologyAclBinding binding1 = new TopologyAclBinding(ResourceType.ANY, "topicA", "*", "ALL", "User:foo", "LITERAL");
+    TopologyAclBinding binding2 = new TopologyAclBinding(ResourceType.ANY, "topicB", "*", "ALL", "User:foo", "LITERAL");
+    Set<TopologyAclBinding> bindings = new HashSet<>(Arrays.asList(binding1, binding2));
+    CreateBindings addBindingsAction = new CreateBindings(aclsProvider, bindings);
+
+    plan.add(addBindingsAction);
+
+    plan.run();
+
+    verify(aclsProvider, times(1)).createBindings(bindings);
+    assertEquals(2, backendController.size());
+  }
+
+  @Test
+  public void deleteBindingsAfterCreateTest() throws IOException {
+    TopologyAclBinding binding1 = new TopologyAclBinding(ResourceType.ANY, "topicA", "*", "ALL", "User:foo", "LITERAL");
+    TopologyAclBinding binding2 = new TopologyAclBinding(ResourceType.ANY, "topicB", "*", "ALL", "User:foo", "LITERAL");
+    Set<TopologyAclBinding> bindings = new HashSet<>(Arrays.asList(binding1, binding2));
+    CreateBindings addBindingsAction = new CreateBindings(aclsProvider, bindings);
+
+    plan.add(addBindingsAction);
+
+    plan.run();
+
+    verify(aclsProvider, times(1)).createBindings(bindings);
+    assertEquals(2, backendController.size());
+
+    BackendController backendController = new BackendController();
+    ExecutionPlan plan = ExecutionPlan.init(backendController, mockPrintStream);
+
+    bindings = new HashSet<>(Arrays.asList(binding2));
+    ClearBindings clearBindingsAction = new ClearBindings(aclsProvider, bindings);
+
+    plan.add(clearBindingsAction);
+
+    plan.run();
+
+    verify(aclsProvider, times(1)).clearBindings(bindings);
+    assertEquals(1, backendController.size());
+
+    backendController = new BackendController();
+    backendController.load();
+    assertEquals(1, backendController.size());
+
+  }
+
+}
diff --git a/src/test/java/com/purbon/kafka/topology/KafkaTopologyBuilderTest.java b/src/test/java/com/purbon/kafka/topology/KafkaTopologyBuilderTest.java
index 8344ad2..a522c11 100644
--- a/src/test/java/com/purbon/kafka/topology/KafkaTopologyBuilderTest.java
+++ b/src/test/java/com/purbon/kafka/topology/KafkaTopologyBuilderTest.java
@@ -12,6 +12,7 @@ import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
+import com.purbon.kafka.topology.BackendController.Mode;
 import com.purbon.kafka.topology.api.adminclient.TopologyBuilderAdminClient;
 import com.purbon.kafka.topology.backend.RedisBackend;
 import com.purbon.kafka.topology.model.Topology;
@@ -210,7 +211,8 @@ public class KafkaTopologyBuilderTest {
     builder.run(plan);
     builder.close();
 
-    verify(stateProcessor, times(2)).createOrOpen();
+    verify(stateProcessor, times(1)).createOrOpen();
+    verify(stateProcessor, times(1)).createOrOpen(Mode.TRUNCATE);
     verify(topicManager, times(1)).apply(anyObject(), anyObject());
     verify(accessControlManager, times(1)).apply(anyObject(), anyObject());
   }
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 fd9c2bc..96f5360 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,6 @@
 package com.purbon.kafka.topology.integration;
 
+import static org.junit.Assert.assertEquals;
 import static org.mockito.Mockito.when;
 
 import com.purbon.kafka.topology.AccessControlManager;
@@ -26,6 +27,8 @@ import com.purbon.kafka.topology.roles.SimpleAclsProvider;
 import com.purbon.kafka.topology.roles.TopologyAclBinding;
 import com.purbon.kafka.topology.roles.acls.AclsBindingsBuilder;
 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;
@@ -73,6 +76,7 @@ public class AccessControlManagerIT {
     kafkaAdminClient = AdminClient.create(config());
     TopologyBuilderAdminClient adminClient = new TopologyBuilderAdminClient(kafkaAdminClient);
     adminClient.clearAcls();
+    Files.deleteIfExists(Paths.get(".cluster-state"));
 
     this.cs = new BackendController();
     this.plan = ExecutionPlan.init(cs, System.out);
@@ -108,7 +112,7 @@ public class AccessControlManagerIT {
     accessControlManager.apply(topology, plan);
     plan.run();
 
-    Assert.assertEquals(6, cs.size());
+    assertEquals(6, cs.size());
     verifyAclsOfSize(8); // Total of 3 acls per consumer + 2 for the producer
 
     consumers.remove(1);
@@ -118,10 +122,50 @@ public class AccessControlManagerIT {
     accessControlManager.apply(topology, plan);
     plan.run();
 
-    Assert.assertEquals(3, cs.size());
+    assertEquals(3, cs.size());
     verifyAclsOfSize(5);
   }
 
+  @Test
+  public void aclsRemovedTest() throws IOException, ExecutionException, InterruptedException {
+
+    TopologyImpl topology = new TopologyImpl();
+    topology.setContext("aclsRemovedTest-Integration");
+
+    List<Consumer> consumers = new ArrayList<>();
+    consumers.add(new Consumer("User:testAclsRemovalUser1"));
+    consumers.add(new Consumer("User:testAclsRemovalUser2"));
+
+    Project project = new ProjectImpl("project");
+    project.setConsumers(consumers);
+    Topic topicA = new TopicImpl("topicA");
+    project.addTopic(topicA);
+    topology.addProject(project);
+
+    accessControlManager.apply(topology, plan);
+    plan.run();
+    assertEquals(6, cs.size());
+
+    // Reset the execution flow to be only with one consumer. Acls expected should be three.
+
+    BackendController cs = new BackendController();
+    ExecutionPlan plan = ExecutionPlan.init(cs, System.out);
+
+    consumers = new ArrayList<>();
+    consumers.add(new Consumer("User:testAclsRemovalUser1"));
+
+    Project project2 = new ProjectImpl("project");
+    project2.setConsumers(consumers);
+    project2.addTopic(topicA);
+    topology.setProjects(Collections.singletonList(project2));
+
+    accessControlManager.apply(topology, plan);
+    plan.run();
+
+    assertEquals(3, cs.size());
+    verifyAclsOfSize(3);
+  }
+
   @Test
   public void consumerAclsCreation() throws ExecutionException, InterruptedException, IOException {
 
@@ -327,7 +371,7 @@ public class AccessControlManagerIT {
     Collection<AclBinding> acls =
         kafkaAdminClient.describeAcls(AclBindingFilter.ANY).values().get();
 
-    Assert.assertEquals(size, acls.size());
+    assertEquals(size, acls.size());
   }
 
   private void verifyConnectAcls(Connector connector)
@@ -344,7 +388,7 @@ public class AccessControlManagerIT {
 
     Collection<AclBinding> acls = kafkaAdminClient.describeAcls(filter).values().get();
 
-    Assert.assertEquals(5, acls.size());
+    assertEquals(5, acls.size());
 
     entryFilter =
         new AccessControlEntryFilter(
@@ -352,7 +396,7 @@ public class AccessControlManagerIT {
     filter = new AclBindingFilter(resourceFilter, entryFilter);
     acls = kafkaAdminClient.describeAcls(filter).values().get();
 
-    Assert.assertEquals(3, acls.size());
+    assertEquals(3, acls.size());
 
     resourceFilter = new ResourcePatternFilter(ResourceType.GROUP, null, PatternType.ANY);
     entryFilter =
@@ -361,7 +405,7 @@ public class AccessControlManagerIT {
     filter = new AclBindingFilter(resourceFilter, entryFilter);
     acls = kafkaAdminClient.describeAcls(filter).values().get();
 
-    Assert.assertEquals(1, acls.size());
+    assertEquals(1, acls.size());
   }
 
   private void verifySchemaRegistryAcls(Platform platform)
@@ -381,7 +425,7 @@ public class AccessControlManagerIT {
 
       Collection<AclBinding> acls = kafkaAdminClient.describeAcls(filter).values().get();
 
-      Assert.assertEquals(3, acls.size());
+      assertEquals(3, acls.size());
     }
   }
 
@@ -402,7 +446,7 @@ public class AccessControlManagerIT {
 
       Collection<AclBinding> acls = kafkaAdminClient.describeAcls(filter).values().get();
 
-      Assert.assertEquals(16, acls.size());
+      assertEquals(16, acls.size());
     }
   }
 
@@ -418,7 +462,7 @@ public class AccessControlManagerIT {
     Collection<AclBinding> acls = kafkaAdminClient.describeAcls(filter).values().get();
 
     // two acls created for the write topics
-    Assert.assertEquals(2, acls.size());
+    assertEquals(2, acls.size());
 
     entryFilter =
         new AccessControlEntryFilter(
@@ -429,7 +473,7 @@ public class AccessControlManagerIT {
     acls = kafkaAdminClient.describeAcls(filter).values().get();
 
     // two acls created for the read topics
-    Assert.assertEquals(2, acls.size());
+    assertEquals(2, acls.size());
 
     entryFilter =
         new AccessControlEntryFilter(
@@ -440,7 +484,7 @@ public class AccessControlManagerIT {
     acls = kafkaAdminClient.describeAcls(filter).values().get();
 
     // 1 acls created for the prefix internal topics
-    Assert.assertEquals(1, acls.size());
+    assertEquals(1, acls.size());
   }
 
   private void verifyProducerAcls(List<Producer> producers, String topic, int aclsCount)
@@ -455,7 +499,7 @@ public class AccessControlManagerIT {
       AclBindingFilter filter = new AclBindingFilter(resourceFilter, entryFilter);
       Collection<AclBinding> acls = kafkaAdminClient.describeAcls(filter).values().get();
 
-      Assert.assertEquals(aclsCount, acls.size());
+      assertEquals(aclsCount, acls.size());
 
       List<ResourceType> types =
           acls.stream()
@@ -486,7 +530,7 @@ public class AccessControlManagerIT {
       AclBindingFilter filter = new AclBindingFilter(resourceFilter, entryFilter);
       Collection<AclBinding> acls = kafkaAdminClient.describeAcls(filter).values().get();
 
-      Assert.assertEquals(3, acls.size());
+      assertEquals(3, acls.size());
 
       List<ResourceType> types =
           acls.stream()