Repository: kafka-ops/kafka-topology-builder (java)

Issue:
Support TransactionID and Idempotence Consumer/Producers in the Topology 
As a user of KTB i would love to setup as well consumers and producers that require transactions and idempotence producing. For this, we need.

* For a producer the DESCRIBE and WRITE operations are given on the topic. The following could be needed as well:
TransactionalId: WRITE, DESCRIBE
Cluster: IDEMPOTENT_WRITE

* For a consumer the READ and DESCRIBE operations are supported on the topic and READ on given consumer group (or default all groups). Not exactly sure at the moment what more we need, but these are possible candidates:
TransactionalId: DESCRIBE
Group: DESCRIBE


Pull request description:
Add support for Transactional and Idempotence producing and consuming in KTB
This PR add ACLs to support Transactional and idempotence producing in KTB. The Consuming part does not need any new ACL as all is setup as expected and required.

close #98  

Patch:
diff --git a/src/main/java/com/purbon/kafka/topology/BindingsBuilderProvider.java b/src/main/java/com/purbon/kafka/topology/BindingsBuilderProvider.java
index 6d4f0a5..97638a5 100644
--- a/src/main/java/com/purbon/kafka/topology/BindingsBuilderProvider.java
+++ b/src/main/java/com/purbon/kafka/topology/BindingsBuilderProvider.java
@@ -4,6 +4,7 @@ import com.purbon.kafka.topology.exceptions.ConfigurationException;
 import com.purbon.kafka.topology.model.Component;
 import com.purbon.kafka.topology.model.users.Connector;
 import com.purbon.kafka.topology.model.users.Consumer;
+import com.purbon.kafka.topology.model.users.Producer;
 import com.purbon.kafka.topology.model.users.platform.SchemaRegistryInstance;
 import com.purbon.kafka.topology.roles.TopologyAclBinding;
 import java.io.IOException;
