diff --git a/Documentation/CRDs/specification.md b/Documentation/CRDs/specification.md
index 6cf3bcde8..dcd304780 100644
--- a/Documentation/CRDs/specification.md
+++ b/Documentation/CRDs/specification.md
@@ -2948,6 +2948,19 @@ int64
 <p>ObservedGeneration is the latest generation observed by the controller.</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>secrets</code><br/>
+<em>
+<a href="#ceph.rook.io/v1.SecretReference">
+[]SecretReference
+</a>
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+</td>
+</tr>
 </tbody>
 </table>
 <h3 id="ceph.rook.io/v1.CIDR">CIDR
@@ -7533,6 +7546,34 @@ string
 <p>The authentication mechanism for this topic (PLAIN/SCRAM-SHA-512/SCRAM-SHA-256/GSSAPI/OAUTHBEARER)</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>UserSecretRef</code><br/>
+<em>
+<a href="https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.24/#secretkeyselector-v1-core">
+Kubernetes core/v1.SecretKeySelector
+</a>
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>The kafka user name to use for authentication</p>
+</td>
+</tr>
+<tr>
+<td>
+<code>PasswordSecretRef</code><br/>
+<em>
+<a href="https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.24/#secretkeyselector-v1-core">
+Kubernetes core/v1.SecretKeySelector
+</a>
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>The kafka password to use for authentication</p>
+</td>
+</tr>
 </tbody>
 </table>
 <h3 id="ceph.rook.io/v1.KerberosConfigFiles">KerberosConfigFiles
@@ -12487,7 +12528,7 @@ int32
 <h3 id="ceph.rook.io/v1.SecretReference">SecretReference
 </h3>
 <p>
-(<em>Appears on:</em><a href="#ceph.rook.io/v1.ObjectStoreUserStatus">ObjectStoreUserStatus</a>)
+(<em>Appears on:</em><a href="#ceph.rook.io/v1.BucketTopicStatus">BucketTopicStatus</a>, <a href="#ceph.rook.io/v1.ObjectStoreUserStatus">ObjectStoreUserStatus</a>)
 </p>
 <div>
 </div>
diff --git a/Documentation/Storage-Configuration/Object-Storage-RGW/ceph-object-bucket-notifications.md b/Documentation/Storage-Configuration/Object-Storage-RGW/ceph-object-bucket-notifications.md
index 7653b0947..8aa76d68d 100644
--- a/Documentation/Storage-Configuration/Object-Storage-RGW/ceph-object-bucket-notifications.md
+++ b/Documentation/Storage-Configuration/Object-Storage-RGW/ceph-object-bucket-notifications.md
@@ -66,50 +66,67 @@ spec:
 #     ackLevel: broker [17]
 #     useSSL: false [18]
 #     mechanism: SCRAM-SHA-512 [19]
+#     userSecretRef: [20]
+#       name: foo-kafka
+#       key: user
+#     passwordSecretRef: [21]
+#       name: foo-kafka
+#       key: pass
+```
+
+```yaml
+---
+apiVersion: v1
+kind: Secret
+metadata:
+  name: foo-kafka
+  namespace: rook-ceph
+stringData:
+  user: bar
+  pass: baz
 ```
 
 1. `name` of the `CephBucketTopic`
-    + In case of AMQP endpoint, the name is used for the AMQP topic (“routing key” for a topic exchange)
+    + In case of AMQP endpoint, the name is used for the AMQP topic ("routing key" for a topic exchange)
     + In case of Kafka endpoint, the name is used as the Kafka topic
 2. `namespace`(optional) of the `CephBucketTopic`. Should match the namespace of the CephBucketNotification associated with this CephBucketTopic, and the OBC with the label referencing the CephBucketNotification
 3. `objectStoreName` is the name of the object store in which the topic should be created. This must be the same object store used for the buckets associated with the notifications referencing this topic.
 4. `objectStoreNamespace` is the namespace of the object store in which the topic should be created
 5. `opaqueData` (optional) is added to all notifications triggered by a notifications associated with the topic
-6. `persistent` (optional) indicates whether notifications to this endpoint are persistent (=asynchronous) or sent synchronously (“false” by default)
+6. `persistent` (optional) indicates whether notifications to this endpoint are persistent (=asynchronous) or sent synchronously (`false` by default)
 7. `endpoint` to which to send the notifications to. Exactly one of the endpoints must be defined: `http`, `amqp`, `kafka`
 8. `http` (optional) hold the spec for an HTTP endpoint. The format of the URI would be: `http[s]://<fqdn>[:<port>][/<resource>]`
     + port defaults to: 80/443 for HTTP/S accordingly
