diff --git a/.gitignore b/.gitignore
index 697247905..c742e3ef1 100644
--- a/.gitignore
+++ b/.gitignore
@@ -3,3 +3,4 @@
 !.gitignore
 build/
 out/
+charts/**/*.tgz
diff --git a/charts/common-app/README.md b/charts/common-app/README.md
index a62a7fd5c..4eaf574db 100644
--- a/charts/common-app/README.md
+++ b/charts/common-app/README.md
@@ -24,15 +24,6 @@ You can then include the library's templates in your own templates/ files. The l
 
 You can include any of the following templates in your application chart.
 
-### `common-app.deployment`
-Renders the main `Deployment` for the application.
-* **Enabled by:** `.Values.deployment: true`.
-* **Usage (`templates/deployment.yaml`):**
-    ```yaml
-    {{- include "common-app.deployment" . -}}
-    ```
-
-
 ### `common-app.service`
 Renders a `Service` to expose the application.
 * **Enabled by:** `.Values.service.enabled: true`.
diff --git a/charts/common-app/templates/_cleanup-job.yaml b/charts/common-app/templates/_cleanup-job.yaml
deleted file mode 100644
index 03a1c2ea2..000000000
--- a/charts/common-app/templates/_cleanup-job.yaml
+++ /dev/null
@@ -1,166 +0,0 @@
-{{- define "common-app.cleanup-job" -}}
-{{- $root := . -}}
-apiVersion: batch/v1
-kind: Job
-metadata:
-  name: {{ include "common-app.fullname" . }}
-  {{- include "common-app.annotations" . }}
-  labels:
-    {{- include "common-app.labels" . | nindent 4 }}
-    {{- range $key, $value := .Values.labels }}
-    {{ $key | quote }}: {{ $value | quote }}
-    {{- end }}
-spec:
-  ttlSecondsAfterFinished: 30
-  template:
-    metadata:
-    {{- if .Values.podAnnotations }}
-      annotations:
-      {{- range $key, $value := .Values.podAnnotations }}
-        {{ $key | quote }}: {{ $value | quote }}
-      {{- end }}
-    {{- end }}
-      labels:
-        {{- include "common-app.selectorLabels" . | nindent 8 }}
-        {{- range $key, $value := .Values.podLabels }}
-        {{ $key }}: {{ $value }}
-        {{- end }}
-    spec:
-      {{- if .Values.serviceAccountName }}
-      serviceAccountName: {{ .Values.serviceAccountName }}
-      {{- end }}
-      restartPolicy: {{ .Values.restartPolicy }}
-      {{- if .Values.tolerations }}
-      tolerations:
-{{ toYaml .Values.tolerations | indent 8 }}
-      {{- end }}
-      {{- with .Values.affinity }}
-      affinity:
-        {{- tpl (toYaml .) $root | nindent 8 }}
-      {{- end }}
-      {{- if .Values.imagePullSecrets }}
-      imagePullSecrets:
-{{- toYaml .Values.imagePullSecrets | nindent 8 }}
-      {{- end }}
-      containers:
-        - name: "kafka-app"
-          image: "{{ .Values.image }}:{{ .Values.imageTag }}"
-          imagePullPolicy: "{{ .Values.imagePullPolicy }}"
-          resources:
-{{ toYaml .Values.resources | indent 12 }}
-          args:
-            {{- if .Values.kafka.deleteOutput }}
-            - clean
-            {{- else }}
-            - reset
-            {{- end }}
-          env:
-            - name: ENV_PREFIX
-              value: {{ .Values.configurationEnvPrefix }}_
-            {{- range $key, $value := .Values.kafka.config }}
-            - name: {{ printf "KAFKA_%s" $key | replace "." "_" | upper | quote }}
-              value: {{ $value | quote }}
-            {{- end }}
-            {{- if hasKey .Values.kafka "bootstrapServers" }}
-            - name: "{{ .Values.configurationEnvPrefix }}_BOOTSTRAP_SERVERS"
-              value: {{ .Values.kafka.bootstrapServers | quote }}
-            {{- end }}
-            {{- if hasKey .Values.kafka "schemaRegistryUrl" }}
-            - name: "{{ .Values.configurationEnvPrefix }}_SCHEMA_REGISTRY_URL"
-              value: {{ .Values.kafka.schemaRegistryUrl | quote }}
-            {{- end }}
-            {{- if and (hasKey .Values.kafka "inputTopics") (.Values.kafka.inputTopics) }}
-            - name: "{{ .Values.configurationEnvPrefix }}_INPUT_TOPICS"
-              value: {{ .Values.kafka.inputTopics | join "," | quote }}
-            {{- end }}
-            {{- if hasKey .Values.kafka "inputPattern" }}
-            - name: "{{ .Values.configurationEnvPrefix }}_INPUT_PATTERN"
-              value: {{ .Values.kafka.inputPattern | join "," | quote }}
-            {{- end }}
-            {{- if hasKey .Values.kafka "outputTopic" }}
-            - name: "{{ .Values.configurationEnvPrefix }}_OUTPUT_TOPIC"
-              value: {{ .Values.kafka.outputTopic | quote }}
-            {{- end }}
-            {{- if hasKey .Values.kafka "errorTopic" }}
-            - name: "{{ .Values.configurationEnvPrefix }}_ERROR_TOPIC"
-              value: {{ .Values.kafka.errorTopic | quote }}
-            {{- end }}
-            {{- if and (hasKey .Values.kafka "labeledOutputTopics") (.Values.kafka.labeledOutputTopics) }}
-            - name: "{{ .Values.configurationEnvPrefix }}_LABELED_OUTPUT_TOPICS"
-              value: "{{- range $key, $value := .Values.kafka.labeledOutputTopics }}{{ $key }}={{ $value }},{{- end }}"
-            {{- end }}
-            {{- $delimiter := ";" }}
-            {{- if and (hasKey .Values.kafka "labeledInputTopics") (.Values.kafka.labeledInputTopics) }}
-            - name: "{{ .Values.configurationEnvPrefix }}_LABELED_INPUT_TOPICS"
-              value: "{{- range $key, $value := .Values.kafka.labeledInputTopics }}{{ $key }}={{ $value | join $delimiter }},{{- end }}"
-            {{- end }}
-            {{- if and (hasKey .Values.kafka "labeledInputPatterns") (.Values.kafka.labeledInputPatterns) }}
-            - name: "{{ .Values.configurationEnvPrefix }}_LABELED_INPUT_PATTERNS"
-              value: "{{- range $key, $value := .Values.kafka.labeledInputPatterns }}{{ $key }}={{ $value }},{{- end }}"
-            {{- end }}
-            {{- if hasKey .Values.kafka "applicationId" }}
-            - name: "{{ .Values.configurationEnvPrefix }}_APPLICATION_ID"
-              value: {{ .Values.kafka.applicationId | quote }}
-            {{- end }}
-            {{- if hasKey .Values.kafka "groupId" }}
-            - name: "{{ .Values.configurationEnvPrefix }}_GROUP_ID"
-              value: {{ .Values.kafka.groupId | quote }}
-            {{- end }}
-            {{- range $key, $value := .Values.secrets }}
-            - name: "{{ $key }}"
-              valueFrom:
-                secretKeyRef:
-                  name: {{ include "common-app.fullname" . }}
-                  key: "{{ $key }}"
-            {{- end }}
-            {{- range $key, $value := .Values.secretRefs }}
-            - name: "{{ $key }}"
-              valueFrom:
-                secretKeyRef:
-                  name: {{ $value.name }}
-                  key: "{{ $value.key }}"
-            {{- end }}
-            {{- range $key, $value := .Values.commandLine }}
-            - name: "{{ $root.Values.configurationEnvPrefix }}_{{ $key }}"
-              value: {{ $value | quote }}
-            {{- end }}
-            {{- range $key, $value := .Values.env }}
-            - name: {{ $key | quote }}
-              value: {{ $value | quote }}
-            {{- end }}
-            - name: JAVA_TOOL_OPTIONS
-              value: '-XX:MaxRAMPercentage={{ printf "%.1f" .Values.javaOptions.maxRAMPercentage }}
-                {{ .Values.javaOptions.others | join " " }}'
-          {{- if or (.Values.files) (.Values.secretFilesRefs) }}
-          volumeMounts:
-            {{- range $key, $value := .Values.files }}
-            - name: config
-              mountPath: {{ printf "%s/%s" $value.mountPath $key | quote }}
-              subPath: {{ $key | quote }}
-            {{- end }}
-            {{- range .Values.secretFilesRefs }}
-            - name: {{ .volume }}
-              mountPath: {{ .mountPath }}
-              {{- if .readOnly }}
-              readOnly: true
-              {{- end }}
-              {{- if .subPath}}
-              subPath: {{.subPath }}
-              {{- end }}
-            {{- end }}
-        {{- end }}
-      {{- if or (.Values.files) (.Values.secretFilesRefs) }}
-      volumes:
-        {{- if .Values.files }}
-        - name: config
-          configMap:
-            name: {{ include "common-app.fullname" . }}
-        {{- end }}
-        {{- range .Values.secretFilesRefs }}
-        - name: {{ .volume }}
-          secret:
-            secretName: {{ .name }}
-        {{- end }}
-      {{- end }}
-  backoffLimit: {{ .Values.backoffLimit }}
-{{- end -}}
diff --git a/charts/common-app/templates/_deployment.yaml b/charts/common-app/templates/_deployment.yaml
deleted file mode 100644
index 8111c6946..000000000
--- a/charts/common-app/templates/_deployment.yaml
+++ /dev/null
@@ -1,282 +0,0 @@
-{{- define "common-app.deployment" -}}
-{{- $root := . -}}
-{{- if and .Values.persistence.enabled (not .Values.persistence.size) }}
-{{- fail "When persistence is enabled, you must set .Values.persistence.size" }}
-{{- end }}
-
-{{- if .Capabilities.APIVersions.Has "apps/v1" }}
-apiVersion: apps/v1
-{{- else }}
-apiVersion: apps/v1beta1
-{{- end }}
-{{- if .Values.statefulSet }}
-kind: StatefulSet
-{{- else }}
-kind: Deployment
-{{- end }}
-metadata:
-  name: {{ include "common-app.fullname" . }}
-  {{- include "common-app.deployment-annotations" . }}
-  labels:
-    {{- include "common-app.labels" . | nindent 4 }}
-    streams-bootstrap/kind: {{ .Chart.Name }}
-    {{- range $key, $value := .Values.labels }}
-    {{ $key | quote }}: {{ $value | quote }}
-    {{- end }}
-spec:
-  {{- if .Values.statefulSet }}
-  serviceName: {{ include "common-app.fullname" . }}
-  podManagementPolicy: Parallel
-  {{- end }}
-  {{- if (not .Values.autoscaling.enabled) }}
-  replicas: {{ .Values.replicaCount }}
-  {{- end }}
-  selector:
-    matchLabels:
-      {{- include "common-app.selectorLabels" . | nindent 6 }}
-  template:
-    metadata:
-    {{- if or .Values.podAnnotations .Values.files }}
-      annotations:
-      {{- if .Values.files }}
-        checksum/config: {{ include (print $.Template.BasePath "/configmap.yaml") . | sha256sum }}
-      {{- end }}
-      {{- range $key, $value := .Values.podAnnotations }}
-        {{ $key | quote }}: {{ $value | quote }}
-      {{- end }}
-    {{- end }}
-      labels:
-        {{- include "common-app.selectorLabels" . | nindent 8 }}
-        streams-bootstrap/kind: {{ .Chart.Name }}
-        {{- range $key, $value := .Values.podLabels }}
-        {{ $key }}: {{ $value }}
-        {{- end }}
-    spec:
-      {{- if .Values.serviceAccountName }}
-      serviceAccountName: {{ .Values.serviceAccountName }}
-      {{- end }}
-      {{- if .Values.tolerations }}
-      tolerations:
-{{ toYaml .Values.tolerations | indent 8 }}
-      {{- end }}
-      {{- with .Values.affinity }}
-      affinity:
-        {{- tpl (toYaml .) $root | nindent 8 }}
-      {{- end }}
-      {{- if .Values.priorityClassName }}
-      priorityClassName: {{ .Values.priorityClassName }}
-      {{- end }}
-      terminationGracePeriodSeconds: {{ .Values.terminationGracePeriodSeconds }}
-      {{- if .Values.imagePullSecrets }}
-      imagePullSecrets:
-{{- toYaml .Values.imagePullSecrets | nindent 8 }}
-      {{- end }}
-      containers:
-        - name: "kafka-app"
-          image: "{{ .Values.image }}:{{ .Values.imageTag }}"
-          imagePullPolicy: "{{ .Values.imagePullPolicy }}"
-          resources:
-{{ toYaml .Values.resources | indent 12 }}
-          env:
-            - name: ENV_PREFIX
-              value: {{ .Values.configurationEnvPrefix }}_
-          {{- range $key, $value := .Values.kafka.config }}
-            - name: {{ printf "KAFKA_%s" $key | replace "." "_" | upper | quote }}
-              value: {{ $value | quote }}
-          {{- end }}
-          {{- range .Values.ports }}
-            {{- if .servicePort }} # TODO verify that there is at most one service port. Currently, if there are multiple service ports, the first one will be used
-            - name: POD_IP
-              valueFrom:
-                fieldRef:
-                  fieldPath: status.podIP
-            - name: KAFKA_APPLICATION_SERVER
-              value: "$(POD_IP):{{ .containerPort }}"
-            {{- end }}
-          {{- end }}
-          {{- if .Values.kafka.staticMembership }}
-            - name: KAFKA_GROUP_INSTANCE_ID
-              valueFrom:
-                fieldRef:
-                  fieldPath: metadata.name
-          {{- end }}
-          {{- if not .Values.statefulSet }}
-            - name: "{{ .Values.configurationEnvPrefix }}_VOLATILE_GROUP_INSTANCE_ID"
-              value: "true"
-          {{- end }}
-          {{- if hasKey .Values.kafka "bootstrapServers" }}
-            - name: "{{ .Values.configurationEnvPrefix }}_BOOTSTRAP_SERVERS"
-              value: {{ .Values.kafka.bootstrapServers | quote }}
-          {{- end }}
-          {{- if hasKey .Values.kafka "schemaRegistryUrl" }}
-            - name: "{{ .Values.configurationEnvPrefix }}_SCHEMA_REGISTRY_URL"
-              value: {{ .Values.kafka.schemaRegistryUrl | quote }}
-          {{- end }}
-          {{- if and (hasKey .Values.kafka "inputTopics") (.Values.kafka.inputTopics) }}
-            - name: "{{ .Values.configurationEnvPrefix }}_INPUT_TOPICS"
-              value: {{ .Values.kafka.inputTopics | join "," | quote }}
-          {{- end }}
-          {{- if hasKey .Values.kafka "inputPattern" }}
-            - name: "{{ .Values.configurationEnvPrefix }}_INPUT_PATTERN"
-              value: {{ .Values.kafka.inputPattern | quote }}
-          {{- end }}
-          {{- if hasKey .Values.kafka "outputTopic" }}
-            - name: "{{ .Values.configurationEnvPrefix }}_OUTPUT_TOPIC"
-              value: {{ .Values.kafka.outputTopic | quote }}
-          {{- end }}
-          {{- if hasKey .Values.kafka "errorTopic" }}
-            - name: "{{ .Values.configurationEnvPrefix }}_ERROR_TOPIC"
-              value: {{ .Values.kafka.errorTopic | quote }}
-          {{- end }}
-          {{- if and (hasKey .Values.kafka "labeledOutputTopics") (.Values.kafka.labeledOutputTopics) }}
-            - name: "{{ .Values.configurationEnvPrefix }}_LABELED_OUTPUT_TOPICS"
-              value: "{{- range $key, $value := .Values.kafka.labeledOutputTopics }}{{ $key }}={{ $value }},{{- end }}"
-          {{- end }}
-          {{- $delimiter := ";" }}
-          {{- if and (hasKey .Values.kafka "labeledInputTopics") (.Values.kafka.labeledInputTopics) }}
-            - name: "{{ .Values.configurationEnvPrefix }}_LABELED_INPUT_TOPICS"
-              value: "{{- range $key, $value := .Values.kafka.labeledInputTopics }}{{ $key }}={{ $value | join $delimiter }},{{- end }}"
-          {{- end }}
-          {{- if and (hasKey .Values.kafka "labeledInputPatterns") (.Values.kafka.labeledInputPatterns) }}
-            - name: "{{ .Values.configurationEnvPrefix }}_LABELED_INPUT_PATTERNS"
-              value: "{{- range $key, $value := .Values.kafka.labeledInputPatterns }}{{ $key }}={{ $value }},{{- end }}"
-          {{- end }}
-          {{- if hasKey .Values.kafka "applicationId" }}
-            - name: "{{ .Values.configurationEnvPrefix }}_APPLICATION_ID"
-              value: {{ .Values.kafka.applicationId | quote }}
-          {{- end }}
-          {{- if hasKey .Values.kafka "groupId" }}
-            - name: "{{ .Values.configurationEnvPrefix }}_GROUP_ID"
-              value: {{ .Values.kafka.groupId | quote }}
-          {{- end }}
-          {{- range $key, $value := .Values.secrets }}
-            - name: "{{ $key }}"
-              valueFrom:
-                secretKeyRef:
-                  name: {{ include "common-app.fullname" . }}
-                  key: "{{ $key }}"
-          {{- end }}
-          {{- range $key, $value := .Values.secretRefs }}
-            - name: "{{ $key }}"
-              valueFrom:
-                secretKeyRef:
-                  name: {{ $value.name }}
-                  key: "{{ $value.key }}"
-          {{- end }}
-          {{- range $key, $value := .Values.commandLine }}
-            - name: "{{ $root.Values.configurationEnvPrefix }}_{{ $key }}"
-              value: {{ $value | quote }}
-          {{- end }}
-          {{- range $key, $value := .Values.env }}
-            - name: {{ $key | quote }}
-              value: {{ $value | quote }}
-          {{- end }}
-            - name: JAVA_TOOL_OPTIONS
-              value: '-Dcom.sun.management.jmxremote.port={{ .Values.jmx.port }}
-                -Dcom.sun.management.jmxremote.authenticate=false
-                -Dcom.sun.management.jmxremote.ssl=false
-                {{- if .Values.jmx.enabled }}
-                -Djava.rmi.server.hostname={{ .Values.jmx.host }}
-                -Dcom.sun.management.jmxremote.rmi.port={{ .Values.jmx.port }}
-                {{- end }}
-                -XX:MaxRAMPercentage={{ printf "%.1f" .Values.javaOptions.maxRAMPercentage }}
-                {{ .Values.javaOptions.others | join " " }}'
-        {{- if or (.Values.files) (and .Values.persistence.enabled .Values.statefulSet) (.Values.secretFilesRefs) }}
-          volumeMounts:
-            {{- range $key, $value := .Values.files }}
-            - name: config
-              mountPath: {{ printf "%s/%s" $value.mountPath $key | quote }}
-              subPath: {{ $key | quote }}
-            {{- end }}
-            {{- range .Values.secretFilesRefs }}
-            - name: {{ .volume }}
-              mountPath: {{ .mountPath }}
-              {{- if .readOnly }}
-              readOnly: true
-              {{- end }}
-              {{- if .subPath}}
-              subPath: {{.subPath }}
-              {{- end }}
-            {{- end }}
-            {{- if and .Values.persistence.enabled .Values.statefulSet }}
-            - name: datadir
-              mountPath: /tmp/kafka-streams
-            {{- end }}
-        {{- end }}
-        {{- if or (.Values.jmx.enabled) (.Values.ports) }}
-          ports:
-            {{- range .Values.ports }}
-            - containerPort: {{ .containerPort }}
-              name: {{ .name | quote }}
-              {{- if .protocol }}
-              protocol: {{ .protocol | quote }}
-              {{- end }}
-            {{- end }}
-            {{- if .Values.jmx.enabled }}
-            - containerPort: {{ .Values.jmx.port }}
-              name: jmx
-            {{- end }}
-        {{- end }}
-          {{- if .Values.livenessProbe }}
-          livenessProbe:
-          {{- .Values.livenessProbe | toYaml | nindent 12 }}
-          {{- end }}
-          {{- if .Values.readinessProbe }}
-          readinessProbe:
-          {{- .Values.readinessProbe | toYaml | nindent 12 }}
-          {{- end }}
-      {{- if .Values.prometheus.jmx.enabled }}
-        - name: prometheus-jmx-exporter
-          image: "{{ .Values.prometheus.jmx.image }}:{{ .Values.prometheus.jmx.imageTag }}"
-          imagePullPolicy: "{{ .Values.prometheus.jmx.imagePullPolicy }}"
-          args:
-            - {{ .Values.prometheus.jmx.port | quote }}
-            - /etc/jmx-{{ include "common-app.name" . }}/jmx-kafka-{{ include "common-app.name" . }}-prometheus.yml
-          env:
-            - name: JAVA_TOOL_OPTIONS
-              value: "-XX:MaxRAMPercentage=90.0"
-          ports:
-            - containerPort: {{ .Values.prometheus.jmx.port }}
-              name: prometheus
-          resources:
-{{ toYaml .Values.prometheus.jmx.resources | indent 12 }}
-          volumeMounts:
-            - name: jmx-config
-              mountPath: /etc/jmx-{{ include "common-app.name" . }}
-      {{- end }}
-      {{- if or (.Values.prometheus.jmx.enabled) (.Values.files) (.Values.secretFilesRefs) }}
-      volumes:
-        {{- if .Values.prometheus.jmx.enabled }}
-        - name: jmx-config
-          configMap:
-            name: {{ include "common-app.fullname" . }}-jmx
-        {{- end }}
-        {{- if .Values.files }}
-        - name: config
-          configMap:
-            name: {{ include "common-app.fullname" . }}
-        {{- end }}
-        {{- range .Values.secretFilesRefs }}
-        - name: {{ .volume }}
-          secret:
-            secretName: {{ .name }}
-        {{- end }}
-      {{- end }}
-  {{- if and .Values.persistence.enabled .Values.statefulSet }}
-  volumeClaimTemplates:
-    - metadata:
-        name: datadir
-      spec:
-        accessModes: [ "ReadWriteOnce" ]
-        resources:
-          requests:
-            storage: "{{ .Values.persistence.size }}"
-        {{- if .Values.persistence.storageClass }}
-        {{- if (eq "-" .Values.persistence.storageClass) }}
-        storageClassName: ""
-        {{- else }}
-        storageClassName: "{{ .Values.persistence.storageClass }}"
-        {{- end }}
-        {{- end }}
-  {{- end }}
-{{- end -}}
diff --git a/charts/common-app/templates/_helpers.tpl b/charts/common-app/templates/_helpers.tpl
index b00be4c74..8d2c35937 100644
--- a/charts/common-app/templates/_helpers.tpl
+++ b/charts/common-app/templates/_helpers.tpl
@@ -84,3 +84,336 @@ Includes default annotations and conditionally adds consumerGroup if applicable.
 {{- end }}
 {{- end }}
 
