Repository: banzaicloud/kafka-operator
Pull request #604 (2021-05-07 16:04:20)

Adding support for broker config and broker level overrides of environment variables
| Q               | A
| --------------- | ---
| Bug fix?        | no
| New feature?    | yes
| API breaks?     | *yes*
| Deprecations?   | no
| Related tickets | breaks #602
| License         | Apache 2.0

### What's in this PR?
<!-- Explain the contents of the PR. Give an overview about the implementation, which decisions were made and why. -->
This PR extends the existing KafkaCluster CR with the following fields:`spec.brokers[].brokerConfig.envs[]`, `spec.brokerConfigGroup{}.envs[]`. These two slices add the capability of specifying environment variables in broker config group and broker level.

### Why?
<!-- Which problem does the PR fix? (Please remove this section if you linked an issue above) -->
There are several use cases when different Kafka brokers needs different settings: e.g. different JVM settings, thread count, limits etc.

### Feature description
When assembling the environment variables the operator takes the following steps (in the following order):
1. Default env vars are added (`CLASSPATH`, `KAFKA_OPTS`, `ENVOY_SIDECAR_STATUS`)
1. Cluster wide env vars are added from `spec.envs`
1. \* Env vars defined in broker config group level are added from `spec.brokerConfigGroup{}.envs[]`
1. \* Env vars defined in broker level are added from `spec.brokers[].brokerConfig.envs[]`
1. Special env vars that can be configured through the KafkaCluster CR are added if not specified in the previous steps (`KAFKA_LOG4J_OPTS`, `KAFKA_HEAP_OPTS`, `KAFKA_JVM_PERFORMANCE_OPTS`).

\* new steps introduced by this feature 

It is now also possible to define environment variables in *prepend mode* and *append mode*. 
- If the user puts the `+` character to the front of the name of the environment variable, it will not overwrite the value added in a previous step, but instead its value gets appended to the existing value.
- Similarly, if the user puts the `+` character after the name of the environment variable, it will not overwrite the value added in a previous step, but instead its value gets prepended to the existing value.

Example:
```yaml
apiVersion: kafka.banzaicloud.io/v1beta1
kind: KafkaCluster
metadata:
  name: kafka
spec:
  envs:
    - name: CLASSPATH+
      value: "/opt/kafka/libs/dev/*:"
    - name: +CLASSPATH
      value: ":/opt/kafka/libs/special-libs/*"
```
results the following `CLASSPATH` env var value to be added to the Kafka broker pod: `/opt/kafka/libs/dev/*:/opt/kafka/libs/extensions/*:/opt/kafka/libs/special-libs/*` where `/opt/kafka/libs/extensions/*` is the default provided by the operator. (Note that leading and trailing `:` is ignored by the JVM).

If an environment variable hasn't been set earlier, and the value is provided in prepend/append mode, the extra `+` character from the front/back is simply removed (interpreted as appending/prepending to the empty string `""`).

