Repository: knative-sandbox/eventing-kafka-broker
Pull request #2748 (2022-10-26 10:23:11)

Auth spec for namespace support with trigger v2 updates for consumer
Signed-off-by: aavarghese <avarghese@us.ibm.com>

/cc @pierDipi 

knative-prow[bot]: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2748#" title="Author self-approved">aavarghese</a>*
**Once this PR has been reviewed and has the lgtm label**, please assign dsimansk for approval by writing `/assign @dsimansk` in a comment. For more information see [the Kubernetes Code Review Process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process).

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka-broker).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/main/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["dsimansk"]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2748?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2748](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2748?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (bd6737a) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/db645fb3613d5c01c035d295928eb6b0d70dddfb?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (db645fb) will **decrease** coverage by `6.07%`.
> The diff coverage is `100.00%`.

```diff
@@             Coverage Diff              @@
##               main    #2748      +/-   ##
============================================
- Coverage     61.45%   55.38%   -6.08%     
============================================
  Files           148       75      -73     
  Lines         10611     8069    -2542     
  Branches        231        0     -231     
============================================
- Hits           6521     4469    -2052     
+ Misses         3629     3276     -353     
+ Partials        461      324     -137     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `?` | |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2748?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...nternals/kafka/eventing/v1alpha1/consumer\_types.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2748/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9pbnRlcm5hbHMva2Fma2EvZXZlbnRpbmcvdjFhbHBoYTEvY29uc3VtZXJfdHlwZXMuZ28=) | `79.41% <ø> (ø)` | |
| [...ntrol-plane/pkg/reconciler/trigger/v2/triggerv2.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2748/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci90cmlnZ2VyL3YyL3RyaWdnZXJ2Mi5nbw==) | `85.95% <100.00%> (+6.52%)` | :arrow_up: |
| [...ive/eventing/kafka/broker/core/utils/Shutdown.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2748/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdXRpbHMvU2h1dGRvd24uamF2YQ==) | | |
| [...ka/broker/core/reconciler/ResourcesReconciler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2748/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvcmVjb25jaWxlci9SZXNvdXJjZXNSZWNvbmNpbGVyLmphdmE=) | | |
| [.../core/reconciler/impl/ResourcesReconcilerImpl.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2748/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvcmVjb25jaWxlci9pbXBsL1Jlc291cmNlc1JlY29uY2lsZXJJbXBsLmphdmE=) | | |
| [...ispatcher/impl/http/WebClientCloudEventSender.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2748/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9odHRwL1dlYkNsaWVudENsb3VkRXZlbnRTZW5kZXIuamF2YQ==) | | |
| [...roker/dispatcher/impl/filter/AttributesFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2748/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvQXR0cmlidXRlc0ZpbHRlci5qYXZh) | | |
| [...ker/receiver/impl/StrictRequestToRecordMapper.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2748/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL1N0cmljdFJlcXVlc3RUb1JlY29yZE1hcHBlci5qYXZh) | | |
| [...ceiver/impl/handler/IngressRequestHandlerImpl.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2748/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL2hhbmRsZXIvSW5ncmVzc1JlcXVlc3RIYW5kbGVySW1wbC5qYXZh) | | |
| [.../kafka/broker/core/eventbus/ContractPublisher.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2748/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvZXZlbnRidXMvQ29udHJhY3RQdWJsaXNoZXIuamF2YQ==) | | |
| ... and [66 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2748/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

knative-prow[bot]: @aavarghese: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | bd6737a2cb562cdb7830804766f87ddb249a8dc5 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2748/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1585215465028849664) | true | `/test channel-integration-tests-sasl-plain`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow[bot]: @aavarghese: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | bd6737a2cb562cdb7830804766f87ddb249a8dc5 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2748/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1585215465028849664) | true | `/test channel-integration-tests-sasl-plain`
unit-tests_eventing-kafka-broker_main | 2c2180225e172e52f3c157230269cc13b3f1f629 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2748/unit-tests_eventing-kafka-broker_main/1585313810581819392) | true | `/test unit-tests`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow[bot]: @aavarghese: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | bd6737a2cb562cdb7830804766f87ddb249a8dc5 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2748/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1585215465028849664) | true | `/test channel-integration-tests-sasl-plain`
unit-tests_eventing-kafka-broker_main | 2c2180225e172e52f3c157230269cc13b3f1f629 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2748/unit-tests_eventing-kafka-broker_main/1585313810581819392) | true | `/test unit-tests`
build-tests_eventing-kafka-broker_main | 2c2180225e172e52f3c157230269cc13b3f1f629 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2748/build-tests_eventing-kafka-broker_main/1585313810179166208) | true | `/test build-tests`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow[bot]: @aavarghese: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | e377a8f21793aae96139daf4342928149369e887 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2748/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1585319957850230784) | true | `/test channel-integration-tests-sasl-plain`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow[bot]: @aavarghese: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | e377a8f21793aae96139daf4342928149369e887 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2748/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1585331900669497344) | true | `/test channel-integration-tests-sasl-plain`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
aavarghese: /test channel-integration-tests-sasl-plain
knative-prow[bot]: @aavarghese: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | e377a8f21793aae96139daf4342928149369e887 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2748/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1585339666406576128) | true | `/test channel-integration-tests-sasl-plain`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
aavarghese: /test channel-integration-tests-sasl-plain
knative-prow[bot]: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2748#" title="Author self-approved">aavarghese</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2748#pullrequestreview-1157689206" title="Approved">pierDipi</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka-broker).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/main/OWNERS)~~ [pierDipi]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->

Patch:
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
index bbfd5b293..bc303253d 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
@@ -27,7 +27,6 @@ import (
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
-	eventingv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 )
 
@@ -125,8 +124,23 @@ type DestinationReply struct {
 }
 
 type Auth struct {
-	NetSpec  *bindings.KafkaNetSpec
-	AuthSpec *eventingv1alpha1.Auth
+	NetSpec    *bindings.KafkaNetSpec
+	SecretSpec *SecretSpec
+}
+
+type SecretSpec struct {
+	// Secret reference for SASL and SSL configurations.
+	Ref *SecretReference `json:"ref,omitempty"`
+}
+
+type SecretReference struct {
+	Name      string `json:"name"`
+	Namespace string `json:"namespace"`
+}
+
+func (a *SecretSpec) HasSecret() bool {
+	return a != nil && a.Ref != nil &&
+		a.Ref.Name != "" && a.Ref.Namespace != ""
 }
 
 type DeliverySpec struct {
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go
index 3a68974e8..01104965c 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go
@@ -23,7 +23,6 @@ package v1alpha1
 
 import (
 	runtime "k8s.io/apimachinery/pkg/runtime"
-	eventingv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
 	v1beta1 "knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
 	duckv1 "knative.dev/eventing/pkg/apis/duck/v1"
 	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
@@ -39,9 +38,9 @@ func (in *Auth) DeepCopyInto(out *Auth) {
 		*out = new(v1beta1.KafkaNetSpec)
 		(*in).DeepCopyInto(*out)
 	}
-	if in.AuthSpec != nil {
-		in, out := &in.AuthSpec, &out.AuthSpec
-		*out = new(eventingv1alpha1.Auth)
+	if in.SecretSpec != nil {
+		in, out := &in.SecretSpec, &out.SecretSpec
+		*out = new(SecretSpec)
 		(*in).DeepCopyInto(*out)
 	}
 	return
@@ -519,6 +518,43 @@ func (in *ReplyStrategy) DeepCopy() *ReplyStrategy {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *SecretReference) DeepCopyInto(out *SecretReference) {
+	*out = *in
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new SecretReference.
+func (in *SecretReference) DeepCopy() *SecretReference {
+	if in == nil {
+		return nil
+	}
+	out := new(SecretReference)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *SecretSpec) DeepCopyInto(out *SecretSpec) {
+	*out = *in
+	if in.Ref != nil {
+		in, out := &in.Ref, &out.Ref
+		*out = new(SecretReference)
+		**out = **in
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new SecretSpec.
+func (in *SecretSpec) DeepCopy() *SecretSpec {
+	if in == nil {
+		return nil
+	}
+	out := new(SecretSpec)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *TopicReply) DeepCopyInto(out *TopicReply) {
 	*out = *in
diff --git a/control-plane/pkg/keda/keda.go b/control-plane/pkg/keda/keda.go
index e2049af3c..d6637654b 100644
--- a/control-plane/pkg/keda/keda.go
+++ b/control-plane/pkg/keda/keda.go
@@ -156,10 +156,10 @@ func GenerateTriggerAuthentication(cg *kafkainternals.ConsumerGroup, saslType *s
 		}
 	}
 
-	if cg.Spec.Template.Spec.Auth.AuthSpec != nil && cg.Spec.Template.Spec.Auth.AuthSpec.Secret.Ref.Name != "" {
+	if cg.Spec.Template.Spec.Auth.SecretSpec != nil && cg.Spec.Template.Spec.Auth.SecretSpec.Ref.Name != "" {
 		host := kedav1alpha1.AuthSecretTargetRef{
 			Parameter: "host", //TODO: parameter name?
-			Name:      cg.Spec.Template.Spec.Auth.AuthSpec.Secret.Ref.Name,
+			Name:      cg.Spec.Template.Spec.Auth.SecretSpec.Ref.Name,
 			Key:       "", //TODO: key value?
 		}
 		secretTargetRefs = append(secretTargetRefs, host)
diff --git a/control-plane/pkg/reconciler/consumer/consumer.go b/control-plane/pkg/reconciler/consumer/consumer.go
index f85d5b56c..5c88214b8 100644
--- a/control-plane/pkg/reconciler/consumer/consumer.go
+++ b/control-plane/pkg/reconciler/consumer/consumer.go
@@ -199,7 +199,7 @@ func (r Reconciler) reconcileAuth(ctx context.Context, c *kafkainternals.Consume
 		return nil
 	}
 
-	if c.Spec.Auth.AuthSpec != nil {
+	if c.Spec.Auth.SecretSpec != nil {
 		secret, err := security.Secret(ctx, &SecretLocator{Consumer: c}, r.SecretProviderFunc())
 		if err != nil {
 			return fmt.Errorf("failed to get secret: %w", err)
@@ -426,12 +426,12 @@ func (r *Reconciler) trackAuthContext(c *kafkainternals.Consumer, auth *kafkaint
 		return nil
 	}
 
-	if auth.AuthSpec.HasAuth() {
+	if auth.SecretSpec.HasSecret() {
 		ref := tracker.Reference{
 			APIVersion: "v1",
 			Kind:       "Secret",
-			Namespace:  c.GetNamespace(),
-			Name:       auth.AuthSpec.Secret.Ref.Name,
+			Namespace:  auth.SecretSpec.Ref.Namespace,
+			Name:       auth.SecretSpec.Ref.Name,
 		}
 		if err := r.Tracker.TrackReference(ref, c); err != nil {
 			return fmt.Errorf("failed to track secret for rotation %s/%s: %w", ref.Namespace, ref.Name, err)
diff --git a/control-plane/pkg/reconciler/consumer/secret_locator.go b/control-plane/pkg/reconciler/consumer/secret_locator.go
index b83a3873a..640ec28e1 100644
--- a/control-plane/pkg/reconciler/consumer/secret_locator.go
+++ b/control-plane/pkg/reconciler/consumer/secret_locator.go
@@ -25,20 +25,23 @@ type SecretLocator struct {
 }
 
 func (s *SecretLocator) SecretName() (string, bool) {
-	if s.hasNoAuth() {
+	if s.hasNoSecret() {
 		return "", false
 	}
-	return s.Consumer.Spec.Auth.AuthSpec.Secret.Ref.Name, true
+	return s.Consumer.Spec.Auth.SecretSpec.Ref.Name, true
 }
 
-func (s *SecretLocator) hasNoAuth() bool {
+func (s *SecretLocator) hasNoSecret() bool {
 	return s.Consumer.Spec.Auth == nil ||
-		s.Consumer.Spec.Auth.AuthSpec == nil ||
-		s.Consumer.Spec.Auth.AuthSpec.Secret == nil ||
-		s.Consumer.Spec.Auth.AuthSpec.Secret.Ref == nil ||
-		s.Consumer.Spec.Auth.AuthSpec.Secret.Ref.Name == ""
+		s.Consumer.Spec.Auth.SecretSpec == nil ||
+		s.Consumer.Spec.Auth.SecretSpec.Ref == nil ||
+		s.Consumer.Spec.Auth.SecretSpec.Ref.Name == "" ||
+		s.Consumer.Spec.Auth.SecretSpec.Ref.Namespace == ""
 }
 
 func (s *SecretLocator) SecretNamespace() (string, bool) {
-	return s.GetNamespace(), !s.hasNoAuth()
+	if s.hasNoSecret() {
+		return "", false
+	}
+	return s.Consumer.Spec.Auth.SecretSpec.Ref.Namespace, true
 }
diff --git a/control-plane/pkg/reconciler/consumergroup/auth.go b/control-plane/pkg/reconciler/consumergroup/auth.go
index 54cceccef..90bea0389 100644
--- a/control-plane/pkg/reconciler/consumergroup/auth.go
+++ b/control-plane/pkg/reconciler/consumergroup/auth.go
@@ -30,8 +30,8 @@ import (
 func (r Reconciler) newAuthConfigOption(ctx context.Context, cg *kafkainternals.ConsumerGroup) (kafka.ConfigOption, error) {
 	var secret *corev1.Secret
 
-	if hasAuthSpecAuthConfig(cg.Spec.Template.Spec.Auth) {
-		secret, err := security.Secret(ctx, &AuthSpecSecretLocator{cg}, security.DefaultSecretProviderFunc(r.SecretLister, r.KubeClient))
+	if hasSecretSpecConfig(cg.Spec.Template.Spec.Auth) {
+		secret, err := security.Secret(ctx, &SecretSpecSecretLocator{cg}, security.DefaultSecretProviderFunc(r.SecretLister, r.KubeClient))
 		if err != nil {
 			return nil, err
 		}
@@ -73,12 +73,12 @@ func (sl *NetSpecSecretLocator) SecretNamespace() (string, bool) {
 	return sl.Namespace, hasNetSpecAuthConfig(sl.ConsumerGroup.Spec.Template.Spec.Auth)
 }
 
-func hasAuthSpecAuthConfig(auth *kafkainternals.Auth) bool {
+func hasSecretSpecConfig(auth *kafkainternals.Auth) bool {
 	return auth != nil &&
-		auth.AuthSpec != nil &&
-		auth.AuthSpec.Secret != nil &&
-		auth.AuthSpec.Secret.Ref != nil &&
-		auth.AuthSpec.Secret.Ref.Name != ""
+		auth.SecretSpec != nil &&
+		auth.SecretSpec.Ref != nil &&
+		auth.SecretSpec.Ref.Name != "" &&
+		auth.SecretSpec.Ref.Namespace != ""
 }
 
 func hasNetSpecAuthConfig(auth *kafkainternals.Auth) bool {
@@ -93,21 +93,22 @@ func hasNetSpecAuthConfig(auth *kafkainternals.Auth) bool {
 			auth.NetSpec.SASL.Type.SecretKeyRef != nil)
 }
 
-type AuthSpecSecretLocator struct {
+type SecretSpecSecretLocator struct {
 	*kafkainternals.ConsumerGroup
 }
 
-func (sl *AuthSpecSecretLocator) SecretNamespace() (string, bool) {
-	if !hasAuthSpecAuthConfig(sl.ConsumerGroup.Spec.Template.Spec.Auth) {
+func (sl *SecretSpecSecretLocator) SecretNamespace() (string, bool) {
+	if !hasSecretSpecConfig(sl.ConsumerGroup.Spec.Template.Spec.Auth) {
 		return "", false
 	}
-	return sl.ConsumerGroup.GetNamespace(), hasAuthSpecAuthConfig(sl.ConsumerGroup.Spec.Template.Spec.Auth)
+	namespace := sl.ConsumerGroup.Spec.Template.Spec.Auth.SecretSpec.Ref.Namespace
+	return namespace, hasSecretSpecConfig(sl.ConsumerGroup.Spec.Template.Spec.Auth)
 }
 
-func (sl *AuthSpecSecretLocator) SecretName() (string, bool) {
-	if !hasAuthSpecAuthConfig(sl.ConsumerGroup.Spec.Template.Spec.Auth) {
+func (sl *SecretSpecSecretLocator) SecretName() (string, bool) {
+	if !hasSecretSpecConfig(sl.ConsumerGroup.Spec.Template.Spec.Auth) {
 		return "", false
 	}
-	name := sl.ConsumerGroup.Spec.Template.Spec.Auth.AuthSpec.Secret.Ref.Name
-	return name, hasAuthSpecAuthConfig(sl.ConsumerGroup.Spec.Template.Spec.Auth)
+	name := sl.ConsumerGroup.Spec.Template.Spec.Auth.SecretSpec.Ref.Name
+	return name, hasSecretSpecConfig(sl.ConsumerGroup.Spec.Template.Spec.Auth)
 }
diff --git a/control-plane/pkg/reconciler/consumergroup/consumergroup.go b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
index 58d1da7d6..cbef3c3c9 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
@@ -449,7 +449,7 @@ func (r Reconciler) reconcileKedaObjects(ctx context.Context, cg *kafkainternals
 	var triggerAuthentication *kedav1alpha1.TriggerAuthentication
 	var secret *corev1.Secret
 
-	if hasAuthSpecAuthConfig(cg.Spec.Template.Spec.Auth) || hasNetSpecAuthConfig(cg.Spec.Template.Spec.Auth) {
+	if hasSecretSpecConfig(cg.Spec.Template.Spec.Auth) || hasNetSpecAuthConfig(cg.Spec.Template.Spec.Auth) {
 		saslType, err := r.retrieveSaslTypeIfPresent(ctx, cg)
 		if err != nil {
 			return err
diff --git a/control-plane/pkg/reconciler/trigger/v2/triggerv2.go b/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
index 77d43c746..1afc5960f 100644
--- a/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
+++ b/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
@@ -46,8 +46,6 @@ import (
 	kafkalogging "knative.dev/eventing-kafka-broker/control-plane/pkg/logging"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
 
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
-
 	kedafunc "knative.dev/eventing-kafka-broker/control-plane/pkg/keda"
 	brokerreconciler "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
 )
@@ -136,7 +134,12 @@ func (r Reconciler) reconcileConsumerGroup(ctx context.Context, broker *eventing
 		offset = sources.OffsetEarliest
 	}
 
-	secret, err := security.Secret(ctx, &security.AnnotationsSecretLocator{Annotations: broker.Status.Annotations, Namespace: trigger.Namespace}, security.DefaultSecretProviderFunc(r.SecretLister, r.KubeClient))
+	namespace := broker.GetNamespace()
+	if broker.Spec.Config.Namespace != "" {
+		namespace = broker.Spec.Config.Namespace
+	}
+
+	secret, err := security.Secret(ctx, &security.AnnotationsSecretLocator{Annotations: broker.Status.Annotations, Namespace: namespace}, security.DefaultSecretProviderFunc(r.SecretLister, r.KubeClient))
 	if err != nil {
 		return nil, fmt.Errorf("failed to get secret: %w", err)
 	}
@@ -189,11 +192,10 @@ func (r Reconciler) reconcileConsumerGroup(ctx context.Context, broker *eventing
 
 	if secret != nil {
 		expectedCg.Spec.Template.Spec.Auth = &internalscg.Auth{
-			AuthSpec: &v1alpha1.Auth{
-				Secret: &v1alpha1.Secret{
-					Ref: &v1alpha1.SecretReference{
-						Name: secret.Name,
-					},
+			SecretSpec: &internalscg.SecretSpec{
+				Ref: &internalscg.SecretReference{
+					Name:      secret.Name,
+					Namespace: secret.Namespace,
 				},
 			},
 		}


Test patch:
diff --git a/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go b/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
index d1bbb2bb8..69b538908 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
@@ -41,7 +41,6 @@ import (
 	eventingduckv1alpha1 "knative.dev/eventing/pkg/apis/duck/v1alpha1"
 	"knative.dev/eventing/pkg/scheduler"
 
-	eventing "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
 	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 	fakekafkainternalsclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client/fake"
@@ -172,9 +171,11 @@ func TestReconcileKind(t *testing.T) {
 							ConsumerGroupIdConfig("my.group.id"),
 						),
 						ConsumerAuth(&kafkainternals.Auth{
-							AuthSpec: &eventing.Auth{Secret: &eventing.Secret{Ref: &eventing.SecretReference{
-								Name: "non-existing secret",
-							}}},
+							SecretSpec: &kafkainternals.SecretSpec{
+								Ref: &kafkainternals.SecretReference{
+									Name:      "non-existing secret",
+									Namespace: "non-existing secret",
+								}},
 						}),
 						ConsumerDelivery(NewConsumerSpecDelivery("", ConsumerInitialOffset(sources.OffsetLatest))),
 						ConsumerVReplicas(1),
@@ -194,9 +195,11 @@ func TestReconcileKind(t *testing.T) {
 						),
 						ConsumerDelivery(NewConsumerSpecDelivery("", ConsumerInitialOffset(sources.OffsetLatest))),
 						ConsumerAuth(&kafkainternals.Auth{
-							AuthSpec: &eventing.Auth{Secret: &eventing.Secret{Ref: &eventing.SecretReference{
-								Name: "non-existing secret",
-							}}},
+							SecretSpec: &kafkainternals.SecretSpec{
+								Ref: &kafkainternals.SecretReference{
+									Name:      "non-existing secret",
+									Namespace: "non-existing secret",
+								}},
 						}),
 					)),
 					ConsumerForTrigger(),
@@ -218,7 +221,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			WantEvents: []string{
 				finalizerUpdatedEvent,
-				"Warning InternalError failed to initialize consumer group offset: failed to create config options for Kafka cluster auth: failed to get secret test-cg-ns/non-existing secret: secrets \"non-existing secret\" not found",
+				"Warning InternalError failed to initialize consumer group offset: failed to create config options for Kafka cluster auth: failed to get secret non-existing secret/non-existing secret: secrets \"non-existing secret\" not found",
 			},
 			WantCreates: []runtime.Object{},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
@@ -233,16 +236,18 @@ func TestReconcileKind(t *testing.T) {
 								),
 								ConsumerDelivery(NewConsumerSpecDelivery("", ConsumerInitialOffset(sources.OffsetLatest))),
 								ConsumerAuth(&kafkainternals.Auth{
-									AuthSpec: &eventing.Auth{Secret: &eventing.Secret{Ref: &eventing.SecretReference{
-										Name: "non-existing secret",
-									}}},
+									SecretSpec: &kafkainternals.SecretSpec{
+										Ref: &kafkainternals.SecretReference{
+											Name:      "non-existing secret",
+											Namespace: "non-existing secret",
+										}},
 								}),
 							)),
 							ConsumerGroupReplicas(2),
 							ConsumerForTrigger(),
 						)
 						cg.InitializeConditions()
-						_ = cg.MarkInitializeOffsetFailed("InitializeOffset", errors.New("failed to create config options for Kafka cluster auth: failed to get secret test-cg-ns/non-existing secret: secrets \"non-existing secret\" not found"))
+						_ = cg.MarkInitializeOffsetFailed("InitializeOffset", errors.New("failed to create config options for Kafka cluster auth: failed to get secret non-existing secret/non-existing secret: secrets \"non-existing secret\" not found"))
 						return cg
 					}(),
 				},
@@ -251,7 +256,7 @@ func TestReconcileKind(t *testing.T) {
 		{
 			Name: "Consumers in multiple pods, propagate consumer not ready condition",
 			Objects: []runtime.Object{
-				NewSSLSecret(ConsumerGroupNamespace, SecretName),
+				NewSSLSecret(SecretNamespace, SecretName),
 				NewConsumer(2,
 					ConsumerSpec(NewConsumerSpec(
 						ConsumerTopics("t1", "t2"),
@@ -261,9 +266,11 @@ func TestReconcileKind(t *testing.T) {
 						),
 						ConsumerDelivery(NewConsumerSpecDelivery("", ConsumerInitialOffset(sources.OffsetLatest))),
 						ConsumerAuth(&kafkainternals.Auth{
-							AuthSpec: &eventing.Auth{Secret: &eventing.Secret{Ref: &eventing.SecretReference{
-								Name: SecretName,
-							}}},
+							SecretSpec: &kafkainternals.SecretSpec{
+								Ref: &kafkainternals.SecretReference{
+									Name:      SecretName,
+									Namespace: SecretNamespace,
+								}},
 						}),
 						ConsumerVReplicas(1),
 						ConsumerPlacement(kafkainternals.PodBind{
@@ -282,9 +289,11 @@ func TestReconcileKind(t *testing.T) {
 						),
 						ConsumerDelivery(NewConsumerSpecDelivery("", ConsumerInitialOffset(sources.OffsetLatest))),
 						ConsumerAuth(&kafkainternals.Auth{
-							AuthSpec: &eventing.Auth{Secret: &eventing.Secret{Ref: &eventing.SecretReference{
-								Name: SecretName,
-							}}},
+							SecretSpec: &kafkainternals.SecretSpec{
+								Ref: &kafkainternals.SecretReference{
+									Name:      SecretName,
+									Namespace: SecretNamespace,
+								}},
 						}),
 					)),
 					ConsumerForTrigger(),
@@ -311,9 +320,11 @@ func TestReconcileKind(t *testing.T) {
 						),
 						ConsumerDelivery(NewConsumerSpecDelivery("", ConsumerInitialOffset(sources.OffsetLatest))),
 						ConsumerAuth(&kafkainternals.Auth{
-							AuthSpec: &eventing.Auth{Secret: &eventing.Secret{Ref: &eventing.SecretReference{
-								Name: SecretName,
-							}}},
+							SecretSpec: &kafkainternals.SecretSpec{
+								Ref: &kafkainternals.SecretReference{
+									Name:      SecretName,
+									Namespace: SecretNamespace,
+								}},
 						}),
 						ConsumerVReplicas(1),
 						ConsumerPlacement(kafkainternals.PodBind{PodName: "p1", PodNamespace: systemNamespace}),
@@ -332,9 +343,11 @@ func TestReconcileKind(t *testing.T) {
 								),
 								ConsumerDelivery(NewConsumerSpecDelivery("", ConsumerInitialOffset(sources.OffsetLatest))),
 								ConsumerAuth(&kafkainternals.Auth{
-									AuthSpec: &eventing.Auth{Secret: &eventing.Secret{Ref: &eventing.SecretReference{
-										Name: SecretName,
-									}}},
+									SecretSpec: &kafkainternals.SecretSpec{
+										Ref: &kafkainternals.SecretReference{
+											Name:      SecretName,
+											Namespace: SecretNamespace,
+										}},
 								}),
 							)),
 							ConsumerGroupReplicas(2),
@@ -368,7 +381,7 @@ func TestReconcileKind(t *testing.T) {
 		{
 			Name: "Consumers in multiple pods, with auth spec, one exists - ready",
 			Objects: []runtime.Object{
-				NewSSLSecret(ConsumerGroupNamespace, SecretName),
+				NewSSLSecret(SecretNamespace, SecretName),
 				NewConsumer(2,
 					ConsumerSpec(NewConsumerSpec(
 						ConsumerTopics("t1", "t2"),
@@ -378,9 +391,11 @@ func TestReconcileKind(t *testing.T) {
 						),
 						ConsumerDelivery(NewConsumerSpecDelivery("", ConsumerInitialOffset(sources.OffsetLatest))),
 						ConsumerAuth(&kafkainternals.Auth{
-							AuthSpec: &eventing.Auth{Secret: &eventing.Secret{Ref: &eventing.SecretReference{
-								Name: SecretName,
-							}}},
+							SecretSpec: &kafkainternals.SecretSpec{
+								Ref: &kafkainternals.SecretReference{
+									Name:      SecretName,
+									Namespace: SecretNamespace,
+								}},
 						}),
 						ConsumerVReplicas(1),
 						ConsumerPlacement(kafkainternals.PodBind{
@@ -399,9 +414,11 @@ func TestReconcileKind(t *testing.T) {
 						),
 						ConsumerDelivery(NewConsumerSpecDelivery("", ConsumerInitialOffset(sources.OffsetLatest))),
 						ConsumerAuth(&kafkainternals.Auth{
-							AuthSpec: &eventing.Auth{Secret: &eventing.Secret{Ref: &eventing.SecretReference{
-								Name: SecretName,
-							}}},
+							SecretSpec: &kafkainternals.SecretSpec{
+								Ref: &kafkainternals.SecretReference{
+									Name:      SecretName,
+									Namespace: SecretNamespace,
+								}},
 						}),
 					)),
 					ConsumerForTrigger(),
@@ -427,9 +444,11 @@ func TestReconcileKind(t *testing.T) {
 						),
 						ConsumerDelivery(NewConsumerSpecDelivery("", ConsumerInitialOffset(sources.OffsetLatest))),
 						ConsumerAuth(&kafkainternals.Auth{
-							AuthSpec: &eventing.Auth{Secret: &eventing.Secret{Ref: &eventing.SecretReference{
-								Name: SecretName,
-							}}},
+							SecretSpec: &kafkainternals.SecretSpec{
+								Ref: &kafkainternals.SecretReference{
+									Name:      SecretName,
+									Namespace: SecretNamespace,
+								}},
 						}),
 						ConsumerVReplicas(1),
 						ConsumerPlacement(kafkainternals.PodBind{PodName: "p1", PodNamespace: systemNamespace}),
@@ -448,9 +467,11 @@ func TestReconcileKind(t *testing.T) {
 								),
 								ConsumerDelivery(NewConsumerSpecDelivery("", ConsumerInitialOffset(sources.OffsetLatest))),
 								ConsumerAuth(&kafkainternals.Auth{
-									AuthSpec: &eventing.Auth{Secret: &eventing.Secret{Ref: &eventing.SecretReference{
-										Name: SecretName,
-									}}},
+									SecretSpec: &kafkainternals.SecretSpec{
+										Ref: &kafkainternals.SecretReference{
+											Name:      SecretName,
+											Namespace: SecretNamespace,
+										}},
 								}),
 							)),
 							ConsumerGroupReplicas(2),
diff --git a/control-plane/pkg/reconciler/testing/objects_broker.go b/control-plane/pkg/reconciler/testing/objects_broker.go
index cb3404deb..8194717e4 100644
--- a/control-plane/pkg/reconciler/testing/objects_broker.go
+++ b/control-plane/pkg/reconciler/testing/objects_broker.go
@@ -194,6 +194,15 @@ func WithBootstrapServerStatusAnnotation(servers string) reconcilertesting.Broke
 	}
 }
 
+func WithSecretStatusAnnotation(name string) reconcilertesting.BrokerOption {
+	return func(broker *eventing.Broker) {
+		if broker.Status.Annotations == nil {
+			broker.Status.Annotations = make(map[string]string, 10)
+		}
+		broker.Status.Annotations[security.AuthSecretNameKey] = name
+	}
+}
+
 func WithAutoscalingAnnotationsTrigger() reconcilertesting.TriggerOption {
 	return func(trigger *eventing.Trigger) {
 		if trigger.Annotations == nil {
diff --git a/control-plane/pkg/reconciler/testing/objects_consumer.go b/control-plane/pkg/reconciler/testing/objects_consumer.go
index e254b3723..b5f504007 100644
--- a/control-plane/pkg/reconciler/testing/objects_consumer.go
+++ b/control-plane/pkg/reconciler/testing/objects_consumer.go
@@ -39,7 +39,8 @@ const (
 	ConsumerNamePrefix = "test-cg"
 	ConsumerNamespace  = "test-cg-ns"
 
-	SecretName = "secret-name"
+	SecretName      = "secret-name"
+	SecretNamespace = SystemNamespace
 )
 
 var (
diff --git a/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go b/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go
index a800c1754..47a7c65e2 100644
--- a/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go
+++ b/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go
@@ -26,6 +26,7 @@ import (
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 
 	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
+	internalscg "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 	brokerreconciler "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
@@ -488,6 +489,88 @@ func TestReconcileKind(t *testing.T) {
 				},
 			},
 		},
+		{
+			Name: "Reconciled normal - with auth - SASL",
+			Objects: []runtime.Object{
+				NewBroker(
+					BrokerReady,
+					WithTopicStatusAnnotation(BrokerTopic()),
+					WithBootstrapServerStatusAnnotation(bootstrapServers),
+					WithSecretStatusAnnotation("secret-1"),
+				),
+				DataPlaneConfigMap(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigConfigMapName, brokerreconciler.ConsumerConfigKey,
+					DataPlaneConfigInitialOffset(brokerreconciler.ConsumerConfigKey, sources.OffsetLatest),
+				),
+				newTrigger(),
+				NewConsumerGroup(
+					WithConsumerGroupName(TriggerUUID),
+					WithConsumerGroupNamespace(triggerNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(newTrigger())),
+					WithConsumerGroupMetaLabels(OwnerAsTriggerLabel),
+					WithConsumerGroupLabels(ConsumerTriggerLabel),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(ChannelTopic()),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(TriggerUUID),
+							ConsumerBootstrapServersConfig(bootstrapServers),
+						),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+						ConsumerSubscriber(NewConsumerSpecSubscriber(Subscription1URI)),
+						ConsumerAuth(&internalscg.Auth{
+							SecretSpec: &internalscg.SecretSpec{
+								Ref: &internalscg.SecretReference{
+									Name:      "secret-1",
+									Namespace: ConfigMapNamespace,
+								},
+							},
+						}),
+					)),
+				),
+				NewLegacySASLSecret(ConfigMapNamespace, "secret-1"),
+			},
+			Key: testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewConsumerGroup(
+						WithConsumerGroupName(TriggerUUID),
+						WithConsumerGroupNamespace(triggerNamespace),
+						WithConsumerGroupOwnerRef(kmeta.NewControllerRef(newTrigger())),
+						WithConsumerGroupMetaLabels(OwnerAsTriggerLabel),
+						WithConsumerGroupLabels(ConsumerTriggerLabel),
+						ConsumerGroupConsumerSpec(NewConsumerSpec(
+							ConsumerTopics(BrokerTopics[0]),
+							ConsumerConfigs(
+								ConsumerGroupIdConfig(TriggerUUID),
+								ConsumerBootstrapServersConfig(bootstrapServers),
+							),
+							ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered, ConsumerInitialOffset(sources.OffsetLatest))),
+							ConsumerFilters(NewConsumerSpecFilters()),
+							ConsumerReply(ConsumerTopicReply()),
+							ConsumerAuth(&internalscg.Auth{
+								SecretSpec: &internalscg.SecretSpec{
+									Ref: &internalscg.SecretReference{
+										Name:      "secret-1",
+										Namespace: ConfigMapNamespace,
+									},
+								},
+							}),
+						)),
+					),
+				},
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerSubscribed(),
+						reconcilertesting.WithTriggerBrokerReady(),
+						withTriggerSubscriberResolvedSucceeded(),
+						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group is not ready"),
+						withDeadLetterSinkURI(""),
+					),
+				},
+			},
+		},
 		{
 			Name: "Reconciled normal - existing cg without update",
 			Objects: []runtime.Object{