Add --unique-app-id parameter

azure-pipelines.yml

@@ -23,6 +23,8 @@ resources:
 
 jobs:
 - template: azure/gradle/build.yml@templates
+  parameters:
+    jdkVersion: '1.17'
 - template: azure/gradle/create_tag_version.yml@templates
 - template: azure/gradle/upload_release.yml@templates
 - template: azure/gradle/upload_snapshot.yml@templates

build.gradle.kts

@@ -1,7 +1,7 @@
 plugins {
     id("net.researchgate.release") version "3.0.2"
-    id("com.bakdata.sonar") version "1.1.7"
-    id("com.bakdata.sonatype") version "1.1.7"
+    id("com.bakdata.sonar") version "1.1.9"
+    id("com.bakdata.sonatype") version "1.1.9"
     id("org.hildan.github.changelog") version "1.12.1"
     id("io.freefair.lombok") version "6.6.1"
 }

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 hasKey .Values.autoscaling "consumerGroup" }}
+            - name: "{{ .Values.configurationEnvPrefix }}_APPLICATION_ID"
+              value: {{ .Values.autoscaling.consumerGroup | quote }}
+            {{- end }}
             {{- range $key, $value := .Values.secrets }}
             - name: "{{ $key }}"
               valueFrom:

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

@@ -59,6 +59,9 @@ labels: {}
 
 # serviceAccountName: foo
 
+autoscaling: {}
+  # consumerGroup: foo
+
 tolerations: []
 #   - key: "foo"
 #     operator: "Exists"

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 hasKey .Values.autoscaling "consumerGroup" }}
+            - name: "{{ .Values.configurationEnvPrefix }}_APPLICATION_ID"
+              value: {{ .Values.autoscaling.consumerGroup | quote }}
+          {{- end }}
           {{- range $key, $value := .Values.secrets }}
             - name: "{{ $key }}"
               valueFrom:

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,8 @@ 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;
+    @CommandLine.Option(names = "--application-id", description = "Application ID to use for Kafka Streams")
+    private String uniqueAppId;
     private KafkaStreams streams;
     private Throwable lastException;
 
@@ -155,12 +157,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 +267,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 --application-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 --application-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 +336,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();