diff --git a/controllers/cloud.redhat.com/clowdenvironment_reconciliation.go b/controllers/cloud.redhat.com/clowdenvironment_reconciliation.go
index 916f03928..ff12c573b 100644
--- a/controllers/cloud.redhat.com/clowdenvironment_reconciliation.go
+++ b/controllers/cloud.redhat.com/clowdenvironment_reconciliation.go
@@ -130,11 +130,12 @@ func (r *ClowdEnvironmentReconciliation) markedForDeletion() (ctrl.Result, error
 func (r *ClowdEnvironmentReconciliation) finalizeEnvironmentImplementation() error {
 
 	provider := providers.Provider{
-		Ctx:    r.ctx,
-		Client: r.client,
-		Env:    r.env,
-		Cache:  r.cache,
-		Log:    *r.log,
+		Ctx:       r.ctx,
+		Client:    r.client,
+		Env:       r.env,
+		Cache:     r.cache,
+		Log:       *r.log,
+		HashCache: r.hashCache,
 	}
 
 	err := runProvidersForEnvFinalize(*r.log, provider)
@@ -290,11 +291,12 @@ func (r *ClowdEnvironmentReconciliation) runProviders() (ctrl.Result, error) {
 	r.hashCache.RemoveClowdObjectFromObjects(r.env)
 
 	provider := providers.Provider{
-		Ctx:    r.ctx,
-		Client: r.client,
-		Env:    r.env,
-		Cache:  r.cache,
-		Log:    *r.log,
+		Ctx:       r.ctx,
+		Client:    r.client,
+		Env:       r.env,
+		Cache:     r.cache,
+		Log:       *r.log,
+		HashCache: r.hashCache,
 	}
 	provErr := runProvidersForEnv(*r.log, provider)
 
diff --git a/controllers/cloud.redhat.com/providers/kafka/msk.go b/controllers/cloud.redhat.com/providers/kafka/msk.go
index 54edf0145..1cb730516 100644
--- a/controllers/cloud.redhat.com/providers/kafka/msk.go
+++ b/controllers/cloud.redhat.com/providers/kafka/msk.go
@@ -242,6 +242,14 @@ func (s *mskProvider) configureListeners() error {
 		return err
 	}
 
+	if _, err := s.HashCache.CreateOrUpdateObject(secret, true); err != nil {
+		return err
+	}
+
+	if err := s.HashCache.AddClowdObjectToObject(s.Env, secret); err != nil {
+		return err
+	}
+
 	brokers, err = getBrokerConfig(secret)
 	if err != nil {
 		return err
diff --git a/tests/kuttl/test-clowdapp-watcher-kafka-msk/00-install.yaml b/tests/kuttl/test-clowdapp-watcher-kafka-msk/00-install.yaml
new file mode 100644
index 000000000..ab768d0e8
--- /dev/null
+++ b/tests/kuttl/test-clowdapp-watcher-kafka-msk/00-install.yaml
@@ -0,0 +1,23 @@
+apiVersion: v1
+kind: Namespace
+metadata:
+  name: test-clowdapp-watcher-kafka-msk
+spec:
+  finalizers:
+  - kubernetes
+---
+apiVersion: v1
+kind: Namespace
+metadata:
+  name: test-clowdapp-watcher-kafka-msk-sec-source
+spec:
+  finalizers:
+  - kubernetes
+---
+apiVersion: v1
+kind: Namespace
+metadata:
+  name: test-clowdapp-watcher-kafka-msk-env
+spec:
+  finalizers:
+  - kubernetes
diff --git a/tests/kuttl/test-clowdapp-watcher-kafka-msk/01-assert.yaml b/tests/kuttl/test-clowdapp-watcher-kafka-msk/01-assert.yaml
new file mode 100644
index 000000000..0257b5550
--- /dev/null
+++ b/tests/kuttl/test-clowdapp-watcher-kafka-msk/01-assert.yaml
@@ -0,0 +1,26 @@
+---
+apiVersion: kafka.strimzi.io/v1beta2
+kind: Kafka
+metadata:
+  name: test-clowdapp-watcher-kafka-msk
+  namespace: test-clowdapp-watcher-kafka-msk
+status:
+  conditions:
+  - reason: ZooKeeperStorage
+    status: "True"
+    type: Warning
+  - reason: KafkaStorage
+    status: "True"
+    type: Warning
+  - status: "True"
+    type: Ready
+---
+apiVersion: kafka.strimzi.io/v1beta2
+kind: KafkaUser
+metadata:
+  labels:
+    strimzi.io/cluster: test-clowdapp-watcher-kafka-msk
+  name: test-clowdapp-watcher-kafka-msk-connect
+  namespace: test-clowdapp-watcher-kafka-msk
+status:
+  username: test-clowdapp-watcher-kafka-msk-connect
diff --git a/tests/kuttl/test-clowdapp-watcher-kafka-msk/01-pods.yaml b/tests/kuttl/test-clowdapp-watcher-kafka-msk/01-pods.yaml
new file mode 100644
index 000000000..5fd345a4f
--- /dev/null
+++ b/tests/kuttl/test-clowdapp-watcher-kafka-msk/01-pods.yaml
@@ -0,0 +1,335 @@
+---
+apiVersion: kafka.strimzi.io/v1beta2
+kind: Kafka
+metadata:
+  name: test-clowdapp-watcher-kafka-msk
+  namespace: test-clowdapp-watcher-kafka-msk
+spec:
+  entityOperator:
+    tlsSidecar:
+      resources:
+        limits:
+          cpu: 100m
+          memory: 100Mi
+        requests:
+          cpu: 50m
+          memory: 50Mi
+    userOperator:
+      resources:
+        limits:
+          cpu: 400m
+          memory: 500Mi
+        requests:
+          cpu: 50m
+          memory: 250Mi
+  kafka:
+    authorization:
+      type: simple
+    config:
+      offsets.topic.replication.factor: 1
+    jvmOptions: {}
+    listeners:
+    - authentication:
+        type: scram-sha-512
+      name: tls
+      port: 9093
+      tls: true
+      type: internal
+    metricsConfig:
+      type: jmxPrometheusExporter
+      valueFrom:
+        configMapKeyRef:
+          key: metrics
+          name: test-clowdapp-watcher-kafka-msk-metrics
+          optional: false
+    replicas: 1
+    resources:
+      limits:
+        cpu: 500m
+        memory: 1Gi
+      requests:
+        cpu: 250m
+        memory: 600Mi
+    storage:
+      type: ephemeral
+    version: 3.4.0
+  zookeeper:
+    replicas: 1
+    resources:
+      limits:
+        cpu: 350m
+        memory: 800Mi
+      requests:
+        cpu: 200m
+        memory: 400Mi
+    storage:
+      type: ephemeral
+---
+apiVersion: kafka.strimzi.io/v1beta2
+kind: KafkaUser
+metadata:
+  labels:
+    strimzi.io/cluster: test-clowdapp-watcher-kafka-msk
+  name: test-clowdapp-watcher-kafka-msk-connect
+  namespace: test-clowdapp-watcher-kafka-msk
+spec:
+  authentication:
+    type: scram-sha-512
+  authorization:
+    acls:
+    - host: '*'
+      operation: All
+      resource:
+        name: '*'
+        patternType: literal
+        type: topic
+    - host: '*'
+      operation: All
+      resource:
+        name: '*'
+        patternType: literal
+        type: group
+    - host: '*'
+      operation: All
+      resource:
+        name: '*'
+        patternType: literal
+        type: cluster
+    type: simple
+---
+apiVersion: v1
+data:
+  metrics: |-
+    {
+        "metrics": {
+          "lowercaseOutputName": true,
+          "rules": [
+            {
+              "labels": {
+                "clientId": "$3",
+                "partition": "$5",
+                "topic": "$4"
+              },
+              "name": "kafka_server_$1_$2",
+              "pattern": "kafka.server<type=(.+), name=(.+), clientId=(.+), topic=(.+), partition=(.*)><>Value",
+              "type": "GAUGE"
+            },
+            {
+              "labels": {
+                "broker": "$4:$5",
+                "clientId": "$3"
+              },
+              "name": "kafka_server_$1_$2",
+              "pattern": "kafka.server<type=(.+), name=(.+), clientId=(.+), brokerHost=(.+), brokerPort=(.+)><>Value",
+              "type": "GAUGE"
+            },
+            {
+              "labels": {
+                "cipher": "$5",
+                "listener": "$2",
+                "networkProcessor": "$3",
+                "protocol": "$4"
+              },
+              "name": "kafka_server_$1_connections_tls_info",
+              "pattern": "kafka.server<type=(.+), cipher=(.+), protocol=(.+), listener=(.+), networkProcessor=(.+)><>connections",
+              "type": "GAUGE"
+            },
+            {
+              "labels": {
+                "clientSoftwareName": "$2",
+                "clientSoftwareVersion": "$3",
+                "listener": "$4",
+                "networkProcessor": "$5"
+              },
+              "name": "kafka_server_$1_connections_software",
+              "pattern": "kafka.server<type=(.+), clientSoftwareName=(.+), clientSoftwareVersion=(.+), listener=(.+), networkProcessor=(.+)><>connections",
+              "type": "GAUGE"
+            },
+            {
+              "labels": {
+                "listener": "$2",
+                "networkProcessor": "$3"
+              },
+              "name": "kafka_server_$1_$4",
+              "pattern": "kafka.server<type=(.+), listener=(.+), networkProcessor=(.+)><>(.+):",
+              "type": "GAUGE"
+            },
+            {
+              "labels": {
+                "listener": "$2",
+                "networkProcessor": "$3"
+              },
+              "name": "kafka_server_$1_$4",
+              "pattern": "kafka.server<type=(.+), listener=(.+), networkProcessor=(.+)><>(.+)",
+              "type": "GAUGE"
+            },
+            {
+              "name": "kafka_$1_$2_$3_percent",
+              "pattern": "kafka.(\\w+)<type=(.+), name=(.+)Percent\\w*><>MeanRate",
+              "type": "GAUGE"
+            },
+            {
+              "name": "kafka_$1_$2_$3_percent",
+              "pattern": "kafka.(\\w+)<type=(.+), name=(.+)Percent\\w*><>Value",
+              "type": "GAUGE"
+            },
+            {
+              "labels": {
+                "$4": "$5"
+              },
+              "name": "kafka_$1_$2_$3_percent",
+              "pattern": "kafka.(\\w+)<type=(.+), name=(.+)Percent\\w*, (.+)=(.+)><>Value",
+              "type": "GAUGE"
+            },
+            {
+              "labels": {
+                "$4": "$5",
+                "$6": "$7"
+              },
+              "name": "kafka_$1_$2_$3_total",
+              "pattern": "kafka.(\\w+)<type=(.+), name=(.+)PerSec\\w*, (.+)=(.+), (.+)=(.+)><>Count",
+              "type": "COUNTER"
+            },
+            {
+              "labels": {
+                "$4": "$5"
+              },
+              "name": "kafka_$1_$2_$3_total",
+              "pattern": "kafka.(\\w+)<type=(.+), name=(.+)PerSec\\w*, (.+)=(.+)><>Count",
+              "type": "COUNTER"
+            },
+            {
+              "name": "kafka_$1_$2_$3_total",
+              "pattern": "kafka.(\\w+)<type=(.+), name=(.+)PerSec\\w*><>Count",
+              "type": "COUNTER"
+            },
+            {
+              "labels": {
+                "$4": "$5",
+                "$6": "$7"
+              },
+              "name": "kafka_$1_$2_$3",
+              "pattern": "kafka.(\\w+)<type=(.+), name=(.+), (.+)=(.+), (.+)=(.+)><>Value",
+              "type": "GAUGE"
+            },
+            {
+              "labels": {
+                "$4": "$5"
+              },
+              "name": "kafka_$1_$2_$3",
+              "pattern": "kafka.(\\w+)<type=(.+), name=(.+), (.+)=(.+)><>Value",
+              "type": "GAUGE"
+            },
+            {
+              "name": "kafka_$1_$2_$3",
+              "pattern": "kafka.(\\w+)<type=(.+), name=(.+)><>Value",
+              "type": "GAUGE"
+            },
+            {
+              "labels": {
+                "$4": "$5",
+                "$6": "$7"
+              },
+              "name": "kafka_$1_$2_$3_count",
+              "pattern": "kafka.(\\w+)<type=(.+), name=(.+), (.+)=(.+), (.+)=(.+)><>Count",
+              "type": "COUNTER"
+            },
+            {
+              "labels": {
+                "$4": "$5",
+                "$6": "$7",
+                "quantile": "0.$8"
+              },
+              "name": "kafka_$1_$2_$3",
+              "pattern": "kafka.(\\w+)<type=(.+), name=(.+), (.+)=(.*), (.+)=(.+)><>(\\d+)thPercentile",
+              "type": "GAUGE"
+            },
+            {
+              "labels": {
+                "$4": "$5"
+              },
+              "name": "kafka_$1_$2_$3_count",
+              "pattern": "kafka.(\\w+)<type=(.+), name=(.+), (.+)=(.+)><>Count",
+              "type": "COUNTER"
+            },
+            {
+              "labels": {
+                "$4": "$5",
+                "quantile": "0.$6"
+              },
+              "name": "kafka_$1_$2_$3",
+              "pattern": "kafka.(\\w+)<type=(.+), name=(.+), (.+)=(.*)><>(\\d+)thPercentile",
+              "type": "GAUGE"
+            },
+            {
+              "name": "kafka_$1_$2_$3_count",
+              "pattern": "kafka.(\\w+)<type=(.+), name=(.+)><>Count",
+              "type": "COUNTER"
+            },
+            {
+              "labels": {
+                "quantile": "0.$4"
+              },
+              "name": "kafka_$1_$2_$3",
+              "pattern": "kafka.(\\w+)<type=(.+), name=(.+)><>(\\d+)thPercentile",
+              "type": "GAUGE"
+            }
+          ]
+        }
+      }
+kind: ConfigMap
+metadata:
+  name: test-clowdapp-watcher-kafka-msk-metrics
+  namespace: test-clowdapp-watcher-kafka-msk
+---
+kind: ServiceAccount
+apiVersion: v1
+metadata:
+  name: strimzi-topic-operator
+  namespace: test-clowdapp-watcher-kafka-msk-sec-source
+  labels:
+    app: strimzi
+---
+kind: Role
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: strimzi-topic-operator
+  namespace: test-clowdapp-watcher-kafka-msk-sec-source
+  labels:
+    app: strimzi
+rules:
+  - verbs:
+      - get
+      - list
+      - watch
+      - create
+      - patch
+      - update
+      - delete
+    apiGroups:
+      - kafka.strimzi.io
+    resources:
+      - kafkatopics
+      - kafkatopics/status
+  - verbs:
+      - create
+    apiGroups:
+      - ''
+    resources:
+      - events
+---
+kind: RoleBinding
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: strimzi-topic-operator
+  namespace: test-clowdapp-watcher-kafka-msk-sec-source
+  labels:
+    app: strimzi
+subjects:
+  - kind: ServiceAccount
+    name: strimzi-topic-operator
+    namespace: test-clowdapp-watcher-kafka-msk-sec-source
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: Role
+  name: strimzi-topic-operator
diff --git a/tests/kuttl/test-clowdapp-watcher-kafka-msk/02-json-asserts.yaml b/tests/kuttl/test-clowdapp-watcher-kafka-msk/02-json-asserts.yaml
new file mode 100644
index 000000000..769be0070
--- /dev/null
+++ b/tests/kuttl/test-clowdapp-watcher-kafka-msk/02-json-asserts.yaml
@@ -0,0 +1,13 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+commands:
+- script: sleep 5
+- script: kubectl get secret --namespace=test-clowdapp-watcher-kafka-msk test-clowdapp-watcher-kafka-msk-connect -o json > /tmp/test-clowdapp-watcher-kafka-msk-user
+- script: kubectl get secret --namespace=test-clowdapp-watcher-kafka-msk test-clowdapp-watcher-kafka-msk-cluster-ca-cert -o json > /tmp/test-clowdapp-watcher-kafka-msk-cluster-ca-cert
+
+- script: sh create_json.sh
+- script: sh create_certs.sh
+- script: kubectl apply -f /tmp/watcher-managed-secret.yaml -n test-clowdapp-watcher-kafka-msk-sec-source
+- script: kubectl apply -f /tmp/test-clowdapp-watcher-kafka-msk-ca-cert.yaml -n test-clowdapp-watcher-kafka-msk-sec-source
+- script: kubectl apply -f /tmp/test-clowdapp-watcher-kafka-msk-connect-user.yaml -n test-clowdapp-watcher-kafka-msk-sec-source
diff --git a/tests/kuttl/test-clowdapp-watcher-kafka-msk/03-assert.yaml b/tests/kuttl/test-clowdapp-watcher-kafka-msk/03-assert.yaml
new file mode 100644
index 000000000..8ca4985b5
--- /dev/null
+++ b/tests/kuttl/test-clowdapp-watcher-kafka-msk/03-assert.yaml
@@ -0,0 +1,75 @@
+---
+apiVersion: v1
+kind: Secret
+metadata:
+  name: puptoo
+  namespace: test-clowdapp-watcher-kafka-msk-env
+  labels:
+    app: puptoo
+  ownerReferences:
+  - apiVersion: cloud.redhat.com/v1alpha1
+    kind: ClowdApp
+    name: puptoo
+type: Opaque
+---
+kind: Deployment
+apiVersion: apps/v1
+metadata:
+  name: strimzi-topic-operator
+  namespace: test-clowdapp-watcher-kafka-msk-sec-source
+  labels:
+    app: strimzi
+spec:
+  template:
+    spec:
+      serviceAccountName: strimzi-topic-operator
+      containers:
+      - name: strimzi-topic-operator
+        env:
+        - name: STRIMZI_NAMESPACE
+          value: test-clowdapp-watcher-kafka-msk-sec-source
+        - name: STRIMZI_RESOURCE_LABELS
+          value: strimzi.io/cluster=test-clowdapp-watcher-kafka-msk
+        - name: STRIMZI_KAFKA_BOOTSTRAP_SERVERS
+          value: test-clowdapp-watcher-kafka-msk-kafka-bootstrap.test-clowdapp-watcher-kafka-msk.svc:9093
+        - name: STRIMZI_FULL_RECONCILIATION_INTERVAL_MS
+          value: '120000'
+        - name: STRIMZI_LOG_LEVEL
+          value: INFO
+        - name: STRIMZI_TLS_ENABLED
+          value: 'false'
+        - name: STRIMZI_TLS_AUTH_ENABLED
+          value: 'false'
+        - name: STRIMZI_JAVA_OPTS
+          value: '-Xmx512M -Xms256M'
+        - name: STRIMZI_PUBLIC_CA
+          value: 'false'
+        - name: STRIMZI_SASL_ENABLED
+          value: 'true'
+        - name: STRIMZI_SASL_USERNAME
+          value: test-clowdapp-watcher-kafka-msk-connect
+        - name: STRIMZI_SASL_PASSWORD
+        - name: STRIMZI_SASL_MECHANISM
+          value: scram-sha-512
+        - name: STRIMZI_SECURITY_PROTOCOL
+          value: SASL_SSL
+        - name: STRIMZI_USE_FINALIZERS
+          value: 'true'
+status:
+  conditions:
+  - reason: MinimumReplicasAvailable
+    status: "True"
+    type: Available
+  - reason: NewReplicaSetAvailable
+    status: "True"
+    type: Progressing
+---
+kind: KafkaTopic
+apiVersion: kafka.strimzi.io/v1beta2
+metadata:
+  name: test-clowdapp-watcher-kafka-msk-topic-one
+  namespace: test-clowdapp-watcher-kafka-msk-sec-source
+status:
+  conditions:
+  - status: "True"
+    type: Ready
diff --git a/tests/kuttl/test-clowdapp-watcher-kafka-msk/03-pods.yaml b/tests/kuttl/test-clowdapp-watcher-kafka-msk/03-pods.yaml
new file mode 100644
index 000000000..5a3cc9cda
--- /dev/null
+++ b/tests/kuttl/test-clowdapp-watcher-kafka-msk/03-pods.yaml
@@ -0,0 +1,172 @@
+---
+apiVersion: cloud.redhat.com/v1alpha1
+kind: ClowdEnvironment
+metadata:
+  name: test-clowdapp-watcher-kafka-msk
+spec:
+  targetNamespace: test-clowdapp-watcher-kafka-msk-env
+  providers:
+    web:
+      port: 8000
+      mode: operator
+    metrics:
+      port: 9000
+      mode: operator
+      path: "/metrics"
+    kafka:
+      mode: ephem-msk
+      managedSecretRef:
+        name: managed-secret
+        namespace: test-clowdapp-watcher-kafka-msk-sec-source
+      clusterAnnotation: test-clowdapp-watcher-kafka-msk
+      topicNamespace: test-clowdapp-watcher-kafka-msk-sec-source
+      kafkaConnectReplicaCount: 1
+    db:
+      mode: none
+    logging:
+      mode: none
+    objectStore:
+      mode: none
+    inMemoryDb:
+      mode: none
+    featureFlags:
+      mode: none
+  resourceDefaults:
+    limits:
+      cpu: 400m
+      memory: 1024Mi
+    requests:
+      cpu: 30m
+      memory: 512Mi
+---
+apiVersion: cloud.redhat.com/v1alpha1
+kind: ClowdApp
+metadata:
+  name: puptoo
+  namespace: test-clowdapp-watcher-kafka-msk-env
+spec:
+  envName: test-clowdapp-watcher-kafka-msk
+  deployments:
+  - name: processor
+    podSpec:
+      image: quay.io/psav/clowder-hello
+  kafkaTopics:
+    - replicas: 3
+      partitions: 64
+      topicName: topic-one
+    - replicas: 5
+      partitions: 32
+      topicName: topic-two
+---
+kind: Deployment
+apiVersion: apps/v1
+metadata:
+  name: strimzi-topic-operator
+  namespace: test-clowdapp-watcher-kafka-msk-sec-source
+  labels:
+    app: strimzi
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      name: strimzi-topic-operator
+  template:
+    metadata:
+      creationTimestamp: null
+      labels:
+        name: strimzi-topic-operator
+    spec:
+      serviceAccountName: strimzi-topic-operator
+      serviceAccount: strimzi-topic-operator
+      restartPolicy: Always
+      schedulerName: default-scheduler
+      terminationGracePeriodSeconds: 30
+      securityContext: {}
+      containers:
+        - resources:
+            limits:
+              cpu: 500m
+              memory: 256Mi
+            requests:
+              cpu: 100m
+              memory: 256Mi
+          readinessProbe:
+            httpGet:
+              path: /ready
+              port: 8080
+              scheme: HTTP
+            initialDelaySeconds: 10
+            timeoutSeconds: 1
+            periodSeconds: 30
+            successThreshold: 1
+            failureThreshold: 3
+          terminationMessagePath: /dev/termination-log
+          name: strimzi-topic-operator
+          livenessProbe:
+            httpGet:
+              path: /healthy
+              port: 8080
+              scheme: HTTP
+            initialDelaySeconds: 10
+            timeoutSeconds: 1
+            periodSeconds: 30
+            successThreshold: 1
+            failureThreshold: 3
+          env:
+            - name: STRIMZI_NAMESPACE
+              value: test-clowdapp-watcher-kafka-msk-sec-source
+            - name: STRIMZI_RESOURCE_LABELS
+              value: strimzi.io/cluster=test-clowdapp-watcher-kafka-msk
+            - name: STRIMZI_KAFKA_BOOTSTRAP_SERVERS
+              value: test-clowdapp-watcher-kafka-msk-kafka-bootstrap.test-clowdapp-watcher-kafka-msk.svc:9093
+            - name: STRIMZI_FULL_RECONCILIATION_INTERVAL_MS
+              value: '120000'
+            - name: STRIMZI_LOG_LEVEL
+              value: INFO
+            - name: STRIMZI_TLS_ENABLED
+              value: 'false'
+            - name: STRIMZI_TLS_AUTH_ENABLED
+              value: 'false'
+            - name: STRIMZI_JAVA_OPTS
+              value: '-Xmx512M -Xms256M'
+            - name: STRIMZI_PUBLIC_CA
+              value: 'false'
+            - name: STRIMZI_SASL_ENABLED
+              value: 'true'
+            - name: STRIMZI_SASL_USERNAME
+              value: test-clowdapp-watcher-kafka-msk-connect
+            - name: STRIMZI_SASL_PASSWORD
+              valueFrom:
+                secretKeyRef:
+                  name: test-clowdapp-watcher-kafka-msk-connect
+                  key: password
+            - name: STRIMZI_SASL_MECHANISM
+              value: scram-sha-512
+            - name: STRIMZI_SECURITY_PROTOCOL
+              value: SASL_SSL
+            - name: STRIMZI_USE_FINALIZERS
+              value: 'true'
+          imagePullPolicy: IfNotPresent
+          volumeMounts:
+            - name: strimzi-tmp
+              mountPath: /tmp
+            - name: ca
+              mountPath: /etc/tls-sidecar/cluster-ca-certs
+          terminationMessagePolicy: File
+          image: >-
+            quay.io/strimzi/operator:0.37.0
+          args:
+            - /opt/strimzi/bin/topic_operator_run.sh
+      volumes:
+        - name: strimzi-tmp
+          emptyDir:
+            medium: Memory
+            sizeLimit: 5Mi
+        - name: ca
+          secret:
+            secretName: test-clowdapp-watcher-kafka-msk-cluster-ca-cert
+      dnsPolicy: ClusterFirst
+  strategy:
+    type: Recreate
+  revisionHistoryLimit: 10
+  progressDeadlineSeconds: 600
diff --git a/tests/kuttl/test-clowdapp-watcher-kafka-msk/04-json-asserts.yaml b/tests/kuttl/test-clowdapp-watcher-kafka-msk/04-json-asserts.yaml
new file mode 100644
index 000000000..0f4e66154
--- /dev/null
+++ b/tests/kuttl/test-clowdapp-watcher-kafka-msk/04-json-asserts.yaml
@@ -0,0 +1,10 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+commands:
+- script: sleep 5
+- script: kubectl get secret --namespace=test-clowdapp-watcher-kafka-msk-env puptoo -o json > /tmp/test-clowdapp-watcher-kafka-msk-env
+- script: jq -r '.data["cdappconfig.json"]' < /tmp/test-clowdapp-watcher-kafka-msk-env | base64 -d > /tmp/test-clowdapp-watcher-kafka-msk-env-json
+
+- script: jq -r '.kafka.brokers[0].hostname == "test-clowdapp-watcher-kafka-msk-kafka-bootstrap.test-clowdapp-watcher-kafka-msk.svc"' -e < /tmp/test-clowdapp-watcher-kafka-msk-env-json
+- script: jq -r '.kafka.brokers[0].sasl.username == "test-clowdapp-watcher-kafka-msk-connect"' -e < /tmp/test-clowdapp-watcher-kafka-msk-env-json
diff --git a/tests/kuttl/test-clowdapp-watcher-kafka-msk/05-assert.yaml b/tests/kuttl/test-clowdapp-watcher-kafka-msk/05-assert.yaml
new file mode 100644
index 000000000..34698a200
--- /dev/null
+++ b/tests/kuttl/test-clowdapp-watcher-kafka-msk/05-assert.yaml
@@ -0,0 +1,23 @@
+---
+apiVersion: v1
+kind: Secret
+metadata:
+  name: puptoo
+  namespace: test-clowdapp-watcher-kafka-msk-env
+  labels:
+    app: puptoo
+  ownerReferences:
+  - apiVersion: cloud.redhat.com/v1alpha1
+    kind: ClowdApp
+    name: puptoo
+type: Opaque
+---
+apiVersion: kafka.strimzi.io/v1beta2
+kind: KafkaUser
+metadata:
+  labels:
+    strimzi.io/cluster: test-clowdapp-watcher-kafka-msk
+  name: test-clowdapp-watcher-kafka-msk-connect2
+  namespace: test-clowdapp-watcher-kafka-msk
+status:
+  username: test-clowdapp-watcher-kafka-msk-connect2
diff --git a/tests/kuttl/test-clowdapp-watcher-kafka-msk/05-pods.yaml b/tests/kuttl/test-clowdapp-watcher-kafka-msk/05-pods.yaml
new file mode 100644
index 000000000..9e3b5a1c3
--- /dev/null
+++ b/tests/kuttl/test-clowdapp-watcher-kafka-msk/05-pods.yaml
@@ -0,0 +1,32 @@
+---
+apiVersion: kafka.strimzi.io/v1beta2
+kind: KafkaUser
+metadata:
+  labels:
+    strimzi.io/cluster: test-clowdapp-watcher-kafka-msk
+  name: test-clowdapp-watcher-kafka-msk-connect2
+  namespace: test-clowdapp-watcher-kafka-msk
+spec:
+  authentication:
+    type: scram-sha-512
+  authorization:
+    acls:
+    - host: '*'
+      operation: All
+      resource:
+        name: '*'
+        patternType: literal
+        type: topic
+    - host: '*'
+      operation: All
+      resource:
+        name: '*'
+        patternType: literal
+        type: group
+    - host: '*'
+      operation: All
+      resource:
+        name: '*'
+        patternType: literal
+        type: cluster
+    type: simple
diff --git a/tests/kuttl/test-clowdapp-watcher-kafka-msk/06-json-asserts.yaml b/tests/kuttl/test-clowdapp-watcher-kafka-msk/06-json-asserts.yaml
new file mode 100644
index 000000000..a84d6de2a
--- /dev/null
+++ b/tests/kuttl/test-clowdapp-watcher-kafka-msk/06-json-asserts.yaml
@@ -0,0 +1,13 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+commands:
+- script: sleep 5
+- script: kubectl get secret --namespace=test-clowdapp-watcher-kafka-msk test-clowdapp-watcher-kafka-msk-connect2 -o json > /tmp/test-clowdapp-watcher-kafka-msk-user
+- script: kubectl get secret --namespace=test-clowdapp-watcher-kafka-msk test-clowdapp-watcher-kafka-msk-cluster-ca-cert -o json > /tmp/test-clowdapp-watcher-kafka-msk-cluster-ca-cert
+
+- script: sh create_json.sh
+- script: sh create_certs.sh
+- script: kubectl apply -f /tmp/watcher-managed-secret.yaml -n test-clowdapp-watcher-kafka-msk-sec-source
+- script: kubectl apply -f /tmp/test-clowdapp-watcher-kafka-msk-ca-cert.yaml -n test-clowdapp-watcher-kafka-msk-sec-source
+- script: kubectl apply -f /tmp/test-clowdapp-watcher-kafka-msk-connect-user.yaml -n test-clowdapp-watcher-kafka-msk-sec-source
diff --git a/tests/kuttl/test-clowdapp-watcher-kafka-msk/07-assert.yaml b/tests/kuttl/test-clowdapp-watcher-kafka-msk/07-assert.yaml
new file mode 100644
index 000000000..10d6ccd5c
--- /dev/null
+++ b/tests/kuttl/test-clowdapp-watcher-kafka-msk/07-assert.yaml
@@ -0,0 +1,75 @@
+---
+apiVersion: v1
+kind: Secret
+metadata:
+  name: puptoo
+  namespace: test-clowdapp-watcher-kafka-msk-env
+  labels:
+    app: puptoo
+  ownerReferences:
+  - apiVersion: cloud.redhat.com/v1alpha1
+    kind: ClowdApp
+    name: puptoo
+type: Opaque
+---
+kind: Deployment
+apiVersion: apps/v1
+metadata:
+  name: strimzi-topic-operator
+  namespace: test-clowdapp-watcher-kafka-msk-sec-source
+  labels:
+    app: strimzi
+spec:
+  template:
+    spec:
+      serviceAccountName: strimzi-topic-operator
+      containers:
+      - name: strimzi-topic-operator
+        env:
+        - name: STRIMZI_NAMESPACE
+          value: test-clowdapp-watcher-kafka-msk-sec-source
+        - name: STRIMZI_RESOURCE_LABELS
+          value: strimzi.io/cluster=test-clowdapp-watcher-kafka-msk
+        - name: STRIMZI_KAFKA_BOOTSTRAP_SERVERS
+          value: test-clowdapp-watcher-kafka-msk-kafka-bootstrap.test-clowdapp-watcher-kafka-msk.svc:9093
+        - name: STRIMZI_FULL_RECONCILIATION_INTERVAL_MS
+          value: '120000'
+        - name: STRIMZI_LOG_LEVEL
+          value: INFO
+        - name: STRIMZI_TLS_ENABLED
+          value: 'false'
+        - name: STRIMZI_TLS_AUTH_ENABLED
+          value: 'false'
+        - name: STRIMZI_JAVA_OPTS
+          value: '-Xmx512M -Xms256M'
+        - name: STRIMZI_PUBLIC_CA
+          value: 'false'
+        - name: STRIMZI_SASL_ENABLED
+          value: 'true'
+        - name: STRIMZI_SASL_USERNAME
+          value: test-clowdapp-watcher-kafka-msk-connect2
+        - name: STRIMZI_SASL_PASSWORD
+        - name: STRIMZI_SASL_MECHANISM
+          value: scram-sha-512
+        - name: STRIMZI_SECURITY_PROTOCOL
+          value: SASL_SSL
+        - name: STRIMZI_USE_FINALIZERS
+          value: 'true'
+status:
+  conditions:
+  - reason: MinimumReplicasAvailable
+    status: "True"
+    type: Available
+  - reason: NewReplicaSetAvailable
+    status: "True"
+    type: Progressing
+---
+kind: KafkaTopic
+apiVersion: kafka.strimzi.io/v1beta2
+metadata:
+  name: test-clowdapp-watcher-kafka-msk-topic-one
+  namespace: test-clowdapp-watcher-kafka-msk-sec-source
+status:
+  conditions:
+  - status: "True"
+    type: Ready
diff --git a/tests/kuttl/test-clowdapp-watcher-kafka-msk/07-pods.yaml b/tests/kuttl/test-clowdapp-watcher-kafka-msk/07-pods.yaml
new file mode 100644
index 000000000..af7370118
--- /dev/null
+++ b/tests/kuttl/test-clowdapp-watcher-kafka-msk/07-pods.yaml
@@ -0,0 +1,113 @@
+---
+kind: Deployment
+apiVersion: apps/v1
+metadata:
+  name: strimzi-topic-operator
+  namespace: test-clowdapp-watcher-kafka-msk-sec-source
+  labels:
+    app: strimzi
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      name: strimzi-topic-operator
+  template:
+    metadata:
+      creationTimestamp: null
+      labels:
+        name: strimzi-topic-operator
+    spec:
+      serviceAccountName: strimzi-topic-operator
+      serviceAccount: strimzi-topic-operator
+      restartPolicy: Always
+      schedulerName: default-scheduler
+      terminationGracePeriodSeconds: 30
+      securityContext: {}
+      containers:
+        - resources:
+            limits:
+              cpu: 500m
+              memory: 256Mi
+            requests:
+              cpu: 100m
+              memory: 256Mi
+          readinessProbe:
+            httpGet:
+              path: /ready
+              port: 8080
+              scheme: HTTP
+            initialDelaySeconds: 10
+            timeoutSeconds: 1
+            periodSeconds: 30
+            successThreshold: 1
+            failureThreshold: 3
+          terminationMessagePath: /dev/termination-log
+          name: strimzi-topic-operator
+          livenessProbe:
+            httpGet:
+              path: /healthy
+              port: 8080
+              scheme: HTTP
+            initialDelaySeconds: 10
+            timeoutSeconds: 1
+            periodSeconds: 30
+            successThreshold: 1
+            failureThreshold: 3
+          env:
+            - name: STRIMZI_NAMESPACE
+              value: test-clowdapp-watcher-kafka-msk-sec-source
+            - name: STRIMZI_RESOURCE_LABELS
+              value: strimzi.io/cluster=test-clowdapp-watcher-kafka-msk
+            - name: STRIMZI_KAFKA_BOOTSTRAP_SERVERS
+              value: test-clowdapp-watcher-kafka-msk-kafka-bootstrap.test-clowdapp-watcher-kafka-msk.svc:9093
+            - name: STRIMZI_FULL_RECONCILIATION_INTERVAL_MS
+              value: '120000'
+            - name: STRIMZI_LOG_LEVEL
+              value: INFO
+            - name: STRIMZI_TLS_ENABLED
+              value: 'false'
+            - name: STRIMZI_TLS_AUTH_ENABLED
+              value: 'false'
+            - name: STRIMZI_JAVA_OPTS
+              value: '-Xmx512M -Xms256M'
+            - name: STRIMZI_PUBLIC_CA
+              value: 'false'
+            - name: STRIMZI_SASL_ENABLED
+              value: 'true'
+            - name: STRIMZI_SASL_USERNAME
+              value: test-clowdapp-watcher-kafka-msk-connect2
+            - name: STRIMZI_SASL_PASSWORD
+              valueFrom:
+                secretKeyRef:
+                  name: test-clowdapp-watcher-kafka-msk-connect2
+                  key: password
+            - name: STRIMZI_SASL_MECHANISM
+              value: scram-sha-512
+            - name: STRIMZI_SECURITY_PROTOCOL
+              value: SASL_SSL
+            - name: STRIMZI_USE_FINALIZERS
+              value: 'true'
+          imagePullPolicy: IfNotPresent
+          volumeMounts:
+            - name: strimzi-tmp
+              mountPath: /tmp
+            - name: ca
+              mountPath: /etc/tls-sidecar/cluster-ca-certs
+          terminationMessagePolicy: File
+          image: >-
+            quay.io/strimzi/operator:0.37.0
+          args:
+            - /opt/strimzi/bin/topic_operator_run.sh
+      volumes:
+        - name: strimzi-tmp
+          emptyDir:
+            medium: Memory
+            sizeLimit: 5Mi
+        - name: ca
+          secret:
+            secretName: test-clowdapp-watcher-kafka-msk-cluster-ca-cert
+      dnsPolicy: ClusterFirst
+  strategy:
+    type: Recreate
+  revisionHistoryLimit: 10
+  progressDeadlineSeconds: 600
diff --git a/tests/kuttl/test-clowdapp-watcher-kafka-msk/08-json-asserts.yaml b/tests/kuttl/test-clowdapp-watcher-kafka-msk/08-json-asserts.yaml
new file mode 100644
index 000000000..a00a594c4
--- /dev/null
+++ b/tests/kuttl/test-clowdapp-watcher-kafka-msk/08-json-asserts.yaml
@@ -0,0 +1,5 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+commands:
+- script: sh kafka_secret_check.sh 300
diff --git a/tests/kuttl/test-clowdapp-watcher-kafka-msk/09-delete.yaml b/tests/kuttl/test-clowdapp-watcher-kafka-msk/09-delete.yaml
new file mode 100644
index 000000000..4abbbeb8c
--- /dev/null
+++ b/tests/kuttl/test-clowdapp-watcher-kafka-msk/09-delete.yaml
@@ -0,0 +1,19 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+delete:
+- apiVersion: cloud.redhat.com/v1alpha1
+  kind: ClowdEnvironment
+  name: test-clowdapp-watcher-kafka-msk
+- apiVersion: kafka.strimzi.io/v1beta2
+  kind: KafkaTopic
+  name: test-clowdapp-watcher-kafka-msk
+- apiVersion: v1
+  kind: Namespace
+  name: test-clowdapp-watcher-kafka-msk-env
+- apiVersion: v1
+  kind: Namespace
+  name: test-clowdapp-watcher-kafka-msk-sec-source
+- apiVersion: v1
+  kind: Namespace
+  name: test-clowdapp-watcher-kafka-msk
diff --git a/tests/kuttl/test-clowdapp-watcher-kafka-msk/create_certs.sh b/tests/kuttl/test-clowdapp-watcher-kafka-msk/create_certs.sh
new file mode 100755
index 000000000..380d2dc97
--- /dev/null
+++ b/tests/kuttl/test-clowdapp-watcher-kafka-msk/create_certs.sh
@@ -0,0 +1,35 @@
+#!/bin/bash
+
+# Set the file paths
+cacrt=$(cat /tmp/test-clowdapp-watcher-kafka-msk-cluster-ca-cert | jq -r '.data["ca.crt"]' | base64 -d)
+cap12=$(cat /tmp/test-clowdapp-watcher-kafka-msk-cluster-ca-cert | jq -r '.data["ca.p12"]' | base64 -d)
+capass=$(cat /tmp/test-clowdapp-watcher-kafka-msk-cluster-ca-cert | jq -r '.data["ca.password"]' | base64 -d)
+
+# Create the Kubernetes Secret YAML
+cat <<EOF > /tmp/test-clowdapp-watcher-kafka-msk-ca-cert.yaml
+apiVersion: v1
+kind: Secret
+metadata:
+  name: test-clowdapp-watcher-kafka-msk-cluster-ca-cert
+type: Opaque
+data:
+  ca.crt: $(echo -n "$cacrt" | base64 | tr -d '\n')
+  ca.p12: $(echo -n "$cap12"| base64 | tr -d '\n')
+  password: $(echo -n "$capass" | base64)
+EOF
+
+# Set the file paths
+password=$(cat /tmp/test-clowdapp-watcher-kafka-msk-user | jq -r '.data["password"]' | base64 -d)
+jaas=$(cat /tmp/test-clowdapp-watcher-kafka-msk-user | jq -r '.data["sasl.jaas.config"]' | base64 -d)
+
+# Create the Kubernetes Secret YAML
+cat <<EOF > /tmp/test-clowdapp-watcher-kafka-msk-connect-user.yaml
+apiVersion: v1
+kind: Secret
+metadata:
+  name: test-clowdapp-watcher-kafka-msk-connect
+type: Opaque
+data:
+  sasl.jaas.config: $(echo -n "$jaas"| base64 | tr -d '\n')
+  password: $(echo -n "$password" | base64)
+EOF
diff --git a/tests/kuttl/test-clowdapp-watcher-kafka-msk/create_json.sh b/tests/kuttl/test-clowdapp-watcher-kafka-msk/create_json.sh
new file mode 100755
index 000000000..980e65771
--- /dev/null
+++ b/tests/kuttl/test-clowdapp-watcher-kafka-msk/create_json.sh
@@ -0,0 +1,26 @@
+#!/bin/bash
+
+# Set the file paths
+username=$(cat /tmp/test-clowdapp-watcher-kafka-msk-user | jq -r '.metadata.name')
+password=$(cat /tmp/test-clowdapp-watcher-kafka-msk-user | jq -r '.data.password' | base64 -d)
+cert=$(cat /tmp/test-clowdapp-watcher-kafka-msk-cluster-ca-cert | jq -r '.data["ca.crt"]' | base64 -d)
+port=9093
+saslMechanism=SCRAM-SHA-512
+hostname=test-clowdapp-watcher-kafka-msk-kafka-bootstrap.test-clowdapp-watcher-kafka-msk.svc
+
+# Create the Kubernetes Secret YAML
+cat <<EOF > /tmp/watcher-managed-secret.yaml
+apiVersion: v1
+kind: Secret
+metadata:
+  name: managed-secret
+type: Opaque
+data:
+  username: $(echo -n "$username" | base64)
+  password: $(echo -n "$password"| base64)
+  saslMechanism: $(echo -n "$saslMechanism" | base64)
+  port: $(echo -n "$port" | base64)
+  hostname: $(echo -n "$hostname" | base64 -w 0)
+  ca.crt: $(echo -n "$cert" | base64 | tr -d '\n')
+  cacert: $(echo -n "$cert" | base64 | tr -d '\n')
+EOF
diff --git a/tests/kuttl/test-clowdapp-watcher-kafka-msk/kafka_secret_check.sh b/tests/kuttl/test-clowdapp-watcher-kafka-msk/kafka_secret_check.sh
new file mode 100755
index 000000000..23e08dca8
--- /dev/null
+++ b/tests/kuttl/test-clowdapp-watcher-kafka-msk/kafka_secret_check.sh
@@ -0,0 +1,71 @@
+#!/bin/bash
+
+if [ $# -ne 1 ]; then
+    echo "Usage: $0 <timeout_in_seconds>"
+    exit 1
+fi
+
+TIMEOUT=$1
+START_TIME=$(date +%s)
+PREV_HOSTNAME=""
+PREV_USERNAME=""
+PREV_HASHCACHE=""
+USERNAME_MATCH=false
+HASHCACHE_CHANGED=false
+
+# Load before changes hashCache value
+jq -r '.hashCache' -e < /tmp/test-clowdapp-watcher-kafka-msk-env-json > /tmp/test-clowdapp-watcher-kafka-msk-env-hash-cache
+
+while true; do
+    # Check elapsed time
+    CURRENT_TIME=$(date +%s)
+    ELAPSED_TIME=$((CURRENT_TIME - START_TIME))
+    if [ "$ELAPSED_TIME" -ge "$TIMEOUT" ]; then
+        echo "Kafka SASL username check: FALSE"
+        echo "HashCache diff comparison: FALSE"
+        echo "Script timed out after $TIMEOUT seconds."
+        exit 1
+    fi
+
+    # Execute commands
+    sleep 5
+    kubectl get secret --namespace=test-clowdapp-watcher-kafka-msk-env puptoo -o json > /tmp/test-clowdapp-watcher-kafka-msk-env2
+    jq -r '.data["cdappconfig.json"]' < /tmp/test-clowdapp-watcher-kafka-msk-env2 | base64 -d > /tmp/test-clowdapp-watcher-kafka-msk-env2-json
+
+    CURRENT_HOSTNAME=$(jq -r '.kafka.brokers[0].hostname' < /tmp/test-clowdapp-watcher-kafka-msk-env2-json)
+    CURRENT_USERNAME=$(jq -r '.kafka.brokers[0].sasl.username' < /tmp/test-clowdapp-watcher-kafka-msk-env2-json)
+    CURRENT_HASHCACHE=$(jq -r '.hashCache' < /tmp/test-clowdapp-watcher-kafka-msk-env2-json)
+
+    if [ "$CURRENT_HOSTNAME" != "$PREV_HOSTNAME" ]; then
+        echo "Kafka broker hostname check: $CURRENT_HOSTNAME"
+        PREV_HOSTNAME=$CURRENT_HOSTNAME
+    fi
+
+    if [ "$CURRENT_USERNAME" != "$PREV_USERNAME" ]; then
+        if [ "$CURRENT_USERNAME" == "test-clowdapp-watcher-kafka-msk-connect2" ]; then
+            echo "Kafka SASL username check: TRUE"
+            USERNAME_MATCH=true
+        else
+            USERNAME_MATCH=false
+        fi
+        PREV_USERNAME=$CURRENT_USERNAME
+    fi
+
+    if [ "$CURRENT_HASHCACHE" != "$PREV_HASHCACHE" ]; then
+        echo "$CURRENT_HASHCACHE" > /tmp/test-clowdapp-watcher-kafka-msk-env-hash-cache2
+        if diff /tmp/test-clowdapp-watcher-kafka-msk-env-hash-cache /tmp/test-clowdapp-watcher-kafka-msk-env-hash-cache2 > /dev/null; then
+            HASHCACHE_CHANGED=false
+        else
+            echo "HashCache diff comparison: TRUE"
+            HASHCACHE_CHANGED=true
+        fi
+        PREV_HASHCACHE=$CURRENT_HASHCACHE
+    fi
+
+    # Exit if both conditions are met
+    if [ "$USERNAME_MATCH" = true ] && [ "$HASHCACHE_CHANGED" = true ]; then
+        echo "Both conditions met, exiting with status 0."
+        exit 0
+    fi
+
+done