feat: allow custom kafka connect storage topics (#26)

Author: devshawn

docs/services.md

@@ -101,15 +101,6 @@ This would allow your streams application to access kafka using `my-stream-appli
 
 A basic example which defines one Kafka Connect cluster that has one connector running. 
 
-**Connect Topics**
-Currently, kafka connect's internal topics must be defined in the following format:
-
-- `connect-configs-{service-name}`
-- `connect-offsets-{service-name}`
-- `connect-status-{service-name}`
-
-In this example, the connect cluster `group.id` should be `my-connect-cluster`. 
-
 ```yaml
 topics:
   connect-configs-my-connect-cluster:
@@ -151,6 +142,31 @@ Behind the scenes, this generates ACLs such as:
 - `READ` and `WRITE` for the internal kafka connect topics
 - `READ` for the consumer group `my-connect-cluster`
 
+#### Storage Topics
+
+By default, `kafka-gitops` generates ACLs for the internal storage topics following this format:
+
+- `connect-configs-{service-name}`
+- `connect-offsets-{service-name}`
+- `connect-status-{service-name}`
+
+You can specify custom internal storage topics using the `storage-topics` property:
+
+```yaml
+services:
+  my-connect-cluster:
+    type: kafka-connect
+    principal: User:myconnectcluster
+    storage-topics:
+      config: custom-config-topic
+      offset: custom-offset-topic
+      status: custom-status-topic
+    connectors:
+      rabbitmq-sink:
+        consumes:
+          - rabbitmq-data
+```
+
 #### Group ID
 
 The `group.id` used for the connect cluster ACLs defaults to the service name. You can override this by specifying the `group-id` property, as shown below:

src/main/java/com/devshawn/kafka/gitops/domain/state/service/KafkaConnectService.java

@@ -21,6 +21,9 @@ public abstract class KafkaConnectService extends ServiceDetails {
 
     public abstract Optional<String> getPrincipal();
 
+    @JsonProperty("storage-topics")
+    public abstract Optional<KafkaConnectStorageTopics> getStorageTopics();
+
     public abstract List<String> getProduces();
 
     public abstract Map<String, KafkaConnectorDetails> getConnectors();
@@ -35,18 +38,43 @@ public List<AclDetails.Builder> getAcls(String serviceName) {
 
     private List<AclDetails.Builder> getConnectWorkerAcls(String serviceName) {
         String groupId = getGroupId().isPresent() ? getGroupId().get() : serviceName;
+        String configTopic = getConfigTopic(serviceName);
+        String offsetTopic = getOffsetTopic(serviceName);
+        String statusTopic = getStatusTopic(serviceName);
+
         List<AclDetails.Builder> acls = new ArrayList<>();
-        acls.add(generateReadAcl(String.format("connect-configs-%s", serviceName), getPrincipal()));
-        acls.add(generateReadAcl(String.format("connect-offsets-%s", serviceName), getPrincipal()));
-        acls.add(generateReadAcl(String.format("connect-status-%s", serviceName), getPrincipal()));
-        acls.add(generateWriteACL(String.format("connect-configs-%s", serviceName), getPrincipal()));
-        acls.add(generateWriteACL(String.format("connect-offsets-%s", serviceName), getPrincipal()));
-        acls.add(generateWriteACL(String.format("connect-status-%s", serviceName), getPrincipal()));
+        acls.add(generateReadAcl(configTopic, getPrincipal()));
+        acls.add(generateReadAcl(offsetTopic, getPrincipal()));
+        acls.add(generateReadAcl(statusTopic, getPrincipal()));
+        acls.add(generateWriteACL(configTopic, getPrincipal()));
+        acls.add(generateWriteACL(offsetTopic, getPrincipal()));
+        acls.add(generateWriteACL(statusTopic, getPrincipal()));
         acls.add(generateConsumerGroupAcl(groupId, getPrincipal(), "READ"));
         getConnectors().forEach((connectorName, connector) -> acls.addAll(connector.getAcls(connectorName, getPrincipal())));
         return acls;
     }
 
+    private String getConfigTopic(String serviceName) {
+        if (getStorageTopics().isPresent() && getStorageTopics().get().getConfig().isPresent()) {
+            return getStorageTopics().get().getConfig().get();
+        }
+        return String.format("connect-configs-%s", serviceName);
+    }
+
+    private String getOffsetTopic(String serviceName) {
+        if (getStorageTopics().isPresent() && getStorageTopics().get().getOffset().isPresent()) {
+            return getStorageTopics().get().getOffset().get();
+        }
+        return String.format("connect-offsets-%s", serviceName);
+    }
+
+    private String getStatusTopic(String serviceName) {
+        if (getStorageTopics().isPresent() && getStorageTopics().get().getStatus().isPresent()) {
+            return getStorageTopics().get().getStatus().get();
+        }
+        return String.format("connect-status-%s", serviceName);
+    }
+
     public static class Builder extends KafkaConnectService_Builder {
 
     }

src/main/java/com/devshawn/kafka/gitops/domain/state/service/KafkaConnectStorageTopics.java

@@ -0,0 +1,20 @@
+package com.devshawn.kafka.gitops.domain.state.service;
+
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import org.inferred.freebuilder.FreeBuilder;
+
+import java.util.Optional;
+
+@FreeBuilder
+@JsonDeserialize(builder = KafkaConnectStorageTopics.Builder.class)
+public interface KafkaConnectStorageTopics {
+
+    Optional<String> getConfig();
+
+    Optional<String> getOffset();
+
+    Optional<String> getStatus();
+
+    class Builder extends KafkaConnectStorageTopics_Builder {
+    }
+}

src/test/groovy/com/devshawn/kafka/gitops/ApplyCommandIntegrationSpec.groovy

@@ -56,7 +56,9 @@ class ApplyCommandIntegrationSpec extends Specification {
                 "custom-user-acls",
                 "custom-group-id-application",
                 "custom-group-id-connect",
-                "custom-application-id-streams"
+                "custom-application-id-streams",
+                "custom-storage-topic",
+                "custom-storage-topics"
         ]
     }
 

src/test/groovy/com/devshawn/kafka/gitops/PlanCommandIntegrationSpec.groovy

@@ -61,7 +61,9 @@ class PlanCommandIntegrationSpec extends Specification {
                 "custom-user-acls",
                 "custom-group-id-application",
                 "custom-group-id-connect",
-                "custom-application-id-streams"
+                "custom-application-id-streams",
+                "custom-storage-topic",
+                "custom-storage-topics"
         ]
     }
 
@@ -130,7 +132,8 @@ class PlanCommandIntegrationSpec extends Specification {
                 "invalid-topic",
                 "unrecognized-property",
                 "invalid-format",
-                "invalid-missing-user-principal"
+                "invalid-missing-user-principal",
+                "invalid-storage-topics"
         ]
     }
 