### Additional context
<!-- Additional information we should know about (eg. edge cases, steps you followed to test the implementation) (Please remove this section if you don't need it) -->
Tested locally by specifying broker config level `env` (uncommenting the `env` config in `simplekafkacluster.yaml`) and after the rolling upgrade the pods were deployed correctly.

**IMPORTANT:**
This PR breaks the API defined in #602 and released in API package `v0.8.1`. 
The following function declaration in `kafkacluster_types.go`
```go
func (b *Broker) GetBrokerConfig(brokerConfigGroups map[string]BrokerConfig) (*BrokerConfig, error)
```
has been changed to
```go
func (b *Broker) GetBrokerConfig(kafkaClusterSpec KafkaClusterSpec) (*BrokerConfig, error)
```
Applications depending on this function will break.
Taking into account that this was released few days ago we can release a new major version (`v0.9.0`) with this change causing minimal disruption.

### Checklist
<!-- Place an '[x]' (no spaces) in all applicable fields. Please remove unrelated fields. -->

- [x] Implementation tested
- [x] Error handling code meets the [guideline](https://github.com/banzaicloud/developer-guide/blob/master/docs/coding-style/error-handling-guide.md)
- [x] Logging code meets the guideline
- [x] Updated implementation based on feedbacks


adamantal: Updated the implementation based on offline feedbacks.
Changes in latest patch:
- shifted implementation from a value-based append to a name-based append (see description)
- introduced prepend mode which can be used for prepending to the `CLASSPATH` env var
- fixed tests, documentation and PR description to reflect this change

Patch:
diff --git a/charts/kafka-operator/templates/crds.yaml b/charts/kafka-operator/templates/crds.yaml
index 6317e2d8..4cdeeb8e 100644
--- a/charts/kafka-operator/templates/crds.yaml
+++ b/charts/kafka-operator/templates/crds.yaml
@@ -724,6 +724,120 @@ spec:
                       type: array
                     config:
                       type: string
+                    envs:
+                      description: Envs defines environment variables for Kafka broker
+                        Pods. Adding the "+" prefix to the name prepends the value
+                        to that environment variable instead of overwriting it. Add
+                        the "+" suffix to append.
+                      items:
+                        description: EnvVar represents an environment variable present
+                          in a Container.
+                        properties:
+                          name:
+                            description: Name of the environment variable. Must be
+                              a C_IDENTIFIER.
+                            type: string
+                          value:
+                            description: 'Variable references $(VAR_NAME) are expanded
+                              using the previous defined environment variables in
+                              the container and any service environment variables.
+                              If a variable cannot be resolved, the reference in the
+                              input string will be unchanged. The $(VAR_NAME) syntax
+                              can be escaped with a double $$, ie: $$(VAR_NAME). Escaped
+                              references will never be expanded, regardless of whether
+                              the variable exists or not. Defaults to "".'
+                            type: string
+                          valueFrom:
+                            description: Source for the environment variable's value.
+                              Cannot be used if value is not empty.
+                            properties:
+                              configMapKeyRef:
+                                description: Selects a key of a ConfigMap.
+                                properties:
+                                  key:
+                                    description: The key to select.
+                                    type: string
+                                  name:
+                                    description: 'Name of the referent. More info:
+                                      https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names
+                                      TODO: Add other useful fields. apiVersion, kind,
+                                      uid?'
+                                    type: string
+                                  optional:
+                                    description: Specify whether the ConfigMap or
+                                      its key must be defined
+                                    type: boolean
+                                required:
+                                - key
+                                type: object
+                              fieldRef:
+                                description: 'Selects a field of the pod: supports
+                                  metadata.name, metadata.namespace, metadata.labels,
+                                  metadata.annotations, spec.nodeName, spec.serviceAccountName,
+                                  status.hostIP, status.podIP, status.podIPs.'
+                                properties:
+                                  apiVersion:
+                                    description: Version of the schema the FieldPath
+                                      is written in terms of, defaults to "v1".
+                                    type: string
+                                  fieldPath:
+                                    description: Path of the field to select in the
+                                      specified API version.
+                                    type: string
+                                required:
+                                - fieldPath
+                                type: object
+                              resourceFieldRef:
+                                description: 'Selects a resource of the container:
+                                  only resources limits and requests (limits.cpu,
+                                  limits.memory, limits.ephemeral-storage, requests.cpu,
+                                  requests.memory and requests.ephemeral-storage)
+                                  are currently supported.'
+                                properties:
+                                  containerName:
+                                    description: 'Container name: required for volumes,
+                                      optional for env vars'
+                                    type: string
+                                  divisor:
+                                    anyOf:
+                                    - type: integer
+                                    - type: string
+                                    description: Specifies the output format of the
+                                      exposed resources, defaults to "1"
+                                    pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                                    x-kubernetes-int-or-string: true
+                                  resource:
+                                    description: 'Required: resource to select'
+                                    type: string
+                                required:
+                                - resource
+                                type: object
+                              secretKeyRef:
+                                description: Selects a key of a secret in the pod's
+                                  namespace
+                                properties:
+                                  key:
+                                    description: The key of the secret to select from.  Must
+                                      be a valid secret key.
+                                    type: string
+                                  name:
+                                    description: 'Name of the referent. More info:
+                                      https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names
+                                      TODO: Add other useful fields. apiVersion, kind,
+                                      uid?'
+                                    type: string
+                                  optional:
+                                    description: Specify whether the Secret or its
+                                      key must be defined
+                                    type: boolean
+                                required:
+                                - key
+                                type: object
+                            type: object
+                        required:
+                        - name
+                        type: object
+                      type: array
                     image:
                       type: string
                     imagePullSecrets:
@@ -4361,6 +4475,121 @@ spec:
                           type: array
                         config:
                           type: string
+                        envs:
+                          description: Envs defines environment variables for Kafka
+                            broker Pods. Adding the "+" prefix to the name prepends
+                            the value to that environment variable instead of overwriting
+                            it. Add the "+" suffix to append.
+                          items:
+                            description: EnvVar represents an environment variable
+                              present in a Container.
+                            properties:
+                              name:
+                                description: Name of the environment variable. Must
+                                  be a C_IDENTIFIER.
+                                type: string
+                              value:
+                                description: 'Variable references $(VAR_NAME) are
+                                  expanded using the previous defined environment
+                                  variables in the container and any service environment
+                                  variables. If a variable cannot be resolved, the
+                                  reference in the input string will be unchanged.
+                                  The $(VAR_NAME) syntax can be escaped with a double
+                                  $$, ie: $$(VAR_NAME). Escaped references will never
+                                  be expanded, regardless of whether the variable
+                                  exists or not. Defaults to "".'
+                                type: string
+                              valueFrom:
+                                description: Source for the environment variable's
+                                  value. Cannot be used if value is not empty.
+                                properties:
+                                  configMapKeyRef:
+                                    description: Selects a key of a ConfigMap.
+                                    properties:
+                                      key:
+                                        description: The key to select.
+                                        type: string
+                                      name:
+                                        description: 'Name of the referent. More info:
+                                          https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names
+                                          TODO: Add other useful fields. apiVersion,
+                                          kind, uid?'
+                                        type: string
+                                      optional:
+                                        description: Specify whether the ConfigMap
+                                          or its key must be defined
+                                        type: boolean
+                                    required:
+                                    - key
+                                    type: object
+                                  fieldRef:
+                                    description: 'Selects a field of the pod: supports
+                                      metadata.name, metadata.namespace, metadata.labels,
+                                      metadata.annotations, spec.nodeName, spec.serviceAccountName,
+                                      status.hostIP, status.podIP, status.podIPs.'
+                                    properties:
+                                      apiVersion:
+                                        description: Version of the schema the FieldPath
+                                          is written in terms of, defaults to "v1".
+                                        type: string
+                                      fieldPath:
+                                        description: Path of the field to select in
+                                          the specified API version.
+                                        type: string
+                                    required:
+                                    - fieldPath
+                                    type: object
+                                  resourceFieldRef:
+                                    description: 'Selects a resource of the container:
+                                      only resources limits and requests (limits.cpu,
+                                      limits.memory, limits.ephemeral-storage, requests.cpu,
+                                      requests.memory and requests.ephemeral-storage)
+                                      are currently supported.'
+                                    properties:
+                                      containerName:
+                                        description: 'Container name: required for
+                                          volumes, optional for env vars'
+                                        type: string
+                                      divisor:
+                                        anyOf:
+                                        - type: integer
+                                        - type: string
+                                        description: Specifies the output format of
+                                          the exposed resources, defaults to "1"
+                                        pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                                        x-kubernetes-int-or-string: true
+                                      resource:
+                                        description: 'Required: resource to select'
+                                        type: string
+                                    required:
+                                    - resource
+                                    type: object
+                                  secretKeyRef:
+                                    description: Selects a key of a secret in the
+                                      pod's namespace
+                                    properties:
+                                      key:
+                                        description: The key of the secret to select
+                                          from.  Must be a valid secret key.
+                                        type: string
+                                      name:
+                                        description: 'Name of the referent. More info:
+                                          https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names
+                                          TODO: Add other useful fields. apiVersion,
+                                          kind, uid?'
+                                        type: string
+                                      optional:
+                                        description: Specify whether the Secret or
+                                          its key must be defined
+                                        type: boolean
+                                    required:
+                                    - key
+                                    type: object
+                                type: object
+                            required:
+                            - name
+                            type: object
+                          type: array
                         image:
                           type: string
                         imagePullSecrets:
@@ -10333,6 +10562,9 @@ spec:
                     type: array
                 type: object
               envs:
+                description: Envs defines environment variables for Kafka broker Pods.
+                  Adding the "+" prefix to the name prepends the value to that environment
+                  variable instead of overwriting it. Add the "+" suffix to append.
                 items:
                   description: EnvVar represents an environment variable present in
                     a Container.
diff --git a/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml b/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml
index abc55d65..d0e523b0 100644
--- a/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml
+++ b/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml
@@ -723,6 +723,120 @@ spec:
                       type: array
                     config:
                       type: string
+                    envs:
+                      description: Envs defines environment variables for Kafka broker
+                        Pods. Adding the "+" prefix to the name prepends the value
+                        to that environment variable instead of overwriting it. Add
+                        the "+" suffix to append.
+                      items:
+                        description: EnvVar represents an environment variable present
+                          in a Container.
+                        properties:
+                          name:
+                            description: Name of the environment variable. Must be
+                              a C_IDENTIFIER.
+                            type: string
+                          value:
+                            description: 'Variable references $(VAR_NAME) are expanded
+                              using the previous defined environment variables in
+                              the container and any service environment variables.
+                              If a variable cannot be resolved, the reference in the
+                              input string will be unchanged. The $(VAR_NAME) syntax
+                              can be escaped with a double $$, ie: $$(VAR_NAME). Escaped
+                              references will never be expanded, regardless of whether
+                              the variable exists or not. Defaults to "".'
+                            type: string
+                          valueFrom:
+                            description: Source for the environment variable's value.
+                              Cannot be used if value is not empty.
+                            properties:
+                              configMapKeyRef:
+                                description: Selects a key of a ConfigMap.
+                                properties:
+                                  key:
+                                    description: The key to select.
+                                    type: string
+                                  name:
+                                    description: 'Name of the referent. More info:
+                                      https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names
+                                      TODO: Add other useful fields. apiVersion, kind,
+                                      uid?'
+                                    type: string
+                                  optional:
+                                    description: Specify whether the ConfigMap or
+                                      its key must be defined
+                                    type: boolean
+                                required:
+                                - key
+                                type: object
+                              fieldRef:
+                                description: 'Selects a field of the pod: supports
+                                  metadata.name, metadata.namespace, metadata.labels,
+                                  metadata.annotations, spec.nodeName, spec.serviceAccountName,
+                                  status.hostIP, status.podIP, status.podIPs.'
+                                properties:
+                                  apiVersion:
+                                    description: Version of the schema the FieldPath
+                                      is written in terms of, defaults to "v1".
+                                    type: string
+                                  fieldPath:
+                                    description: Path of the field to select in the
+                                      specified API version.
+                                    type: string
+                                required:
+                                - fieldPath
+                                type: object
+                              resourceFieldRef:
+                                description: 'Selects a resource of the container:
+                                  only resources limits and requests (limits.cpu,
+                                  limits.memory, limits.ephemeral-storage, requests.cpu,
+                                  requests.memory and requests.ephemeral-storage)
+                                  are currently supported.'
+                                properties:
+                                  containerName:
+                                    description: 'Container name: required for volumes,
+                                      optional for env vars'
+                                    type: string
+                                  divisor:
+                                    anyOf:
+                                    - type: integer
+                                    - type: string
+                                    description: Specifies the output format of the
+                                      exposed resources, defaults to "1"
+                                    pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                                    x-kubernetes-int-or-string: true
+                                  resource:
+                                    description: 'Required: resource to select'
+                                    type: string
+                                required:
+                                - resource
+                                type: object
+                              secretKeyRef:
+                                description: Selects a key of a secret in the pod's
+                                  namespace
+                                properties:
+                                  key:
+                                    description: The key of the secret to select from.  Must
+                                      be a valid secret key.
+                                    type: string
+                                  name:
+                                    description: 'Name of the referent. More info:
+                                      https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names
+                                      TODO: Add other useful fields. apiVersion, kind,
+                                      uid?'
+                                    type: string
+                                  optional:
+                                    description: Specify whether the Secret or its
+                                      key must be defined
+                                    type: boolean
+                                required:
+                                - key
+                                type: object
+                            type: object
+                        required:
+                        - name
+                        type: object
+                      type: array
                     image:
                       type: string
                     imagePullSecrets:
@@ -4360,6 +4474,121 @@ spec:
                           type: array
                         config:
                           type: string
+                        envs:
+                          description: Envs defines environment variables for Kafka
+                            broker Pods. Adding the "+" prefix to the name prepends
+                            the value to that environment variable instead of overwriting
+                            it. Add the "+" suffix to append.
+                          items:
+                            description: EnvVar represents an environment variable
+                              present in a Container.
+                            properties:
+                              name:
+                                description: Name of the environment variable. Must
+                                  be a C_IDENTIFIER.
+                                type: string
+                              value:
+                                description: 'Variable references $(VAR_NAME) are
+                                  expanded using the previous defined environment
+                                  variables in the container and any service environment
+                                  variables. If a variable cannot be resolved, the
+                                  reference in the input string will be unchanged.
+                                  The $(VAR_NAME) syntax can be escaped with a double
+                                  $$, ie: $$(VAR_NAME). Escaped references will never
+                                  be expanded, regardless of whether the variable
+                                  exists or not. Defaults to "".'
+                                type: string
+                              valueFrom:
+                                description: Source for the environment variable's
+                                  value. Cannot be used if value is not empty.
+                                properties:
+                                  configMapKeyRef:
+                                    description: Selects a key of a ConfigMap.
+                                    properties:
+                                      key:
+                                        description: The key to select.
+                                        type: string
+                                      name:
+                                        description: 'Name of the referent. More info:
+                                          https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names
+                                          TODO: Add other useful fields. apiVersion,
+                                          kind, uid?'
+                                        type: string
+                                      optional:
+                                        description: Specify whether the ConfigMap
+                                          or its key must be defined
+                                        type: boolean
+                                    required:
+                                    - key
+                                    type: object
+                                  fieldRef:
+                                    description: 'Selects a field of the pod: supports
+                                      metadata.name, metadata.namespace, metadata.labels,
+                                      metadata.annotations, spec.nodeName, spec.serviceAccountName,
+                                      status.hostIP, status.podIP, status.podIPs.'
+                                    properties:
+                                      apiVersion:
+                                        description: Version of the schema the FieldPath
+                                          is written in terms of, defaults to "v1".
+                                        type: string
+                                      fieldPath:
+                                        description: Path of the field to select in
+                                          the specified API version.
+                                        type: string
+                                    required:
+                                    - fieldPath
+                                    type: object
+                                  resourceFieldRef:
+                                    description: 'Selects a resource of the container:
+                                      only resources limits and requests (limits.cpu,
+                                      limits.memory, limits.ephemeral-storage, requests.cpu,
+                                      requests.memory and requests.ephemeral-storage)
+                                      are currently supported.'
+                                    properties:
+                                      containerName:
+                                        description: 'Container name: required for
+                                          volumes, optional for env vars'
+                                        type: string
+                                      divisor:
+                                        anyOf:
+                                        - type: integer
+                                        - type: string
+                                        description: Specifies the output format of
+                                          the exposed resources, defaults to "1"
+                                        pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                                        x-kubernetes-int-or-string: true
+                                      resource:
+                                        description: 'Required: resource to select'
+                                        type: string
+                                    required:
+                                    - resource
+                                    type: object
+                                  secretKeyRef:
+                                    description: Selects a key of a secret in the
+                                      pod's namespace
+                                    properties:
+                                      key:
+                                        description: The key of the secret to select
+                                          from.  Must be a valid secret key.
+                                        type: string
+                                      name:
+                                        description: 'Name of the referent. More info:
+                                          https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names
+                                          TODO: Add other useful fields. apiVersion,
+                                          kind, uid?'
+                                        type: string
+                                      optional:
+                                        description: Specify whether the Secret or
+                                          its key must be defined
+                                        type: boolean
+                                    required:
+                                    - key
+                                    type: object
+                                type: object
+                            required:
+                            - name
+                            type: object
+                          type: array
                         image:
                           type: string
                         imagePullSecrets:
@@ -10332,6 +10561,9 @@ spec:
                     type: array
                 type: object
               envs:
+                description: Envs defines environment variables for Kafka broker Pods.
+                  Adding the "+" prefix to the name prepends the value to that environment
+                  variable instead of overwriting it. Add the "+" suffix to append.
                 items:
                   description: EnvVar represents an environment variable present in
                     a Container.
diff --git a/config/samples/simplekafkacluster.yaml b/config/samples/simplekafkacluster.yaml
index 6995bce0..e62cf9d3 100644
--- a/config/samples/simplekafkacluster.yaml
+++ b/config/samples/simplekafkacluster.yaml
@@ -36,6 +36,12 @@ spec:
   brokers:
     - id: 0
       brokerConfigGroup: "default"
+      # brokerConfig:
+      #   envs:
+      #     - name: CLASSPATH+
+      #       value: "/opt/kafka/libs/dev/*:"
+      #     - name: +CLASSPATH
+      #       value: ":/opt/kafka/libs/extra-jars/*"
     - id: 1
       brokerConfigGroup: "default"
     - id: 2
diff --git a/internal/alertmanager/currentalert/process.go b/internal/alertmanager/currentalert/process.go
index 05fe96f6..08ff4353 100644
--- a/internal/alertmanager/currentalert/process.go
+++ b/internal/alertmanager/currentalert/process.go
@@ -253,7 +253,7 @@ func resizePvc(log logr.Logger, labels model.LabelSet, annotiations model.LabelS
 
 	for i, broker := range cr.Spec.Brokers {
 		if strconv.Itoa(int(broker.Id)) == pvc.Labels["brokerId"] {
-			brokerConfig, err := broker.GetBrokerConfig(cr.Spec.BrokerConfigGroups)
+			brokerConfig, err := broker.GetBrokerConfig(cr.Spec)
 			if err != nil {
 				return errors.WrapIf(err, "failed to determine broker config")
 			}
diff --git a/pkg/resources/cruisecontrol/configmap.go b/pkg/resources/cruisecontrol/configmap.go
index 0827e07c..ff469c98 100644
--- a/pkg/resources/cruisecontrol/configmap.go
+++ b/pkg/resources/cruisecontrol/configmap.go
@@ -235,7 +235,7 @@ func generateDefaultBrokerCapacity() BrokerCapacity {
 }
 
 func generateBrokerNetworkIn(broker v1beta1.Broker, kafkaClusterSpec v1beta1.KafkaClusterSpec, log logr.Logger) string {
-	brokerConfig, err := broker.GetBrokerConfig(kafkaClusterSpec.BrokerConfigGroups)
+	brokerConfig, err := broker.GetBrokerConfig(kafkaClusterSpec)
 	if err != nil {
 		log.V(warnLevel).Info("could not get incoming network resource limits falling back to default value")
 		return storageConfigNWINDefaultValue
@@ -249,7 +249,7 @@ func generateBrokerNetworkIn(broker v1beta1.Broker, kafkaClusterSpec v1beta1.Kaf
 }
 
 func generateBrokerNetworkOut(broker v1beta1.Broker, kafkaClusterSpec v1beta1.KafkaClusterSpec, log logr.Logger) string {
-	brokerConfig, err := broker.GetBrokerConfig(kafkaClusterSpec.BrokerConfigGroups)
+	brokerConfig, err := broker.GetBrokerConfig(kafkaClusterSpec)
 	if err != nil {
 		log.V(warnLevel).Info("could not get outgoing network resource limits falling back to default value")
 		return storageConfigNWOUTDefaultValue
@@ -263,7 +263,7 @@ func generateBrokerNetworkOut(broker v1beta1.Broker, kafkaClusterSpec v1beta1.Ka
 }
 
 func generateBrokerCPU(broker v1beta1.Broker, kafkaClusterSpec v1beta1.KafkaClusterSpec, log logr.Logger) string {
-	brokerConfig, err := broker.GetBrokerConfig(kafkaClusterSpec.BrokerConfigGroups)
+	brokerConfig, err := broker.GetBrokerConfig(kafkaClusterSpec)
 	if err != nil {
 		log.V(warnLevel).Info("could not get cpu resource limits falling back to default value")
 		return storageConfigCPUDefaultValue
diff --git a/pkg/resources/kafka/kafka.go b/pkg/resources/kafka/kafka.go
index 3754fecf..640e80c3 100644
--- a/pkg/resources/kafka/kafka.go
+++ b/pkg/resources/kafka/kafka.go
@@ -207,7 +207,7 @@ func (r *Reconciler) Reconcile(log logr.Logger) error {
 
 	brokersVolumes := make(map[string][]*corev1.PersistentVolumeClaim, len(r.KafkaCluster.Spec.Brokers))
 	for _, broker := range r.KafkaCluster.Spec.Brokers {
-		brokerConfig, err := broker.GetBrokerConfig(r.KafkaCluster.Spec.BrokerConfigGroups)
+		brokerConfig, err := broker.GetBrokerConfig(r.KafkaCluster.Spec)
 		if err != nil {
 			return errors.WrapIf(err, "failed to reconcile resource")
 		}
@@ -230,7 +230,7 @@ func (r *Reconciler) Reconcile(log logr.Logger) error {
 
 	reorderedBrokers := r.reorderBrokers(log, r.KafkaCluster.Spec.Brokers)
 	for _, broker := range reorderedBrokers {
-		brokerConfig, err := broker.GetBrokerConfig(r.KafkaCluster.Spec.BrokerConfigGroups)
+		brokerConfig, err := broker.GetBrokerConfig(r.KafkaCluster.Spec)
 		if err != nil {
 			return errors.WrapIf(err, "failed to reconcile resource")
 		}
@@ -888,7 +888,7 @@ func (r *Reconciler) createExternalListenerStatuses(log logr.Logger) (map[string
 				portNumber := eListener.ExternalStartingPort + broker.Id
 
 				if eListener.GetAccessMethod() != corev1.ServiceTypeLoadBalancer {
-					bConfig, err := broker.GetBrokerConfig(r.KafkaCluster.Spec.BrokerConfigGroups)
+					bConfig, err := broker.GetBrokerConfig(r.KafkaCluster.Spec)
 					if err != nil {
 						return nil, err
 					}
@@ -911,7 +911,7 @@ func (r *Reconciler) createExternalListenerStatuses(log logr.Logger) (map[string
 						brokerHost = fmt.Sprintf("%s-%d-%s.%s%s", r.KafkaCluster.Name, broker.Id, eListener.Name, r.KafkaCluster.Namespace, brokerHost)
 					}
 				}
-				brokerConfig, err := broker.GetBrokerConfig(r.KafkaCluster.Spec.BrokerConfigGroups)
+				brokerConfig, err := broker.GetBrokerConfig(r.KafkaCluster.Spec)
 				if err != nil {
 					return nil, err
 				}
diff --git a/pkg/resources/kafka/pod.go b/pkg/resources/kafka/pod.go
index f3c1d9e6..b65513dc 100644
--- a/pkg/resources/kafka/pod.go
+++ b/pkg/resources/kafka/pod.go
@@ -149,7 +149,8 @@ fi`},
 								},
 							},
 						},
-					}, r.KafkaCluster.Spec.Envs),
+					}),
+
 					Command: command,
 					Ports: append(kafkaBrokerContainerPorts, []corev1.ContainerPort{
 						{
@@ -404,15 +405,42 @@ func generateVolumeMountForSSL() []corev1.VolumeMount {
 	}
 }
 
-func generateEnvConfig(brokerConfig *v1beta1.BrokerConfig, defaultEnvVars, clusterEnvVars []corev1.EnvVar) []corev1.EnvVar {
+func generateEnvConfig(brokerConfig *v1beta1.BrokerConfig, defaultEnvVars []corev1.EnvVar) []corev1.EnvVar {
 	envs := map[string]corev1.EnvVar{}
 
 	for _, v := range defaultEnvVars {
 		envs[v.Name] = v
 	}
 
-	for _, v := range clusterEnvVars {
-		envs[v.Name] = v
+	// merge the env variables
+	for _, envVar := range brokerConfig.Envs {
+		envVarName := strings.TrimSpace(envVar.Name)
+		switch {
+		default:
+			fallthrough
+		case envVar.ValueFrom != nil:
+			// append/prepend based on sources is not supported
+			envVar.Name = envVarName
+			envs[envVarName] = envVar
+		case strings.HasPrefix(envVarName, "+"):
+			envVarName = envVarName[1:]
+			if envVarFromMap, ok := envs[envVarName]; ok {
+				envVarFromMap.Value += envVar.Value
+				envs[envVarName] = envVarFromMap
+			} else {
+				envVar.Name = envVarName
+				envs[envVarName] = envVar
+			}
+		case strings.HasSuffix(envVarName, "+"):
+			envVarName = envVarName[:len(envVarName)-1]
+			if envVarFromMap, ok := envs[envVarName]; ok {
+				envVarFromMap.Value = envVar.Value + envVarFromMap.Value
+				envs[envVarName] = envVarFromMap
+			} else {
+				envVar.Name = envVarName
+				envs[envVarName] = envVar
+			}
+		}
 	}
 
 	if brokerConfig.Log4jConfig != "" {
diff --git a/pkg/resources/nodeportexternalaccess/nodeportExternalAccess.go b/pkg/resources/nodeportexternalaccess/nodeportExternalAccess.go
index 87a962cd..1fb49812 100644
--- a/pkg/resources/nodeportexternalaccess/nodeportExternalAccess.go
+++ b/pkg/resources/nodeportexternalaccess/nodeportExternalAccess.go
@@ -52,7 +52,7 @@ func (r *Reconciler) Reconcile(log logr.Logger) error {
 		for _, eListener := range r.KafkaCluster.Spec.ListenersConfig.ExternalListeners {
 			if eListener.GetAccessMethod() == corev1.ServiceTypeNodePort {
 				for _, broker := range r.KafkaCluster.Spec.Brokers {
-					brokerConfig, err := broker.GetBrokerConfig(r.KafkaCluster.Spec.BrokerConfigGroups)
+					brokerConfig, err := broker.GetBrokerConfig(r.KafkaCluster.Spec)
 					if err != nil {
 						return err
 					}
diff --git a/pkg/sdk/v1beta1/kafkacluster_types.go b/pkg/sdk/v1beta1/kafkacluster_types.go
index 3af62614..439877db 100644
--- a/pkg/sdk/v1beta1/kafkacluster_types.go
+++ b/pkg/sdk/v1beta1/kafkacluster_types.go
@@ -62,16 +62,19 @@ type KafkaClusterSpec struct {
 	IngressController string `json:"ingressController,omitempty"`
 	// If true OneBrokerPerNode ensures that each kafka broker will be placed on a different node unless a custom
 	// Affinity definition overrides this behavior
-	OneBrokerPerNode        bool                `json:"oneBrokerPerNode"`
-	PropagateLabels         bool                `json:"propagateLabels,omitempty"`
-	CruiseControlConfig     CruiseControlConfig `json:"cruiseControlConfig"`
-	EnvoyConfig             EnvoyConfig         `json:"envoyConfig,omitempty"`
-	MonitoringConfig        MonitoringConfig    `json:"monitoringConfig,omitempty"`
-	VaultConfig             VaultConfig         `json:"vaultConfig,omitempty"`
-	AlertManagerConfig      *AlertManagerConfig `json:"alertManagerConfig,omitempty"`
-	IstioIngressConfig      IstioIngressConfig  `json:"istioIngressConfig,omitempty"`
-	Envs                    []corev1.EnvVar     `json:"envs,omitempty"`
-	KubernetesClusterDomain string              `json:"kubernetesClusterDomain,omitempty"`
+	OneBrokerPerNode    bool                `json:"oneBrokerPerNode"`
+	PropagateLabels     bool                `json:"propagateLabels,omitempty"`
+	CruiseControlConfig CruiseControlConfig `json:"cruiseControlConfig"`
+	EnvoyConfig         EnvoyConfig         `json:"envoyConfig,omitempty"`
+	MonitoringConfig    MonitoringConfig    `json:"monitoringConfig,omitempty"`
+	VaultConfig         VaultConfig         `json:"vaultConfig,omitempty"`
+	AlertManagerConfig  *AlertManagerConfig `json:"alertManagerConfig,omitempty"`
+	IstioIngressConfig  IstioIngressConfig  `json:"istioIngressConfig,omitempty"`
+	// Envs defines environment variables for Kafka broker Pods.
+	// Adding the "+" prefix to the name prepends the value to that environment variable instead of overwriting it.
+	// Add the "+" suffix to append.
+	Envs                    []corev1.EnvVar `json:"envs,omitempty"`
+	KubernetesClusterDomain string          `json:"kubernetesClusterDomain,omitempty"`
 }
 
 // KafkaClusterStatus defines the observed state of KafkaCluster
@@ -157,6 +160,10 @@ type BrokerConfig struct {
 	Volumes []corev1.Volume `json:"volumes,omitempty"`
 	// VolumeMounts define some extra Kubernetes VolumeMounts for the Kafka broker Pods.
 	VolumeMounts []corev1.VolumeMount `json:"volumeMounts,omitempty"`
+	// Envs defines environment variables for Kafka broker Pods.
+	// Adding the "+" prefix to the name prepends the value to that environment variable instead of overwriting it.
+	// Add the "+" suffix to append.
+	Envs []corev1.EnvVar `json:"envs,omitempty"`
 }
 
 type NetworkConfig struct {
@@ -1061,7 +1068,9 @@ func (mConfig *MonitoringConfig) GetCCJMXExporterConfig() string {
 }
 
 // GetBrokerConfig composes the brokerConfig for a given broker using the broker's config group
-func (b *Broker) GetBrokerConfig(brokerConfigGroups map[string]BrokerConfig) (*BrokerConfig, error) {
+func (b *Broker) GetBrokerConfig(kafkaClusterSpec KafkaClusterSpec) (*BrokerConfig, error) {
+	brokerConfigGroups := kafkaClusterSpec.BrokerConfigGroups
+
 	bConfig := &BrokerConfig{}
 	if b.BrokerConfigGroup == "" {
 		return b.BrokerConfig, nil
@@ -1074,37 +1083,56 @@ func (b *Broker) GetBrokerConfig(brokerConfigGroups map[string]BrokerConfig) (*B
 		return nil, errors.NewWithDetails("missing brokerConfigGroup", "key", b.BrokerConfigGroup)
 	}
 
-	dstAffinity := &corev1.Affinity{}
-	srcAffinity := &corev1.Affinity{}
-
-	if groupConfig.Affinity != nil {
-		dstAffinity = groupConfig.Affinity.DeepCopy()
-	}
-	if bConfig.Affinity != nil {
-		srcAffinity = bConfig.Affinity
-	}
-
-	if err := mergo.Merge(dstAffinity, srcAffinity, mergo.WithOverride); err != nil {
+	dstAffinity, err := mergeAffinity(groupConfig, bConfig)
+	if err != nil {
 		return nil, errors.WrapIf(err, "could not merge brokerConfig.Affinity with ConfigGroup.Affinity")
 	}
+	envs := mergeEnvs(kafkaClusterSpec, &groupConfig, bConfig)
 
-	err := mergo.Merge(bConfig, groupConfig, mergo.WithAppendSlice)
+	err = mergo.Merge(bConfig, groupConfig, mergo.WithAppendSlice)
 	if err != nil {
 		return nil, errors.WrapIf(err, "could not merge brokerConfig with ConfigGroup")
 	}
 
 	bConfig.StorageConfigs = dedupStorageConfigs(bConfig.StorageConfigs)
-
 	if groupConfig.Affinity != nil || bConfig.Affinity != nil {
 		bConfig.Affinity = dstAffinity
 	}
+	bConfig.Envs = envs
 
 	return bConfig, nil
 }
 
+func mergeEnvs(kafkaClusterSpec KafkaClusterSpec, groupConfig, bConfig *BrokerConfig) []corev1.EnvVar {
+	var envs []corev1.EnvVar
+	envs = append(envs, kafkaClusterSpec.Envs...)
+	envs = append(envs, groupConfig.Envs...)
+	if bConfig != nil {
+		envs = append(envs, bConfig.Envs...)
+	}
+	return envs
+}
+
+func mergeAffinity(groupConfig BrokerConfig, bConfig *BrokerConfig) (*corev1.Affinity, error) {
+	dstAffinity := &corev1.Affinity{}
+	srcAffinity := &corev1.Affinity{}
+
+	if groupConfig.Affinity != nil {
+		dstAffinity = groupConfig.Affinity.DeepCopy()
+	}
+	if bConfig.Affinity != nil {
+		srcAffinity = bConfig.Affinity
+	}
+
+	if err := mergo.Merge(dstAffinity, srcAffinity, mergo.WithOverride); err != nil {
+		return nil, err
+	}
+	return dstAffinity, nil
+}
+
 func dedupStorageConfigs(elements []StorageConfig) []StorageConfig {
 	encountered := make(map[string]struct{})
-	result := make([]StorageConfig, 0)
+	var result []StorageConfig
 
 	for _, v := range elements {
 		if _, ok := encountered[v.MountPath]; !ok {
diff --git a/pkg/sdk/v1beta1/zz_generated.deepcopy.go b/pkg/sdk/v1beta1/zz_generated.deepcopy.go
index c8c97772..7f400b48 100644
--- a/pkg/sdk/v1beta1/zz_generated.deepcopy.go
+++ b/pkg/sdk/v1beta1/zz_generated.deepcopy.go
@@ -154,6 +154,13 @@ func (in *BrokerConfig) DeepCopyInto(out *BrokerConfig) {
 			(*in)[i].DeepCopyInto(&(*out)[i])
 		}
 	}
+	if in.Envs != nil {
+		in, out := &in.Envs, &out.Envs
+		*out = make([]v1.EnvVar, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new BrokerConfig.
diff --git a/pkg/util/kafka/common.go b/pkg/util/kafka/common.go
index 30a42e77..75e5f6e2 100644
--- a/pkg/util/kafka/common.go
+++ b/pkg/util/kafka/common.go
@@ -245,7 +245,7 @@ func GatherBrokerConfigIfAvailable(kafkaClusterSpec v1beta1.KafkaClusterSpec, br
 	// but the broker spec does not contain the required config values.
 	for _, broker := range kafkaClusterSpec.Brokers {
 		if int(broker.Id) == brokerID {
-			brokerConfig, err := broker.GetBrokerConfig(kafkaClusterSpec.BrokerConfigGroups)
+			brokerConfig, err := broker.GetBrokerConfig(kafkaClusterSpec)
 			if err != nil {
 				return nil, err
 			}
diff --git a/pkg/util/util.go b/pkg/util/util.go
index 14447f16..a2a23ca6 100644
--- a/pkg/util/util.go
+++ b/pkg/util/util.go
@@ -205,7 +205,7 @@ func IsIngressConfigInUse(iConfigName, defaultConfigName string, cluster *v1beta
 	}
 	// Check if the given iConfigName is bound to a given broker
 	for _, broker := range cluster.Spec.Brokers {
-		brokerConfig, err := broker.GetBrokerConfig(cluster.Spec.BrokerConfigGroups)
+		brokerConfig, err := broker.GetBrokerConfig(cluster.Spec)
 		if err != nil {
 			log.Error(err, "could not determine if ingressConfig is in use")
 			return false


Test patch:
diff --git a/controllers/tests/kafkacluster_controller_kafka_test.go b/controllers/tests/kafkacluster_controller_kafka_test.go
index 1be9abbd..de71bd6d 100644
--- a/controllers/tests/kafkacluster_controller_kafka_test.go
+++ b/controllers/tests/kafkacluster_controller_kafka_test.go
@@ -256,7 +256,6 @@ func expectKafkaBrokerPod(kafkaCluster *v1beta1.KafkaCluster, broker v1beta1.Bro
 	getEnvName := func(c corev1.EnvVar) string { return c.Name }
 	Expect(container.Env).To(ConsistOf(
 		// the exact value is not interesting
-		WithTransform(getEnvName, Equal("CLASSPATH")),
 		WithTransform(getEnvName, Equal("KAFKA_OPTS")),
 		WithTransform(getEnvName, Equal("KAFKA_JVM_PERFORMANCE_OPTS")),
 
@@ -274,6 +273,18 @@ func expectKafkaBrokerPod(kafkaCluster *v1beta1.KafkaCluster, broker v1beta1.Bro
 			Name:  "KAFKA_HEAP_OPTS",
 			Value: "-Xmx2G -Xms2G",
 		},
+		corev1.EnvVar{
+			Name:  "ENVVAR1",
+			Value: "VALUE1 VALUE2",
+		},
+		corev1.EnvVar{
+			Name:  "ENVVAR2",
+			Value: "VALUE2",
+		},
+		corev1.EnvVar{
+			Name:  "CLASSPATH",
+			Value: "/opt/kafka/libs/extensions/*:/test/class/path",
+		},
 	))
 	Expect(container.VolumeMounts).To(HaveLen(8))
 	Expect(container.VolumeMounts[0]).To(Equal(corev1.VolumeMount{
diff --git a/controllers/tests/kafkacluster_controller_test.go b/controllers/tests/kafkacluster_controller_test.go
index 3bc6fa5c..51f53a64 100644
--- a/controllers/tests/kafkacluster_controller_test.go
+++ b/controllers/tests/kafkacluster_controller_test.go
@@ -103,6 +103,20 @@ var _ = Describe("KafkaCluster", func() {
 				MountPath: "/a/test/path",
 			},
 		}
+		defaultGroup.Envs = []corev1.EnvVar{
+			{
+				Name:  "+ENVVAR1",
+				Value: " VALUE2",
+			},
+			{
+				Name:  "ENVVAR2",
+				Value: "VALUE2",
+			},
+			{
+				Name:  "+CLASSPATH",
+				Value: ":/test/class/path",
+			},
+		}
 		kafkaCluster.Spec.BrokerConfigGroups[defaultBrokerConfigGroup] = defaultGroup
 		// Set some CruiseControl pod and container related SecurityContext values
 		kafkaCluster.Spec.CruiseControlConfig.PodSecurityContext = &corev1.PodSecurityContext{
@@ -114,6 +128,16 @@ var _ = Describe("KafkaCluster", func() {
 		kafkaCluster.Spec.EnvoyConfig.Annotations = map[string]string{
 			"envoy-annotation-key": "envoy-annotation-value",
 		}
+		kafkaCluster.Spec.Envs = []corev1.EnvVar{
+			{
+				Name:  "ENVVAR1",
+				Value: "VALUE1",
+			},
+			{
+				Name:  "+ENVVAR2",
+				Value: "VALUE1",
+			},
+		}
 	})
 
 	JustBeforeEach(func() {
diff --git a/pkg/resources/kafka/pod_test.go b/pkg/resources/kafka/pod_test.go
index e23adb20..a0d17c0a 100644
--- a/pkg/resources/kafka/pod_test.go
+++ b/pkg/resources/kafka/pod_test.go
@@ -15,6 +15,7 @@
 package kafka
 
 import (
+	"reflect"
 	"testing"
 
 	"gotest.tools/assert"
@@ -67,7 +68,7 @@ func TestGetAffinity(t *testing.T) {
 		BrokerConfig: &nilAffinityBrokerConfig,
 	}
 
-	mergedAffinityBrokerConfig, _ := broker.GetBrokerConfig(cluster.Spec.BrokerConfigGroups)
+	mergedAffinityBrokerConfig, _ := broker.GetBrokerConfig(cluster.Spec)
 
 	// expecting old behavior
 	affinity = getAffinity(mergedAffinityBrokerConfig, &cluster)
@@ -78,7 +79,7 @@ func TestGetAffinity(t *testing.T) {
 		BrokerConfig:      &nilAffinityBrokerConfig,
 	}
 
-	mergedAffinityBrokerConfig2, _ := broker.GetBrokerConfig(cluster.Spec.BrokerConfigGroups)
+	mergedAffinityBrokerConfig2, _ := broker.GetBrokerConfig(cluster.Spec)
 
 	// expecting old behavior
 	affinity = getAffinity(mergedAffinityBrokerConfig2, &cluster)
@@ -98,3 +99,89 @@ func TestGetAffinity(t *testing.T) {
 	affinity = getAffinity(&nonNilAffinityBrokerConfig, &cluster)
 	assert.DeepEqual(t, affinity, nonNilAffinityBrokerConfig.Affinity.DeepCopy())
 }
+
+func Test_generateEnvConfig(t *testing.T) {
+	expected := []corev1.EnvVar{
+		{Name: "KAFKA_HEAP_OPTS", Value: "-Xmx2G -Xms2G"},
+		{Name: "KAFKA_JVM_PERFORMANCE_OPTS", Value: "-server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -Djava.awt.headless=true -Dsun.net.inetaddr.ttl=60"},
+		{Name: "a", Value: "global"},
+		{Name: "b", Value: "group"},
+		{Name: "c", Value: "broker"},
+		{Name: "d", Value: "global"},
+		{Name: "e", Value: "codegroup"},
+		{Name: "f", Value: "broker"},
+		{Name: "g", Value: "group"},
+		{Name: "h", Value: "broker"},
+		{Name: "i", Value: "groupbroker"},
+		{Name: "j", Value: "broker"},
+		{Name: "k", Value: "brokerglobal group"},
+		{Name: "l", Value: "broker group"},
+		{Name: "m", Value: "globalgroupbroker"},
+		{Name: "n", Value: ""},
+		{Name: "o", Value: "brokergroupglobal"},
+	}
+
+	broker := v1beta1.Broker{
+		Id:                0,
+		BrokerConfigGroup: "default",
+		BrokerConfig: &v1beta1.BrokerConfig{
+			Envs: []corev1.EnvVar{
+				{Name: "c", Value: "broker"},
+				{Name: " +f", Value: "broker"},
+				{Name: "h ", Value: "broker"},
+				{Name: "+i", Value: "broker"},
+				{Name: "j", Value: "broker"},
+				{Name: "k+ ", Value: "broker"},
+				{Name: "l+", Value: "broker"},
+				{Name: " +m", Value: "broker"},
+				{Name: "o+ ", Value: "broker"},
+			},
+		},
+	}
+
+	spec := v1beta1.KafkaClusterSpec{
+		Envs: []corev1.EnvVar{
+			{Name: "a", Value: "global"},
+			{Name: "+d", Value: "global"},
+			{Name: "g", Value: "global"},
+			{Name: "+i", Value: "global"},
+			{Name: "j", Value: "global"},
+			{Name: "k", Value: "global"},
+			{Name: "+m", Value: "global"},
+			{Name: "n", Value: ""},
+			{Name: " o+ ", Value: "global"},
+		},
+		BrokerConfigGroups: map[string]v1beta1.BrokerConfig{
+			"default": {
+				Envs: []corev1.EnvVar{
+					{Name: "b", Value: "group"},
+					{Name: "+e", Value: "group"},
+					{Name: "g", Value: "group"},
+					{Name: "h", Value: "group"},
+					{Name: "i", Value: "group"},
+					{Name: " +j", Value: "group"},
+					{Name: "+k", Value: " group"},
+					{Name: " +l", Value: " group"},
+					{Name: "+m", Value: "group"},
+					{Name: " o+", Value: "group"},
+				},
+			},
+		},
+	}
+
+	brokerConfig, err := broker.GetBrokerConfig(spec)
+	if err != nil {
+		t.Error("GetBrokerConfig failed")
+	}
+
+	result := generateEnvConfig(brokerConfig, []corev1.EnvVar{
+		{Name: "b", Value: "code"},
+		{Name: "e", Value: "code"},
+	})
+	if err != nil {
+		t.Error("Error GetBrokerConfig throw an unexpected error")
+	}
+	if !reflect.DeepEqual(result, expected) {
+		t.Error("Expected:", expected, "Got:", result)
+	}
+}
diff --git a/pkg/sdk/v1beta1/kafkacluster_types_test.go b/pkg/sdk/v1beta1/kafkacluster_types_test.go
index c85be9b4..30601acd 100644
--- a/pkg/sdk/v1beta1/kafkacluster_types_test.go
+++ b/pkg/sdk/v1beta1/kafkacluster_types_test.go
@@ -52,19 +52,21 @@ func TestGetBrokerConfigAffinityMergeBrokerNodeAffinityWithGroupsAntiAffinity(t
 		},
 	}
 
-	brokerConfigGroups := map[string]BrokerConfig{
-		"default": {
-			Affinity: &corev1.Affinity{
-				NodeAffinity: &corev1.NodeAffinity{
-					RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{
-						NodeSelectorTerms: []corev1.NodeSelectorTerm{
-							{
-								MatchExpressions: nil,
-								MatchFields: []corev1.NodeSelectorRequirement{
-									{
-										Key:      "fruit",
-										Operator: "in",
-										Values:   []string{"apple"},
+	spec := KafkaClusterSpec{
+		BrokerConfigGroups: map[string]BrokerConfig{
+			"default": {
+				Affinity: &corev1.Affinity{
+					NodeAffinity: &corev1.NodeAffinity{
+						RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{
+							NodeSelectorTerms: []corev1.NodeSelectorTerm{
+								{
+									MatchExpressions: nil,
+									MatchFields: []corev1.NodeSelectorRequirement{
+										{
+											Key:      "fruit",
+											Operator: "in",
+											Values:   []string{"apple"},
+										},
 									},
 								},
 							},
@@ -94,7 +96,7 @@ func TestGetBrokerConfigAffinityMergeBrokerNodeAffinityWithGroupsAntiAffinity(t
 		},
 	}
 
-	config, err := broker.GetBrokerConfig(brokerConfigGroups)
+	config, err := broker.GetBrokerConfig(spec)
 	if err != nil {
 		t.Error(err)
 	}
@@ -128,29 +130,31 @@ func TestGetBrokerConfigAffinityMergeEmptyBrokerConfigWithDefaultConfig(t *testi
 		},
 	}
 
-	brokerConfigGroups := map[string]BrokerConfig{
-		"default": {
-			Affinity: &corev1.Affinity{
-				PodAntiAffinity: &corev1.PodAntiAffinity{
-					RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
-						{
-							LabelSelector: &metav1.LabelSelector{
-								MatchLabels: map[string]string{"app": "kafka", "kafka_cr": "kafka_config_group"},
+	spec := KafkaClusterSpec{
+		BrokerConfigGroups: map[string]BrokerConfig{
+			"default": {
+				Affinity: &corev1.Affinity{
+					PodAntiAffinity: &corev1.PodAntiAffinity{
+						RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
+							{
+								LabelSelector: &metav1.LabelSelector{
+									MatchLabels: map[string]string{"app": "kafka", "kafka_cr": "kafka_config_group"},
+								},
+								TopologyKey: "kubernetes.io/hostname",
 							},
-							TopologyKey: "kubernetes.io/hostname",
 						},
 					},
-				},
-				NodeAffinity: &corev1.NodeAffinity{
-					RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{
-						NodeSelectorTerms: []corev1.NodeSelectorTerm{
-							{
-								MatchExpressions: nil,
-								MatchFields: []corev1.NodeSelectorRequirement{
-									{
-										Key:      "fruit",
-										Operator: "in",
-										Values:   []string{"apple"},
+					NodeAffinity: &corev1.NodeAffinity{
+						RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{
+							NodeSelectorTerms: []corev1.NodeSelectorTerm{
+								{
+									MatchExpressions: nil,
+									MatchFields: []corev1.NodeSelectorRequirement{
+										{
+											Key:      "fruit",
+											Operator: "in",
+											Values:   []string{"apple"},
+										},
 									},
 								},
 							},
@@ -166,7 +170,7 @@ func TestGetBrokerConfigAffinityMergeEmptyBrokerConfigWithDefaultConfig(t *testi
 		BrokerConfigGroup: "default",
 	}
 
-	config, err := broker.GetBrokerConfig(brokerConfigGroups)
+	config, err := broker.GetBrokerConfig(spec)
 	if err != nil {
 		t.Error(err)
 	}
@@ -211,16 +215,18 @@ func TestGetBrokerConfigAffinityMergeEqualPodAntiAffinity(t *testing.T) {
 		},
 	}
 
-	brokerConfigGroups := map[string]BrokerConfig{
-		"default": {
-			Affinity: &corev1.Affinity{
-				PodAntiAffinity: &corev1.PodAntiAffinity{
-					RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
-						{
-							LabelSelector: &metav1.LabelSelector{
-								MatchLabels: map[string]string{"app": "kafka", "kafka_cr": "kafka_config_group"},
+	spec := KafkaClusterSpec{
+		BrokerConfigGroups: map[string]BrokerConfig{
+			"default": {
+				Affinity: &corev1.Affinity{
+					PodAntiAffinity: &corev1.PodAntiAffinity{
+						RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
+							{
+								LabelSelector: &metav1.LabelSelector{
+									MatchLabels: map[string]string{"app": "kafka", "kafka_cr": "kafka_config_group"},
+								},
+								TopologyKey: "kubernetes.io/hostname",
 							},
-							TopologyKey: "kubernetes.io/hostname",
 						},
 					},
 				},
@@ -228,7 +234,7 @@ func TestGetBrokerConfigAffinityMergeEqualPodAntiAffinity(t *testing.T) {
 		},
 	}
 
-	config, err := broker.GetBrokerConfig(brokerConfigGroups)
+	config, err := broker.GetBrokerConfig(spec)
 	if err != nil {
 		t.Error(err)
 	}
@@ -275,16 +281,17 @@ func TestGetBrokerConfig(t *testing.T) {
 			},
 		},
 	}
-
-	brokerConfigGroups := map[string]BrokerConfig{
-		"default": {
-			StorageConfigs: []StorageConfig{
-				{
-					MountPath: "kafka-test1/log",
-					PvcSpec: &corev1.PersistentVolumeClaimSpec{
-						AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
-						Resources: corev1.ResourceRequirements{
-							Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("10Gi")},
+	spec := KafkaClusterSpec{
+		BrokerConfigGroups: map[string]BrokerConfig{
+			"default": {
+				StorageConfigs: []StorageConfig{
+					{
+						MountPath: "kafka-test1/log",
+						PvcSpec: &corev1.PersistentVolumeClaimSpec{
+							AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
+							Resources: corev1.ResourceRequirements{
+								Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("10Gi")},
+							},
 						},
 					},
 				},
@@ -292,7 +299,7 @@ func TestGetBrokerConfig(t *testing.T) {
 		},
 	}
 
-	result, err := broker.GetBrokerConfig(brokerConfigGroups)
+	result, err := broker.GetBrokerConfig(spec)
 	if err != nil {
 		t.Error("Error GetBrokerConfig throw an unexpected error")
 	}
@@ -342,24 +349,26 @@ func TestGetBrokerConfigUniqueStorage(t *testing.T) {
 		},
 	}
 
-	brokerConfigGroups := map[string]BrokerConfig{
-		"default": {
-			StorageConfigs: []StorageConfig{
-				{
-					MountPath: "kafka-test1/log",
-					PvcSpec: &corev1.PersistentVolumeClaimSpec{
-						AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
-						Resources: corev1.ResourceRequirements{
-							Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("10Gi")},
+	spec := KafkaClusterSpec{
+		BrokerConfigGroups: map[string]BrokerConfig{
+			"default": {
+				StorageConfigs: []StorageConfig{
+					{
+						MountPath: "kafka-test1/log",
+						PvcSpec: &corev1.PersistentVolumeClaimSpec{
+							AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
+							Resources: corev1.ResourceRequirements{
+								Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("10Gi")},
+							},
 						},
 					},
-				},
-				{
-					MountPath: "kafka-test/log",
-					PvcSpec: &corev1.PersistentVolumeClaimSpec{
-						AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
-						Resources: corev1.ResourceRequirements{
-							Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("10Gi")},
+					{
+						MountPath: "kafka-test/log",
+						PvcSpec: &corev1.PersistentVolumeClaimSpec{
+							AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
+							Resources: corev1.ResourceRequirements{
+								Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("10Gi")},
+							},
 						},
 					},
 				},
@@ -367,7 +376,48 @@ func TestGetBrokerConfigUniqueStorage(t *testing.T) {
 		},
 	}
 
-	result, err := broker.GetBrokerConfig(brokerConfigGroups)
+	result, err := broker.GetBrokerConfig(spec)
+	if err != nil {
+		t.Error("Error GetBrokerConfig throw an unexpected error")
+	}
+	if !reflect.DeepEqual(result, expected) {
+		t.Error("Expected:", expected, "Got:", result)
+	}
+}
+
+func TestGetBrokerConfigEnvs(t *testing.T) {
+	expected := &BrokerConfig{
+		Envs: []corev1.EnvVar{
+			{Name: "VAR", Value: "cluster"},
+			{Name: "VAR", Value: "group"},
+			{Name: "VAR", Value: "broker"},
+		},
+	}
+
+	broker := Broker{
+		Id:                0,
+		BrokerConfigGroup: "default",
+		BrokerConfig: &BrokerConfig{
+			Envs: []corev1.EnvVar{
+				{Name: "VAR", Value: "broker"},
+			},
+		},
+	}
+
+	spec := KafkaClusterSpec{
+		Envs: []corev1.EnvVar{
+			{Name: "VAR", Value: "cluster"},
+		},
+		BrokerConfigGroups: map[string]BrokerConfig{
+			"default": {
+				Envs: []corev1.EnvVar{
+					{Name: "VAR", Value: "group"},
+				},
+			},
+		},
+	}
+
+	result, err := broker.GetBrokerConfig(spec)
 	if err != nil {
 		t.Error("Error GetBrokerConfig throw an unexpected error")
 	}