+{{- define "common-app.common-env" -}}
+{{- $root := . -}}
+- name: ENV_PREFIX
+  value: {{ .Values.configurationEnvPrefix }}_
+{{- range $key, $value := .Values.kafka.config }}
+- name: {{ printf "KAFKA_%s" $key | replace "." "_" | upper | quote }}
+  value: {{ $value | quote }}
+{{- end }}
+{{- if hasKey .Values.kafka "bootstrapServers" }}
+- name: "{{ .Values.configurationEnvPrefix }}_BOOTSTRAP_SERVERS"
+  value: {{ .Values.kafka.bootstrapServers | quote }}
+{{- end }}
+{{- if hasKey .Values.kafka "schemaRegistryUrl" }}
+- name: "{{ .Values.configurationEnvPrefix }}_SCHEMA_REGISTRY_URL"
+  value: {{ .Values.kafka.schemaRegistryUrl | quote }}
+{{- end }}
+{{- range $key, $value := .Values.secrets }}
+- name: "{{ $key }}"
+  valueFrom:
+    secretKeyRef:
+      name: {{ include "common-app.fullname" $root }}
+      key: "{{ $key }}"
+{{- end }}
+{{- range $key, $value := .Values.secretRefs }}
+- name: "{{ $key }}"
+  valueFrom:
+    secretKeyRef:
+      name: {{ $value.name }}
+      key: "{{ $value.key }}"
+{{- end }}
+{{- range $key, $value := .Values.commandLine }}
+- name: "{{ $root.Values.configurationEnvPrefix }}_{{ $key }}"
+  value: {{ $value | quote }}
+{{- end }}
+{{- range $key, $value := .Values.env }}
+- name: {{ $key | quote }}
+  value: {{ $value | quote }}
+{{- end }}
+{{- end }}
+
+{{- define "common-app.input-env" -}}
+{{- if and (hasKey .Values.kafka "inputTopics") (.Values.kafka.inputTopics) }}
+- name: "{{ .Values.configurationEnvPrefix }}_INPUT_TOPICS"
+  value: {{ .Values.kafka.inputTopics | join "," | quote }}
+{{- end }}
+{{- if hasKey .Values.kafka "inputPattern" }}
+- name: "{{ .Values.configurationEnvPrefix }}_INPUT_PATTERN"
+  value: {{ .Values.kafka.inputPattern | quote }}
+{{- end }}
+{{- $delimiter := ";" }}
+{{- if and (hasKey .Values.kafka "labeledInputTopics") (.Values.kafka.labeledInputTopics) }}
+- name: "{{ .Values.configurationEnvPrefix }}_LABELED_INPUT_TOPICS"
+  value: "{{- range $key, $value := .Values.kafka.labeledInputTopics }}{{ $key }}={{ $value | join $delimiter }},{{- end }}"
+{{- end }}
+{{- if and (hasKey .Values.kafka "labeledInputPatterns") (.Values.kafka.labeledInputPatterns) }}
+- name: "{{ .Values.configurationEnvPrefix }}_LABELED_INPUT_PATTERNS"
+  value: "{{- range $key, $value := .Values.kafka.labeledInputPatterns }}{{ $key }}={{ $value }},{{- end }}"
+{{- end }}
+{{- end }}
+
+{{- define "common-app.output-env" -}}
+{{- if hasKey .Values.kafka "outputTopic" }}
+- name: "{{ .Values.configurationEnvPrefix }}_OUTPUT_TOPIC"
+  value: {{ .Values.kafka.outputTopic | quote }}
+{{- end }}
+{{- if and (hasKey .Values.kafka "labeledOutputTopics") (.Values.kafka.labeledOutputTopics) }}
+- name: "{{ .Values.configurationEnvPrefix }}_LABELED_OUTPUT_TOPICS"
+  value: "{{- range $key, $value := .Values.kafka.labeledOutputTopics }}{{ $key }}={{ $value }},{{- end }}"
+{{- end }}
+{{- end }}
+
+{{- define "common-app.error-env" -}}
+{{- if hasKey .Values.kafka "errorTopic" }}
+- name: "{{ .Values.configurationEnvPrefix }}_ERROR_TOPIC"
+  value: {{ .Values.kafka.errorTopic | quote }}
+{{- end }}
+{{- end }}
+
+{{- define "common-app.group-instance-id-env" -}}
+{{- if .Values.kafka.staticMembership }}
+- name: KAFKA_GROUP_INSTANCE_ID
+  valueFrom:
+    fieldRef:
+      fieldPath: metadata.name
+{{- end }}
+{{- if not .Values.statefulSet }}
+- name: "{{ .Values.configurationEnvPrefix }}_VOLATILE_GROUP_INSTANCE_ID"
+  value: "true"
+{{- end }}
+{{- end }}
+
+{{- define "common-app.group-id-env" -}}
+{{- if hasKey .Values.kafka "groupId" }}
+- name: "{{ .Values.configurationEnvPrefix }}_GROUP_ID"
+  value: {{ .Values.kafka.groupId | quote }}
+{{- end }}
+{{- end }}
+
+{{- define "common-app.application-id-env" -}}
+{{- if hasKey .Values.kafka "applicationId" }}
+- name: "{{ .Values.configurationEnvPrefix }}_APPLICATION_ID"
+  value: {{ .Values.kafka.applicationId | quote }}
+{{- end }}
+{{- end }}
+
+{{- define "common-app.volume-mounts" -}}
+{{- range $key, $value := .Values.files }}
+- name: config
+  mountPath: {{ printf "%s/%s" $value.mountPath $key | quote }}
+  subPath: {{ $key | quote }}
+{{- end }}
+{{- range .Values.secretFilesRefs }}
+- name: {{ .volume }}
+  mountPath: {{ .mountPath }}
+  {{- if .readOnly }}
+  readOnly: true
+  {{- end }}
+  {{- if .subPath}}
+  subPath: {{.subPath }}
+  {{- end }}
+{{- end }}
+{{- end }}
+
+{{- define "common-app.volumes" -}}
+{{- if .Values.files }}
+- name: config
+  configMap:
+    name: {{ include "common-app.fullname" . }}
+{{- end }}
+{{- range .Values.secretFilesRefs }}
+- name: {{ .volume }}
+  secret:
+    secretName: {{ .name }}
+{{- end }}
+{{- end }}
+
+{{- define "common-app.jmx-volume" -}}
+{{- if .Values.prometheus.jmx.enabled }}
+- name: jmx-config
+  configMap:
+    name: {{ include "common-app.fullname" . }}-jmx
+{{- end }}
+{{- end }}
+
+{{- define "common-app.ports" -}}
+{{- range .Values.ports }}
+- containerPort: {{ .containerPort }}
+  name: {{ .name | quote }}
+  {{- if .protocol }}
+  protocol: {{ .protocol | quote }}
+  {{- end }}
+{{- end }}
+{{- end }}
+
+{{- define "common-app.jmx-port" -}}
+{{- if .Values.jmx.enabled }}
+- containerPort: {{ .Values.jmx.port }}
+  name: jmx
+{{- end }}
+{{- end }}
+
+{{- define "common-app.common-pod-spec" -}}
+{{- $root := . -}}
+  {{- if .Values.serviceAccountName }}
+  serviceAccountName: {{ .Values.serviceAccountName }}
+  {{- end }}
+  {{- if .Values.tolerations }}
+  tolerations:
+{{ toYaml .Values.tolerations | indent 4 }}
+  {{- end }}
+  {{- with .Values.affinity }}
+  affinity:
+    {{- tpl (toYaml .) $root | nindent 4 }}
+  {{- end }}
+  {{- if .Values.imagePullSecrets }}
+  imagePullSecrets:
+{{- toYaml .Values.imagePullSecrets | nindent 4 }}
+  {{- end }}
+{{- end }}
+
+{{- define "common-app.cleanup-pod-spec" -}}
+{{- include "common-app.common-pod-spec" . }}
+  restartPolicy: {{ .Values.restartPolicy }}
+  {{- if or (.Values.files) (.Values.secretFilesRefs) }}
+  volumes:
+    {{- include "common-app.volumes" . | nindent 8 }}
+  {{- end }}
+{{- end }}
+
+{{- define "common-app.pod-spec" -}}
+{{- include "common-app.common-pod-spec" . }}
+  {{- if .Values.priorityClassName }}
+  priorityClassName: {{ .Values.priorityClassName }}
+  {{- end }}
+  terminationGracePeriodSeconds: {{ .Values.terminationGracePeriodSeconds }}
+{{- end }}
+
+{{- define "common-app.pod-metadata" -}}
+{{- if or .Values.podAnnotations .Values.files }}
+  annotations:
+  {{- if .Values.files }}
+    checksum/config: {{ include (print $.Template.BasePath "/configmap.yaml") . | sha256sum }}
+  {{- end }}
+  {{- range $key, $value := .Values.podAnnotations }}
+    {{ $key | quote }}: {{ $value | quote }}
+  {{- end }}
+{{- end }}
+  labels:
+    {{- include "common-app.selectorLabels" . | nindent 4 }}
+    streams-bootstrap/kind: {{ .Chart.Name }}
+    {{- range $key, $value := .Values.podLabels }}
+    {{ $key }}: {{ $value }}
+    {{- end }}
+{{- end }}
+
+{{- define "common-app.cleanup-pod-metadata" -}}
+{{- if .Values.podAnnotations }}
+  annotations:
+  {{- range $key, $value := .Values.podAnnotations }}
+    {{ $key | quote }}: {{ $value | quote }}
+  {{- end }}
+{{- end }}
+  labels:
+    {{- include "common-app.selectorLabels" . | nindent 4 }}
+    {{- range $key, $value := .Values.podLabels }}
+    {{ $key }}: {{ $value }}
+    {{- end }}
+{{- end }}
+
+{{- define "common-app.common-kafka-container" -}}
+- name: "kafka-app"
+  image: "{{ .Values.image }}:{{ .Values.imageTag }}"
+  imagePullPolicy: "{{ .Values.imagePullPolicy }}"
+  resources:
+{{ toYaml .Values.resources | indent 4 }}
+{{- end }}
+
+{{- define "common-app.kafka-container" -}}
+{{- include "common-app.common-kafka-container" . }}
+  {{- if .Values.livenessProbe }}
+  livenessProbe:
+{{- .Values.livenessProbe | toYaml | nindent 4 }}
+  {{- end }}
+  {{- if .Values.readinessProbe }}
+  readinessProbe:
+{{- .Values.readinessProbe | toYaml | nindent 4 }}
+  {{- end }}
+{{- end }}
+
+{{- define "common-app.java-tool-options" -}}
+-XX:MaxRAMPercentage={{ printf "%.1f" .Values.javaOptions.maxRAMPercentage }}
+{{ .Values.javaOptions.others | join " " }}
+{{- end }}
+
+{{- define "common-app.java-tool-jmx-options" -}}
+-Dcom.sun.management.jmxremote.port={{ .Values.jmx.port }}
+-Dcom.sun.management.jmxremote.authenticate=false
+-Dcom.sun.management.jmxremote.ssl=false
+{{- if .Values.jmx.enabled }}
+-Djava.rmi.server.hostname={{ .Values.jmx.host }}
+-Dcom.sun.management.jmxremote.rmi.port={{ .Values.jmx.port }}
+{{- end }}
+{{- end }}
+
+{{- define "common-app.cleanup-job-spec" -}}
+ttlSecondsAfterFinished: 30
+backoffLimit: {{ .Values.backoffLimit }}
+{{- end }}
+
+{{- define "common-app.deployment-spec" -}}
+  {{- if .Values.statefulSet }}
+  serviceName: {{ include "common-app.fullname" . }}
+  podManagementPolicy: Parallel
+  {{- end }}
+  {{- if (not .Values.autoscaling.enabled) }}
+  replicas: {{ .Values.replicaCount }}
+  {{- end }}
+  selector:
+    matchLabels:
+      {{- include "common-app.selectorLabels" . | nindent 6 }}
+  {{- if and .Values.persistence.enabled .Values.statefulSet }}
+  volumeClaimTemplates:
+    - metadata:
+        name: datadir
+      spec:
+        accessModes: [ "ReadWriteOnce" ]
+        resources:
+          requests:
+            storage: "{{ .Values.persistence.size }}"
+        {{- if .Values.persistence.storageClass }}
+        {{- if (eq "-" .Values.persistence.storageClass) }}
+        storageClassName: ""
+        {{- else }}
+        storageClassName: "{{ .Values.persistence.storageClass }}"
+        {{- end }}
+        {{- end }}
+  {{- end }}
+{{- end }}
+
+{{- define "common-app.deployment" -}}
+{{- if .Capabilities.APIVersions.Has "apps/v1" }}
+apiVersion: apps/v1
+{{- else }}
+apiVersion: apps/v1beta1
+{{- end }}
+{{- if .Values.statefulSet }}
+kind: StatefulSet
+{{- else }}
+kind: Deployment
+{{- end }}
+metadata:
+  name: {{ include "common-app.fullname" . }}
+  {{- include "common-app.deployment-annotations" . }}
+  labels:
+    {{- include "common-app.labels" . | nindent 4 }}
+    streams-bootstrap/kind: {{ .Chart.Name }}
+    {{- range $key, $value := .Values.labels }}
+    {{ $key | quote }}: {{ $value | quote }}
+    {{- end }}
+{{- end }}
+
+{{- define "common-app.cleanup-job" -}}
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: {{ include "common-app.fullname" . }}
+  {{- include "common-app.annotations" . }}
+  labels:
+    {{- include "common-app.labels" . | nindent 4 }}
+    {{- range $key, $value := .Values.labels }}
+    {{ $key | quote }}: {{ $value | quote }}
+    {{- end }}
+{{- end }}
diff --git a/charts/common-app/templates/_jmx-configmap.yaml b/charts/common-app/templates/_jmx-configmap.yaml
index 2a8897f17..67220a085 100644
--- a/charts/common-app/templates/_jmx-configmap.yaml
+++ b/charts/common-app/templates/_jmx-configmap.yaml
@@ -11,7 +11,7 @@ metadata:
     {{ $key | quote }}: {{ $value | quote }}
     {{- end }}
 data:
-  jmx-kafka-{{ include "common-app.name" . }}-prometheus.yml: |+
+  jmx-kafka-{{ .Chart.Name }}-prometheus.yml: |+
     jmxUrl: service:jmx:rmi:///jndi/rmi://localhost:{{ .Values.jmx.port }}/jmxrmi
     lowercaseOutputName: true
     lowercaseOutputLabelNames: true
diff --git a/charts/common-app/templates/_prometheus-container.yaml b/charts/common-app/templates/_prometheus-container.yaml
new file mode 100644
index 000000000..c85f226a9
--- /dev/null
+++ b/charts/common-app/templates/_prometheus-container.yaml
@@ -0,0 +1,19 @@
+{{- define "common-app.prometheus-container" -}}
+- name: prometheus-jmx-exporter
+  image: "{{ .Values.prometheus.jmx.image }}:{{ .Values.prometheus.jmx.imageTag }}"
+  imagePullPolicy: "{{ .Values.prometheus.jmx.imagePullPolicy }}"
+  args:
+    - {{ .Values.prometheus.jmx.port | quote }}
+    - /etc/jmx-{{ .Chart.Name }}/jmx-kafka-{{ .Chart.Name }}-prometheus.yml
+  env:
+    - name: JAVA_TOOL_OPTIONS
+      value: "-XX:MaxRAMPercentage=90.0"
+  ports:
+    - containerPort: {{ .Values.prometheus.jmx.port }}
+      name: prometheus
+  resources:
+{{ toYaml .Values.prometheus.jmx.resources | indent 4 }}
+  volumeMounts:
+    - name: jmx-config
+      mountPath: /etc/jmx-{{ .Chart.Name }}
+{{- end -}}
diff --git a/charts/consumer-app-cleanup-job/templates/job.yaml b/charts/consumer-app-cleanup-job/templates/job.yaml
index fc2a74b3c..e02d46409 100644
--- a/charts/consumer-app-cleanup-job/templates/job.yaml
+++ b/charts/consumer-app-cleanup-job/templates/job.yaml
@@ -1 +1,26 @@
-{{- include "common-app.cleanup-job" . -}}
+{{- include "common-app.cleanup-job" . }}
+spec:
+{{- include "common-app.cleanup-job-spec" . | nindent 2 }}
+  template:
+    metadata:
+{{- include "common-app.cleanup-pod-metadata" . | indent 4 }}
+    spec:
+{{- include "common-app.cleanup-pod-spec" . | indent 4 }}
+      containers:
+        {{- include "common-app.common-kafka-container" . | nindent 8 }}
+          args:
+              {{- if .Values.kafka.deleteOutput }}
+            - clean
+              {{- else }}
+            - reset
+              {{- end }}
+          env:
+            {{- include "common-app.common-env" . | nindent 12 }}
+            {{- include "common-app.input-env" . | indent 12 }}
+            {{- include "common-app.group-id-env" . | indent 12 }}
+            - name: JAVA_TOOL_OPTIONS
+              value: '{{ include "common-app.java-tool-options" . }}'
+          {{- if or (.Values.files) (.Values.secretFilesRefs) }}
+          volumeMounts:
+              {{- include "common-app.volume-mounts" . | nindent 12 }}
+          {{- end }}
diff --git a/charts/consumer-app/templates/deployment.yaml b/charts/consumer-app/templates/deployment.yaml
index 385f927bc..f3d511c91 100644
--- a/charts/consumer-app/templates/deployment.yaml
+++ b/charts/consumer-app/templates/deployment.yaml
@@ -1,176 +1,28 @@
-{{- $root := . -}}
 {{- if and .Values.persistence.enabled (not .Values.persistence.size) }}
 {{- fail "When persistence is enabled, you must set .Values.persistence.size" }}
 {{- end }}
 
-{{- if .Capabilities.APIVersions.Has "apps/v1" }}
-apiVersion: apps/v1
-{{- else }}
-apiVersion: apps/v1beta1
-{{- end }}
-{{- if .Values.statefulSet }}
-kind: StatefulSet
-{{- else }}
-kind: Deployment
-{{- end }}
-metadata:
-  name: {{ include "common-app.fullname" . }}
-  {{- include "common-app.deployment-annotations" . }}
-  labels:
-    {{- include "common-app.labels" . | nindent 4 }}
-    streams-bootstrap/kind: {{ .Chart.Name }}
-    {{- range $key, $value := .Values.labels }}
-    {{ $key | quote }}: {{ $value | quote }}
-    {{- end }}
+{{- include "common-app.deployment" . }}
 spec:
-  {{- if .Values.statefulSet }}
-  serviceName: {{ include "common-app.fullname" . }}
-  podManagementPolicy: Parallel
-  {{- end }}
-  {{- if (not .Values.autoscaling.enabled) }}
-  replicas: {{ .Values.replicaCount }}
-  {{- end }}
-  selector:
-    matchLabels:
-      {{- include "common-app.selectorLabels" . | nindent 6 }}
+{{- include "common-app.deployment-spec" . }}
   template:
     metadata:
-    {{- if or .Values.podAnnotations .Values.files }}
-      annotations:
-      {{- if .Values.files }}
-        checksum/config: {{ include (print $.Template.BasePath "/configmap.yaml") . | sha256sum }}
-      {{- end }}
-      {{- range $key, $value := .Values.podAnnotations }}
-        {{ $key | quote }}: {{ $value | quote }}
-      {{- end }}
-    {{- end }}
-      labels:
-        {{- include "common-app.selectorLabels" . | nindent 8 }}
-        streams-bootstrap/kind: {{ .Chart.Name }}
-        {{- range $key, $value := .Values.podLabels }}
-        {{ $key }}: {{ $value }}
-        {{- end }}
+{{- include "common-app.pod-metadata" . | indent 4 }}
     spec:
-      {{- if .Values.serviceAccountName }}
-      serviceAccountName: {{ .Values.serviceAccountName }}
-      {{- end }}
-      {{- if .Values.tolerations }}
-      tolerations:
-{{ toYaml .Values.tolerations | indent 8 }}
-      {{- end }}
-      {{- with .Values.affinity }}
-      affinity:
-        {{- tpl (toYaml .) $root | nindent 8 }}
-      {{- end }}
-      {{- if .Values.priorityClassName }}
-      priorityClassName: {{ .Values.priorityClassName }}
-      {{- end }}
-      terminationGracePeriodSeconds: {{ .Values.terminationGracePeriodSeconds }}
-      {{- if .Values.imagePullSecrets }}
-      imagePullSecrets:
-{{- toYaml .Values.imagePullSecrets | nindent 8 }}
-      {{- end }}
+{{- include "common-app.pod-spec" . | indent 4 }}
       containers:
-        - name: "kafka-app"
-          image: "{{ .Values.image }}:{{ .Values.imageTag }}"
-          imagePullPolicy: "{{ .Values.imagePullPolicy }}"
-          resources:
-{{ toYaml .Values.resources | indent 12 }}
+        {{- include "common-app.kafka-container" . | nindent 8 }}
           env:
-            - name: ENV_PREFIX
-              value: {{ .Values.configurationEnvPrefix }}_
-          {{- range $key, $value := .Values.kafka.config }}
-            - name: {{ printf "KAFKA_%s" $key | replace "." "_" | upper | quote }}
-              value: {{ $value | quote }}
-          {{- end }}
-          {{- if .Values.kafka.staticMembership }}
-            - name: KAFKA_GROUP_INSTANCE_ID
-              valueFrom:
-                fieldRef:
-                  fieldPath: metadata.name
-          {{- end }}
-          {{- if not .Values.statefulSet }}
-            - name: "{{ .Values.configurationEnvPrefix }}_VOLATILE_GROUP_INSTANCE_ID"
-              value: "true"
-          {{- end }}
-          {{- if hasKey .Values.kafka "bootstrapServers" }}
-            - name: "{{ .Values.configurationEnvPrefix }}_BOOTSTRAP_SERVERS"
-              value: {{ .Values.kafka.bootstrapServers | quote }}
-          {{- end }}
-          {{- if hasKey .Values.kafka "schemaRegistryUrl" }}
-            - name: "{{ .Values.configurationEnvPrefix }}_SCHEMA_REGISTRY_URL"
-              value: {{ .Values.kafka.schemaRegistryUrl | quote }}
-          {{- end }}
-          {{- if and (hasKey .Values.kafka "inputTopics") (.Values.kafka.inputTopics) }}
-            - name: "{{ .Values.configurationEnvPrefix }}_INPUT_TOPICS"
-              value: {{ .Values.kafka.inputTopics | join "," | quote }}
-          {{- end }}
-          {{- if hasKey .Values.kafka "inputPattern" }}
-            - name: "{{ .Values.configurationEnvPrefix }}_INPUT_PATTERN"
-              value: {{ .Values.kafka.inputPattern | quote }}
-          {{- end }}
-          {{- $delimiter := ";" }}
-          {{- if and (hasKey .Values.kafka "labeledInputTopics") (.Values.kafka.labeledInputTopics) }}
-            - name: "{{ .Values.configurationEnvPrefix }}_LABELED_INPUT_TOPICS"
-              value: "{{- range $key, $value := .Values.kafka.labeledInputTopics }}{{ $key }}={{ $value | join $delimiter }},{{- end }}"
-          {{- end }}
-          {{- if and (hasKey .Values.kafka "labeledInputPatterns") (.Values.kafka.labeledInputPatterns) }}
-            - name: "{{ .Values.configurationEnvPrefix }}_LABELED_INPUT_PATTERNS"
-              value: "{{- range $key, $value := .Values.kafka.labeledInputPatterns }}{{ $key }}={{ $value }},{{- end }}"
-          {{- end }}
-          {{- if hasKey .Values.kafka "groupId" }}
-            - name: "{{ .Values.configurationEnvPrefix }}_GROUP_ID"
-              value: {{ .Values.kafka.groupId | quote }}
-          {{- end }}
-          {{- range $key, $value := .Values.secrets }}
-            - name: "{{ $key }}"
-              valueFrom:
-                secretKeyRef:
-                  name: {{ include "common-app.fullname" . }}
-                  key: "{{ $key }}"
-          {{- end }}
-          {{- range $key, $value := .Values.secretRefs }}
-            - name: "{{ $key }}"
-              valueFrom:
-                secretKeyRef:
-                  name: {{ $value.name }}
-                  key: "{{ $value.key }}"
-          {{- end }}
-          {{- range $key, $value := .Values.commandLine }}
-            - name: "{{ $root.Values.configurationEnvPrefix }}_{{ $key }}"
-              value: {{ $value | quote }}
-          {{- end }}
-          {{- range $key, $value := .Values.env }}
-            - name: {{ $key | quote }}
-              value: {{ $value | quote }}
-          {{- end }}
+          {{- include "common-app.common-env" . | nindent 12 }}
+          {{- include "common-app.group-instance-id-env" . | indent 12 }}
+          {{- include "common-app.input-env" . | indent 12 }}
+          {{- include "common-app.group-id-env" . | indent 12 }}
             - name: JAVA_TOOL_OPTIONS
-              value: '-Dcom.sun.management.jmxremote.port={{ .Values.jmx.port }}
-                -Dcom.sun.management.jmxremote.authenticate=false
-                -Dcom.sun.management.jmxremote.ssl=false
-                {{- if .Values.jmx.enabled }}
-                -Djava.rmi.server.hostname={{ .Values.jmx.host }}
-                -Dcom.sun.management.jmxremote.rmi.port={{ .Values.jmx.port }}
-                {{- end }}
-                -XX:MaxRAMPercentage={{ printf "%.1f" .Values.javaOptions.maxRAMPercentage }}
-                {{ .Values.javaOptions.others | join " " }}'
+              value: '{{ include "common-app.java-tool-jmx-options" . }}
+                {{ include "common-app.java-tool-options" . }}'
         {{- if or (.Values.files) (and .Values.persistence.enabled .Values.statefulSet) (.Values.secretFilesRefs) }}
           volumeMounts:
-            {{- range $key, $value := .Values.files }}
-            - name: config
-              mountPath: {{ printf "%s/%s" $value.mountPath $key | quote }}
-              subPath: {{ $key | quote }}
-            {{- end }}
-            {{- range .Values.secretFilesRefs }}
-            - name: {{ .volume }}
-              mountPath: {{ .mountPath }}
-              {{- if .readOnly }}
-              readOnly: true
-              {{- end }}
-              {{- if .subPath}}
-              subPath: {{.subPath }}
-              {{- end }}
-            {{- end }}
+            {{- include "common-app.volume-mounts" . | indent 12 }}
             {{- if and .Values.persistence.enabled .Values.statefulSet }}
             - name: datadir
               mountPath: /tmp/kafka-consumer
@@ -178,77 +30,14 @@ spec:
         {{- end }}
         {{- if or (.Values.jmx.enabled) (.Values.ports) }}
           ports:
-            {{- range .Values.ports }}
-            - containerPort: {{ .containerPort }}
-              name: {{ .name | quote }}
-              {{- if .protocol }}
-              protocol: {{ .protocol | quote }}
-              {{- end }}
-            {{- end }}
-            {{- if .Values.jmx.enabled }}
-            - containerPort: {{ .Values.jmx.port }}
-              name: jmx
-            {{- end }}
+            {{- include "common-app.ports" . | indent 12 }}
+            {{- include "common-app.jmx-port" . | indent 12 }}
         {{- end }}
-          {{- if .Values.livenessProbe }}
-          livenessProbe:
-          {{- .Values.livenessProbe | toYaml | nindent 12 }}
-          {{- end }}
-          {{- if .Values.readinessProbe }}
-          readinessProbe:
-          {{- .Values.readinessProbe | toYaml | nindent 12 }}
-          {{- end }}
       {{- if .Values.prometheus.jmx.enabled }}
-        - name: prometheus-jmx-exporter
-          image: "{{ .Values.prometheus.jmx.image }}:{{ .Values.prometheus.jmx.imageTag }}"
-          imagePullPolicy: "{{ .Values.prometheus.jmx.imagePullPolicy }}"
-          args:
-            - {{ .Values.prometheus.jmx.port | quote }}
-            - /etc/jmx-consumer-app/jmx-kafka-consumer-app-prometheus.yml
-          env:
-            - name: JAVA_TOOL_OPTIONS
-              value: "-XX:MaxRAMPercentage=90.0"
-          ports:
-            - containerPort: {{ .Values.prometheus.jmx.port }}
-              name: prometheus
-          resources:
-{{ toYaml .Values.prometheus.jmx.resources | indent 12 }}
-          volumeMounts:
-            - name: jmx-config
-              mountPath: /etc/jmx-consumer-app
+{{- include "common-app.prometheus-container" . | nindent 8 }}
       {{- end }}
       {{- if or (.Values.prometheus.jmx.enabled) (.Values.files) (.Values.secretFilesRefs) }}
       volumes:
-        {{- if .Values.prometheus.jmx.enabled }}
-        - name: jmx-config
-          configMap:
-            name: {{ include "common-app.fullname" . }}-jmx
-        {{- end }}
-        {{- if .Values.files }}
-        - name: config
-          configMap:
-            name: {{ include "common-app.fullname" . }}
-        {{- end }}
-        {{- range .Values.secretFilesRefs }}
-        - name: {{ .volume }}
-          secret:
-            secretName: {{ .name }}
-        {{- end }}
+        {{- include "common-app.jmx-volume" . | indent 8 }}
+        {{- include "common-app.volumes" . | indent 8 }}
       {{- end }}
-  {{- if and .Values.persistence.enabled .Values.statefulSet }}
-  volumeClaimTemplates:
-    - metadata:
-        name: datadir
-      spec:
-        accessModes: [ "ReadWriteOnce" ]
-        resources:
-          requests:
-            storage: "{{ .Values.persistence.size }}"
-        {{- if .Values.persistence.storageClass }}
-        {{- if (eq "-" .Values.persistence.storageClass) }}
-        storageClassName: ""
-        {{- else }}
-        storageClassName: "{{ .Values.persistence.storageClass }}"
-        {{- end }}
-        {{- end }}
-  {{- end }}
diff --git a/charts/consumerproducer-app-cleanup-job/templates/job.yaml b/charts/consumerproducer-app-cleanup-job/templates/job.yaml
index fc2a74b3c..5608a31b5 100644
--- a/charts/consumerproducer-app-cleanup-job/templates/job.yaml
+++ b/charts/consumerproducer-app-cleanup-job/templates/job.yaml
@@ -1 +1,28 @@
-{{- include "common-app.cleanup-job" . -}}
+{{- include "common-app.cleanup-job" . }}
+spec:
+{{- include "common-app.cleanup-job-spec" . | nindent 2 }}
+  template:
+    metadata:
+{{- include "common-app.cleanup-pod-metadata" . | indent 4 }}
+    spec:
+{{- include "common-app.cleanup-pod-spec" . | indent 4 }}
+      containers:
+        {{- include "common-app.common-kafka-container" . | nindent 8 }}
+          args:
+              {{- if .Values.kafka.deleteOutput }}
+            - clean
+              {{- else }}
+            - reset
+              {{- end }}
+          env:
+            {{- include "common-app.common-env" . | nindent 12 }}
+            {{- include "common-app.input-env" . | indent 12 }}
+            {{- include "common-app.output-env" . | indent 12 }}
+            {{- include "common-app.error-env" . | indent 12 }}
+            {{- include "common-app.group-id-env" . | indent 12 }}
+            - name: JAVA_TOOL_OPTIONS
+              value: '{{ include "common-app.java-tool-options" . }}'
+          {{- if or (.Values.files) (.Values.secretFilesRefs) }}
+          volumeMounts:
+              {{- include "common-app.volume-mounts" . | nindent 12 }}
+          {{- end }}
diff --git a/charts/consumerproducer-app/templates/deployment.yaml b/charts/consumerproducer-app/templates/deployment.yaml
index db5ca18ed..4fde0623c 100644
--- a/charts/consumerproducer-app/templates/deployment.yaml
+++ b/charts/consumerproducer-app/templates/deployment.yaml
@@ -1 +1,55 @@
-{{- include "common-app.deployment" . -}}
+{{- if and .Values.persistence.enabled (not .Values.persistence.size) }}
+{{- fail "When persistence is enabled, you must set .Values.persistence.size" }}
+{{- end }}
+
+{{- include "common-app.deployment" . }}
+spec:
+{{- include "common-app.deployment-spec" . }}
+  template:
+    metadata:
+    {{- include "common-app.pod-metadata" . | indent 4 }}
+    spec:
+    {{- include "common-app.pod-spec" . | indent 4 }}
+      containers:
+        {{- include "common-app.kafka-container" . | nindent 8 }}
+          env:
+          {{- include "common-app.common-env" . | nindent 12 }}
+          {{- range .Values.ports }}
+            {{- if .servicePort }} # TODO verify that there is at most one service port. Currently, if there are multiple service ports, the first one will be used
+            - name: POD_IP
+              valueFrom:
+                fieldRef:
+                  fieldPath: status.podIP
+            - name: KAFKA_APPLICATION_SERVER
+              value: "$(POD_IP):{{ .containerPort }}"
+            {{- end }}
+          {{- end }}
+          {{- include "common-app.group-instance-id-env" . | indent 12 }}
+          {{- include "common-app.input-env" . | indent 12 }}
+          {{- include "common-app.output-env" . | indent 12 }}
+          {{- include "common-app.error-env" . | indent 12 }}
+          {{- include "common-app.group-id-env" . | indent 12 }}
+            - name: JAVA_TOOL_OPTIONS
+              value: '{{ include "common-app.java-tool-jmx-options" . }}
+                {{ include "common-app.java-tool-options" . }}'
+        {{- if or (.Values.files) (and .Values.persistence.enabled .Values.statefulSet) (.Values.secretFilesRefs) }}
+          volumeMounts:
+            {{- include "common-app.volume-mounts" . | indent 12 }}
+            {{- if and .Values.persistence.enabled .Values.statefulSet }}
+            - name: datadir
+              mountPath: /tmp/kafka-streams
+            {{- end }}
+        {{- end }}
+        {{- if or (.Values.jmx.enabled) (.Values.ports) }}
+          ports:
+            {{- include "common-app.ports" . | indent 12 }}
+            {{- include "common-app.jmx-port" . | indent 12 }}
+        {{- end }}
+      {{- if .Values.prometheus.jmx.enabled }}
+      {{- include "common-app.prometheus-container" . | nindent 8 }}
+      {{- end }}
+      {{- if or (.Values.prometheus.jmx.enabled) (.Values.files) (.Values.secretFilesRefs) }}
+      volumes:
+        {{- include "common-app.jmx-volume" . | indent 8 }}
+        {{- include "common-app.volumes" . | indent 8 }}
+      {{- end }}
diff --git a/charts/producer-app-cleanup-job/templates/job.yaml b/charts/producer-app-cleanup-job/templates/job.yaml
index d9ad528d0..40e6aa863 100644
--- a/charts/producer-app-cleanup-job/templates/job.yaml
+++ b/charts/producer-app-cleanup-job/templates/job.yaml
@@ -1,131 +1,21 @@
-{{- $root := . -}}
-apiVersion: batch/v1
-kind: Job
-metadata:
-  name: {{ include "common-app.fullname" . }}
-  {{- include "common-app.annotations" . }}
-  labels:
-    {{- include "common-app.labels" . | nindent 4 }}
-    {{- range $key, $value := .Values.labels }}
-    {{ $key | quote }}: {{ $value | quote }}
-    {{- end }}
+{{- include "common-app.cleanup-job" . }}
 spec:
-  ttlSecondsAfterFinished: 30
+{{- include "common-app.cleanup-job-spec" . | nindent 2 }}
   template:
     metadata:
-    {{- if .Values.podAnnotations }}
-      annotations:
-      {{- range $key, $value := .Values.podAnnotations }}
-        {{ $key | quote }}: {{ $value | quote }}
-      {{- end }}
-    {{- end }}
-      labels:
-        {{- include "common-app.selectorLabels" . | nindent 8 }}
-        {{- range $key, $value := .Values.podLabels }}
-        {{ $key }}: {{ $value }}
-        {{- end }}
+{{- include "common-app.cleanup-pod-metadata" . | indent 4 }}
     spec:
-      restartPolicy: {{ .Values.restartPolicy }}
-      {{- if .Values.serviceAccountName }}
-      serviceAccountName: {{ .Values.serviceAccountName }}
-      {{- end }}
-      {{- if .Values.tolerations }}
-      tolerations:
-{{ toYaml .Values.tolerations | indent 8 }}
-      {{- end }}
-      {{- with .Values.affinity }}
-      affinity:
-        {{- tpl (toYaml .) $root | nindent 8 }}
-      {{- end }}
-      {{- if .Values.imagePullSecrets }}
-      imagePullSecrets:
-{{- toYaml .Values.imagePullSecrets | nindent 8 }}
-      {{- end }}
+{{- include "common-app.cleanup-pod-spec" . | indent 4 }}
       containers:
-        - name: "kafka-app"
-          image: "{{ .Values.image }}:{{ .Values.imageTag }}"
-          imagePullPolicy: "{{ .Values.imagePullPolicy }}"
-          resources:
-{{ toYaml .Values.resources | indent 12 }}
+        {{- include "common-app.common-kafka-container" . | nindent 8 }}
           args:
             - clean
           env:
-            - name: ENV_PREFIX
-              value: {{ .Values.configurationEnvPrefix }}_
-            {{- range $key, $value := .Values.kafka.config }}
-            - name: {{ printf "KAFKA_%s" $key | replace "." "_" | upper | quote }}
-              value: {{ $value | quote }}
-            {{- end }}
-            {{- if hasKey .Values.kafka "bootstrapServers" }}
-            - name: "{{ .Values.configurationEnvPrefix }}_BOOTSTRAP_SERVERS"
-              value: {{ .Values.kafka.bootstrapServers | quote }}
-            {{- end }}
-            {{- if hasKey .Values.kafka "schemaRegistryUrl" }}
-            - name: "{{ .Values.configurationEnvPrefix }}_SCHEMA_REGISTRY_URL"
-              value: {{ .Values.kafka.schemaRegistryUrl | quote }}
-            {{- end }}
-            {{- if hasKey .Values.kafka "outputTopic" }}
-            - name: "{{ .Values.configurationEnvPrefix }}_OUTPUT_TOPIC"
-              value: {{ .Values.kafka.outputTopic | quote }}
-            {{- end }}
-            {{- if and (hasKey .Values.kafka "labeledOutputTopics") (.Values.kafka.labeledOutputTopics) }}
-            - name: "{{ .Values.configurationEnvPrefix }}_LABELED_OUTPUT_TOPICS"
-              value: "{{- range $key, $value := .Values.kafka.labeledOutputTopics }}{{ $key }}={{ $value }},{{- end }}"
-            {{- end }}
-            {{- range $key, $value := .Values.secrets }}
-            - name: "{{ $key }}"
-              valueFrom:
-                secretKeyRef:
-                  name: {{ include "common-app.fullname" . }}
-                  key: "{{ $key }}"
-            {{- end }}
-            {{- range $key, $value := .Values.secretRefs }}
-            - name: "{{ $key }}"
-              valueFrom:
-                secretKeyRef:
-                  name: {{ $value.name }}
-                  key: "{{ $value.key }}"
-            {{- end }}
-            {{- range $key, $value := .Values.commandLine }}
-            - name: "{{ $root.Values.configurationEnvPrefix }}_{{ $key }}"
-              value: {{ $value | quote }}
-            {{- end }}
-            {{- range $key, $value := .Values.env }}
-            - name: {{ $key | quote }}
-              value: {{ $value | quote }}
-            {{- end }}
+          {{- include "common-app.common-env" . | nindent 12 }}
+          {{- include "common-app.output-env" . | indent 12 }}
             - name: JAVA_TOOL_OPTIONS
-              value: '-XX:MaxRAMPercentage={{ printf "%.1f" .Values.javaOptions.maxRAMPercentage }}
-                {{ .Values.javaOptions.others | join " " }}'
+              value: '{{ include "common-app.java-tool-options" . }}'
           {{- if or (.Values.files) (.Values.secretFilesRefs) }}
           volumeMounts:
-            {{- range $key, $value := .Values.files }}
-            - name: config
-              mountPath: {{ printf "%s/%s" $value.mountPath $key | quote }}
-              subPath: {{ $key | quote }}
-            {{- end }}
-            {{- range .Values.secretFilesRefs }}
-            - name: {{ .volume }}
-              mountPath: {{ .mountPath }}
-              {{- if .readOnly }}
-              readOnly: true
-              {{- end }}
-              {{- if .subPath}}
-              subPath: {{.subPath }}
-              {{- end }}
-            {{- end }}
+            {{- include "common-app.volume-mounts" . | nindent 12 }}
           {{- end }}
-      {{- if or (.Values.files) (.Values.secretFilesRefs) }}
-      volumes:
-        {{- if .Values.files }}
-        - name: config
-          configMap:
-            name: {{ include "common-app.fullname" . }}
-        {{- end }}
-        {{- range .Values.secretFilesRefs }}
-        - name: {{ .volume }}
-          secret:
-            secretName: {{ .name }}
-        {{- end }}
-      {{- end }}
-  backoffLimit: {{ .Values.backoffLimit }}
diff --git a/charts/producer-app/templates/_job-spec.yaml b/charts/producer-app/templates/_job-spec.yaml
new file mode 100644
index 000000000..a29a5eb7c
--- /dev/null
+++ b/charts/producer-app/templates/_job-spec.yaml
@@ -0,0 +1,8 @@
+{{- define "producer-app.jobSpec" -}}
+{{- if .Values.ttlSecondsAfterFinished }}
+ttlSecondsAfterFinished: {{ .Values.ttlSecondsAfterFinished }}
+{{- end }}
+template:
+{{ include "producer-app.podTemplate" . | indent 2 }}
+backoffLimit: {{ .Values.backoffLimit }}
+{{- end -}}
diff --git a/charts/producer-app/templates/_pod.yaml b/charts/producer-app/templates/_pod.yaml
index 19a118b60..6efb01874 100644
--- a/charts/producer-app/templates/_pod.yaml
+++ b/charts/producer-app/templates/_pod.yaml
@@ -1,139 +1,28 @@
 {{- define "producer-app.podTemplate" -}}
-{{- $root := . -}}
 metadata:
-{{- if or .Values.podAnnotations (and .Values.deployment .Values.files) }}
-  annotations:
-  {{- if and .Values.deployment .Values.files }}
-    checksum/config: {{ include (print $.Template.BasePath "/configmap.yaml") . | sha256sum }}
-  {{- end }}
-  {{- range $key, $value := .Values.podAnnotations }}
-    {{ $key | quote }}: {{ $value | quote }}
-  {{- end }}
-{{- end }}
-  labels:
-    {{- include "common-app.selectorLabels" . | nindent 4 }}
-    streams-bootstrap/kind: {{ .Chart.Name }}
-    {{- range $key, $value := .Values.podLabels }}
-    {{ $key }}: {{ $value }}
-    {{- end }}
+{{- include "common-app.pod-metadata" . }}
 spec:
+{{- include "common-app.pod-spec" . }}
   {{- if not .Values.deployment }}
   restartPolicy: {{ .Values.restartPolicy }}
-  {{- end }}
-  {{- if .Values.serviceAccountName }}
-  serviceAccountName: {{ .Values.serviceAccountName }}
-  {{- end }}
-  {{- if .Values.tolerations }}
-  tolerations:
-{{ toYaml .Values.tolerations | indent 4 }}
-  {{- end }}
-  {{- with .Values.affinity }}
-  affinity:
-    {{- tpl (toYaml .) $root | nindent 4 }}
-  {{- end }}
-  {{- if .Values.imagePullSecrets }}
-  imagePullSecrets:
-{{- toYaml .Values.imagePullSecrets | nindent 8 }}
   {{- end }}
   containers:
-  - name: "kafka-app"
-    image: "{{ .Values.image }}:{{ .Values.imageTag }}"
-    imagePullPolicy: "{{ .Values.imagePullPolicy }}"
-    resources:
-{{ toYaml .Values.resources | indent 6 }}
+  {{- include "common-app.kafka-container" . | nindent 2 }}
     env:
-      - name: ENV_PREFIX
-        value: {{ .Values.configurationEnvPrefix }}_
-      {{- range $key, $value := .Values.kafka.config }}
-      - name: {{ printf "KAFKA_%s" $key | replace "." "_" | upper | quote }}
-        value: {{ $value | quote }}
-        {{- end }}
-        {{- if hasKey .Values.kafka "bootstrapServers" }}
-      - name: "{{ .Values.configurationEnvPrefix }}_BOOTSTRAP_SERVERS"
-        value: {{ .Values.kafka.bootstrapServers | quote }}
-        {{- end }}
-        {{- if hasKey .Values.kafka "schemaRegistryUrl" }}
-      - name: "{{ .Values.configurationEnvPrefix }}_SCHEMA_REGISTRY_URL"
-        value: {{ .Values.kafka.schemaRegistryUrl | quote }}
-        {{- end }}
-        {{- if hasKey .Values.kafka "outputTopic" }}
-      - name: "{{ .Values.configurationEnvPrefix }}_OUTPUT_TOPIC"
-        value: {{ .Values.kafka.outputTopic | quote }}
-        {{- end }}
-        {{- if and (hasKey .Values.kafka "labeledOutputTopics") (.Values.kafka.labeledOutputTopics) }}
-      - name: "{{ .Values.configurationEnvPrefix }}_LABELED_OUTPUT_TOPICS"
-        value: "{{- range $key, $value := .Values.kafka.labeledOutputTopics }}{{ $key }}={{ $value }},{{- end }}"
-        {{- end }}
-        {{- range $key, $value := .Values.secrets }}
-      - name: "{{ $key }}"
-        valueFrom:
-          secretKeyRef:
-            name: {{ include "common-app.fullname" . }}
-            key: "{{ $key }}"
-        {{- end }}
-        {{- range $key, $value := .Values.secretRefs }}
-      - name: "{{ $key }}"
-        valueFrom:
-          secretKeyRef:
-            name: {{ $value.name }}
-            key: "{{ $value.key }}"
-        {{- end }}
-        {{- range $key, $value := .Values.commandLine }}
-      - name: "{{ $root.Values.configurationEnvPrefix }}_{{ $key }}"
-        value: {{ $value | quote }}
-        {{- end }}
-        {{- range $key, $value := .Values.env }}
-      - name: {{ $key | quote }}
-        value: {{ $value | quote }}
-        {{- end }}
+    {{- include "common-app.common-env" . | nindent 6 }}
+    {{- include "common-app.output-env" . | indent 6 }}
       - name: JAVA_TOOL_OPTIONS
-        value: '-XX:MaxRAMPercentage={{ printf "%.1f" .Values.javaOptions.maxRAMPercentage }}
-                {{ .Values.javaOptions.others | join " " }}'
-    {{- range .Values.ports }}
+        value: '{{ include "common-app.java-tool-options" . }}'
+    {{- if .Values.ports }}
     ports:
-      - containerPort: {{ .containerPort }}
-        name: {{ .name | quote }}
-        {{- if .protocol }}
-        protocol: {{ .protocol | quote }}
-        {{- end }}
-    {{- end }}
-    {{- if .Values.livenessProbe }}
-    livenessProbe:
-    {{- .Values.livenessProbe | toYaml | nindent 6 }}
-    {{- end }}
-    {{- if .Values.readinessProbe }}
-    readinessProbe:
-    {{- .Values.readinessProbe | toYaml | nindent 6 }}
+      {{- include "common-app.ports" . | nindent 6 }}
     {{- end }}
     {{- if or (.Values.files) (.Values.secretFilesRefs) }}
     volumeMounts:
-      {{- range $key, $value := .Values.files }}
-      - name: config
-        mountPath: {{ printf "%s/%s" $value.mountPath $key | quote }}
-        subPath: {{ $key | quote }}
-      {{- end }}
-      {{- range .Values.secretFilesRefs }}
-      - name: {{ .volume }}
-        mountPath: {{ .mountPath }}
-        {{- if .readOnly }}
-        readOnly: true
-        {{- end }}
-        {{- if .subPath}}
-        subPath: {{.subPath }}
-        {{- end }}
-      {{- end }}
+      {{- include "common-app.volume-mounts" . | nindent 6 }}
     {{- end }}
   {{- if or (.Values.files) (.Values.secretFilesRefs) }}
   volumes:
-    {{- if .Values.files }}
-    - name: config
-      configMap:
-        name: {{ include "common-app.fullname" . }}
-    {{- end }}
-    {{- range .Values.secretFilesRefs }}
-    - name: {{ .volume }}
-      secret:
-        secretName: {{ .name }}
-    {{- end }}
+    {{- include "common-app.volumes" . | nindent 4 }}
   {{- end }}
 {{- end -}}
diff --git a/charts/producer-app/templates/configmap.yaml b/charts/producer-app/templates/configmap.yaml
index 040678bc5..efbeffd77 100644
--- a/charts/producer-app/templates/configmap.yaml
+++ b/charts/producer-app/templates/configmap.yaml
@@ -1,16 +1 @@
-{{ if .Values.files }}
-apiVersion: v1
-kind: ConfigMap
-metadata:
-  name: {{ include "common-app.fullname" . }}
-  {{- include "common-app.annotations" . }}
-  labels:
-    {{- include "common-app.labels" . | nindent 4 }}
-    {{- range $key, $value := .Values.labels }}
-    {{ $key | quote }}: {{ $value | quote }}
-    {{- end }}
-data:
-  {{- range $key, $value := .Values.files }}
-  {{ $key }}: {{ $value.content | quote }}
-  {{- end }}
-{{ end }}
+{{- include "common-app.configmap" . -}}
diff --git a/charts/producer-app/templates/deployment.yaml b/charts/producer-app/templates/deployment.yaml
index 4e6e3be10..4f1c68e07 100644
--- a/charts/producer-app/templates/deployment.yaml
+++ b/charts/producer-app/templates/deployment.yaml
@@ -1,5 +1,4 @@
 {{- if .Values.deployment }}
-{{- $root := . -}}
 {{- if .Capabilities.APIVersions.Has "apps/v1" }}
 apiVersion: apps/v1
 {{- else }}
diff --git a/charts/producer-app/templates/job.yaml b/charts/producer-app/templates/job.yaml
index 28d8941a7..1f30df949 100644
--- a/charts/producer-app/templates/job.yaml
+++ b/charts/producer-app/templates/job.yaml
@@ -1,5 +1,4 @@
 {{- if not .Values.deployment -}}
-{{- $root := . -}}
   {{- if .Values.schedule -}}
     {{- if .Capabilities.APIVersions.Has "batch/v1/CronJob" }}
 apiVersion: batch/v1
@@ -37,18 +36,8 @@ spec:
         {{ $key | quote }}: {{ $value | quote }}
         {{- end }}
     spec:
-      {{- if .Values.ttlSecondsAfterFinished }}
-      ttlSecondsAfterFinished: {{ .Values.ttlSecondsAfterFinished }}
-      {{- end }}
-      template:
-{{ include "producer-app.podTemplate" . | indent 8 }}
-      backoffLimit: {{ .Values.backoffLimit }}
+      {{- include "producer-app.jobSpec" . | indent 6 }}
   {{ else }}
-  {{- if .Values.ttlSecondsAfterFinished }}
-  ttlSecondsAfterFinished: {{ .Values.ttlSecondsAfterFinished }}
-  {{- end }}
-  template:
-{{ include "producer-app.podTemplate" . | indent 4 }}
-  backoffLimit: {{ .Values.backoffLimit }}
+  {{- include "producer-app.jobSpec" . | indent 2 }}
   {{- end -}}
 {{- end }}
diff --git a/charts/producer-app/values.yaml b/charts/producer-app/values.yaml
index ba8db2432..40f9653f3 100644
--- a/charts/producer-app/values.yaml
+++ b/charts/producer-app/values.yaml
@@ -36,6 +36,9 @@ tolerations: []
 #     operator: "Exists"
 #     effect: "NoSchedule"
 
+# priorityClass
+# priorityClassName: ""
+
 ## Affinity for pod assignment (evaluated as template)
 ## ref: https://kubernetes.io/docs/concepts/configuration/assign-pod-node/#affinity-and-anti-affinity
 ##
@@ -119,3 +122,5 @@ readinessProbe: {}
 podAnnotations: {}
 
 podLabels: {}
+
+terminationGracePeriodSeconds: 300
diff --git a/charts/streams-app-cleanup-job/templates/job.yaml b/charts/streams-app-cleanup-job/templates/job.yaml
index fc2a74b3c..4915ee29d 100644
--- a/charts/streams-app-cleanup-job/templates/job.yaml
+++ b/charts/streams-app-cleanup-job/templates/job.yaml
@@ -1 +1,28 @@
-{{- include "common-app.cleanup-job" . -}}
+{{- include "common-app.cleanup-job" . }}
+spec:
+{{- include "common-app.cleanup-job-spec" . | nindent 2 }}
+  template:
+    metadata:
+{{- include "common-app.cleanup-pod-metadata" . | indent 4 }}
+    spec:
+{{- include "common-app.cleanup-pod-spec" . | indent 4 }}
+      containers:
+        {{- include "common-app.common-kafka-container" . | nindent 8 }}
+          args:
+              {{- if .Values.kafka.deleteOutput }}
+            - clean
+              {{- else }}
+            - reset
+              {{- end }}
+          env:
+            {{- include "common-app.common-env" . | nindent 12 }}
+            {{- include "common-app.input-env" . | indent 12 }}
+            {{- include "common-app.output-env" . | indent 12 }}
+            {{- include "common-app.error-env" . | indent 12 }}
+            {{- include "common-app.application-id-env" . | indent 12 }}
+            - name: JAVA_TOOL_OPTIONS
+              value: '{{ include "common-app.java-tool-options" . }}'
+          {{- if or (.Values.files) (.Values.secretFilesRefs) }}
+          volumeMounts:
+              {{- include "common-app.volume-mounts" . | nindent 12 }}
+          {{- end }}
diff --git a/charts/streams-app/templates/deployment.yaml b/charts/streams-app/templates/deployment.yaml
index db5ca18ed..b3d914494 100644
--- a/charts/streams-app/templates/deployment.yaml
+++ b/charts/streams-app/templates/deployment.yaml
@@ -1 +1,55 @@
-{{- include "common-app.deployment" . -}}
+{{- if and .Values.persistence.enabled (not .Values.persistence.size) }}
+{{- fail "When persistence is enabled, you must set .Values.persistence.size" }}
+{{- end }}
+
+{{- include "common-app.deployment" . }}
+spec:
+{{- include "common-app.deployment-spec" . }}
+  template:
+    metadata:
+    {{- include "common-app.pod-metadata" . | indent 4 }}
+    spec:
+    {{- include "common-app.pod-spec" . | indent 4 }}
+      containers:
+        {{- include "common-app.kafka-container" . | nindent 8 }}
+          env:
+          {{- include "common-app.common-env" . | nindent 12 }}
+          {{- range .Values.ports }}
+            {{- if .servicePort }} # TODO verify that there is at most one service port. Currently, if there are multiple service ports, the first one will be used
+            - name: POD_IP
+              valueFrom:
+                fieldRef:
+                  fieldPath: status.podIP
+            - name: KAFKA_APPLICATION_SERVER
+              value: "$(POD_IP):{{ .containerPort }}"
+            {{- end }}
+          {{- end }}
+          {{- include "common-app.group-instance-id-env" . | indent 12 }}
+          {{- include "common-app.input-env" . | indent 12 }}
+          {{- include "common-app.output-env" . | indent 12 }}
+          {{- include "common-app.error-env" . | indent 12 }}
+          {{- include "common-app.application-id-env" . | indent 12 }}
+            - name: JAVA_TOOL_OPTIONS
+              value: '{{ include "common-app.java-tool-jmx-options" . }}
+                {{ include "common-app.java-tool-options" . }}'
+        {{- if or (.Values.files) (and .Values.persistence.enabled .Values.statefulSet) (.Values.secretFilesRefs) }}
+          volumeMounts:
+            {{- include "common-app.volume-mounts" . | indent 12 }}
+            {{- if and .Values.persistence.enabled .Values.statefulSet }}
+            - name: datadir
+              mountPath: /tmp/kafka-streams
+            {{- end }}
+        {{- end }}
+        {{- if or (.Values.jmx.enabled) (.Values.ports) }}
+          ports:
+            {{- include "common-app.ports" . | indent 12 }}
+            {{- include "common-app.jmx-port" . | indent 12 }}
+        {{- end }}
+      {{- if .Values.prometheus.jmx.enabled }}
+      {{- include "common-app.prometheus-container" . | nindent 8 }}
+      {{- end }}
+      {{- if or (.Values.prometheus.jmx.enabled) (.Values.files) (.Values.secretFilesRefs) }}
+      volumes:
+        {{- include "common-app.jmx-volume" . | indent 8 }}
+        {{- include "common-app.volumes" . | indent 8 }}
+      {{- end }}