src/test/resources/plans/custom-storage-topic-apply-output.txt

@@ -0,0 +1,143 @@
+Executing apply...
+
+Applying: [CREATE]
+
++ [ACL] test-connect-cluster-0
+	 + resource_name: test-connect-cluster-configs
+	 + resource_type: TOPIC
+	 + resource_pattern: LITERAL
+	 + resource_principal: User:test
+	 + host: *
+	 + operation: READ
+	 + permission: ALLOW
+
+
+Successfully applied.
+
+Applying: [CREATE]
+
++ [ACL] test-connect-cluster-1
+	 + resource_name: connect-offsets-test-connect-cluster
+	 + resource_type: TOPIC
+	 + resource_pattern: LITERAL
+	 + resource_principal: User:test
+	 + host: *
+	 + operation: READ
+	 + permission: ALLOW
+
+
+Successfully applied.
+
+Applying: [CREATE]
+
++ [ACL] test-connect-cluster-2
+	 + resource_name: connect-status-test-connect-cluster
+	 + resource_type: TOPIC
+	 + resource_pattern: LITERAL
+	 + resource_principal: User:test
+	 + host: *
+	 + operation: READ
+	 + permission: ALLOW
+
+
+Successfully applied.
+
+Applying: [CREATE]
+
++ [ACL] test-connect-cluster-3
+	 + resource_name: test-connect-cluster-configs
+	 + resource_type: TOPIC
+	 + resource_pattern: LITERAL
+	 + resource_principal: User:test
+	 + host: *
+	 + operation: WRITE
+	 + permission: ALLOW
+
+
+Successfully applied.
+
+Applying: [CREATE]
+
++ [ACL] test-connect-cluster-4
+	 + resource_name: connect-offsets-test-connect-cluster
+	 + resource_type: TOPIC
+	 + resource_pattern: LITERAL
+	 + resource_principal: User:test
+	 + host: *
+	 + operation: WRITE
+	 + permission: ALLOW
+
+
+Successfully applied.
+
+Applying: [CREATE]
+
++ [ACL] test-connect-cluster-5
+	 + resource_name: connect-status-test-connect-cluster
+	 + resource_type: TOPIC
+	 + resource_pattern: LITERAL
+	 + resource_principal: User:test
+	 + host: *
+	 + operation: WRITE
+	 + permission: ALLOW
+
+
+Successfully applied.
+
+Applying: [CREATE]
+
++ [ACL] test-connect-cluster-6
+	 + resource_name: test-connect-cluster
+	 + resource_type: GROUP
+	 + resource_pattern: LITERAL
+	 + resource_principal: User:test
+	 + host: *
+	 + operation: READ
+	 + permission: ALLOW
+
+
+Successfully applied.
+
+Applying: [CREATE]
+
++ [ACL] test-connect-cluster-7
+	 + resource_name: production-topic
+	 + resource_type: TOPIC
+	 + resource_pattern: LITERAL
+	 + resource_principal: User:test
+	 + host: *
+	 + operation: WRITE
+	 + permission: ALLOW
+
+
+Successfully applied.
+
+Applying: [CREATE]
+
++ [ACL] test-connect-cluster-8
+	 + resource_name: consumption-topic
+	 + resource_type: TOPIC
+	 + resource_pattern: LITERAL
+	 + resource_principal: User:test
+	 + host: *
+	 + operation: READ
+	 + permission: ALLOW
+
+
+Successfully applied.
+
+Applying: [CREATE]
+
++ [ACL] test-connect-cluster-9
+	 + resource_name: connect-test-sink
+	 + resource_type: GROUP
+	 + resource_pattern: LITERAL
+	 + resource_principal: User:test
+	 + host: *
+	 + operation: READ
+	 + permission: ALLOW
+
+
+Successfully applied.
+
+[SUCCESS] Apply complete! Resources: 10 created, 0 updated, 0 deleted.