@@ -20,7 +21,7 @@ public interface BindingsBuilderProvider {
 
   List<TopologyAclBinding> buildBindingsForConsumers(Collection<Consumer> consumers, String topic);
 
-  List<TopologyAclBinding> buildBindingsForProducers(Collection<String> principals, String topic);
+  List<TopologyAclBinding> buildBindingsForProducers(Collection<Producer> producers, String topic);
 
   default TopologyAclBinding setPredefinedRole(
       String principal, String predefinedRole, String topicPrefix) {
diff --git a/src/main/java/com/purbon/kafka/topology/actions/access/builders/BuildBindingsForProducer.java b/src/main/java/com/purbon/kafka/topology/actions/access/builders/BuildBindingsForProducer.java
index d6791ed..8e9b78f 100644
--- a/src/main/java/com/purbon/kafka/topology/actions/access/builders/BuildBindingsForProducer.java
+++ b/src/main/java/com/purbon/kafka/topology/actions/access/builders/BuildBindingsForProducer.java
@@ -8,7 +8,6 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
 public class BuildBindingsForProducer extends BaseAccessControlAction {
 
@@ -26,10 +25,7 @@ public class BuildBindingsForProducer extends BaseAccessControlAction {
 
   @Override
   protected void execute() throws IOException {
-    Stream<String> producersStream = producers.stream().map(p -> p.getPrincipal());
-    bindings =
-        builderProvider.buildBindingsForProducers(
-            producersStream.collect(Collectors.toList()), fullTopicName);
+    bindings = builderProvider.buildBindingsForProducers(producers, fullTopicName);
   }
 
   @Override
diff --git a/src/main/java/com/purbon/kafka/topology/model/users/Producer.java b/src/main/java/com/purbon/kafka/topology/model/users/Producer.java
index 844be9d..9479985 100644
--- a/src/main/java/com/purbon/kafka/topology/model/users/Producer.java
+++ b/src/main/java/com/purbon/kafka/topology/model/users/Producer.java
@@ -1,14 +1,38 @@
 package com.purbon.kafka.topology.model.users;
 
 import com.purbon.kafka.topology.model.User;
+import java.util.Optional;
 
 public class Producer extends User {
 
+  Optional<String> transactionId;
+  Optional<Boolean> idempotence;
+
   public Producer() {
     super();
+    transactionId = Optional.empty();
+    idempotence = Optional.empty();
   }
 
   public Producer(String principal) {
     super(principal);
+    transactionId = Optional.empty();
+    idempotence = Optional.empty();
+  }
+
+  public Optional<String> getTransactionId() {
+    return transactionId;
+  }
+
+  public void setTransactionId(Optional<String> transactionId) {
+    this.transactionId = transactionId;
+  }
+
+  public Optional<Boolean> getIdempotence() {
+    return idempotence;
+  }
+
+  public void setIdempotence(Optional<Boolean> idempotence) {
+    this.idempotence = idempotence;
   }
 }
diff --git a/src/main/java/com/purbon/kafka/topology/roles/acls/AclsBindingsBuilder.java b/src/main/java/com/purbon/kafka/topology/roles/acls/AclsBindingsBuilder.java
index 820fd2b..ed8b2e2 100644
--- a/src/main/java/com/purbon/kafka/topology/roles/acls/AclsBindingsBuilder.java
+++ b/src/main/java/com/purbon/kafka/topology/roles/acls/AclsBindingsBuilder.java
@@ -7,9 +7,11 @@ import com.purbon.kafka.topology.TopologyBuilderConfig;
 import com.purbon.kafka.topology.api.adminclient.AclBuilder;
 import com.purbon.kafka.topology.model.users.Connector;
 import com.purbon.kafka.topology.model.users.Consumer;
+import com.purbon.kafka.topology.model.users.Producer;
 import com.purbon.kafka.topology.model.users.platform.SchemaRegistryInstance;
 import com.purbon.kafka.topology.roles.TopologyAclBinding;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
 import java.util.stream.Collectors;
@@ -92,8 +94,8 @@ public class AclsBindingsBuilder implements BindingsBuilderProvider {
 
   @Override
   public List<TopologyAclBinding> buildBindingsForProducers(
-      Collection<String> principals, String topic) {
-    return toList(principals.stream().flatMap(principal -> producerAclsStream(principal, topic)));
+      Collection<Producer> producers, String topic) {
+    return toList(producers.stream().flatMap(p -> producerAclsStream(p, topic)));
   }
 
   @Override
@@ -111,10 +113,45 @@ public class AclsBindingsBuilder implements BindingsBuilderProvider {
     return bindingStream.map(TopologyAclBinding::new).collect(Collectors.toList());
   }
 
-  private Stream<AclBinding> producerAclsStream(String principal, String topic) {
-    return Stream.of(
-        buildTopicLevelAcl(principal, topic, PatternType.LITERAL, AclOperation.DESCRIBE),
-        buildTopicLevelAcl(principal, topic, PatternType.LITERAL, AclOperation.WRITE));
+  private Stream<AclBinding> producerAclsStream(Producer producer, String topic) {
+
+    List<AclBinding> bindings = new ArrayList<>();
+
+    bindings.addAll(
+        Arrays.asList(
+            buildTopicLevelAcl(
+                producer.getPrincipal(), topic, PatternType.LITERAL, AclOperation.DESCRIBE),
+            buildTopicLevelAcl(
+                producer.getPrincipal(), topic, PatternType.LITERAL, AclOperation.WRITE)));
+
+    producer
+        .getTransactionId()
+        .ifPresent(
+            transactionId -> {
+              bindings.add(
+                  buildTransactionIdLevelAcl(
+                      producer.getPrincipal(),
+                      transactionId,
+                      PatternType.LITERAL,
+                      AclOperation.DESCRIBE));
+              bindings.add(
+                  buildTransactionIdLevelAcl(
+                      producer.getPrincipal(),
+                      transactionId,
+                      PatternType.LITERAL,
+                      AclOperation.WRITE));
+            });
+
+    if (producer.getTransactionId().isPresent() || producer.getIdempotence().isPresent()) {
+      ResourcePattern resourcePattern =
+          new ResourcePattern(ResourceType.CLUSTER, "kafka-cluster", PatternType.LITERAL);
+      AccessControlEntry entry =
+          new AccessControlEntry(
+              producer.getPrincipal(), "*", AclOperation.IDEMPOTENT_WRITE, AclPermissionType.ALLOW);
+      bindings.add(new AclBinding(resourcePattern, entry));
+    }
+
+    return bindings.stream();
   }
 
   private Stream<AclBinding> consumerAclsStream(Consumer consumer, String topic) {
@@ -211,6 +248,14 @@ public class AclsBindingsBuilder implements BindingsBuilderProvider {
         .build();
   }
 
+  private AclBinding buildTransactionIdLevelAcl(
+      String principal, String transactionId, PatternType patternType, AclOperation op) {
+    return new AclBuilder(principal)
+        .addResource(ResourceType.TRANSACTIONAL_ID, transactionId, patternType)
+        .addControlEntry("*", op, AclPermissionType.ALLOW)
+        .build();
+  }
+
   private AclBinding buildGroupLevelAcl(
       String principal, String group, PatternType patternType, AclOperation op) {
     return new AclBuilder(principal)
diff --git a/src/main/java/com/purbon/kafka/topology/roles/rbac/RBACBindingsBuilder.java b/src/main/java/com/purbon/kafka/topology/roles/rbac/RBACBindingsBuilder.java
index d7b600b..9c2fb5f 100644
--- a/src/main/java/com/purbon/kafka/topology/roles/rbac/RBACBindingsBuilder.java
+++ b/src/main/java/com/purbon/kafka/topology/roles/rbac/RBACBindingsBuilder.java
@@ -11,6 +11,7 @@ import com.purbon.kafka.topology.api.mds.MDSApiClient;
 import com.purbon.kafka.topology.model.Component;
 import com.purbon.kafka.topology.model.users.Connector;
 import com.purbon.kafka.topology.model.users.Consumer;
+import com.purbon.kafka.topology.model.users.Producer;
 import com.purbon.kafka.topology.model.users.platform.SchemaRegistryInstance;
 import com.purbon.kafka.topology.roles.TopologyAclBinding;
 import java.io.IOException;
@@ -140,11 +141,12 @@ public class RBACBindingsBuilder implements BindingsBuilderProvider {
 
   @Override
   public List<TopologyAclBinding> buildBindingsForProducers(
-      Collection<String> principals, String topic) {
+      Collection<Producer> producers, String topic) {
     List<TopologyAclBinding> bindings = new ArrayList<>();
-    principals.forEach(
-        principal -> {
-          TopologyAclBinding binding = apiClient.bind(principal, DEVELOPER_WRITE, topic, LITERAL);
+    producers.forEach(
+        producer -> {
+          TopologyAclBinding binding =
+              apiClient.bind(producer.getPrincipal(), DEVELOPER_WRITE, topic, LITERAL);
           bindings.add(binding);
         });
     return bindings;
diff --git a/src/main/java/com/purbon/kafka/topology/serdes/TopologyCustomDeserializer.java b/src/main/java/com/purbon/kafka/topology/serdes/TopologyCustomDeserializer.java
index c328c88..e80c4e5 100644
--- a/src/main/java/com/purbon/kafka/topology/serdes/TopologyCustomDeserializer.java
+++ b/src/main/java/com/purbon/kafka/topology/serdes/TopologyCustomDeserializer.java
@@ -151,7 +151,7 @@ public class TopologyCustomDeserializer extends StdDeserializer<Topology> {
 
     List<String> keys =
         Arrays.asList(
-            CONSUMERS_KEY, PRODUCERS_KEY, PROJECTS_KEY, CONNECTORS_KEY, STREAMS_KEY, SCHEMAS_KEY);
+            CONSUMERS_KEY, PROJECTS_KEY, PRODUCERS_KEY, CONNECTORS_KEY, STREAMS_KEY, SCHEMAS_KEY);
 
     Map<String, JsonNode> rootNodes = Maps.asMap(new HashSet<>(keys), (key) -> rootNode.get(key));
 


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 f8d0ad9..c2523fe 100644
--- a/src/test/java/com/purbon/kafka/topology/AccessControlManagerTest.java
+++ b/src/test/java/com/purbon/kafka/topology/AccessControlManagerTest.java
@@ -80,7 +80,7 @@ public class AccessControlManagerTest {
   }
 
   @Test
-  public void newConsumerACLsCreation() throws IOException {
+  public void newConsumerACLsCreation() {
 
     List<Consumer> consumers = new ArrayList<>();
     consumers.add(new Consumer("User:app1"));
@@ -103,7 +103,7 @@ public class AccessControlManagerTest {
   }
 
   @Test
-  public void newProducerACLsCreation() throws IOException {
+  public void newProducerACLsCreation() {
 
     List<Producer> producers = new ArrayList<>();
     producers.add(new Producer("User:app1"));
@@ -116,13 +116,11 @@ public class AccessControlManagerTest {
     Topology topology = new TopologyImpl();
     topology.addProject(project);
 
-    List<String> users = asList(new String[] {"User:app1"});
-
     doReturn(new ArrayList<TopologyAclBinding>())
         .when(aclsBuilder)
-        .buildBindingsForProducers(users, topicA.toString());
+        .buildBindingsForProducers(producers, topicA.toString());
     accessControlManager.apply(topology, plan);
-    verify(aclsBuilder, times(1)).buildBindingsForProducers(eq(users), eq(topicA.toString()));
+    verify(aclsBuilder, times(1)).buildBindingsForProducers(eq(producers), eq(topicA.toString()));
   }
 
   @Test
diff --git a/src/test/java/com/purbon/kafka/topology/AclsBindingsBuilderTest.java b/src/test/java/com/purbon/kafka/topology/AclsBindingsBuilderTest.java
new file mode 100644
index 0000000..cd22517
--- /dev/null
+++ b/src/test/java/com/purbon/kafka/topology/AclsBindingsBuilderTest.java
@@ -0,0 +1,235 @@
+package com.purbon.kafka.topology;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.purbon.kafka.topology.api.adminclient.AclBuilder;
+import com.purbon.kafka.topology.model.users.Connector;
+import com.purbon.kafka.topology.model.users.Consumer;
+import com.purbon.kafka.topology.model.users.KStream;
+import com.purbon.kafka.topology.model.users.Producer;
+import com.purbon.kafka.topology.roles.TopologyAclBinding;
+import com.purbon.kafka.topology.roles.acls.AclsBindingsBuilder;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Optional;
+import org.apache.kafka.common.acl.AccessControlEntry;
+import org.apache.kafka.common.acl.AclBinding;
+import org.apache.kafka.common.acl.AclOperation;
+import org.apache.kafka.common.acl.AclPermissionType;
+import org.apache.kafka.common.resource.PatternType;
+import org.apache.kafka.common.resource.ResourcePattern;
+import org.apache.kafka.common.resource.ResourceType;
+import org.junit.Before;
+import org.junit.Test;
+
+public class AclsBindingsBuilderTest {
+
+  TopologyBuilderConfig config;
+  AclsBindingsBuilder builder;
+
+  @Before
+  public void before() {
+    config = new TopologyBuilderConfig();
+    builder = new AclsBindingsBuilder(config);
+  }
+
+  @Test
+  public void testConsumerAclsBuilder() {
+
+    Consumer consumer = new Consumer("User:foo");
+
+    List<TopologyAclBinding> aclBindings =
+        builder.buildBindingsForConsumers(Collections.singleton(consumer), "bar");
+    assertThat(aclBindings.size()).isEqualTo(3);
+    assertThat(aclBindings)
+        .contains(buildTopicLevelAcl("User:foo", "bar", PatternType.LITERAL, AclOperation.READ));
+    assertThat(aclBindings)
+        .contains(
+            buildTopicLevelAcl("User:foo", "bar", PatternType.LITERAL, AclOperation.DESCRIBE));
+    assertThat(aclBindings)
+        .contains(buildGroupLevelAcl("User:foo", "*", PatternType.LITERAL, AclOperation.READ));
+  }
+
+  @Test
+  public void testProducerAclsBuilder() {
+    Producer producer = new Producer("User:foo");
+    List<TopologyAclBinding> aclBindings =
+        builder.buildBindingsForProducers(Collections.singleton(producer), "bar");
+    assertThat(aclBindings.size()).isEqualTo(2);
+    assertThat(aclBindings)
+        .contains(buildTopicLevelAcl("User:foo", "bar", PatternType.LITERAL, AclOperation.WRITE));
+    assertThat(aclBindings)
+        .contains(
+            buildTopicLevelAcl("User:foo", "bar", PatternType.LITERAL, AclOperation.DESCRIBE));
+  }
+
+  @Test
+  public void testProducerWithTxIdAclsBuilder() {
+    Producer producer = new Producer("User:foo");
+    producer.setTransactionId(Optional.of("1234"));
+    List<TopologyAclBinding> aclBindings =
+        builder.buildBindingsForProducers(Collections.singleton(producer), "bar");
+    assertThat(aclBindings.size()).isEqualTo(5);
+
+    assertThat(aclBindings)
+        .contains(buildTopicLevelAcl("User:foo", "bar", PatternType.LITERAL, AclOperation.WRITE));
+    assertThat(aclBindings)
+        .contains(
+            buildTopicLevelAcl("User:foo", "bar", PatternType.LITERAL, AclOperation.DESCRIBE));
+
+    assertThat(aclBindings)
+        .contains(
+            buildTransactionIdLevelAcl(
+                producer.getPrincipal(),
+                producer.getTransactionId().get(),
+                PatternType.LITERAL,
+                AclOperation.DESCRIBE));
+
+    assertThat(aclBindings)
+        .contains(
+            buildTransactionIdLevelAcl(
+                producer.getPrincipal(),
+                producer.getTransactionId().get(),
+                PatternType.LITERAL,
+                AclOperation.WRITE));
+
+    assertThat(aclBindings)
+        .contains(buildClusterLevelAcl(producer.getPrincipal(), AclOperation.IDEMPOTENT_WRITE));
+  }
+
+  @Test
+  public void testIdempotenceProducerAclsBuilder() {
+    Producer producer = new Producer("User:foo");
+    producer.setIdempotence(Optional.of(true));
+    List<TopologyAclBinding> aclBindings =
+        builder.buildBindingsForProducers(Collections.singleton(producer), "bar");
+    assertThat(aclBindings.size()).isEqualTo(3);
+
+    assertThat(aclBindings)
+        .contains(buildTopicLevelAcl("User:foo", "bar", PatternType.LITERAL, AclOperation.WRITE));
+    assertThat(aclBindings)
+        .contains(
+            buildTopicLevelAcl("User:foo", "bar", PatternType.LITERAL, AclOperation.DESCRIBE));
+
+    assertThat(aclBindings)
+        .contains(buildClusterLevelAcl(producer.getPrincipal(), AclOperation.IDEMPOTENT_WRITE));
+  }
+
+  @Test
+  public void testConnectorAclsBuilder() {
+    Connector connector = new Connector("User:foo");
+
+    List<TopologyAclBinding> bindings = builder.buildBindingsForConnect(connector, "prefix");
+    assertThat(bindings.size()).isEqualTo(8);
+
+    List<TopologyAclBinding> shouldContainBindings = new ArrayList<>();
+    shouldContainBindings.add(
+        buildTopicLevelAcl(
+            connector.getPrincipal(),
+            connector.statusTopicString(),
+            PatternType.LITERAL,
+            AclOperation.READ));
+    shouldContainBindings.add(
+        buildTopicLevelAcl(
+            connector.getPrincipal(),
+            connector.offsetTopicString(),
+            PatternType.LITERAL,
+            AclOperation.READ));
+    shouldContainBindings.add(
+        buildTopicLevelAcl(
+            connector.getPrincipal(),
+            connector.configsTopicString(),
+            PatternType.LITERAL,
+            AclOperation.READ));
+    shouldContainBindings.add(
+        buildTopicLevelAcl(
+            connector.getPrincipal(),
+            connector.statusTopicString(),
+            PatternType.LITERAL,
+            AclOperation.WRITE));
+    shouldContainBindings.add(
+        buildTopicLevelAcl(
+            connector.getPrincipal(),
+            connector.offsetTopicString(),
+            PatternType.LITERAL,
+            AclOperation.WRITE));
+    shouldContainBindings.add(
+        buildTopicLevelAcl(
+            connector.getPrincipal(),
+            connector.configsTopicString(),
+            PatternType.LITERAL,
+            AclOperation.WRITE));
+    shouldContainBindings.add(
+        buildGroupLevelAcl(
+            connector.getPrincipal(),
+            connector.groupString(),
+            PatternType.LITERAL,
+            AclOperation.READ));
+    shouldContainBindings.add(buildClusterLevelAcl(connector.getPrincipal(), AclOperation.CREATE));
+
+    assertThat(shouldContainBindings.size()).isEqualTo(8);
+    assertThat(bindings).containsAll(shouldContainBindings);
+  }
+
+  @Test
+  public void testStreamsAclsBuilder() {
+    KStream stream = new KStream("User:foo", new HashMap<>());
+
+    List<String> readTopics = Collections.singletonList("foo");
+    List<String> writeTopics = Collections.singletonList("bar");
+
+    List<TopologyAclBinding> bindings =
+        builder.buildBindingsForStreamsApp(
+            stream.getPrincipal(), "prefix", readTopics, writeTopics);
+
+    assertThat(bindings.size()).isEqualTo(3);
+    assertThat(bindings)
+        .contains(
+            buildTopicLevelAcl(
+                stream.getPrincipal(), "foo", PatternType.LITERAL, AclOperation.READ));
+    assertThat(bindings)
+        .contains(
+            buildTopicLevelAcl(
+                stream.getPrincipal(), "bar", PatternType.LITERAL, AclOperation.WRITE));
+    assertThat(bindings)
+        .contains(
+            buildTopicLevelAcl(
+                stream.getPrincipal(), "prefix", PatternType.PREFIXED, AclOperation.ALL));
+  }
+
+  private TopologyAclBinding buildTopicLevelAcl(
+      String principal, String topic, PatternType patternType, AclOperation op) {
+    return new TopologyAclBinding(
+        new AclBuilder(principal)
+            .addResource(ResourceType.TOPIC, topic, patternType)
+            .addControlEntry("*", op, AclPermissionType.ALLOW)
+            .build());
+  }
+
+  private TopologyAclBinding buildTransactionIdLevelAcl(
+      String principal, String transactionId, PatternType patternType, AclOperation op) {
+    return new TopologyAclBinding(
+        new AclBuilder(principal)
+            .addResource(ResourceType.TRANSACTIONAL_ID, transactionId, patternType)
+            .addControlEntry("*", op, AclPermissionType.ALLOW)
+            .build());
+  }
+
+  private TopologyAclBinding buildClusterLevelAcl(String principal, AclOperation op) {
+    ResourcePattern resourcePattern =
+        new ResourcePattern(ResourceType.CLUSTER, "kafka-cluster", PatternType.LITERAL);
+    AccessControlEntry entry = new AccessControlEntry(principal, "*", op, AclPermissionType.ALLOW);
+    return new TopologyAclBinding(new AclBinding(resourcePattern, entry));
+  }
+
+  private TopologyAclBinding buildGroupLevelAcl(
+      String principal, String group, PatternType patternType, AclOperation op) {
+    return new TopologyAclBinding(
+        new AclBuilder(principal)
+            .addResource(ResourceType.GROUP, group, patternType)
+            .addControlEntry("*", op, AclPermissionType.ALLOW)
+            .build());
+  }
+}
diff --git a/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java b/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java
index 50214cb..19bf65b 100644
--- a/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java
+++ b/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java
@@ -191,10 +191,14 @@ public class TopologySerdesTest {
     Topology topology = parser.deserialise(Paths.get(topologyDescriptor.toURI()).toFile());
 
     Project project = topology.getProjects().get(0);
-    assertThat(project.getProducers()).hasSize(1);
+    assertThat(project.getProducers()).hasSize(3);
     assertThat(project.getConsumers()).hasSize(2);
     assertThat(project.getStreams()).hasSize(1);
     assertThat(project.getConnectors()).hasSize(2);
+
+    assertThat(project.getProducers().get(0).getIdempotence()).isEmpty();
+    assertThat(project.getProducers().get(1).getTransactionId()).isEqualTo(Optional.of("1234"));
+    assertThat(project.getProducers().get(2).getIdempotence()).isNotEmpty();
   }
 
   @Test
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 310862b..e9b2e42 100644
--- a/src/test/java/com/purbon/kafka/topology/integration/AccessControlManagerIT.java
+++ b/src/test/java/com/purbon/kafka/topology/integration/AccessControlManagerIT.java
@@ -33,6 +33,7 @@ import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Optional;
 import java.util.Properties;
 import java.util.concurrent.ExecutionException;
 import java.util.stream.Collectors;
@@ -86,7 +87,8 @@ public class AccessControlManagerIT {
 
     // Crate an ACL outside of the control of the state manager.
     List<TopologyAclBinding> bindings =
-        bindingsBuilder.buildBindingsForProducers(Collections.singleton("User:foo"), "bar");
+        bindingsBuilder.buildBindingsForProducers(
+            Collections.singleton(new Producer("User:foo")), "bar");
     aclsProvider.createBindings(new HashSet<>(bindings));
 
     List<Consumer> consumers = new ArrayList<>();
@@ -146,7 +148,8 @@ public class AccessControlManagerIT {
   public void producerAclsCreation() throws ExecutionException, InterruptedException, IOException {
 
     List<Producer> producers = new ArrayList<>();
-    producers.add(new Producer("User:Producer1"));
+    Producer producer = new Producer("User:Producer1");
+    producers.add(producer);
 
     Project project = new ProjectImpl("project");
     project.setProducers(producers);
@@ -161,7 +164,57 @@ public class AccessControlManagerIT {
     accessControlManager.apply(topology, plan);
     plan.run(false);
 
-    verifyProducerAcls(producers, topicA.toString());
+    verifyProducerAcls(producers, topicA.toString(), 2);
+  }
+
+  @Test
+  public void producerWithTxAclsCreation()
+      throws ExecutionException, InterruptedException, IOException {
+
+    List<Producer> producers = new ArrayList<>();
+    Producer producer = new Producer("User:Producer12");
+    producer.setTransactionId(Optional.of("1234"));
+    producers.add(producer);
+
+    Project project = new ProjectImpl("project");
+    project.setProducers(producers);
+    Topic topicA = new TopicImpl("topicA");
+    project.addTopic(topicA);
+
+    Topology topology = new TopologyImpl();
+    topology.setContext("integration-test");
+    topology.addOther("source", "producerAclsCreation");
+    topology.addProject(project);
+
+    accessControlManager.apply(topology, plan);
+    plan.run(false);
+
+    verifyProducerAcls(producers, topicA.toString(), 5);
+  }
+
+  @Test
+  public void producerWithIdempotenceAclsCreation()
+      throws ExecutionException, InterruptedException, IOException {
+
+    List<Producer> producers = new ArrayList<>();
+    Producer producer = new Producer("User:Producer13");
+    producer.setIdempotence(Optional.of(true));
+    producers.add(producer);
+
+    Project project = new ProjectImpl("project");
+    project.setProducers(producers);
+    Topic topicA = new TopicImpl("topicA2");
+    project.addTopic(topicA);
+
+    Topology topology = new TopologyImpl();
+    topology.setContext("integration-test");
+    topology.addOther("source", "producerAclsCreation");
+    topology.addProject(project);
+
+    accessControlManager.apply(topology, plan);
+    plan.run(false);
+
+    verifyProducerAcls(producers, topicA.toString(), 3);
   }
 
   @Test
@@ -390,7 +443,7 @@ public class AccessControlManagerIT {
     Assert.assertEquals(1, acls.size());
   }
 
-  private void verifyProducerAcls(List<Producer> producers, String topic)
+  private void verifyProducerAcls(List<Producer> producers, String topic, int aclsCount)
       throws InterruptedException, ExecutionException {
 
     for (Producer producer : producers) {
@@ -402,7 +455,7 @@ public class AccessControlManagerIT {
       AclBindingFilter filter = new AclBindingFilter(resourceFilter, entryFilter);
       Collection<AclBinding> acls = kafkaAdminClient.describeAcls(filter).values().get();
 
-      Assert.assertEquals(2, acls.size());
+      Assert.assertEquals(aclsCount, acls.size());
 
       List<ResourceType> types =
           acls.stream()
diff --git a/src/test/resources/descriptor.yaml b/src/test/resources/descriptor.yaml
index 859c5df..82eb3f9 100644
--- a/src/test/resources/descriptor.yaml
+++ b/src/test/resources/descriptor.yaml
@@ -9,6 +9,10 @@ projects:
       - principal: "User:App1"
     producers:
       - principal: "User:App0"
+      - principal: "User:App2"
+        transactionId: "1234"
+      - principal: "User:App2"
+        idempotence: "true"
     streams:
       - principal: "User:App0"
         topics: