feat: add exactly-once to kstreams and create transactionalid ACLs

src/main/java/com/purbon/kafka/topology/BindingsBuilderProvider.java

@@ -21,7 +21,11 @@ public interface BindingsBuilderProvider {
   List<TopologyAclBinding> buildBindingsForConnect(Connector connector, String topicPrefix);
 
   List<TopologyAclBinding> buildBindingsForStreamsApp(
-      String principal, String topicPrefix, List<String> readTopics, List<String> writeTopics);
+      String principal,
+      String topicPrefix,
+      List<String> readTopics,
+      List<String> writeTopics,
+      boolean eos);
 
   List<TopologyAclBinding> buildBindingsForConsumers(
       Collection<Consumer> consumers, String resource, boolean prefixed);

src/main/java/com/purbon/kafka/topology/actions/access/builders/BuildBindingsForKStreams.java

@@ -35,7 +35,11 @@ protected void execute() throws IOException {
 
     bindings =
         builderProvider.buildBindingsForStreamsApp(
-            app.getPrincipal(), prefix, readTopics, writeTopics);
+            app.getPrincipal(),
+            prefix,
+            readTopics,
+            writeTopics,
+            app.getExactlyOnce().get());
   }
 
   @Override

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

@@ -5,32 +5,54 @@
 import com.purbon.kafka.topology.model.DynamicUser;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
+import org.apache.zookeeper.Op;
 
 public class KStream extends DynamicUser {
 
   @JsonInclude(Include.NON_EMPTY)
   private Optional<String> applicationId;
 
+  @JsonInclude(Include.NON_EMPTY)
+  private Optional<Boolean> exactlyOnce;
+
   public KStream() {
     this("", new HashMap<>());
   }
 
   public KStream(
-      String principal, HashMap<String, List<String>> topics, Optional<String> applicationId) {
+      String principal,
+      Map<String, List<String>> topics,
+      Optional<String> applicationId,
+      Optional<Boolean> exactlyOnce) {
     super(principal, topics);
     this.applicationId = applicationId;
+    this.exactlyOnce = exactlyOnce;
+  }
+
+  public KStream(
+      String principal, HashMap<String, List<String>> topics, Optional<String> applicationId) {
+    this(principal, topics, applicationId,  Optional.of(Boolean.FALSE));
   }
 
   public KStream(String principal, HashMap<String, List<String>> topics) {
-    this(principal, topics, Optional.empty());
+    this(principal, topics, Optional.empty(), Optional.of(Boolean.FALSE));
   }
 
   public Optional<String> getApplicationId() {
     return applicationId;
   }
 
+  public Optional<Boolean> getExactlyOnce() {
+    return exactlyOnce;
+  }
+
   public void setApplicationId(Optional<String> applicationId) {
     this.applicationId = applicationId;
   }
+
+  public void setExactlyOnce(Optional<Boolean> exactlyOnce) {
+    this.exactlyOnce = exactlyOnce;
+  }
 }

src/main/java/com/purbon/kafka/topology/roles/acls/AclsBindingsBuilder.java

