feat: allow custom group.id and application.id (#25)

Author: devshawn

docs/confluent-cloud.md

@@ -109,7 +109,4 @@ Congrats! You're now using `kafka-gitops` to manage your Confluent Cloud cluster
 
 Welcome to GitOps! 
 
-
-
-
 [installation]: /installation.md

docs/permissions.md

@@ -1,6 +1,6 @@
 # Permissions
 
-When running against a secure Kafka cluster, kafka-gitops needs to be authorized to perform actions against the cluster. This can be either a super user defined by the Kafka cluster or a custom user with specific permissions.
+When running against a secured Kafka cluster, `kafka-gitops` needs to be authorized to perform actions against the cluster. This can either be a super user defined by the Kafka cluster or a custom user with specific permissions.
 
 ## Example
 

docs/quick-start.md

@@ -33,6 +33,8 @@ For our quick start example, open a terminal where your `state.yaml` file is loc
 export KAFKA_BOOTSTRAP_SERVERS=localhost:9092
 ```
 
+!> **NOTE**: If running `kafka-gitops` against a secured Kafka cluster, it must be run with super user credentials or a user with special ACLs. [Read more on our permissions page][permissions].
+
 ## Validate
 We can validate the desired state file conforms to the [specification][specification]. To do this, run:
 
@@ -141,5 +143,6 @@ org.apache.kafka.common.errors.PolicyViolationException: Topic replication facto
 Congrats! You've successfully started using GitOps strategies to manage your cluster. If you have security on your cluster, read the [services][services] page to start defining services. 
 
 [ccloud]: /confluent-cloud.md
+[permissions]: /permissions.md
 [specification]: /specification.md
 [services]: /services.md

docs/services.md

@@ -8,7 +8,7 @@ A basic example shown below defines one topic, `test-topic`, and one service, `m
 
 The service `my-application` both consumes from and produces to `test-topic`. This will generate the necessary ACLs for `my-application` to do this.
 
-?> **Note**: If using Confluent Cloud, omit the principal field.
+?> **NOTE**: If using Confluent Cloud, omit the principal field.
 
 ```yaml
 topics:
@@ -32,7 +32,23 @@ Behind the scenes, this will generate three ACLs:
 - `WRITE` for topic `test-topic`
 - `READ` for consumer group `my-application`
 
-!> Currently, consumer `group.id` must match the service name.
+#### Group ID
+
+The `group.id` used for consumer ACLs defaults to the service name. You can override this by specifying the `group-id` property, as shown below:
+
+```yaml
+services:
+  my-application:
+    type: application
+    principal: User:myapp
+    group-id: my-application-service
+    consumes:
+      - test-topic
+    produces:
+      - test-topic
+```
+
+This would allow your consumer group to access kafka using `my-application-service` as the `group.id`.
 
 ## Kafka Streams Example
 
@@ -63,6 +79,24 @@ Behind the scenes, this generates ACLs such as:
 - `READ` for consumer group `my-stream`
 - Various ACLs for Kafka streams internal topic management
 
+#### Application ID
+
+The `application.id` used for streams ACLs defaults to the service name. You can override this by specifying the `application-id` property, as shown below:
+
+```yaml
+services:
+  my-stream:
+    type: kafka-streams
+    principal: User:mystream
+    application-id: my-stream-application
+    consumes:
+      - test-topic
+    produces:
+      - test-topic
+```
+
+This would allow your streams application to access kafka using `my-stream-application` as the `application.id`.
+
 ## Kafka Connect Example
 
 A basic example which defines one Kafka Connect cluster that has one connector running. 
@@ -116,3 +150,23 @@ Behind the scenes, this generates ACLs such as:
 - `READ` for the consumer group `connect-rabbitmq-sink`
 - `READ` and `WRITE` for the internal kafka connect topics
 - `READ` for the consumer group `my-connect-cluster`
+
+#### 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:
+
+```yaml
+services:
+  my-connect-cluster:
+    type: kafka-connect
+    principal: User:myconnectcluster
+    group-id: kafka-connect-cluster
+    connectors:
+      rabbitmq-sink:
+        consumes:
+          - rabbitmq-data
+```
+
+This allows your connect cluster to access kafka using `kafka-connect-cluster` as the `group.id`. 
+
+!> **NOTE**: The `group-id` setting only affects the connect cluster `group.id`, and not any sink connector group IDs.

docs/specification.md

@@ -2,7 +2,7 @@
 
 This document describes the specification for how to write your Kafka cluster's desired state file. This currently must be a `YAML` file. 
 
-?> Current version: `1`
+?> Current version: `1.0.1`
 
 The desired state file consists of:
 
@@ -63,17 +63,18 @@ There are currently three service types:
 - `kafka-connect`
 - `kafka-streams`
 
-?> **NOTE**: If using Confluent Cloud, omit the `principal` fields.
+!> **NOTE**: If using Confluent Cloud, omit the `principal` fields.
 
 **Example application**:
 
-!> **NOTE**: Currently, the service name must match the consumer `group.id`.
+?> **NOTE**: The `group-id` property is optional and defaults to the service name.
 
 ```yaml
 services:
   my-application-name:
     type: application
     principal: User:my-application-principal
+    group-id: optional-group-id-override
     produces:
       - topic-name-one
     consumes:
@@ -83,13 +84,14 @@ services:
 
 **Example kafka connect cluster**:
 
-!> **NOTE**: Currently, the service name must match the connect cluster `group.id`.
+?> **NOTE**: The `group-id` property is optional and defaults to the service name.
 
 ```yaml
 services:
   my-kafka-connect-cluster-name:
     type: kafka-connect
     principal: User:my-connect-principal
+    group-id: optional-group-id-override
     connectors:
       my-source-connector-name:
         produces:
@@ -101,20 +103,21 @@ services:
 
 **Example kafka streams application**:
 
-!> **NOTE**: Currently, the service name must match the streams `application.id`.
+?> **NOTE**: The `application-id` property is optional and defaults to the service name.
 
 ```yaml
 services:
   my-kafka-streams-name:
     type: kafka-streams
     principal: User:my-streams-principal
+    application-id: optional-application-id-override
     produces:
       - topic-name-one
     consumes:
       - topic-name-two
 ```
 
-Under the cover, `kafka-gitops` generates ACLs based on these definitions.
+Behind the scenes, `kafka-gitops` generates ACLs based on these definitions.
 
 ## Users
 

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

@@ -2,6 +2,7 @@
 
 import com.devshawn.kafka.gitops.domain.state.AclDetails;
 import com.devshawn.kafka.gitops.domain.state.ServiceDetails;
+import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
 import org.inferred.freebuilder.FreeBuilder;
 
@@ -15,6 +16,9 @@ public abstract class ApplicationService extends ServiceDetails {
 
     public abstract Optional<String> getPrincipal();
 
+    @JsonProperty("group-id")
+    public abstract Optional<String> getGroupId();
+
     public abstract List<String> getProduces();
 
     public abstract List<String> getConsumes();
@@ -25,7 +29,8 @@ public List<AclDetails.Builder> getAcls(String serviceName) {
         getProduces().forEach(topic -> acls.add(generateWriteACL(topic, getPrincipal())));
         getConsumes().forEach(topic -> acls.add(generateReadAcl(topic, getPrincipal())));
         if (!getConsumes().isEmpty()) {
-            acls.add(generateConsumerGroupAcl(serviceName, getPrincipal(), "READ"));
+            String groupId = getGroupId().isPresent() ? getGroupId().get() : serviceName;
+            acls.add(generateConsumerGroupAcl(groupId, getPrincipal(), "READ"));
         }
         return acls;
     }

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

@@ -3,6 +3,7 @@
 
 import com.devshawn.kafka.gitops.domain.state.AclDetails;
 import com.devshawn.kafka.gitops.domain.state.ServiceDetails;
+import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
 import org.inferred.freebuilder.FreeBuilder;
 
@@ -15,6 +16,9 @@
 @JsonDeserialize(builder = KafkaConnectService.Builder.class)
 public abstract class KafkaConnectService extends ServiceDetails {
 
+    @JsonProperty("group-id")
+    public abstract Optional<String> getGroupId();
+
     public abstract Optional<String> getPrincipal();
 
     public abstract List<String> getProduces();
@@ -30,14 +34,15 @@ public List<AclDetails.Builder> getAcls(String serviceName) {
     }
 
     private List<AclDetails.Builder> getConnectWorkerAcls(String serviceName) {
+        String groupId = getGroupId().isPresent() ? getGroupId().get() : 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(generateConsumerGroupAcl(serviceName, getPrincipal(), "READ"));
+        acls.add(generateConsumerGroupAcl(groupId, getPrincipal(), "READ"));
         getConnectors().forEach((connectorName, connector) -> acls.addAll(connector.getAcls(connectorName, getPrincipal())));
         return acls;
     }

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

@@ -2,6 +2,7 @@
 
 import com.devshawn.kafka.gitops.domain.state.AclDetails;
 import com.devshawn.kafka.gitops.domain.state.ServiceDetails;
+import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
 import org.inferred.freebuilder.FreeBuilder;
 
@@ -15,6 +16,9 @@ public abstract class KafkaStreamsService extends ServiceDetails {
 
     public abstract Optional<String> getPrincipal();
 
+    @JsonProperty("application-id")
+    public abstract Optional<String> getApplicationId();
+
     public abstract List<String> getProduces();
 
     public abstract List<String> getConsumes();
@@ -29,17 +33,18 @@ public List<AclDetails.Builder> getAcls(String serviceName) {
     }
 
     private List<AclDetails.Builder> getInternalAcls(String serviceName) {
+        String applicationId = getApplicationId().isPresent() ? getApplicationId().get() : serviceName;
         List<AclDetails.Builder> acls = new ArrayList<>();
-        acls.add(generatePrefixedTopicACL(serviceName, getPrincipal(), "READ"));
-        acls.add(generatePrefixedTopicACL(serviceName, getPrincipal(), "WRITE"));
-        acls.add(generatePrefixedTopicACL(serviceName, getPrincipal(), "DESCRIBE"));
-        acls.add(generatePrefixedTopicACL(serviceName, getPrincipal(), "DELETE"));
-        acls.add(generatePrefixedTopicACL(serviceName, getPrincipal(), "CREATE"));
-        acls.add(generatePrefixedTopicACL(serviceName, getPrincipal(), "ALTER"));
-        acls.add(generatePrefixedTopicACL(serviceName, getPrincipal(), "ALTER_CONFIGS"));
-        acls.add(generatePrefixedTopicACL(serviceName, getPrincipal(), "DESCRIBE_CONFIGS"));
-        acls.add(generateConsumerGroupAcl(serviceName, getPrincipal(), "READ"));
-        acls.add(generateConsumerGroupAcl(serviceName, getPrincipal(), "DESCRIBE"));
+        acls.add(generatePrefixedTopicACL(applicationId, getPrincipal(), "READ"));
+        acls.add(generatePrefixedTopicACL(applicationId, getPrincipal(), "WRITE"));
+        acls.add(generatePrefixedTopicACL(applicationId, getPrincipal(), "DESCRIBE"));
+        acls.add(generatePrefixedTopicACL(applicationId, getPrincipal(), "DELETE"));
+        acls.add(generatePrefixedTopicACL(applicationId, getPrincipal(), "CREATE"));
+        acls.add(generatePrefixedTopicACL(applicationId, getPrincipal(), "ALTER"));
+        acls.add(generatePrefixedTopicACL(applicationId, getPrincipal(), "ALTER_CONFIGS"));
+        acls.add(generatePrefixedTopicACL(applicationId, getPrincipal(), "DESCRIBE_CONFIGS"));
+        acls.add(generateConsumerGroupAcl(applicationId, getPrincipal(), "READ"));
+        acls.add(generateConsumerGroupAcl(applicationId, getPrincipal(), "DESCRIBE"));
         return acls;
     }
 

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

@@ -48,10 +48,15 @@ class ApplyCommandIntegrationSpec extends Specification {
         planFile << [
                 "simple",
                 "application-service",
+                "kafka-streams-service",
+                "kafka-connect-service",
                 "multi-file",
                 "simple-users",
                 "custom-service-acls",
-                "custom-user-acls"
+                "custom-user-acls",
+                "custom-group-id-application",
+                "custom-group-id-connect",
+                "custom-application-id-streams"
         ]
     }
 

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

@@ -58,7 +58,10 @@ class PlanCommandIntegrationSpec extends Specification {
                 "multi-file",
                 "simple-users",
                 "custom-service-acls",
-                "custom-user-acls"
+                "custom-user-acls",
+                "custom-group-id-application",
+                "custom-group-id-connect",
+                "custom-application-id-streams"
         ]
     }