Repository: kafka-ops/kafka-topology-builder
Pull request #210 (2021-02-18 13:25:44)

File backend fix
* **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)
- [ ] Docs have been added / updated (for bug fixes / features)
- [x ] 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, ...)

Bug fix

* **What is the current behavior?** (You can also link to an open issue here)

https://github.com/kafka-ops/kafka-topology-builder/issues/209

* **What is the new behavior (if this is a feature change)?**



* **Does this PR introduce a breaking change?** (What changes might users need to make in their application due to this PR?)

No.

* **Other information**:

Probably not a good idea to use spaces for separators anyway, but I kept them in order not to introduce incompatibilities. The fix allows space characters in the principal column, but not in the other columns.


akselh: It seems integration test pipelines are broken on master now?

A lot of PRs have broken integration tests with this in the log:
```
...
com.github.dockerjava.api.exception.NotFoundException: Status 404: {"message":"No such image: testcontainers/ryuk:0.3.0"}
```
sverrehu: I noticed the failing integration tests too. Had a plan to look into that later.

sverrehu: Integration tests work again, after upgrading testcontainers. Ref https://github.com/testcontainers/testcontainers-java/issues/3574
akselh: > Integration tests work again, after upgrading testcontainers. Ref [testcontainers/testcontainers-java#3574](https://github.com/testcontainers/testcontainers-java/issues/3574)

Nice!
purbon: thanks gents! I will make sure to get this in soon.
sverrehu: Wish you all a nice weekend.

sverrehu: I have noticed that there are pull requests with reference to rules, and since it is Friday night and I had some beers, I got to think about the "rules" set out in my 20 year old book. In particular, this one: "Rule 8: Always handle metacharacters when passing data to subsystems". In FileBackend, space has been made a metacharacter. So if valid data contains a space, it needs to be treated in some way. I don't think making a greedy sub-regexp by hard-anchoring at the end is the correct way to handle metacharacters, but at least it solves our current problem. Cheers!
sverrehu: Just found that the RedisBackend uses the same regexp, and suffers from the same issue. Applied the same semi-fix there.


Patch:
diff --git a/pom.xml b/pom.xml
index 299cf1d..31d8d7c 100644
--- a/pom.xml
+++ b/pom.xml
@@ -430,7 +430,7 @@
     <commons.version>1.4</commons.version>
     <mockito.version>3.6.0</mockito.version>
     <junit.version>4.13.1</junit.version>
-    <testcontainers.version>1.15.0</testcontainers.version>
+    <testcontainers.version>1.15.2</testcontainers.version>
     <jedis.version>3.2.0</jedis.version>
     <confluent.version>5.5.0</confluent.version>
     <confluent-ce.version>5.5.1-ce</confluent-ce.version>
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 08d386c..4c59038 100644
--- a/src/main/java/com/purbon/kafka/topology/backend/FileBackend.java
+++ b/src/main/java/com/purbon/kafka/topology/backend/FileBackend.java
@@ -34,7 +34,7 @@ public class FileBackend implements Backend {
 
   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+)\\'\"?$";
   private Pattern regexp;
 
   public FileBackend() {
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 65eed88..7e68679 100644
--- a/src/main/java/com/purbon/kafka/topology/backend/RedisBackend.java
+++ b/src/main/java/com/purbon/kafka/topology/backend/RedisBackend.java
@@ -20,7 +20,7 @@ public class RedisBackend implements Backend {
   static final String KAFKA_TOPOLOGY_BUILDER_TYPE = "kafka.topology.builder.type";
 
   private String expression =
-      "^\\'(\\S+)\\',\\s*\\'(\\S+)\\',\\s*\\'(\\S+)\\',\\s*\\'(\\S+)\\',\\s*\\'(\\S+)\\',\\s*\\'(\\S+)\\'";
+      "^\\'(\\S+)\\',\\s*\\'(\\S+)\\',\\s*\\'(\\S+)\\',\\s*\\'(\\S+)\\',\\s*\\'(.+)\\',\\s*\\'(\\S+)\\'$";
   private Pattern regexp;
   private Jedis jedis;
 


Test patch:
diff --git a/src/test/java/com/purbon/kafka/topology/backend/FileBackendTest.java b/src/test/java/com/purbon/kafka/topology/backend/FileBackendTest.java
index 700a046..2e18b63 100644
--- a/src/test/java/com/purbon/kafka/topology/backend/FileBackendTest.java
+++ b/src/test/java/com/purbon/kafka/topology/backend/FileBackendTest.java
@@ -4,16 +4,11 @@ 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.TestTopologyBuilder;
 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;
@@ -43,25 +38,36 @@ public class FileBackendTest {
 
   @Test
   public void testStoreAndLoadBindingsAndTopics() throws IOException {
+    verifyStoreAndLoadWithPrincipal("principal");
+  }
+
+  @Test
+  public void shouldHandlePrincipalWithSpace() throws IOException {
+    verifyStoreAndLoadWithPrincipal("User:C=NO,CN=John Doe,emailAddress=john.doe@example.com");
+  }
+
+  @Test
+  public void shouldHandlePrincipalWithUri() throws IOException {
+    verifyStoreAndLoadWithPrincipal("SPIFFE:spiffe://example.com/foo/bar");
+  }
+
+  private void verifyStoreAndLoadWithPrincipal(final String principal) throws IOException {
     TopologyAclBinding binding =
         TopologyAclBinding.build(
-            ResourceType.CLUSTER.name(), "Topic", "host", "op", "principal", "LITERAL");
+            ResourceType.CLUSTER.name(), "Topic", "host", "op", principal, "LITERAL");
 
-    Topology topology = new TopologyImpl();
-    topology.setContext("context");
-    Project project = new ProjectImpl("project");
-    topology.setProjects(singletonList(project));
+    TestTopologyBuilder builder =
+        TestTopologyBuilder.createProject().addTopic("foo").addTopic("bar");
 
-    Topic topic = new TopicImpl("foo");
-    Topic topicBar = new TopicImpl("bar");
-    project.setTopics(Arrays.asList(topic, topicBar));
+    Topic fooTopic = builder.getTopic("foo");
+    Topic barTopic = builder.getTopic("bar");
 
     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.saveTopics(new HashSet<>(Arrays.asList(fooTopic.toString(), barTopic.toString())));
     backend.close();
 
     backend = new FileBackend();
@@ -73,7 +79,7 @@ public class FileBackendTest {
     assertThat(bindings).hasSize(1);
     assertThat(bindings).contains(binding);
     assertThat(topics).hasSize(2);
-    assertThat(topics).contains(topic.toString());
-    assertThat(topics).contains(topicBar.toString());
+    assertThat(topics).contains(fooTopic.toString());
+    assertThat(topics).contains(barTopic.toString());
   }
 }
diff --git a/src/test/java/com/purbon/kafka/topology/backend/RedisBackendTest.java b/src/test/java/com/purbon/kafka/topology/backend/RedisBackendTest.java
index b006733..3df7829 100644
--- a/src/test/java/com/purbon/kafka/topology/backend/RedisBackendTest.java
+++ b/src/test/java/com/purbon/kafka/topology/backend/RedisBackendTest.java
@@ -56,7 +56,7 @@ public class RedisBackendTest {
     when(jedis.get(eq(KAFKA_TOPOLOGY_BUILDER_TYPE))).thenReturn("acls");
     when(jedis.scard(eq(KAFKA_TOPOLOGY_BUILDER_BINDINGS))).thenReturn(10l);
     when(jedis.spop(eq(KAFKA_TOPOLOGY_BUILDER_BINDINGS)))
-        .thenReturn("'TOPIC', 'topicA', '*', 'READ', 'User:Connect1', 'LITERAL'")
+        .thenReturn("'TOPIC', 'topicA', '*', 'READ', 'User:C=NO,CN=John Doe,emailAddress=john.doe@example.com', 'LITERAL'")
         .thenReturn("'TOPIC', 'topicB', '*', 'READ', 'User:Connect1', 'LITERAL'");
 
     Set<TopologyAclBinding> bindings = stateProcessor.loadBindings();