Add --application-id parameter

bakdata · Jan 26, 2024 · ec6d53d · ec6d53d
1 parent 32634a9
commit ec6d53d
Show file tree

Hide file tree

Showing 3 changed files with 31 additions and 17 deletions.
diff --git a/charts/streams-app/README.md b/charts/streams-app/README.md
@@ -98,19 +98,20 @@ Alternatively, a YAML file that specifies the values for the parameters can be p
 
 ### Auto-Scaling
 
-| Parameter                        | Description                                                                                              | Default    |
-| -------------------------------- | -------------------------------------------------------------------------------------------------------- | ---------- |
-| `autoscaling.enabled`            | Whether to enable auto-scaling using [KEDA](https://keda.sh/docs/latest/scalers/apache-kafka/).          | `false`    |
-| `autoscaling.consumerGroup`      | Name of the consumer group used for checking the offset on the topic and processing the related lag.     |            |
-| `autoscaling.lagThreshold`       | Average target value to trigger scaling actions.                                                         |            |
-| `autoscaling.pollingInterval`    | https://keda.sh/docs/2.10/concepts/scaling-deployments/#pollinginterval                                  | `30`       |
-| `autoscaling.cooldownPeriod`     | https://keda.sh/docs/2.10/concepts/scaling-deployments/#cooldownperiod                                   | `300`      |
-| `autoscaling.offsetResetPolicy`  | The offset reset policy for the consumer if the the consumer group is not yet subscribed to a partition. | `earliest` |
-| `autoscaling.minReplicas`        | https://keda.sh/docs/2.10/concepts/scaling-deployments/#minreplicacount                                  | `0`        |
-| `autoscaling.maxReplicas`        | https://keda.sh/docs/2.10/concepts/scaling-deployments/#maxreplicacount                                  | `1`        |
-| `autoscaling.idleReplicas`       | https://keda.sh/docs/2.10/concepts/scaling-deployments/#idlereplicacount                                 |            |
-| `autoscaling.topics`             | List of auto-generated Kafka Streams topics used by the streams app.                                     | `[]`       |
-| `autoscaling.additionalTriggers` | List of additional KEDA triggers, see https://keda.sh/docs/latest/scalers/                               | `[]`       |
+| Parameter                        | Description                                                                                                        | Default    |
+|----------------------------------|--------------------------------------------------------------------------------------------------------------------|------------|
+| `autoscaling.enabled`            | Whether to enable auto-scaling using [KEDA](https://keda.sh/docs/latest/scalers/apache-kafka/).                    | `false`    |
+| `autoscaling.consumerGroup`      | Name of the consumer group used for checking the offset on the topic and processing the related lag.               |            |
+| `autoscaling.lagThreshold`       | Average target value to trigger scaling actions.                                                                   |            |
+| `autoscaling.pollingInterval`    | https://keda.sh/docs/2.10/concepts/scaling-deployments/#pollinginterval                                            | `30`       |
+| `autoscaling.cooldownPeriod`     | https://keda.sh/docs/2.10/concepts/scaling-deployments/#cooldownperiod                                             | `300`      |
+| `autoscaling.offsetResetPolicy`  | The offset reset policy for the consumer if the the consumer group is not yet subscribed to a partition.           | `earliest` |
+| `autoscaling.minReplicas`        | https://keda.sh/docs/2.10/concepts/scaling-deployments/#minreplicacount                                            | `0`        |
+| `autoscaling.maxReplicas`        | https://keda.sh/docs/2.10/concepts/scaling-deployments/#maxreplicacount                                            | `1`        |
+| `autoscaling.idleReplicas`       | https://keda.sh/docs/2.10/concepts/scaling-deployments/#idlereplicacount                                           |            |
+| `autoscaling.internalTopics`     | List of auto-generated Kafka Streams topics used by the streams app. Consumer group prefix is added automatically. | `[]`       |
+| `autoscaling.topics`             | List of topics used by the streams app.                                                                            | `[]`       |
+| `autoscaling.additionalTriggers` | List of additional KEDA triggers, see https://keda.sh/docs/latest/scalers/                                         | `[]`       |
 
 ### JVM
 

diff --git a/charts/streams-app/templates/scaled-object.yaml b/charts/streams-app/templates/scaled-object.yaml
@@ -26,8 +26,8 @@ spec:
   idleReplicaCount: {{ .Values.autoscaling.idleReplicas }}
   {{- end }}
   triggers:
-    {{- if not (or .Values.streams.inputTopics .Values.autoscaling.topics .Values.streams.extraInputTopics .Values.autoscaling.additionalTriggers) }}
-    {{- fail "To use autoscaling, you must define one of .Values.streams.inputTopics, .Values.autoscaling.topics, .Values.streams.extraInputTopics or .Values.autoscaling.additionalTriggers" }}
+    {{- if not (or .Values.streams.inputTopics .Values.autoscaling.internalTopics .Values.autoscaling.topics .Values.streams.extraInputTopics .Values.autoscaling.additionalTriggers) }}
+    {{- fail "To use autoscaling, you must define one of .Values.streams.inputTopics, .Values.autoscaling.internalTopics, .Values.autoscaling.topics, .Values.streams.extraInputTopics or .Values.autoscaling.additionalTriggers" }}
     {{- end}}
     # todo: concat .Values.streams.inputTopics and .Values.autoscaling.topics to
     # minimize number of loops when we don't need to support helm 2 anymore
@@ -40,6 +40,15 @@ spec:
         lagThreshold: {{ $root.Values.autoscaling.lagThreshold | quote }}
         offsetResetPolicy: {{ $root.Values.autoscaling.offsetResetPolicy }}
     {{- end }}
+    {{- range .Values.autoscaling.internalTopics }}
+    - type: kafka
+      metadata:
+        bootstrapServers: {{ $root.Values.streams.brokers }}
+        consumerGroup: {{ $root.Values.autoscaling.consumerGroup }}
+        topic: {{ printf "%s-%s" $root.Values.autoscaling.consumerGroup . | quote }}
+        lagThreshold: {{ $root.Values.autoscaling.lagThreshold | quote }}
+        offsetResetPolicy: {{ $root.Values.autoscaling.offsetResetPolicy }}
+    {{- end }}
     {{- range .Values.autoscaling.topics }}
     - type: kafka
       metadata:

diff --git a/charts/streams-app/values.yaml b/charts/streams-app/values.yaml
@@ -103,8 +103,12 @@ autoscaling:
   minReplicas: 0
   maxReplicas: 1
   # idleReplicas: 0
-  ## all topics from streams.inputTopics and streams.extraInputTopics are automatically taken, so
-  ## only use the 'topics' option for adding more topics, e.g., internal topics
+  ## all topics from streams.inputTopics and streams.extraInputTopics are automatically taken
+  ## only use the 'internalTopics' option for adding internal topics, i.e., auto-generated topics by Kafka Streams. Consumer group name will automatically be added as a prefix
+  internalTopics: [ ]
+  #   - bar-repartition                       # results in foo-bar-repartition
+  #   - baz-join-subscription-response-topic  # results in foo-baz-join-subscription-response-topic
+  ## only use the 'topics' option for adding more topics, e.g., intermediate topics, not covered by `internalTopics`
   topics: []
   #   - bar
   #   - baz