@@ -96,8 +96,13 @@ public List<TopologyAclBinding> buildBindingsForConnect(
 
   @Override
   public List<TopologyAclBinding> buildBindingsForStreamsApp(
-      String principal, String topicPrefix, List<String> readTopics, List<String> writeTopics) {
-    return toList(streamsAppStream(translate(principal), topicPrefix, readTopics, writeTopics));
+      String principal,
+      String topicPrefix,
+      List<String> readTopics,
+      List<String> writeTopics,
+      boolean eos) {
+    return toList(
+        streamsAppStream(translate(principal), topicPrefix, readTopics, writeTopics, eos));
   }
 
   @Override
@@ -229,7 +234,11 @@ private Stream<AclBinding> consumerAclsStream(Consumer consumer, String topic, b
   }
 
   private Stream<AclBinding> streamsAppStream(
-      String principal, String prefix, List<String> readTopics, List<String> writeTopics) {
+      String principal,
+      String prefix,
+      List<String> readTopics,
+      List<String> writeTopics,
+      boolean eos) {
 
     List<AclBinding> acls = new ArrayList<>();
 
@@ -246,6 +255,14 @@ private Stream<AclBinding> streamsAppStream(
 
     acls.add(buildGroupLevelAcl(principal, prefix, PatternType.PREFIXED, AclOperation.READ));
 
+    if (eos) {
+      acls.add(
+          buildTransactionIdLevelAcl(principal, prefix, PatternType.PREFIXED, AclOperation.WRITE));
+      acls.add(
+          buildTransactionIdLevelAcl(
+              principal, prefix, PatternType.PREFIXED, AclOperation.DESCRIBE));
+    }
+
     return acls.stream();
   }
 

src/main/java/com/purbon/kafka/topology/roles/rbac/RBACBindingsBuilder.java

@@ -99,7 +99,11 @@ public List<TopologyAclBinding> buildBindingsForConnect(Connector connector, Str
 
   @Override
   public List<TopologyAclBinding> buildBindingsForStreamsApp(
-      String principal, String topicPrefix, List<String> readTopics, List<String> writeTopics) {
+      String principal,
+      String topicPrefix,
+      List<String> readTopics,
+      List<String> writeTopics,
+      boolean eos) {
     List<TopologyAclBinding> bindings = new ArrayList<>();
 
     TopologyAclBinding binding = apiClient.bind(principal, DEVELOPER_READ, topicPrefix, PREFIX);
@@ -118,6 +122,11 @@ public List<TopologyAclBinding> buildBindingsForStreamsApp(
           bindings.add(writeBinding);
         });
 
+    if (eos) {
+      bindings.add(
+          apiClient.bind(principal, DEVELOPER_WRITE, topicPrefix, "TransactionalId", PREFIX));
+    }
+
     binding = apiClient.bind(principal, RESOURCE_OWNER, topicPrefix, PREFIX);
     bindings.add(binding);
     binding = apiClient.bind(principal, RESOURCE_OWNER, topicPrefix, "Group", PREFIX);
@@ -318,7 +327,7 @@ public Collection<TopologyAclBinding> buildBindingsForKSqlApp(KSqlApp app, Strin
     // schema access
     List<String> subjects =
         readTopics.stream()
-            .flatMap((Function<List<String>, Stream<String>>) topics -> topics.stream())
+            .flatMap((Function<List<String>, Stream<String>>) Collection::stream)
             .map(topicName -> String.format("%s-value", topicName))
             .collect(Collectors.toList());
 

src/main/java/com/purbon/kafka/topology/serdes/TopologyCustomDeserializer.java

@@ -379,28 +379,40 @@ private Optional<PlatformSystem> doKSqlElements(JsonParser parser, JsonNode node
   }
 
   private Optional<PlatformSystem> doStreamsElements(JsonParser parser, JsonNode node)
-          throws IOException {
+      throws IOException {
     List<KStream> streams =
-        new JsonSerdesUtils<KStream>().parseApplicationUser(parser, node, KStream.class)
-            .stream()
-            .map(ks -> {
-              ks.getTopics().putIfAbsent(KStream.READ_TOPICS, Collections.emptyList());
-              ks.getTopics().putIfAbsent(KStream.WRITE_TOPICS, Collections.emptyList());
-              return ks;
-            }).collect(Collectors.toList());
-
-    for(KStream ks : streams) {
+        new JsonSerdesUtils<KStream>()
+            .parseApplicationUser(parser, node, KStream.class).stream()
+                .map(
+                    ks -> {
+                      ks.getTopics().putIfAbsent(KStream.READ_TOPICS, Collections.emptyList());
+                      ks.getTopics().putIfAbsent(KStream.WRITE_TOPICS, Collections.emptyList());
+                      return ks;
+                    })
+                .collect(Collectors.toList());
+
+    for (KStream ks : streams) {
       var topics = ks.getTopics();
       if (topics.get(KStream.READ_TOPICS).isEmpty() || topics.get(KStream.WRITE_TOPICS).isEmpty()) {
-        LOGGER.warn("A Kafka Streams application with Id ("+ks.getApplicationId()+") and Principal ("+ks.getPrincipal()+")" +
-                " might require both read and write topics as per its " +
-                "nature it is always reading and writing into Apache Kafka, be aware if you notice problems.");
+        LOGGER.warn(
+            "A Kafka Streams application with Id ("
+                + ks.getApplicationId()
+                + ") and Principal ("
+                + ks.getPrincipal()
+                + ")"
+                + " might require both read and write topics as per its "
+                + "nature it is always reading and writing into Apache Kafka, be aware if you notice problems.");
       }
       if (topics.get(KStream.READ_TOPICS).isEmpty()) {
-        // should have at minimum read topics defined as we could think of write topics as internal topics being
+        // should have at minimum read topics defined as we could think of write topics as internal
+        // topics being
         // auto created.
-        throw new IOException("Kafka Streams application with Id "+ks.getApplicationId()+" and principal "+ks.getPrincipal()+
-                " have missing read topics. This field is required.");
+        throw new IOException(
+            "Kafka Streams application with Id "
+                + ks.getApplicationId()
+                + " and principal "
+                + ks.getPrincipal()
+                + " have missing read topics. This field is required.");
       }
     }
     return Optional.of(new PlatformSystem(streams));

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

@@ -246,13 +246,15 @@ public void newKafkaStreamsAppACLsCreation() throws IOException {
             "User:App0",
             project.namePrefix(),
             topics.get(KStream.READ_TOPICS),
-            topics.get(KStream.WRITE_TOPICS));
+            topics.get(KStream.WRITE_TOPICS),
+            false);
     verify(aclsBuilder, times(1))
         .buildBindingsForStreamsApp(
             eq("User:App0"),
             eq(project.namePrefix()),
             eq(topics.get(KStream.READ_TOPICS)),
-            eq(topics.get(KStream.WRITE_TOPICS)));
+            eq(topics.get(KStream.WRITE_TOPICS)),
+            eq(false));
   }
 
   @Test

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

@@ -16,6 +16,7 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Optional;
 import java.util.Properties;
 import org.apache.kafka.common.acl.AccessControlEntry;
 import org.apache.kafka.common.acl.AclBinding;
@@ -85,6 +86,39 @@ public void testProducerAclsBuilder() {
             buildTopicLevelAcl("User:foo", "bar", PatternType.LITERAL, AclOperation.DESCRIBE));
   }
 
+  @Test
+  public void testStreamsWithTxIdAclsBuilder() {
+    KStream producer =
+        new KStream(
+            "User:foo",
+            singletonMap("read", singletonList("bar")),
+            Optional.of("app1"),
+            Optional.of(true));
+    List<TopologyAclBinding> aclBindings =
+        builder.buildBindingsForStreamsApp(
+            "User:foo", "app1", singletonList("bar"), emptyList(), true);
+    assertThat(aclBindings.size()).isEqualTo(5);
+
+    assertThat(aclBindings)
+        .contains(buildTopicLevelAcl("User:foo", "bar", PatternType.LITERAL, AclOperation.READ));
+
+    assertThat(aclBindings)
+        .contains(
+            buildTransactionIdLevelAcl(
+                producer.getPrincipal(),
+                producer.getApplicationId().get(),
+                PatternType.PREFIXED,
+                AclOperation.DESCRIBE));
+
+    assertThat(aclBindings)
+        .contains(
+            buildTransactionIdLevelAcl(
+                producer.getPrincipal(),
+                producer.getApplicationId().get(),
+                PatternType.PREFIXED,
+                AclOperation.WRITE));
+  }
+
   @Test
   public void testProducerWithTxIdAclsBuilder() {
     Producer producer = new Producer("User:foo", "1234", true);
@@ -282,7 +316,7 @@ public void testStreamsAclsBuilder() {
 
     List<TopologyAclBinding> bindings =
         builder.buildBindingsForStreamsApp(
-            stream.getPrincipal(), "prefix", readTopics, writeTopics);
+            stream.getPrincipal(), "prefix", readTopics, writeTopics, false);
 
     assertThat(bindings.size()).isEqualTo(4);
     assertThat(bindings)

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

@@ -73,11 +73,11 @@ public void testMetadata() {
   @Test
   public void testStreamsParsingOnlyReadTopicsShouldNotParseAsNull() {
     Topology topology =
-            parser.deserialise(TestUtils.getResourceFile("/descriptor-streams-only-read.yaml"));
+        parser.deserialise(TestUtils.getResourceFile("/descriptor-streams-only-read.yaml"));
 
     Project p = topology.getProjects().get(0);
 
-    for(KStream s : p.getStreams()) {
+    for (KStream s : p.getStreams()) {
       assertThat(s.getTopics().get(KStream.READ_TOPICS)).isNotNull();
       assertThat(s.getTopics().get(KStream.READ_TOPICS)).isInstanceOf(List.class);
       assertThat(s.getTopics().get(KStream.READ_TOPICS)).contains("topicA", "topicB");

src/test/java/com/purbon/kafka/topology/actions/access/builders/BuildBindingsForKStreamsTest.java

@@ -37,12 +37,12 @@ public void testStreamsWithoutApplicationId() throws IOException {
     assertThat(action.getBindings())
         .anyMatch(
             b ->
-                b.getResourceType() == ResourceType.TOPIC.name()
+                b.getResourceType().equals(ResourceType.TOPIC.name())
                     && b.getResourceName().equals(topicPrefix));
     assertThat(action.getBindings())
         .anyMatch(
             b ->
-                b.getResourceType() == ResourceType.GROUP.name()
+                b.getResourceType().equals(ResourceType.GROUP.name())
                     && b.getResourceName().equals(topicPrefix));
   }
 
@@ -59,12 +59,12 @@ public void testStreamWithApplicationId() throws IOException {
     assertThat(action.getBindings())
         .anyMatch(
             b ->
-                b.getResourceType() == ResourceType.TOPIC.name()
+                b.getResourceType().equals(ResourceType.TOPIC.name())
                     && b.getResourceName().equals(applicationId));
     assertThat(action.getBindings())
         .anyMatch(
             b ->
-                b.getResourceType() == ResourceType.GROUP.name()
+                b.getResourceType().equals(ResourceType.GROUP.name())
                     && b.getResourceName().equals(applicationId));
   }
 }