fix:add support in the custom roles for subject and connect attributes

src/main/java/com/purbon/kafka/topology/model/users/Other.java

@@ -12,10 +12,14 @@ public class Other extends User {
   private Optional<Boolean> idempotence;
   private Optional<String> group;
   private Optional<String> topic;
+  private Optional<String> subject;
+  private Optional<String> connector;
 
   public Other() {
     super();
     topic = Optional.empty();
+    subject = Optional.empty();
+    connector = Optional.empty();
     group = Optional.empty();
     transactionId = Optional.empty();
     idempotence = Optional.empty();
@@ -24,6 +28,12 @@ public Other() {
   public Map<String, Object> asMap() {
     Map<String, Object> map = new HashMap<>();
     map.put("topic", topicString());
+    if (subject.isPresent()) {
+      map.put("subject", subjectString());
+    }
+    if (connector.isPresent()) {
+      map.put("connector", connectorString());
+    }
     map.put("group", groupString());
     if (transactionId.isPresent()) {
       map.put("transactionId", transactionId.get());
@@ -55,6 +65,30 @@ public void setTopic(Optional<String> topic) {
     this.topic = topic;
   }
 
+  public Optional<String> getSubject() {
+    return subject;
+  }
+
+  public String subjectString() {
+    return subject.orElse("");
+  }
+
+  public void setSubject(Optional<String> subject) {
+    this.subject = subject;
+  }
+
+  public Optional<String> getConnector() {
+    return connector;
+  }
+
+  public String connectorString() {
+    return connector.orElse("");
+  }
+
+  public void setConnector(Optional<String> connector) {
+    this.connector = connector;
+  }
+
   public Optional<String> getTransactionId() {
     return transactionId;
   }

src/test/java/com/purbon/kafka/topology/JulieRolesTest.java

@@ -5,12 +5,16 @@
 import com.purbon.kafka.topology.model.JulieRole;
 import com.purbon.kafka.topology.model.JulieRoleAcl;
 import com.purbon.kafka.topology.model.JulieRoles;
+import com.purbon.kafka.topology.model.PlanMap;
 import com.purbon.kafka.topology.model.Topology;
+import com.purbon.kafka.topology.model.users.Other;
 import com.purbon.kafka.topology.serdes.JulieRolesSerdes;
 import com.purbon.kafka.topology.serdes.TopologySerdes;
+import com.purbon.kafka.topology.utils.JinjaUtils;
 import com.purbon.kafka.topology.utils.TestUtils;
 import java.io.IOException;
 import java.util.List;
+import java.util.Map;
 import java.util.stream.Collectors;
 import org.junit.After;
 import org.junit.Before;
@@ -30,7 +34,7 @@ public void after() {}
 
   @Test
   public void testSerdes() throws IOException {
-    JulieRoles roles = parser.deserialise(TestUtils.getResourceFile("/roles.yaml"));
+    JulieRoles roles = parser.deserialise(TestUtils.getResourceFile("/roles-rbac.yaml"));
 
     assertThat(roles.getRoles()).hasSize(2);
     for (JulieRole role : roles.getRoles()) {
@@ -40,17 +44,48 @@ public void testSerdes() throws IOException {
     JulieRole role = roles.get("app");
     List<String> resources =
         role.getAcls().stream().map(JulieRoleAcl::getResourceType).collect(Collectors.toList());
-    assertThat(resources).contains("Topic", "Group");
+    assertThat(resources).contains("Topic", "Group", "Subject", "Connector");
+
     assertThat(role.getName()).isEqualTo("app");
-    assertThat(role.getAcls()).hasSize(4);
-    assertThat(role.getAcls().get(0).getRole()).isEqualTo("ALL");
+    assertThat(role.getAcls()).hasSize(9);
+    assertThat(role.getAcls().get(0).getRole()).isEqualTo("ResourceOwner");
 
     role = roles.get("other");
     resources =
         role.getAcls().stream().map(JulieRoleAcl::getResourceType).collect(Collectors.toList());
     assertThat(resources).contains("Topic");
     assertThat(role.getName()).isEqualTo("other");
     assertThat(role.getAcls()).hasSize(2);
+
+    TopologySerdes topologySerdes =
+        new TopologySerdes(new Configuration(), TopologySerdes.FileType.YAML, new PlanMap());
+    Topology topology = topologySerdes.deserialise(TestUtils.getResourceFile("/descriptor.yaml"));
+
+    var project = topology.getProjects().get(0);
+    for (Map.Entry<String, List<Other>> entry : project.getOthers().entrySet()) {
+      if (!entry.getKey().equals("app")) {
+        continue;
+      }
+      role = roles.get(entry.getKey());
+      var other = entry.getValue().get(0);
+      var acls =
+          role.getAcls().stream()
+              .map(
+                  acl -> {
+                    String resourceName =
+                        JinjaUtils.serialise(acl.getResourceName(), other.asMap());
+                    return new JulieRoleAcl(
+                        acl.getResourceType(),
+                        resourceName,
+                        acl.getPatternType(),
+                        acl.getHost(),
+                        acl.getOperation(),
+                        acl.getPermissionType());
+                  })
+              .collect(Collectors.toList());
+      var names = acls.stream().map(JulieRoleAcl::getResourceName).collect(Collectors.toList());
+      assertThat(names).contains("test.subject", "con");
+    }
   }
 
   @Test(expected = IOException.class)

src/test/java/com/purbon/kafka/topology/TestTopologyBuilder.java

@@ -91,9 +91,16 @@ public TestTopologyBuilder addConsumer(String user, String group) {
   }
 
   public TestTopologyBuilder addOther(String roleName, String principal, String topic) {
+    return addOther(roleName, principal, topic, "", "");
+  }
+
+  public TestTopologyBuilder addOther(
+      String roleName, String principal, String topic, String subject, String connector) {
     Other other = new Other();
     other.setPrincipal(principal);
     other.setTopic(Optional.of(topic));
+    if (!connector.isEmpty()) other.setConnector(Optional.of(connector));
+    if (!subject.isEmpty()) other.setSubject(Optional.of(subject));
     others.add(Map.entry(roleName, Collections.singletonList(other)));
     return this;
   }

src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java

@@ -288,7 +288,7 @@ public void testOtherSerdes() {
 
     var others = project.getOthers();
     assertThat(others).hasSize(2);
-    var foos = others.get("foo");
+    var foos = others.get("app");
     assertThat(foos.get(0).getPrincipal()).isEqualTo("User:banana");
     assertThat(foos.get(0).groupString()).isEqualTo("foo");
     var bars = others.get("bar");

src/test/java/com/purbon/kafka/topology/integration/RBACPRoviderRbacIT.java

@@ -424,7 +424,9 @@ public void testJulieRoleAclCreation() throws IOException {
     String principal = "User:app" + System.currentTimeMillis();
 
     Topology topology =
-        TestTopologyBuilder.createProject().addOther("app", principal, "foo").buildTopology();
+        TestTopologyBuilder.createProject()
+            .addOther("app", principal, "foo", "subj", "con")
+            .buildTopology();
 
     Map<String, String> cliOps = new HashMap<>();
     cliOps.put(BROKERS_OPTION, "");

src/test/resources/descriptor.yaml

@@ -3,9 +3,12 @@ context: "contextOrg"
 source: "source"
 projects:
   - name: "foo"
-    foo:
+    app:
       - principal: "User:banana"
         group: "foo"
+        subject: "test.subject"
+        connector: "con"
+        topic: "top"
     bar:
       - principal: "User:banana"
         group: "bar"

src/test/resources/roles-goodTest.yaml

@@ -1,13 +1,23 @@
 ---
 roles:
-  - name: "foo"
+  - name: "app"
     acls:
       - resourceType: "Topic"
         resourceName: "{{topic}}"
         patternType: "PREFIXED"
         host: "*"
         operation: "ALL"
         permissionType: "ALLOW"
+      - resourceType: "Subject"
+        resourceName: "{{subject}}"
+        patternType: "PREFIXED"
+        host: "*"
+        role: "ResourceOwner"
+      - resourceType: "Connector"
+        resourceName: "{{connector}}"
+        patternType: "PREFIXED"
+        host: "*"
+        role: "ResourceOwner"
       - resourceType: "Topic"
         resourceName: "sourceTopic"
         patternType: "LITERAL"

src/test/resources/roles-rbac.yaml

@@ -7,6 +7,16 @@ roles:
         patternType: "PREFIXED"
         host: "*"
         role: "ResourceOwner"
+      - resourceType: "Subject"
+        resourceName: "{{subject}}"
+        patternType: "PREFIXED"
+        host: "*"
+        role: "ResourceOwner"
+      - resourceType: "Connector"
+        resourceName: "{{connector}}"
+        patternType: "PREFIXED"
+        host: "*"
+        role: "ResourceOwner"
       - resourceType: "Topic"
         resourceName: "sourceTopic"
         patternType: "LITERAL"
@@ -36,4 +46,18 @@ roles:
         resourceName: "KsqlCluster:ksql-cluster"
         patternType: "LITERAL"
         host: "*"
-        role: "ResourceOwner"
+        role: "ResourceOwner"
+  - name: "other"
+    acls:
+      - resourceType: "Topic"
+        resourceName: "{{topic}}"
+        patternType: "LITERAL"
+        host: "*"
+        operation: "READ"
+        permissionType: "ALLOW"
+      - resourceType: "Group"
+        resourceName: ""
+        patternType: ""
+        host: ""
+        operation: ""
+        permissionType: ""