-9. `disableVerifySSL` indicates whether the RGW is going to verify the SSL certificate of the HTTP server in case HTTPS is used ("false" by default)
-10. `sendCloudEvents`: (optional) send the notifications with the [CloudEvents header](https://github.com/cloudevents/spec/blob/main/cloudevents/adapters/aws-s3.md). ("false" by default)
+9. `disableVerifySSL` indicates whether the RGW is going to verify the SSL certificate of the HTTP server in case HTTPS is used (`false` by default)
+10. `sendCloudEvents`: (optional) send the notifications with the [CloudEvents header](https://github.com/cloudevents/spec/blob/main/cloudevents/adapters/aws-s3.md). (`false` by default)
 11. `amqp` (optional) hold the spec for an AMQP endpoint. The format of the URI would be: `amqp[s]://[<user>:<password>@]<fqdn>[:<port>][/<vhost>]`
     + port defaults to: 5672/5671 for AMQP/S accordingly
     + user/password defaults to: guest/guest
     + user/password may only be provided if HTTPS is used with the RGW. If not, topic creation request will be rejected
     + vhost defaults to: “/”
-12. `disableVerifySSL` (optional) indicates whether the RGW is going to verify the SSL certificate of the AMQP server in case AMQPS is used ("false" by default)
+12. `disableVerifySSL` (optional) indicates whether the RGW is going to verify the SSL certificate of the AMQP server in case AMQPS is used (`false` by default)
 13. `ackLevel` (optional) indicates what kind of ack the RGW is waiting for after sending the notifications:
-    + “none”: message is considered “delivered” if sent to broker
-    + “broker”: message is considered “delivered” if acked by broker (default)
-    + “routable”: message is considered “delivered” if broker can route to a consumer
+    + `none`: message is considered "delivered" if sent to broker
+    + `broker`: message is considered "delivered" if acked by broker (default)
+    + `routable`: message is considered "delivered" if broker can route to a consumer
 14. `exchange` in the AMQP broker that would route the notifications. Different topics pointing to the same endpoint must use the same exchange
 15. `kafka` (optional) hold the spec for a Kafka endpoint. The format of the URI would be: `kafka://[<user>:<password>@]<fqdn>[:<port]`
     + port defaults to: 9092
     + user/password may only be provided if HTTPS is used with the RGW. If not, topic creation request will be rejected
     + user/password may only be provided together with `useSSL`, if not, the connection to the broker would fail
-16. `disableVerifySSL` (optional) indicates whether the RGW is going to verify the SSL certificate of the Kafka server in case `useSSL` flag is used ("false" by default)
+16. `disableVerifySSL` (optional) indicates whether the RGW is going to verify the SSL certificate of the Kafka server in case `useSSL` flag is used (`false` by default)
 17. `ackLevel` (optional) indicates what kind of ack the RGW is waiting for after sending the notifications:
-    + “none”: message is considered “delivered” if sent to broker
-    + “broker”: message is considered “delivered” if acked by broker (default)
-18. `useSSL` (optional) indicates that secure connection will be used for connecting with the broker (“false” by default)
+    + `none`: message is considered "delivered" if sent to broker
+    + `broker`: message is considered "delivered" if acked by broker (default)
+18. `useSSL` (optional) indicates that secure connection will be used for connecting with the broker (`false` by default)
 19. `mechanism` (optional) select which authentication mechanism to use, one of:
-    + “PLAIN” (default)
-    + “SCRAM-SHA-512”
-    + “SCRAM-SHA-256”
-    + “GSSAPI”
-    + “OAUTHBEARER”
-
-!!! note
-    In case of Kafka and AMQP, the consumer of the notifications is not required to ack the notifications, since the broker persists the messages before delivering them to their final destinations.
+    + `PLAIN` (default)
+    + `SCRAM-SHA-512`
+    + `SCRAM-SHA-256`
+    + `GSSAPI`
+    + `OAUTHBEARER`
+20. `userSecretRef` (optional) A [SecretKeySelector](https://pkg.go.dev/k8s.io/api/core/v1#SecretKeySelector) to set the Kafka endpoint username
+21. `passwordSecretRef` (optional) A [SecretKeySelector](https://pkg.go.dev/k8s.io/api/core/v1#SecretKeySelector) to set the Kafka endpoint password
 
 ### CephBucketNotification Custom Resource
 
diff --git a/deploy/charts/rook-ceph/templates/resources.yaml b/deploy/charts/rook-ceph/templates/resources.yaml
index a6ae5e771..3048672a6 100644
--- a/deploy/charts/rook-ceph/templates/resources.yaml
+++ b/deploy/charts/rook-ceph/templates/resources.yaml
@@ -954,6 +954,50 @@ spec:
                     kafka:
                       description: Spec of Kafka endpoint
                       properties:
+                        PasswordSecretRef:
+                          description: The kafka password to use for authentication
+                          properties:
+                            key:
+                              description: The key of the secret to select from.  Must be a valid secret key.
+                              type: string
+                            name:
+                              default: ""
+                              description: |-
+                                Name of the referent.
+                                This field is effectively required, but due to backwards compatibility is
+                                allowed to be empty. Instances of this type with an empty value here are
+                                almost certainly wrong.
+                                More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names
+                              type: string
+                            optional:
+                              description: Specify whether the Secret or its key must be defined
+                              type: boolean
+                          required:
+                            - key
+                          type: object
+                          x-kubernetes-map-type: atomic
+                        UserSecretRef:
+                          description: The kafka user name to use for authentication
+                          properties:
+                            key:
+                              description: The key of the secret to select from.  Must be a valid secret key.
+                              type: string
+                            name:
+                              default: ""
+                              description: |-
+                                Name of the referent.
+                                This field is effectively required, but due to backwards compatibility is
+                                allowed to be empty. Instances of this type with an empty value here are
+                                almost certainly wrong.
+                                More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names
+                              type: string
+                            optional:
+                              description: Specify whether the Secret or its key must be defined
+                              type: boolean
+                          required:
+                            - key
+                          type: object
+                          x-kubernetes-map-type: atomic
                         ackLevel:
                           default: broker
                           description: The ack level required for this topic (none/broker)
@@ -1017,6 +1061,26 @@ spec:
                   type: integer
                 phase:
                   type: string
+                secrets:
+                  items:
+                    properties:
+                      name:
+                        description: name is unique within a namespace to reference a secret resource.
+                        type: string
+                      namespace:
+                        description: namespace defines the space within which the secret name must be unique.
+                        type: string
+                      resourceVersion:
+                        type: string
+                      uid:
+                        description: |-
+                          UID is a type that holds unique ID values, including UUIDs.  Because we
+                          don't ONLY use UUIDs, this is an alias to string.  Being a type captures
+                          intent and helps make sure that UIDs and names do not get conflated.
+                        type: string
+                    type: object
+                    x-kubernetes-map-type: atomic
+                  type: array
               type: object
               x-kubernetes-preserve-unknown-fields: true
           required:
diff --git a/deploy/examples/crds.yaml b/deploy/examples/crds.yaml
index 01aa2fb90..7b317bb66 100644
--- a/deploy/examples/crds.yaml
+++ b/deploy/examples/crds.yaml
@@ -954,6 +954,50 @@ spec:
                     kafka:
                       description: Spec of Kafka endpoint
                       properties:
+                        PasswordSecretRef:
+                          description: The kafka password to use for authentication
+                          properties:
+                            key:
+                              description: The key of the secret to select from.  Must be a valid secret key.
+                              type: string
+                            name:
+                              default: ""
+                              description: |-
+                                Name of the referent.
+                                This field is effectively required, but due to backwards compatibility is
+                                allowed to be empty. Instances of this type with an empty value here are
+                                almost certainly wrong.
+                                More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names
+                              type: string
+                            optional:
+                              description: Specify whether the Secret or its key must be defined
+                              type: boolean
+                          required:
+                            - key
+                          type: object
+                          x-kubernetes-map-type: atomic
+                        UserSecretRef:
+                          description: The kafka user name to use for authentication
+                          properties:
+                            key:
+                              description: The key of the secret to select from.  Must be a valid secret key.
+                              type: string
+                            name:
+                              default: ""
+                              description: |-
+                                Name of the referent.
+                                This field is effectively required, but due to backwards compatibility is
+                                allowed to be empty. Instances of this type with an empty value here are
+                                almost certainly wrong.
+                                More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names
+                              type: string
+                            optional:
+                              description: Specify whether the Secret or its key must be defined
+                              type: boolean
+                          required:
+                            - key
+                          type: object
+                          x-kubernetes-map-type: atomic
                         ackLevel:
                           default: broker
                           description: The ack level required for this topic (none/broker)
@@ -1017,6 +1061,26 @@ spec:
                   type: integer
                 phase:
                   type: string
+                secrets:
+                  items:
+                    properties:
+                      name:
+                        description: name is unique within a namespace to reference a secret resource.
+                        type: string
+                      namespace:
+                        description: namespace defines the space within which the secret name must be unique.
+                        type: string
+                      resourceVersion:
+                        type: string
+                      uid:
+                        description: |-
+                          UID is a type that holds unique ID values, including UUIDs.  Because we
+                          don't ONLY use UUIDs, this is an alias to string.  Being a type captures
+                          intent and helps make sure that UIDs and names do not get conflated.
+                        type: string
+                    type: object
+                    x-kubernetes-map-type: atomic
+                  type: array
               type: object
               x-kubernetes-preserve-unknown-fields: true
           required:
diff --git a/go.mod b/go.mod
index bfae4c079..a6cde1178 100644
--- a/go.mod
+++ b/go.mod
@@ -17,6 +17,10 @@ replace (
 require (
 	github.com/IBM/keyprotect-go-client v0.15.1
 	github.com/aws/aws-sdk-go v1.55.6
+	github.com/aws/aws-sdk-go-v2/config v1.29.4
+	github.com/aws/aws-sdk-go-v2/credentials v1.17.57
+	github.com/aws/aws-sdk-go-v2/service/sns v1.34.3
+	github.com/aws/smithy-go v1.22.2
 	github.com/banzaicloud/k8s-objectmatcher v1.8.0
 	github.com/ceph/ceph-csi-operator/api v0.0.0-20241211130321-1c8ad98a7d6e
 	github.com/ceph/ceph-csi/api v0.0.0-20241216133622-88b7e0d6684f
@@ -68,9 +72,16 @@ require (
 	github.com/AzureAD/microsoft-authentication-library-for-go v1.3.2 // indirect
 	github.com/ansel1/merry v1.8.0 // indirect
 	github.com/ansel1/merry/v2 v2.2.1 // indirect
-	github.com/aws/aws-sdk-go-v2 v1.36.0 // indirect
-	github.com/aws/aws-sdk-go-v2/credentials v1.17.57 // indirect
-	github.com/aws/smithy-go v1.22.2 // indirect
+	github.com/aws/aws-sdk-go-v2 v1.36.3 // indirect
+	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.27 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.34 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.34 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.2 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.12.2 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.12 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sso v1.24.14 // indirect
+	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.28.13 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sts v1.33.12 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/blang/semver/v4 v4.0.0 // indirect
 	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
diff --git a/go.sum b/go.sum
index 4ecfe1eef..7513e72cf 100644
--- a/go.sum
+++ b/go.sum
@@ -148,8 +148,8 @@ github.com/asaskevich/govalidator v0.0.0-20190424111038-f61b66f89f4a/go.mod h1:l
 github.com/aws/aws-sdk-go v1.44.164/go.mod h1:aVsgQcEevwlmQ7qHE9I3h+dtQgpqhFB+i8Phjh7fkwI=
 github.com/aws/aws-sdk-go v1.55.6 h1:cSg4pvZ3m8dgYcgqB97MrcdjUmZ1BeMYKUxMMB89IPk=
 github.com/aws/aws-sdk-go v1.55.6/go.mod h1:eRwEWoyTWFMVYVQzKMNHWP5/RV4xIUGMQfXQHfHkpNU=
-github.com/aws/aws-sdk-go-v2 v1.36.0 h1:b1wM5CcE65Ujwn565qcwgtOTT1aT4ADOHHgglKjG7fk=
-github.com/aws/aws-sdk-go-v2 v1.36.0/go.mod h1:5PMILGVKiW32oDzjj6RU52yrNrDPUHcbZQYr1sM7qmM=
+github.com/aws/aws-sdk-go-v2 v1.36.3 h1:mJoei2CxPutQVxaATCzDUjcZEjVRdpsiiXi2o38yqWM=
+github.com/aws/aws-sdk-go-v2 v1.36.3/go.mod h1:LLXuLpgzEbD766Z5ECcRmi8AzSwfZItDtmABVkRLGzg=
 github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.7 h1:lL7IfaFzngfx0ZwUGOZdsFFnQ5uLvR0hWqqhyE7Q9M8=
 github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.7/go.mod h1:QraP0UcVlQJsmHfioCrveWOC1nbiWUl3ej08h4mXWoc=
 github.com/aws/aws-sdk-go-v2/config v1.29.4 h1:ObNqKsDYFGr2WxnoXKOhCvTlf3HhwtoGgc+KmZ4H5yg=
@@ -158,10 +158,10 @@ github.com/aws/aws-sdk-go-v2/credentials v1.17.57 h1:kFQDsbdBAR3GZsB8xA+51ptEnq9
 github.com/aws/aws-sdk-go-v2/credentials v1.17.57/go.mod h1:2kerxPUUbTagAr/kkaHiqvj/bcYHzi2qiJS/ZinllU0=
 github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.27 h1:7lOW8NUwE9UZekS1DYoiPdVAqZ6A+LheHWb+mHbNOq8=
 github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.27/go.mod h1:w1BASFIPOPUae7AgaH4SbjNbfdkxuggLyGfNFTn8ITY=
-github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.31 h1:lWm9ucLSRFiI4dQQafLrEOmEDGry3Swrz0BIRdiHJqQ=
-github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.31/go.mod h1:Huu6GG0YTfbPphQkDSo4dEGmQRTKb9k9G7RdtyQWxuI=
-github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.31 h1:ACxDklUKKXb48+eg5ROZXi1vDgfMyfIA/WyvqHcHI0o=
-github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.31/go.mod h1:yadnfsDwqXeVaohbGc/RaD287PuyRw2wugkh5ZL2J6k=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.34 h1:ZK5jHhnrioRkUNOc+hOgQKlUL5JeC3S6JgLxtQ+Rm0Q=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.34/go.mod h1:p4VfIceZokChbA9FzMbRGz5OV+lekcVtHlPKEO0gSZY=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.34 h1:SZwFm17ZUNNg5Np0ioo/gq8Mn6u9w19Mri8DnJ15Jf0=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.34/go.mod h1:dFZsC0BLo346mvKQLWmoJxT+Sjp+qcVR1tRVHQGOH9Q=
 github.com/aws/aws-sdk-go-v2/internal/ini v1.8.2 h1:Pg9URiobXy85kgFev3og2CuOZ8JZUBENF+dcgWBaYNk=
 github.com/aws/aws-sdk-go-v2/internal/ini v1.8.2/go.mod h1:FbtygfRFze9usAadmnGJNc8KsP346kEe+y2/oyhGAGc=
 github.com/aws/aws-sdk-go-v2/internal/v4a v1.3.26 h1:GeNJsIFHB+WW5ap2Tec4K6dzcVTsRbsT1Lra46Hv9ME=
@@ -176,6 +176,8 @@ github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.18.7 h1:Hi0KGbrnr57bEH
 github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.18.7/go.mod h1:wKNgWgExdjjrm4qvfbTorkvocEstaoDl4WCvGfeCy9c=
 github.com/aws/aws-sdk-go-v2/service/s3 v1.72.0 h1:SAfh4pNx5LuTafKKWR02Y+hL3A+3TX8cTKG1OIAJaBk=
 github.com/aws/aws-sdk-go-v2/service/s3 v1.72.0/go.mod h1:r+xl5yzMk9083rMR+sJ5TYj9Tihvf/l1oxzZXDgGj2Q=
+github.com/aws/aws-sdk-go-v2/service/sns v1.34.3 h1:iJtp/KnPsgMO4TSGfjqi3oGr+R73W7xWqDXHCbqdnv8=
+github.com/aws/aws-sdk-go-v2/service/sns v1.34.3/go.mod h1:PJtxxMdj747j8DeZENRTTYAz/lx/pADn/U0k7YNNiUY=
 github.com/aws/aws-sdk-go-v2/service/sso v1.24.14 h1:c5WJ3iHz7rLIgArznb3JCSQT3uUMiz9DLZhIX+1G8ok=
 github.com/aws/aws-sdk-go-v2/service/sso v1.24.14/go.mod h1:+JJQTxB6N4niArC14YNtxcQtwEqzS3o9Z32n7q33Rfs=
 github.com/aws/aws-sdk-go-v2/service/ssooidc v1.28.13 h1:f1L/JtUkVODD+k1+IiSJUUv8A++2qVr+Xvb3xWXETMU=
diff --git a/pkg/apis/ceph.rook.io/v1/types.go b/pkg/apis/ceph.rook.io/v1/types.go
index 72162221d..0c80afaaa 100755
--- a/pkg/apis/ceph.rook.io/v1/types.go
+++ b/pkg/apis/ceph.rook.io/v1/types.go
@@ -19,6 +19,7 @@ package v1
 import (
 	"time"
 
+	corev1 "k8s.io/api/core/v1"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -2300,6 +2301,8 @@ type BucketTopicStatus struct {
 	// ObservedGeneration is the latest generation observed by the controller.
 	// +optional
 	ObservedGeneration int64 `json:"observedGeneration,omitempty"`
+	// +optional
+	Secrets []SecretReference `json:"secrets,omitempty"`
 }
 
 // CephBucketTopicList represents a list Ceph Object Store Bucket Notification Topics
@@ -2393,6 +2396,12 @@ type KafkaEndpointSpec struct {
 	// +kubebuilder:default=PLAIN
 	// +optional
 	Mechanism string `json:"mechanism,omitempty"`
+	// The kafka user name to use for authentication
+	// +optional
+	UserSecretRef *corev1.SecretKeySelector `json:"UserSecretRef,omitempty"`
+	// The kafka password to use for authentication
+	// +optional
+	PasswordSecretRef *corev1.SecretKeySelector `json:"PasswordSecretRef,omitempty"`
 }
 
 // +genclient
diff --git a/pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go b/pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go
index 556739e25..36c747f75 100644
--- a/pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go
+++ b/pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go
@@ -238,6 +238,11 @@ func (in *BucketTopicStatus) DeepCopyInto(out *BucketTopicStatus) {
 		*out = new(string)
 		**out = **in
 	}
+	if in.Secrets != nil {
+		in, out := &in.Secrets, &out.Secrets
+		*out = make([]SecretReference, len(*in))
+		copy(*out, *in)
+	}
 	return
 }
 
@@ -2831,6 +2836,16 @@ func (in *HybridStorageSpec) DeepCopy() *HybridStorageSpec {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *KafkaEndpointSpec) DeepCopyInto(out *KafkaEndpointSpec) {
 	*out = *in
+	if in.UserSecretRef != nil {
+		in, out := &in.UserSecretRef, &out.UserSecretRef
+		*out = new(corev1.SecretKeySelector)
+		(*in).DeepCopyInto(*out)
+	}
+	if in.PasswordSecretRef != nil {
+		in, out := &in.PasswordSecretRef, &out.PasswordSecretRef
+		*out = new(corev1.SecretKeySelector)
+		(*in).DeepCopyInto(*out)
+	}
 	return
 }
 
@@ -5061,7 +5076,7 @@ func (in *TopicEndpointSpec) DeepCopyInto(out *TopicEndpointSpec) {
 	if in.Kafka != nil {
 		in, out := &in.Kafka, &out.Kafka
 		*out = new(KafkaEndpointSpec)
-		**out = **in
+		(*in).DeepCopyInto(*out)
 	}
 	return
 }
diff --git a/pkg/operator/ceph/object/topic/controller.go b/pkg/operator/ceph/object/topic/controller.go
index 290fff563..84d4d6a71 100644
--- a/pkg/operator/ceph/object/topic/controller.go
+++ b/pkg/operator/ceph/object/topic/controller.go
@@ -19,6 +19,8 @@ package topic
 
 import (
 	"context"
+	"slices"
+	"sort"
 
 	"github.com/coreos/pkg/capnslog"
 	"github.com/pkg/errors"
@@ -28,12 +30,16 @@ import (
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/ceph/reporting"
 	"github.com/rook/rook/pkg/operator/k8sutil"
+	corev1 "k8s.io/api/core/v1"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
+	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/source"
 )
@@ -85,6 +91,95 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error {
 		return err
 	}
 
+	// watch for kafka secrets
+	// watch secrets referenced by CephBucketTopic.spec.endpoint.kafka.{userSecretRef,passwordSecretRef}
+	const (
+		// disable warning: G101: Potential hardcoded credentials (gosec)
+		// nolint:gosec
+		secretNameField = "spec.endpoint.kafka.secretNames"
+	)
+
+	err = mgr.GetFieldIndexer().IndexField(
+		context.TODO(),
+		&cephv1.CephBucketTopic{},
+		secretNameField,
+		func(obj client.Object) []string {
+			var secretNames []string
+			topic, ok := obj.(*cephv1.CephBucketTopic)
+			if !ok {
+				return nil
+			}
+
+			if topic.Spec.Endpoint.Kafka == nil {
+				return nil
+			}
+
+			kafka := topic.Spec.Endpoint.Kafka
+
+			if kafka.UserSecretRef != nil {
+				secretNames = append(secretNames, kafka.UserSecretRef.Name)
+			}
+
+			if kafka.PasswordSecretRef != nil {
+				secretNames = append(secretNames, kafka.PasswordSecretRef.Name)
+			}
+
+			slices.Sort(secretNames)
+			return slices.Compact(secretNames)
+		},
+	)
+	if err != nil {
+		return errors.Wrapf(err, "failed to setup IndexField for CephBucketTopic.Spec.Endpoint.Kafka.{UserSecretRef,PasswordSecretRef}")
+	}
+
+	// Always trigger a reconcile when a secret is deleted. This will cause a
+	// reconciliation failure to happen immediately in hopes of alerting the end
+	// user to the configuration problem.
+	changedOrDeleted := predicate.Or(
+		predicate.TypedResourceVersionChangedPredicate[*corev1.Secret]{},
+		predicate.TypedFuncs[*corev1.Secret]{
+			DeleteFunc: func(e event.TypedDeleteEvent[*corev1.Secret]) bool {
+				return true
+			},
+		},
+	)
+
+	err = c.Watch(
+		source.Kind(
+			mgr.GetCache(),
+			&corev1.Secret{},
+			handler.TypedEnqueueRequestsFromMapFunc(
+				func(ctx context.Context, secret *corev1.Secret) []reconcile.Request {
+					referencingTopics := &cephv1.CephBucketTopicList{}
+					err := r.(*ReconcileBucketTopic).client.List(ctx, referencingTopics, &client.ListOptions{
+						FieldSelector: fields.OneTermEqualSelector(secretNameField, secret.GetName()),
+						Namespace:     secret.GetNamespace(),
+					})
+					if err != nil {
+						logger.Errorf("failed to list CephBucketTopic(s) while handling event for secret %q in namespace %q. %v", secret.GetName(), secret.GetNamespace(), err)
+						return []reconcile.Request{}
+					}
+
+					requests := make([]reconcile.Request, len(referencingTopics.Items))
+					for i, item := range referencingTopics.Items {
+						requests[i] = reconcile.Request{
+							NamespacedName: types.NamespacedName{
+								Name:      item.GetName(),
+								Namespace: item.GetNamespace(),
+							},
+						}
+					}
+					logger.Tracef("CephBucketTopic(s) referencing Secret %q in namespace %q: %v", secret.GetName(), secret.GetNamespace(), requests)
+					return requests
+				},
+			),
+			changedOrDeleted,
+		),
+	)
+	if err != nil {
+		return errors.Wrapf(err, "failed to configure watch for Secret(s)")
+	}
+
 	return nil
 }
 
@@ -96,7 +191,7 @@ func (r *ReconcileBucketTopic) Reconcile(context context.Context, request reconc
 	// workaround because the rook logging mechanism is not compatible with the controller-runtime logging interface
 	reconcileResponse, err := r.reconcile(request)
 	if err != nil {
-		r.updateStatus(k8sutil.ObservedGenerationNotAvailable, request.NamespacedName, k8sutil.ReconcileFailedStatus, nil)
+		r.updateStatus(k8sutil.ObservedGenerationNotAvailable, request.NamespacedName, k8sutil.ReconcileFailedStatus, nil, nil)
 		logger.Errorf("failed to reconcile %v", err)
 	}
 
@@ -130,11 +225,6 @@ func (r *ReconcileBucketTopic) reconcile(request reconcile.Request) (reconcile.R
 		return reconcile.Result{}, nil
 	}
 
-	// The CR was just created, initializing status fields
-	if cephBucketTopic.Status == nil {
-		r.updateStatus(k8sutil.ObservedGenerationNotAvailable, request.NamespacedName, k8sutil.EmptyStatus, nil)
-	}
-
 	// Make sure a CephCluster is present otherwise do nothing
 	cephCluster, isReadyToReconcile, cephClusterExists, reconcileResponse := opcontroller.IsReadyToReconcile(
 		r.opManagerContext,
@@ -188,24 +278,24 @@ func (r *ReconcileBucketTopic) reconcile(request reconcile.Request) (reconcile.R
 	}
 
 	// Start object reconciliation, updating status for this
-	r.updateStatus(k8sutil.ObservedGenerationNotAvailable, request.NamespacedName, k8sutil.ReconcilingStatus, nil)
+	r.updateStatus(k8sutil.ObservedGenerationNotAvailable, request.NamespacedName, k8sutil.ReconcilingStatus, nil, nil)
 
 	// create topic
-	topicARN, err := r.createCephBucketTopic(cephBucketTopic)
+	topicARN, referencedSecrets, err := r.createCephBucketTopic(cephBucketTopic)
 	if err != nil {
 		return reconcile.Result{}, errors.Wrapf(err, "topic creation failed for CephBucketTopic %q", request.NamespacedName)
 	}
 
 	// update ObservedGeneration in status a the end of reconcile
 	// Set Ready status, we are done reconciling
-	r.updateStatus(observedGeneration, request.NamespacedName, k8sutil.ReadyStatus, topicARN)
+	r.updateStatus(observedGeneration, request.NamespacedName, k8sutil.ReadyStatus, topicARN, referencedSecrets)
 
 	// Return and do not requeue
 	return reconcile.Result{}, nil
 }
 
-func (r *ReconcileBucketTopic) createCephBucketTopic(topic *cephv1.CephBucketTopic) (topicARN *string, err error) {
-	topicARN, err = createTopicFunc(
+func (r *ReconcileBucketTopic) createCephBucketTopic(topic *cephv1.CephBucketTopic) (topicARN *string, referencedSecrets *map[types.UID]*corev1.Secret, err error) {
+	topicARN, referencedSecrets, err = createTopicFunc(
 		provisioner{
 			client:           r.client,
 			context:          r.context,
@@ -232,7 +322,7 @@ func (r *ReconcileBucketTopic) deleteCephBucketTopic(topic *cephv1.CephBucketTop
 }
 
 // updateStatus updates the topic with a given status
-func (r *ReconcileBucketTopic) updateStatus(observedGeneration int64, nsName types.NamespacedName, status string, topicARN *string) {
+func (r *ReconcileBucketTopic) updateStatus(observedGeneration int64, nsName types.NamespacedName, status string, topicARN *string, referencedSecrets *map[types.UID]*corev1.Secret) {
 	topic := &cephv1.CephBucketTopic{}
 	if err := r.client.Get(r.opManagerContext, nsName, topic); err != nil {
 		if kerrors.IsNotFound(err) {
@@ -251,6 +341,32 @@ func (r *ReconcileBucketTopic) updateStatus(observedGeneration int64, nsName typ
 	if observedGeneration != k8sutil.ObservedGenerationNotAvailable {
 		topic.Status.ObservedGeneration = observedGeneration
 	}
+
+	logger.Debugf("updating CephBucketTopic %q .status.secrets to %+v.", nsName, referencedSecrets)
+
+	if referencedSecrets != nil {
+		secretsStatus := []cephv1.SecretReference{}
+
+		for _, secret := range *referencedSecrets {
+			secretsStatus = append(secretsStatus, cephv1.SecretReference{
+				SecretReference: corev1.SecretReference{
+					Name:      secret.Name,
+					Namespace: secret.Namespace,
+				},
+				UID:             secret.UID,
+				ResourceVersion: secret.ResourceVersion,
+			})
+		}
+
+		// assume map key ordering is unstable between reconciles and sort the slice
+		// by secret name
+		sort.Slice(secretsStatus, func(i, j int) bool {
+			return secretsStatus[i].Name < secretsStatus[j].Name
+		})
+
+		topic.Status.Secrets = secretsStatus
+	}
+
 	if err := reporting.UpdateStatus(r.client, topic); err != nil {
 		logger.Errorf("failed to set CephBucketTopic %q status to %q. error %v", nsName, status, err)
 		return
diff --git a/pkg/operator/ceph/object/topic/controller_test.go b/pkg/operator/ceph/object/topic/controller_test.go
index 5478e0d7e..20c384785 100644
--- a/pkg/operator/ceph/object/topic/controller_test.go
+++ b/pkg/operator/ceph/object/topic/controller_test.go
@@ -33,6 +33,7 @@ import (
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	exectest "github.com/rook/rook/pkg/util/exec/test"
 	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -233,8 +234,8 @@ func TestCephBucketTopicController(t *testing.T) {
 
 		// mock the provisioner
 		expectedARN := "arn:aws:sns:" + store + "::" + bucketTopic.Name
-		createTopicFunc = func(p provisioner, topic *cephv1.CephBucketTopic) (*string, error) {
-			return &expectedARN, nil
+		createTopicFunc = func(p provisioner, topic *cephv1.CephBucketTopic) (*string, *map[types.UID]*corev1.Secret, error) {
+			return &expectedARN, nil, nil
 		}
 		defer func() { createTopicFunc = createTopic }()
 		res, err := r.Reconcile(ctx, req)
diff --git a/pkg/operator/ceph/object/topic/provisioner.go b/pkg/operator/ceph/object/topic/provisioner.go
index 1f6c844f9..510ba489a 100644
--- a/pkg/operator/ceph/object/topic/provisioner.go
+++ b/pkg/operator/ceph/object/topic/provisioner.go
@@ -20,11 +20,11 @@ package topic
 import (
 	"context"
 	"crypto/hmac"
-
 	//nolint:gosec // sha1 is needed for v2 signatures
 	"crypto/sha1"
 	"encoding/base64"
 	"net/http"
+	"net/url"
 	"strconv"
 	"strings"
 	"time"
@@ -43,6 +43,7 @@ import (
 	"github.com/rook/rook/pkg/clusterd"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
 	"github.com/rook/rook/pkg/operator/ceph/object"
+	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -143,9 +144,12 @@ func createSNSClient(p provisioner, objectStoreName types.NamespacedName) (*sns.
 	return snsClient, nil
 }
 
-func createTopicAttributes(topic *cephv1.CephBucketTopic) map[string]*string {
+func createTopicAttributes(p provisioner, topic *cephv1.CephBucketTopic) (map[string]*string, *map[types.UID]*corev1.Secret, error) {
 	attr := make(map[string]*string)
 	nsName := types.NamespacedName{Name: topic.Name, Namespace: topic.Namespace}
+	// Currently, referencedSecrets is only used by the kafka endpoint but it
+	// could be used by other endpoints in the future.
+	referencedSecrets := make(map[types.UID]*corev1.Secret)
 
 	attr["OpaqueData"] = &topic.Spec.OpaqueData
 	persistent := strconv.FormatBool(topic.Spec.Persistent)
@@ -169,8 +173,48 @@ func createTopicAttributes(topic *cephv1.CephBucketTopic) map[string]*string {
 		attr["cloudevents"] = &cloudEvents
 	}
 	if topic.Spec.Endpoint.Kafka != nil {
-		logger.Infof("creating CephBucketTopic %q with endpoint %q", nsName, topic.Spec.Endpoint.Kafka.URI)
-		attr["push-endpoint"] = &topic.Spec.Endpoint.Kafka.URI
+		kafka := topic.Spec.Endpoint.Kafka
+
+		uri, err := url.Parse(kafka.URI)
+		if err != nil {
+			// URI could contain a passphrase...
+			return nil, nil, errors.Wrapf(err, "failed to parse CephBucketTopic %q .spec.endpoint.kafka.URI %q", nsName, kafka.URI)
+		}
+
+		// If UserSecretRef or PasswordRef is set, we need to parse the URI and insert the
+		// credentials as http basic auth. If basic auth was already set as part of
+		// the URI string, it will be overridden.
+		if kafka.UserSecretRef != nil || kafka.PasswordSecretRef != nil {
+			var user, pass string
+
+			if kafka.UserSecretRef != nil {
+				var secret *corev1.Secret
+				user, secret, err = p.getSecretValue(kafka.UserSecretRef, topic.Namespace)
+				if err != nil {
+					return nil, nil, errors.Wrapf(err, "failed to get secret value from CephBucketTopic %q .spec.endpoint.kafka.userSecretRef %q", nsName, kafka.UserSecretRef)
+				}
+
+				logger.Debugf("CephBucketTopic %q references secret %q", nsName, client.ObjectKeyFromObject(secret))
+				referencedSecrets[secret.UID] = secret
+			}
+			if kafka.PasswordSecretRef != nil {
+				var secret *corev1.Secret
+				pass, secret, err = p.getSecretValue(kafka.PasswordSecretRef, topic.Namespace)
+				if err != nil {
+					return nil, nil, errors.Wrapf(err, "failed to get secret value from CephBucketTopic %q .spec.endpoint.kafka.passwordSecretRef %q", nsName, kafka.PasswordSecretRef)
+				}
+				logger.Debugf("CephBucketTopic %q references secret %q", nsName, client.ObjectKeyFromObject(secret))
+				referencedSecrets[secret.UID] = secret
+			}
+
+			uri.User = url.UserPassword(user, pass)
+		}
+
+		// do not log passphrases, if set
+		logger.Infof("creating CephBucketTopic %q with endpoint %q", nsName, uri.Redacted())
+
+		kafkaUri := uri.String()
+		attr["push-endpoint"] = &kafkaUri
 		useSSL = strconv.FormatBool(topic.Spec.Endpoint.Kafka.UseSSL)
 		attr["use-ssl"] = &useSSL
 		attr["kafka-ack-level"] = &topic.Spec.Endpoint.Kafka.AckLevel
@@ -179,30 +223,34 @@ func createTopicAttributes(topic *cephv1.CephBucketTopic) map[string]*string {
 		attr["mechanism"] = &topic.Spec.Endpoint.Kafka.Mechanism
 	}
 
-	return attr
+	return attr, &referencedSecrets, nil
 }
 
 // Allow overriding this function for unit tests
 var createTopicFunc = createTopic
 
-func createTopic(p provisioner, topic *cephv1.CephBucketTopic) (*string, error) {
+func createTopic(p provisioner, topic *cephv1.CephBucketTopic) (*string, *map[types.UID]*corev1.Secret, error) {
 	nsName := types.NamespacedName{Name: topic.Name, Namespace: topic.Namespace}
 
 	snsClient, err := createSNSClient(p, types.NamespacedName{Name: topic.Spec.ObjectStoreName, Namespace: topic.Spec.ObjectStoreNamespace})
 	if err != nil {
-		return nil, errors.Wrapf(err, "failed to create SNS client for CephBucketTopic %q provisioning", nsName)
+		return nil, nil, errors.Wrapf(err, "failed to create SNS client for CephBucketTopic %q provisioning", nsName)
+	}
+	attr, referencedSecrets, err := createTopicAttributes(p, topic)
+	if err != nil {
+		return nil, nil, errors.Wrapf(err, "failed to generate topic attributes for CephBucketTopic %q", nsName)
 	}
 	topicOutput, err := snsClient.CreateTopic(&sns.CreateTopicInput{
 		Name:       &topic.Name,
-		Attributes: createTopicAttributes(topic),
+		Attributes: attr,
 	})
 	if err != nil {
-		return nil, errors.Wrapf(err, "failed to provision CephBucketTopic %q", nsName)
+		return nil, nil, errors.Wrapf(err, "failed to provision CephBucketTopic %q", nsName)
 	}
 
 	logger.Infof("CephBucketTopic %q created with ARN %q", nsName, *topicOutput.TopicArn)
 
-	return topicOutput.TopicArn, nil
+	return topicOutput.TopicArn, referencedSecrets, nil
 }
 
 // Allow overriding this function for unit tests
@@ -258,3 +306,21 @@ func GetProvisioned(cl client.Client, ctx context.Context, topicName types.Names
 
 	return bucketTopic, nil
 }
+
+// getSecretValue returns the value of key in a kubernetes secret
+func (p *provisioner) getSecretValue(selector *corev1.SecretKeySelector, namespace string) (string, *corev1.Secret, error) {
+	secret := &corev1.Secret{}
+	namespacedName := types.NamespacedName{
+		Name:      selector.Name,
+		Namespace: namespace,
+	}
+	if err := p.client.Get(p.opManagerContext, namespacedName, secret); err != nil {
+		return "", nil, errors.Wrapf(err, "failed to get secret %q", namespacedName)
+	}
+	value, ok := secret.Data[selector.Key]
+	if !ok {
+		return "", nil, errors.Errorf("failed to find key %q in secret %q", selector.Key, namespacedName)
+	}
+
+	return string(value), secret, nil
+}
diff --git a/pkg/operator/ceph/object/topic/provisioner_test.go b/pkg/operator/ceph/object/topic/provisioner_test.go
index 798a7b6c9..fd0920180 100644
--- a/pkg/operator/ceph/object/topic/provisioner_test.go
+++ b/pkg/operator/ceph/object/topic/provisioner_test.go
@@ -25,6 +25,7 @@ import (
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
@@ -62,8 +63,12 @@ func TestTopicAttributesCreation(t *testing.T) {
 				},
 			},
 		}
-		assert.Equal(t, expectedAttrs, createTopicAttributes(bucketTopic))
+
+		attrs, _, err := createTopicAttributes(provisioner{}, bucketTopic)
+		require.NoError(t, err)
+		assert.Equal(t, expectedAttrs, attrs)
 	})
+
 	t.Run("test AMQP attributes", func(t *testing.T) {
 		uri := "amqp://my-rabbitmq-service:5672/vhost1"
 		ackLevel := "broker"
@@ -96,8 +101,12 @@ func TestTopicAttributesCreation(t *testing.T) {
 				},
 			},
 		}
-		assert.Equal(t, expectedAttrs, createTopicAttributes(bucketTopic))
+
+		attrs, _, err := createTopicAttributes(provisioner{}, bucketTopic)
+		require.NoError(t, err)
+		assert.Equal(t, expectedAttrs, attrs)
 	})
+
 	t.Run("test Kafka attributes", func(t *testing.T) {
 		uri := "kafka://my-kafka-service:9092"
 		ackLevel := "broker"
@@ -132,6 +141,9 @@ func TestTopicAttributesCreation(t *testing.T) {
 				},
 			},
 		}
-		assert.Equal(t, expectedAttrs, createTopicAttributes(bucketTopic))
+
+		attrs, _, err := createTopicAttributes(provisioner{}, bucketTopic)
+		require.NoError(t, err)
+		assert.Equal(t, expectedAttrs, attrs)
 	})
 }
diff --git a/tests/framework/installer/ceph_manifests.go b/tests/framework/installer/ceph_manifests.go
index d12ad5f7c..bb2e71cfb 100644
--- a/tests/framework/installer/ceph_manifests.go
+++ b/tests/framework/installer/ceph_manifests.go
@@ -192,6 +192,7 @@ spec:
   cephConfig:
     global:
       mon_data_avail_warn: "10"
+      rgw_allow_notification_secrets_in_cleartext: "true"
   healthCheck:
     daemonHealth:
       mon:
diff --git a/tests/integration/ceph_object_test.go b/tests/integration/ceph_object_test.go
index 3d18c86a3..f55677112 100644
--- a/tests/integration/ceph_object_test.go
+++ b/tests/integration/ceph_object_test.go
@@ -41,6 +41,7 @@ import (
 	"github.com/rook/rook/tests/framework/installer"
 	"github.com/rook/rook/tests/framework/utils"
 	"github.com/rook/rook/tests/integration/object/bucketowner"
+	topickafka "github.com/rook/rook/tests/integration/object/topic/kafka"
 	"github.com/rook/rook/tests/integration/object/user/userkeys"
 )
 
@@ -156,6 +157,7 @@ func runObjectE2ETest(helper *clients.TestClient, k8sh *utils.K8sHelper, install
 
 	bucketowner.TestObjectBucketClaimBucketOwner(s.T(), k8sh, installer, logger, tlsEnable)
 	userkeys.TestObjectStoreUserKeys(s.T(), k8sh, installer, logger, tlsEnable)
+	topickafka.TestBucketTopicKafka(s.T(), k8sh, installer, logger, tlsEnable)
 
 	bucketNotificationTestStoreName := "bucket-notification-" + storeName
 	createCephObjectStore(s.T(), helper, k8sh, installer, namespace, bucketNotificationTestStoreName, 1, tlsEnable, swiftAndKeystone)
diff --git a/tests/integration/object/topic/kafka/kafka.go b/tests/integration/object/topic/kafka/kafka.go
new file mode 100644
index 000000000..3cb23bd24
--- /dev/null
+++ b/tests/integration/object/topic/kafka/kafka.go
@@ -0,0 +1,701 @@
+/*
+Copyright 2025 The Rook Authors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package kafka
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/url"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/service/sns"
+	"github.com/coreos/pkg/capnslog"
+	bktv1alpha1 "github.com/kube-object-storage/lib-bucket-provisioner/pkg/apis/objectbucket.io/v1alpha1"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	storagev1 "k8s.io/api/storage/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
+
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
+	"github.com/rook/rook/tests/framework/installer"
+	"github.com/rook/rook/tests/framework/utils"
+	utilsns "github.com/rook/rook/tests/integration/object/util/sns"
+)
+
+var (
+	defaultName = "test-topickafka"
+
+	ns = &corev1.Namespace{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: defaultName,
+		},
+	}
+
+	objectStore = &cephv1.CephObjectStore{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: defaultName,
+			// the CephObjectstore must be in the same ns as the CephCluster
+			Namespace: "object-ns",
+		},
+		Spec: cephv1.ObjectStoreSpec{
+			MetadataPool: cephv1.PoolSpec{
+				Replicated: cephv1.ReplicatedSpec{
+					Size:                   1,
+					RequireSafeReplicaSize: false,
+				},
+			},
+			DataPool: cephv1.PoolSpec{
+				Replicated: cephv1.ReplicatedSpec{
+					Size:                   1,
+					RequireSafeReplicaSize: false,
+				},
+			},
+			Gateway: cephv1.GatewaySpec{
+				Port:      80,
+				Instances: 1,
+			},
+			AllowUsersInNamespaces: []string{ns.Name},
+		},
+	}
+
+	objectStoreSvc = &corev1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      objectStore.Name,
+			Namespace: objectStore.Namespace,
+		},
+		Spec: corev1.ServiceSpec{
+			Selector: map[string]string{
+				"app":               "rook-ceph-rgw",
+				"rook_cluster":      objectStore.Namespace,
+				"rook_object_store": objectStore.Name,
+			},
+			Ports: []corev1.ServicePort{
+				{
+					Name:       "http",
+					Port:       80,
+					Protocol:   corev1.ProtocolTCP,
+					TargetPort: intstr.FromInt(8080),
+				},
+			},
+			SessionAffinity: corev1.ServiceAffinityNone,
+			Type:            corev1.ServiceTypeNodePort,
+		},
+	}
+
+	secret1 = &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      defaultName + "-secret1",
+			Namespace: ns.Name,
+		},
+		Data: map[string][]byte{
+			"user-name": []byte("kafka-user1"),
+		},
+	}
+
+	secret2 = &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      defaultName + "-secret2",
+			Namespace: ns.Name,
+		},
+		Data: map[string][]byte{
+			"password": []byte("kafka-pass2"),
+		},
+	}
+
+	secret3 = &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      defaultName + "-secret3",
+			Namespace: ns.Name,
+		},
+		Data: map[string][]byte{
+			"user-name": []byte("kafka-user3"),
+		},
+	}
+
+	secret4 = &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      defaultName + "-secret4",
+			Namespace: ns.Name,
+		},
+		Data: map[string][]byte{
+			"password": []byte("kafka-pass4"),
+		},
+	}
+
+	storageClass = &storagev1.StorageClass{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: defaultName,
+		},
+		Provisioner: objectStore.Namespace + ".ceph.rook.io/bucket",
+		Parameters: map[string]string{
+			"objectStoreName":      objectStore.Name,
+			"objectStoreNamespace": objectStore.Namespace,
+		},
+	}
+
+	obc1 = &bktv1alpha1.ObjectBucketClaim{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      defaultName + "-obc1",
+			Namespace: ns.Name,
+			Labels: map[string]string{
+				"bucket-notification-" + bn1.Name: bn1.Name,
+			},
+		},
+		Spec: bktv1alpha1.ObjectBucketClaimSpec{
+			BucketName:       defaultName + "-obc1",
+			StorageClassName: storageClass.Name,
+		},
+	}
+
+	bt1 = &cephv1.CephBucketTopic{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      defaultName + "-topic1",
+			Namespace: ns.Name,
+		},
+		Spec: cephv1.BucketTopicSpec{
+			ObjectStoreName:      objectStore.Name,
+			ObjectStoreNamespace: objectStore.Namespace,
+			Persistent:           false,
+			Endpoint: cephv1.TopicEndpointSpec{
+				Kafka: &cephv1.KafkaEndpointSpec{
+					URI:      "kafka://kafka.example.com:9094",
+					AckLevel: "broker",
+					UseSSL:   false,
+					UserSecretRef: &corev1.SecretKeySelector{
+						LocalObjectReference: corev1.LocalObjectReference{
+							Name: secret1.Name,
+						},
+						Key: "user-name",
+					},
+					PasswordSecretRef: &corev1.SecretKeySelector{
+						LocalObjectReference: corev1.LocalObjectReference{
+							Name: secret2.Name,
+						},
+						Key: "password",
+					},
+				},
+			},
+		},
+	}
+
+	bn1 = &cephv1.CephBucketNotification{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      defaultName + "-notification1",
+			Namespace: ns.Name,
+		},
+		Spec: cephv1.BucketNotificationSpec{
+			Topic: bt1.Name,
+			Events: []cephv1.BucketNotificationEvent{
+				"s3:ObjectCreated:*",
+			},
+		},
+	}
+)
+
+func checkStatusSecrets(t *testing.T, k8sh *utils.K8sHelper, bt *cephv1.CephBucketTopic, expectedSecrets []*corev1.Secret) {
+	t.Run(fmt.Sprintf("cephBucketTopic %q has .status.secrets set", bt.Name), func(t *testing.T) {
+		ctx := context.TODO()
+
+		liveBt, err := k8sh.RookClientset.CephV1().CephBucketTopics(bt.Namespace).Get(ctx, bt.Name, metav1.GetOptions{})
+		require.NoError(t, err)
+
+		require.NotNil(t, liveBt.Status)
+		assert.Len(t, liveBt.Status.Secrets, len(expectedSecrets))
+
+		for _, secret := range expectedSecrets {
+			secretRef, err := func(secretName string, keys []cephv1.SecretReference) (cephv1.SecretReference, error) {
+				for _, secretRef := range keys {
+					if secretRef.Name == secretName {
+						return secretRef, nil
+					}
+				}
+				return cephv1.SecretReference{}, fmt.Errorf("secretReference for secret %q not found in CephObjectStoreUser.status.keys", secret.Name)
+			}(secret.Name, liveBt.Status.Secrets)
+			require.NoError(t, err)
+
+			// fetch the live secret for UID and ResourceVersion
+			liveSecret, err := k8sh.Clientset.CoreV1().Secrets(secret.Namespace).Get(ctx, secret.Name, metav1.GetOptions{})
+			require.NoError(t, err)
+
+			assert.Equal(t, liveSecret.Name, secretRef.Name)
+			assert.Equal(t, liveSecret.Namespace, secretRef.Namespace)
+			assert.Equal(t, liveSecret.UID, secretRef.UID)
+			assert.Equal(t, liveSecret.ResourceVersion, secretRef.ResourceVersion)
+		}
+	})
+}
+
+func checkRgwTopicEndpoint(t *testing.T, snsClient *sns.Client, arn, user, pass string) {
+	t.Run(fmt.Sprintf("rgw topic arn %q has basic auth set", arn), func(t *testing.T) {
+		ctx := context.TODO()
+
+		var uri *url.URL
+
+		inSync := utils.Retry(40, time.Second, "rgw topic basic auth in sync", func() bool {
+			topicAttrs, err := snsClient.GetTopicAttributes(ctx, &sns.GetTopicAttributesInput{
+				TopicArn: &arn,
+			})
+			require.NoError(t, err)
+
+			// the sns endpoint attributes are returned as JSON
+			var endpointJSON map[string]interface{}
+			err = json.Unmarshal([]byte(topicAttrs.Attributes["EndPoint"]), &endpointJSON)
+			require.NoError(t, err)
+
+			uri, err = url.Parse(string(endpointJSON["EndpointAddress"].(string)))
+			require.NoError(t, err)
+
+			uriPassword, _ := uri.User.Password()
+
+			return user == uri.User.Username() && pass == uriPassword
+		})
+		require.True(t, inSync)
+
+		assert.Equal(t, "kafka", uri.Scheme)
+		assert.Equal(t, "kafka.example.com:9094", uri.Host)
+		assert.Equal(t, user, uri.User.Username())
+		uriPassword, _ := uri.User.Password()
+		assert.Equal(t, pass, uriPassword)
+	})
+}
+
+// Note that .status.ARN is nil while .status.phase == Reconciling
+func cephBucketTopicReady(t *testing.T, k8sh *utils.K8sHelper, bt *cephv1.CephBucketTopic) *cephv1.CephBucketTopic {
+	ctx := context.TODO()
+
+	var liveBt *cephv1.CephBucketTopic
+	btReady := utils.Retry(60, time.Second, fmt.Sprintf("CephBucketTopic %q is Ready", bt.Name), func() bool {
+		var err error
+		liveBt, err = k8sh.RookClientset.CephV1().CephBucketTopics(bt.Namespace).Get(ctx, bt.Name, metav1.GetOptions{})
+		if err != nil {
+			return false
+		}
+
+		if liveBt.Status == nil {
+			return false
+		}
+
+		return liveBt.Status.Phase == string(cephv1.ConditionReady)
+	})
+	require.True(t, btReady)
+	require.NotNil(t, liveBt.Status.ARN)
+
+	return liveBt
+}
+
+func TestBucketTopicKafka(t *testing.T, k8sh *utils.K8sHelper, installer *installer.CephInstaller, logger *capnslog.PackageLogger, tlsEnable bool) {
+	ctx := context.TODO()
+	var snsClient *sns.Client
+
+	t.Run("CephBucketTopic kafka", func(t *testing.T) {
+		if tlsEnable {
+			// Skip testing with and without TLS to reduce test time
+			t.Skip("skipping test for TLS enabled clusters")
+		}
+
+		t.Run(fmt.Sprintf("create ns %q", ns.Name), func(t *testing.T) {
+			_, err := k8sh.Clientset.CoreV1().Namespaces().Create(ctx, ns, metav1.CreateOptions{})
+			require.NoError(t, err)
+		})
+
+		t.Run(fmt.Sprintf("create CephObjectStore %q", objectStore.Name), func(t *testing.T) {
+			objectStore, err := k8sh.RookClientset.CephV1().CephObjectStores(objectStore.Namespace).Create(ctx, objectStore, metav1.CreateOptions{})
+			require.NoError(t, err)
+
+			osReady := utils.Retry(180, time.Second, "CephObjectStore is Ready", func() bool {
+				liveOs, err := k8sh.RookClientset.CephV1().CephObjectStores(objectStore.Namespace).Get(ctx, objectStore.Name, metav1.GetOptions{})
+				if err != nil {
+					return false
+				}
+
+				if liveOs.Status == nil {
+					return false
+				}
+
+				return liveOs.Status.Phase == cephv1.ConditionReady
+			})
+			require.True(t, osReady)
+		})
+
+		t.Run(fmt.Sprintf("create svc %q", objectStoreSvc.Name), func(t *testing.T) {
+			_, err := k8sh.Clientset.CoreV1().Services(objectStore.Namespace).Create(ctx, objectStoreSvc, metav1.CreateOptions{})
+			require.NoError(t, err)
+		})
+
+		t.Run(fmt.Sprintf("create secret %q", secret1.Name), func(t *testing.T) {
+			_, err := k8sh.Clientset.CoreV1().Secrets(secret1.Namespace).Create(ctx, secret1, metav1.CreateOptions{})
+			require.NoError(t, err)
+		})
+
+		t.Run(fmt.Sprintf("create secret %q", secret2.Name), func(t *testing.T) {
+			_, err := k8sh.Clientset.CoreV1().Secrets(secret2.Namespace).Create(ctx, secret2, metav1.CreateOptions{})
+			require.NoError(t, err)
+		})
+
+		t.Run(fmt.Sprintf("create secret %q", secret3.Name), func(t *testing.T) {
+			_, err := k8sh.Clientset.CoreV1().Secrets(secret3.Namespace).Create(ctx, secret3, metav1.CreateOptions{})
+			require.NoError(t, err)
+		})
+
+		t.Run(fmt.Sprintf("create secret %q", secret4.Name), func(t *testing.T) {
+			_, err := k8sh.Clientset.CoreV1().Secrets(secret4.Namespace).Create(ctx, secret4, metav1.CreateOptions{})
+			require.NoError(t, err)
+		})
+
+		// the sc, obc, and CephBucketNotification are essentially unused for testing but are created for completeness
+		t.Run(fmt.Sprintf("create sc %q", storageClass.Name), func(t *testing.T) {
+			_, err := k8sh.Clientset.StorageV1().StorageClasses().Create(ctx, storageClass, metav1.CreateOptions{})
+			require.NoError(t, err)
+		})
+
+		t.Run(fmt.Sprintf("create obc %q", obc1.Name), func(t *testing.T) {
+			_, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(obc1.Namespace).Create(ctx, obc1, metav1.CreateOptions{})
+			require.NoError(t, err)
+		})
+
+		t.Run(fmt.Sprintf("create CephBucketTopic %q", bt1.Name), func(t *testing.T) {
+			_, err := k8sh.RookClientset.CephV1().CephBucketTopics(bt1.Namespace).Create(ctx, bt1, metav1.CreateOptions{})
+			require.NoError(t, err)
+
+			// user creation may be slow right after rgw start up
+			cephBucketTopicReady(t, k8sh, bt1)
+		})
+
+		t.Run(fmt.Sprintf("create CephBucketNotification %q", bn1.Name), func(t *testing.T) {
+			_, err := k8sh.RookClientset.CephV1().CephBucketNotifications(bn1.Namespace).Create(ctx, bn1, metav1.CreateOptions{})
+			require.NoError(t, err)
+		})
+
+		t.Run("setup sns client", func(t *testing.T) {
+			var err error
+			snsClient, err = utilsns.NewClient(objectStore, objectStoreSvc, k8sh, installer, tlsEnable)
+			require.NoError(t, err)
+		})
+
+		{
+			liveBt := cephBucketTopicReady(t, k8sh, bt1)
+
+			checkRgwTopicEndpoint(t, snsClient, *liveBt.Status.ARN, string(secret1.Data["user-name"]), string(secret2.Data["password"]))
+
+			secrets := []*corev1.Secret{secret1, secret2}
+
+			checkStatusSecrets(t, k8sh, bt1, secrets)
+		}
+
+		t.Run("updating referenced secrets reconciles rgw topic", func(t *testing.T) {
+			t.Run(fmt.Sprintf("update kafka auth on CephBucketTopic %q", bt1.Name), func(t *testing.T) {
+				liveBt, err := k8sh.RookClientset.CephV1().CephBucketTopics(bt1.Namespace).Get(ctx, bt1.Name, metav1.GetOptions{})
+				require.NoError(t, err)
+
+				liveBt.Spec.Endpoint.Kafka.UserSecretRef = &corev1.SecretKeySelector{
+					LocalObjectReference: corev1.LocalObjectReference{
+						Name: secret3.Name,
+					},
+					Key: "user-name",
+				}
+
+				liveBt.Spec.Endpoint.Kafka.PasswordSecretRef = &corev1.SecretKeySelector{
+					LocalObjectReference: corev1.LocalObjectReference{
+						Name: secret4.Name,
+					},
+					Key: "password",
+				}
+
+				_, err = k8sh.RookClientset.CephV1().CephBucketTopics(liveBt.Namespace).Update(ctx, liveBt, metav1.UpdateOptions{})
+				require.NoError(t, err)
+			})
+
+			{
+				liveBt := cephBucketTopicReady(t, k8sh, bt1)
+
+				checkRgwTopicEndpoint(t, snsClient, *liveBt.Status.ARN, string(secret3.Data["user-name"]), string(secret4.Data["password"]))
+
+				secrets := []*corev1.Secret{secret3, secret4}
+
+				checkStatusSecrets(t, k8sh, bt1, secrets)
+			}
+		})
+
+		t.Run("removing referenced secrets reconciles rgw topic", func(t *testing.T) {
+			t.Run(fmt.Sprintf("delete kafka auth on CephBucketTopic %q", bt1.Name), func(t *testing.T) {
+				liveBt, err := k8sh.RookClientset.CephV1().CephBucketTopics(bt1.Namespace).Get(ctx, bt1.Name, metav1.GetOptions{})
+				require.NoError(t, err)
+
+				liveBt.Spec.Endpoint.Kafka.UserSecretRef = nil
+				liveBt.Spec.Endpoint.Kafka.PasswordSecretRef = nil
+
+				_, err = k8sh.RookClientset.CephV1().CephBucketTopics(liveBt.Namespace).Update(ctx, liveBt, metav1.UpdateOptions{})
+				require.NoError(t, err)
+			})
+
+			{
+				liveBt := cephBucketTopicReady(t, k8sh, bt1)
+
+				checkRgwTopicEndpoint(t, snsClient, *liveBt.Status.ARN, "", "")
+
+				secrets := []*corev1.Secret{}
+
+				checkStatusSecrets(t, k8sh, bt1, secrets)
+			}
+		})
+
+		t.Run("adding new referenced secrets reconciles rgw topic", func(t *testing.T) {
+			t.Run(fmt.Sprintf("add kafka auth on CephBucketTopic %q", bt1.Name), func(t *testing.T) {
+				liveBt, err := k8sh.RookClientset.CephV1().CephBucketTopics(bt1.Namespace).Get(ctx, bt1.Name, metav1.GetOptions{})
+				require.NoError(t, err)
+
+				liveBt.Spec.Endpoint.Kafka.UserSecretRef = &corev1.SecretKeySelector{
+					LocalObjectReference: corev1.LocalObjectReference{
+						Name: secret3.Name,
+					},
+					Key: "user-name",
+				}
+
+				liveBt.Spec.Endpoint.Kafka.PasswordSecretRef = &corev1.SecretKeySelector{
+					LocalObjectReference: corev1.LocalObjectReference{
+						Name: secret4.Name,
+					},
+					Key: "password",
+				}
+
+				_, err = k8sh.RookClientset.CephV1().CephBucketTopics(liveBt.Namespace).Update(ctx, liveBt, metav1.UpdateOptions{})
+				require.NoError(t, err)
+			})
+
+			{
+				liveBt := cephBucketTopicReady(t, k8sh, bt1)
+
+				checkRgwTopicEndpoint(t, snsClient, *liveBt.Status.ARN, string(secret3.Data["user-name"]), string(secret4.Data["password"]))
+
+				secrets := []*corev1.Secret{secret3, secret4}
+
+				checkStatusSecrets(t, k8sh, bt1, secrets)
+			}
+		})
+
+		t.Run("updating secrets reconciles rgw topic", func(t *testing.T) {
+			t.Run(fmt.Sprintf("update secret %q data", secret3.Name), func(t *testing.T) {
+				liveSecret, err := k8sh.Clientset.CoreV1().Secrets(secret3.Namespace).Get(ctx, secret3.Name, metav1.GetOptions{})
+				require.NoError(t, err)
+
+				liveSecret.Data = map[string][]byte{
+					"user-name": []byte("kafka-user3-updated"),
+				}
+
+				_, err = k8sh.Clientset.CoreV1().Secrets(secret3.Namespace).Update(ctx, liveSecret, metav1.UpdateOptions{})
+				require.NoError(t, err)
+			})
+
+			t.Run(fmt.Sprintf("update secret %q data", secret4.Name), func(t *testing.T) {
+				liveSecret, err := k8sh.Clientset.CoreV1().Secrets(secret3.Namespace).Get(ctx, secret4.Name, metav1.GetOptions{})
+				require.NoError(t, err)
+
+				liveSecret.Data = map[string][]byte{
+					"password": []byte("kafka-pass4-updated"),
+				}
+
+				_, err = k8sh.Clientset.CoreV1().Secrets(secret4.Namespace).Update(ctx, liveSecret, metav1.UpdateOptions{})
+				require.NoError(t, err)
+			})
+
+			{
+				liveBt := cephBucketTopicReady(t, k8sh, bt1)
+
+				liveSecret3, err := k8sh.Clientset.CoreV1().Secrets(secret3.Namespace).Get(ctx, secret3.Name, metav1.GetOptions{})
+				require.NoError(t, err)
+
+				liveSecret4, err := k8sh.Clientset.CoreV1().Secrets(secret4.Namespace).Get(ctx, secret4.Name, metav1.GetOptions{})
+				require.NoError(t, err)
+
+				checkRgwTopicEndpoint(t, snsClient, *liveBt.Status.ARN, string(liveSecret3.Data["user-name"]), string(liveSecret4.Data["password"]))
+
+				secrets := []*corev1.Secret{liveSecret3, liveSecret4}
+
+				checkStatusSecrets(t, k8sh, bt1, secrets)
+			}
+		})
+
+		t.Run("deleting a referenced secret triggers a reconcile, which fails", func(t *testing.T) {
+			t.Run(fmt.Sprintf("delete secret %q", secret4.Name), func(t *testing.T) {
+				err := k8sh.Clientset.CoreV1().Secrets(secret4.Namespace).Delete(ctx, secret4.Name, metav1.DeleteOptions{})
+				require.NoError(t, err)
+			})
+
+			t.Run(fmt.Sprintf("CephBucketTopic %q has phase ReconcileFailed", bt1.Name), func(t *testing.T) {
+				btReady := utils.Retry(40, time.Second, "CephBucketTopic is ReconcileFailed", func() bool {
+					liveBt, err := k8sh.RookClientset.CephV1().CephBucketTopics(bt1.Namespace).Get(ctx, bt1.Name, metav1.GetOptions{})
+					if err != nil {
+						return false
+					}
+
+					if liveBt.Status == nil {
+						return false
+					}
+
+					return liveBt.Status.Phase == string(cephv1.ReconcileFailed)
+				})
+				require.True(t, btReady)
+			})
+
+			t.Run(fmt.Sprintf("create secret %q", secret4.Name), func(t *testing.T) {
+				_, err := k8sh.Clientset.CoreV1().Secrets(secret4.Namespace).Create(ctx, secret4, metav1.CreateOptions{})
+				require.NoError(t, err)
+			})
+
+			cephBucketTopicReady(t, k8sh, bt1)
+		})
+
+		t.Run(fmt.Sprintf("delete CephBucketNotification %q", bn1.Name), func(t *testing.T) {
+			err := k8sh.RookClientset.CephV1().CephBucketNotifications(bn1.Namespace).Delete(ctx, bn1.Name, metav1.DeleteOptions{})
+			require.NoError(t, err)
+
+			absent := utils.Retry(40, time.Second, "CephBucketNotification is absent", func() bool {
+				_, err := k8sh.RookClientset.CephV1().CephBucketNotifications(bt1.Namespace).Get(ctx, bt1.Name, metav1.GetOptions{})
+				return err != nil
+			})
+			assert.True(t, absent)
+		})
+
+		t.Run(fmt.Sprintf("no CephBucketNotification(s) in ns %q", ns.Name), func(t *testing.T) {
+			list, err := k8sh.RookClientset.CephV1().CephBucketNotifications(ns.Name).List(ctx, metav1.ListOptions{})
+			require.NoError(t, err)
+
+			assert.Len(t, list.Items, 0)
+		})
+
+		t.Run(fmt.Sprintf("delete CephBucketTopic %q", bt1.Name), func(t *testing.T) {
+			err := k8sh.RookClientset.CephV1().CephBucketTopics(bt1.Namespace).Delete(ctx, bt1.Name, metav1.DeleteOptions{})
+			require.NoError(t, err)
+
+			absent := utils.Retry(40, time.Second, "CephBucketTopic is absent", func() bool {
+				_, err := k8sh.RookClientset.CephV1().CephBucketTopics(bt1.Namespace).Get(ctx, bt1.Name, metav1.GetOptions{})
+				return err != nil
+			})
+			assert.True(t, absent)
+		})
+
+		t.Run(fmt.Sprintf("no CephBucketTopic(s) in ns %q", ns.Name), func(t *testing.T) {
+			list, err := k8sh.RookClientset.CephV1().CephBucketTopics(ns.Name).List(ctx, metav1.ListOptions{})
+			require.NoError(t, err)
+
+			assert.Len(t, list.Items, 0)
+		})
+
+		t.Run(fmt.Sprintf("delete obc %q", obc1.Name), func(t *testing.T) {
+			// lookup ob name
+			liveObc, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(obc1.Namespace).Get(ctx, obc1.Name, metav1.GetOptions{})
+			require.NoError(t, err)
+			obName := liveObc.Spec.ObjectBucketName
+
+			// delete obc
+			err = k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(obc1.Namespace).Delete(ctx, obc1.Name, metav1.DeleteOptions{})
+			require.NoError(t, err)
+
+			absent := utils.Retry(40, time.Second, "OBC is absent", func() bool {
+				_, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(obc1.Namespace).Get(ctx, obc1.Name, metav1.GetOptions{})
+				return err != nil
+			})
+			assert.True(t, absent)
+
+			absent = utils.Retry(40, time.Second, "OB is absent", func() bool {
+				_, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBuckets().Get(ctx, obName, metav1.GetOptions{})
+				return err != nil
+			})
+			assert.True(t, absent)
+		})
+
+		t.Run(fmt.Sprintf("no obc(s) in ns %q", ns.Name), func(t *testing.T) {
+			list, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name).List(ctx, metav1.ListOptions{})
+			require.NoError(t, err)
+
+			assert.Len(t, list.Items, 0)
+		})
+
+		t.Run("CephBucketTopic deletion did not remove any secrets", func(t *testing.T) {
+			t.Run(fmt.Sprintf("secret %q still exists", secret1.Name), func(t *testing.T) {
+				_, err := k8sh.Clientset.CoreV1().Secrets(secret1.Namespace).Get(ctx, secret1.Name, metav1.GetOptions{})
+				require.NoError(t, err)
+			})
+
+			t.Run(fmt.Sprintf("secret %q still exists", secret2.Name), func(t *testing.T) {
+				_, err := k8sh.Clientset.CoreV1().Secrets(secret2.Namespace).Get(ctx, secret2.Name, metav1.GetOptions{})
+				require.NoError(t, err)
+			})
+
+			t.Run(fmt.Sprintf("secret %q still exists", secret3.Name), func(t *testing.T) {
+				_, err := k8sh.Clientset.CoreV1().Secrets(secret3.Namespace).Get(ctx, secret3.Name, metav1.GetOptions{})
+				require.NoError(t, err)
+			})
+
+			t.Run(fmt.Sprintf("secret %q still exists", secret4.Name), func(t *testing.T) {
+				_, err := k8sh.Clientset.CoreV1().Secrets(secret4.Namespace).Get(ctx, secret4.Name, metav1.GetOptions{})
+				require.NoError(t, err)
+			})
+		})
+
+		t.Run(fmt.Sprintf("delete sc %q", storageClass.Name), func(t *testing.T) {
+			err := k8sh.Clientset.StorageV1().StorageClasses().Delete(ctx, storageClass.Name, metav1.DeleteOptions{})
+			require.NoError(t, err)
+		})
+
+		t.Run(fmt.Sprintf("delete secret %q", secret4.Name), func(t *testing.T) {
+			err := k8sh.Clientset.CoreV1().Secrets(secret4.Namespace).Delete(ctx, secret4.Name, metav1.DeleteOptions{})
+			require.NoError(t, err)
+		})
+
+		t.Run(fmt.Sprintf("delete secret %q", secret3.Name), func(t *testing.T) {
+			err := k8sh.Clientset.CoreV1().Secrets(secret3.Namespace).Delete(ctx, secret3.Name, metav1.DeleteOptions{})
+			require.NoError(t, err)
+		})
+
+		t.Run(fmt.Sprintf("delete secret %q", secret2.Name), func(t *testing.T) {
+			err := k8sh.Clientset.CoreV1().Secrets(secret3.Namespace).Delete(ctx, secret2.Name, metav1.DeleteOptions{})
+			require.NoError(t, err)
+		})
+
+		t.Run(fmt.Sprintf("delete secret %q", secret1.Name), func(t *testing.T) {
+			err := k8sh.Clientset.CoreV1().Secrets(secret1.Namespace).Delete(ctx, secret1.Name, metav1.DeleteOptions{})
+			require.NoError(t, err)
+		})
+
+		t.Run(fmt.Sprintf("no secrets in ns %q", ns.Name), func(t *testing.T) {
+			secrets, err := k8sh.Clientset.CoreV1().Secrets(ns.Name).List(ctx, metav1.ListOptions{})
+			require.NoError(t, err)
+
+			assert.Len(t, secrets.Items, 0)
+		})
+
+		t.Run(fmt.Sprintf("delete svc %q", objectStoreSvc.Name), func(t *testing.T) {
+			err := k8sh.Clientset.CoreV1().Services(objectStore.Namespace).Delete(ctx, objectStoreSvc.Name, metav1.DeleteOptions{})
+			require.NoError(t, err)
+		})
+
+		t.Run(fmt.Sprintf("delete CephObjectStore %q", objectStore.Name), func(t *testing.T) {
+			err := k8sh.RookClientset.CephV1().CephObjectStores(objectStore.Namespace).Delete(ctx, objectStore.Name, metav1.DeleteOptions{})
+			require.NoError(t, err)
+		})
+
+		t.Run(fmt.Sprintf("delete ns %q", ns.Name), func(t *testing.T) {
+			err := k8sh.Clientset.CoreV1().Namespaces().Delete(ctx, ns.Name, metav1.DeleteOptions{})
+			require.NoError(t, err)
+		})
+	})
+}
diff --git a/tests/integration/object/util/sns/sns.go b/tests/integration/object/util/sns/sns.go
new file mode 100644
index 000000000..3ea9445bb
--- /dev/null
+++ b/tests/integration/object/util/sns/sns.go
@@ -0,0 +1,147 @@
+/*
+Copyright 2025 The Rook Authors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package sns
+
+import (
+	"context"
+	"crypto/tls"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/url"
+
+	"github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/aws/aws-sdk-go-v2/service/sns"
+	smithyendpoints "github.com/aws/smithy-go/endpoints"
+	"github.com/pkg/errors"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
+	"github.com/rook/rook/tests/framework/installer"
+	"github.com/rook/rook/tests/framework/utils"
+)
+
+// based on the s3 endpoint example in the official docs:
+// https://docs.aws.amazon.com/sdk-for-go/v2/developer-guide/configure-endpoints.html
+type snsResolverV2 struct {
+	endpoint string
+}
+
+func (r *snsResolverV2) ResolveEndpoint(ctx context.Context, params sns.EndpointParameters) (smithyendpoints.Endpoint, error) {
+	u, err := url.Parse(r.endpoint)
+	if err != nil {
+		return smithyendpoints.Endpoint{}, err
+	}
+	return smithyendpoints.Endpoint{
+		URI: *u,
+	}, nil
+}
+
+func GetS3Credentials(objectStore *cephv1.CephObjectStore, installer *installer.CephInstaller) (string, string, error) {
+	err, output := installer.Execute("radosgw-admin", []string{"user", "info", "--uid=dashboard-admin", fmt.Sprintf("--rgw-realm=%s", objectStore.Name)}, objectStore.Namespace)
+	if err != nil {
+		return "", "", errors.Wrap(err, "failed to get user info")
+	}
+
+	// extract api creds from json output
+	var userInfo map[string]interface{}
+	err = json.Unmarshal([]byte(output), &userInfo)
+	if err != nil {
+		return "", "", errors.Wrap(err, "failed to unmarshal user info")
+	}
+
+	accessKey, ok := userInfo["keys"].([]interface{})[0].(map[string]interface{})["access_key"].(string)
+	if !ok {
+		return "", "", errors.New("failed to get access key")
+	}
+
+	secretKey, ok := userInfo["keys"].([]interface{})[0].(map[string]interface{})["secret_key"].(string)
+	if !ok {
+		return "", "", errors.New("failed to get secret key")
+	}
+
+	return accessKey, secretKey, nil
+}
+
+func GetS3Endpoint(objectStore *cephv1.CephObjectStore, k8sh *utils.K8sHelper, tlsEnable bool) (string, error) {
+	ctx := context.TODO()
+
+	// extract rgw endpoint from k8s svc
+	svc, err := k8sh.Clientset.CoreV1().Services(objectStore.Namespace).Get(ctx, objectStore.Name, metav1.GetOptions{})
+	if err != nil {
+		return "", errors.Wrap(err, "failed to get objectstore svc")
+	}
+
+	schema := "http://"
+	httpClient := &http.Client{}
+
+	if tlsEnable {
+		schema = "https://"
+		httpClient.Transport = &http.Transport{
+			TLSClientConfig: &tls.Config{
+				// nolint:gosec // skip TLS verification as this is a test
+				InsecureSkipVerify: true,
+			},
+		}
+	}
+	endpoint := schema + svc.Spec.ClusterIP + ":80"
+
+	return endpoint, nil
+}
+
+func NewClient(objectStore *cephv1.CephObjectStore, svc *corev1.Service, k8sh *utils.K8sHelper, installer *installer.CephInstaller, tlsEnable bool) (*sns.Client, error) {
+	ctx := context.TODO()
+
+	accessKey, secretKey, err := GetS3Credentials(objectStore, installer)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get s3 credentials")
+	}
+
+	cfg, err := config.LoadDefaultConfig(
+		ctx,
+		config.WithRegion("us-east-1"),
+		config.WithCredentialsProvider(
+			credentials.NewStaticCredentialsProvider(
+				accessKey,
+				secretKey,
+				"",
+			),
+		),
+	)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to load default aws config")
+	}
+
+	endpoint, err := GetS3Endpoint(objectStore, k8sh, tlsEnable)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get sns endpoint")
+	}
+
+	snsClient := sns.NewFromConfig(cfg, func(o *sns.Options) {
+		o.EndpointResolverV2 = &snsResolverV2{endpoint: endpoint}
+	})
+
+	// sanity check that client is working
+	_, err = snsClient.ListTopics(ctx, &sns.ListTopicsInput{})
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to sns client sanity check")
+	}
+
+	return snsClient, nil
+}