Add --unique-app-id parameter

README.md

@@ -111,6 +111,9 @@ The following configuration options are available:
 - `--extra-output-topics`: Additional named output topics if you need to specify multiple topics with different message
   types (`String=String>[,<String=String>...]`)
 
+- `--unique-app-id`: Unique application ID to use for Kafka Streams. Can also be provided by implementing
+  #getUniqueAppId()
+
 - `--volatile-group-instance-id`: Whether the group instance id is volatile, i.e., it will change on a Streams shutdown.
 
 - `--clean-up`: Whether the state of the Kafka Streams app, i.e., offsets and state stores and auto-created topics,

azure-pipelines.yml

@@ -19,6 +19,7 @@ resources:
     - repository: templates
       type: github
       name: bakdata/bakdata-project-templates
+      ref: tmp/upload-snapshot
       endpoint: bot
 
 jobs:

charts/streams-app-cleanup-job/templates/job.yaml

@@ -112,6 +112,10 @@ spec:
             - name: "{{ .Values.configurationEnvPrefix }}_EXTRA_INPUT_PATTERNS"
               value: "{{- range $key, $value := .Values.streams.extraInputPatterns }}{{ $key }}={{ $value }},{{- end }}"
             {{- end }}
+            {{- if and (.Values.streams.passApplicationId) (hasKey .Values.autoscaling "consumerGroup") }}
+            - name: "{{ .Values.configurationEnvPrefix }}_UNIQUE_APP_ID"
+              value: {{ .Values.autoscaling.consumerGroup | quote }}
+            {{- end }}
             {{- range $key, $value := .Values.secrets }}
             - name: "{{ $key }}"
               valueFrom:

charts/streams-app-cleanup-job/values.yaml

@@ -15,6 +15,7 @@ configurationEnvPrefix: "APP"
 streams:
 #  brokers: "test:9092"
 #  schemaRegistryUrl: "url:1234"
+  passApplicationId: true
   config: {}
 #    max.poll.records: 500
 #    Note that YAML may convert large integers to scientific notation. Use Strings to avoid this.
@@ -59,6 +60,9 @@ labels: {}
 
 # serviceAccountName: foo
 
+autoscaling: {}
+  # consumerGroup: foo
+
 tolerations: []
 #   - key: "foo"
 #     operator: "Exists"

charts/streams-app/README.md

@@ -49,11 +49,12 @@ Alternatively, a YAML file that specifies the values for the parameters can be p
 ### Streams
 
 | Parameter                            | Description                                                                                                                                                                      | Default |
-| ------------------------------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------- |
+|--------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------|
 | `streams.brokers`                    | Comma separated list of Kafka brokers to connect to.                                                                                                                             |         |
 | `streams.schemaRegistryUrl`          | URL of Schema Registry to connect to.                                                                                                                                            | `null`  |
 | `streams.staticMembership`           | Whether to use [Kafka Static Group Membership](https://cwiki.apache.org/confluence/display/KAFKA/KIP-345%3A+Introduce+static+membership+protocol+to+reduce+consumer+rebalances). | `false` |
 | `streams.optimizeLeaveGroupBehavior` | Enabling this optimizes the leave group behavior when a pod is terminated. Depends on the deployment kind, i.e., `statefulSet`. Requires the app to use streams-bootstrap 2.7+.  | `true`  |
+| `streams.passApplicationId`          | Configure the unique application ID via `autoscaling.consumerGroup`. Requires the app to use streams-bootstrap 2.17+.                                                            | `true`  |
 | `streams.config`                     | Configurations for your [Kafka Streams app](https://kafka.apache.org/documentation/#streamsconfigs).                                                                             | `{}`    |
 | `streams.inputTopics`                | List of input topics for your streams application.                                                                                                                               | `[]`    |
 | `streams.extraInputTopics`           | Map of additional named input topics if you need to specify multiple topics with different message types.                                                                        | `{}`    |

charts/streams-app/templates/deployment.yaml

@@ -175,6 +175,10 @@ spec:
             - name: "{{ .Values.configurationEnvPrefix }}_EXTRA_INPUT_PATTERNS"
               value: "{{- range $key, $value := .Values.streams.extraInputPatterns }}{{ $key }}={{ $value }},{{- end }}"
           {{- end }}
+          {{- if and (.Values.streams.passApplicationId) (hasKey .Values.autoscaling "consumerGroup") }}
+            - name: "{{ .Values.configurationEnvPrefix }}_UNIQUE_APP_ID"
+              value: {{ .Values.autoscaling.consumerGroup | quote }}
+          {{- end }}
           {{- range $key, $value := .Values.secrets }}
             - name: "{{ $key }}"
               valueFrom:

charts/streams-app/values.yaml

@@ -27,6 +27,7 @@ streams:
   # schemaRegistryUrl: "url:1234"
   staticMembership: false
   optimizeLeaveGroupBehavior: true
+  passApplicationId: true
   config: {}
   #   max.poll.records: 500
   #   Note that YAML may convert large integers to scientific notation. Use Strings to avoid this.

streams-bootstrap/src/main/java/com/bakdata/kafka/KafkaStreamsApplication.java

@@ -1,7 +1,7 @@
 /*
  * MIT License
  *
- * Copyright (c) 2023 bakdata
+ * Copyright (c) 2024 bakdata
  *
  * Permission is hereby granted, free of charge, to any person obtaining a copy
  * of this software and associated documentation files (the "Software"), to deal
@@ -94,6 +94,14 @@ public abstract class KafkaStreamsApplication extends KafkaApplication implement
     @CommandLine.Option(names = "--volatile-group-instance-id", arity = "0..1",
             description = "Whether the group instance id is volatile, i.e., it will change on a Streams shutdown.")
     private boolean volatileGroupInstanceId;
+    /**
+     * This must be set to a unique value for every application interacting with your kafka cluster to ensure internal
+     * state encapsulation. Could be set to: className-inputTopic-outputTopic
+     */
+    @CommandLine.Option(names = "--unique-app-id",
+            description = "Unique application ID to use for Kafka Streams. Can also be provided by implementing "
+                          + "#getUniqueAppId()")
+    private String uniqueAppId;
     private KafkaStreams streams;
     private Throwable lastException;
 
@@ -155,12 +163,6 @@ public void close() {
      */
     public abstract void buildTopology(StreamsBuilder builder);
 
-    /**
-     * This must be set to a unique value for every application interacting with your kafka cluster to ensure internal
-     * state encapsulation. Could be set to: className-inputTopic-outputTopic
-     */
-    public abstract String getUniqueAppId();
-
     /**
      * Create the topology of the Kafka Streams app
      *
@@ -271,13 +273,31 @@ protected Properties createKafkaProperties() {
         kafkaConfig.setProperty(StreamsConfig.producerPrefix(ProducerConfig.COMPRESSION_TYPE_CONFIG), "gzip");
 
         // topology
-        kafkaConfig.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, this.getUniqueAppId());
+        kafkaConfig.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, this.getApplicationId());
 
         this.configureDefaultSerde(kafkaConfig);
         kafkaConfig.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, this.getBrokers());
         return kafkaConfig;
     }
 
+    private String getApplicationId() {
+        final String uniqueAppId = this.getUniqueAppId();
+        if (uniqueAppId == null) {
+            if (this.uniqueAppId == null) {
+                throw new IllegalArgumentException("Must pass --unique-app-id or implement #getUniqueAppId()");
+            }
+            return this.uniqueAppId;
+        }
+        if (this.uniqueAppId == null) {
+            return uniqueAppId;
+        }
+        if (!uniqueAppId.equals(this.uniqueAppId)) {
+            throw new IllegalArgumentException(
+                    "Application ID provided via --unique-app-id does not match #getUniqueAppId()");
+        }
+        return uniqueAppId;
+    }
+
     /**
      * Run the Streams application. This method blocks until Kafka Streams has completed shutdown, either because it
      * caught an error or the application has received a shutdown event.
@@ -322,7 +342,7 @@ protected void runCleanUp() {
         try (final ImprovedAdminClient adminClient = this.createAdminClient()) {
             final CleanUpRunner cleanUpRunner = CleanUpRunner.builder()
                     .topology(this.createTopology())
-                    .appId(this.getUniqueAppId())
+                    .appId(this.getApplicationId())
                     .adminClient(adminClient)
                     .streams(this.streams)
                     .build();