Repository: knative-sandbox/eventing-kafka-broker
Pull request #2490 (2022-08-09 13:15:27)

[release-1.6] Track secrets in consumer reconciler  (#2153)
Cherry-pick https://github.com/knative-sandbox/eventing-kafka-broker/commit/d250e80663febb3a58b57308138c6d9bd5a11b24

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2490?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2490](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2490?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (75ff951) into [release-1.6](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/c1bf4acedf9e52ee9ca7e0ebc707000368b82b8a?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (c1bf4ac) will **decrease** coverage by `0.14%`.
> The diff coverage is `3.84%`.

```diff
@@               Coverage Diff               @@
##           release-1.6    #2490      +/-   ##
===============================================
- Coverage        59.95%   59.81%   -0.15%     
===============================================
  Files               69       69              
  Lines             6973     6988      +15     
===============================================
- Hits              4181     4180       -1     
- Misses            2496     2511      +15     
- Partials           296      297       +1     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2490?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...ane/pkg/apis/eventing/v1alpha1/kafka\_sink\_types.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2490/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9rYWZrYV9zaW5rX3R5cGVzLmdv) | `54.54% <0.00%> (-20.46%)` | :arrow_down: |
| [control-plane/pkg/reconciler/consumer/consumer.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2490/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lci9jb25zdW1lci5nbw==) | `50.68% <0.00%> (-1.97%)` | :arrow_down: |
| [control-plane/pkg/security/secrets\_tracker.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2490/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvc2VjdXJpdHkvc2VjcmV0c190cmFja2VyLmdv) | `87.50% <25.00%> (-12.50%)` | :arrow_down: |

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]: @pierDipi: 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
--- | --- | --- | --- | ---
build-tests_eventing-kafka-broker_release-1.6 | 75ff951cce9bad0ca8c112ff73de7841a6b7f29b | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2490/build-tests_eventing-kafka-broker_release-1.6/1556992559841021952) | true | `/test build-tests_eventing-kafka-broker_release-1.6`

[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]: @pierDipi: 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-ssl_eventing-kafka-broker_release-1.6 | 9135035fe73dbe11a118a790bf42a364ac8926f9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2490/channel-integration-tests-ssl_eventing-kafka-broker_release-1.6/1556995116059594752) | true | `/test channel-integration-tests-ssl_eventing-kafka-broker_release-1.6`

[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]: @pierDipi: 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-ssl_eventing-kafka-broker_release-1.6 | 9135035fe73dbe11a118a790bf42a364ac8926f9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2490/channel-integration-tests-ssl_eventing-kafka-broker_release-1.6/1556995116059594752) | true | `/test channel-integration-tests-ssl_eventing-kafka-broker_release-1.6`
integration-tests_eventing-kafka-broker_release-1.6 | 9135035fe73dbe11a118a790bf42a364ac8926f9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2490/integration-tests_eventing-kafka-broker_release-1.6/1556995113362657280) | true | `/test integration-tests_eventing-kafka-broker_release-1.6`

[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 -->
pierDipi: /retest-required
pierDipi: /test integration-tests_eventing-kafka-broker_release-1.6
knative-prow[bot]: @pierDipi: 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
--- | --- | --- | --- | ---
integration-tests_eventing-kafka-broker_release-1.6 | 9135035fe73dbe11a118a790bf42a364ac8926f9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2490/integration-tests_eventing-kafka-broker_release-1.6/1557024269752340480) | true | `/test integration-tests_eventing-kafka-broker_release-1.6`

[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 -->
pierDipi: /retest-required
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/2490#pullrequestreview-1067706955" title="Approved">matzew</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2490#" title="Author self-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/release-1.6/OWNERS)~~ [matzew,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/eventing/v1alpha1/kafka_sink_types.go b/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_types.go
index 1814e2e2f..b5ddaef91 100644
--- a/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_types.go
+++ b/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_types.go
@@ -103,6 +103,11 @@ type Auth struct {
 	Secret *Secret `json:"secret,omitempty"`
 }
 
+func (a *Auth) HasAuth() bool {
+	return a != nil && a.Secret != nil &&
+		a.Secret.Ref != nil && a.Secret.Ref.Name != ""
+}
+
 type Secret struct {
 	// Secret reference for SASL and SSL configurations.
 	Ref *SecretReference `json:"ref,omitempty"`
diff --git a/control-plane/pkg/reconciler/consumer/consumer.go b/control-plane/pkg/reconciler/consumer/consumer.go
index 98248f291..327252078 100644
--- a/control-plane/pkg/reconciler/consumer/consumer.go
+++ b/control-plane/pkg/reconciler/consumer/consumer.go
@@ -179,10 +179,14 @@ func (r Reconciler) reconcileContractEgress(ctx context.Context, c *kafkainterna
 }
 
 func (r Reconciler) reconcileAuth(ctx context.Context, c *kafkainternals.Consumer, resource *contract.Resource) error {
-
 	if c.Spec.Auth == nil {
 		return nil
 	}
+
+	if err := r.trackAuthContext(c, c.Spec.Auth); err != nil {
+		return err
+	}
+
 	if c.Spec.Auth.NetSpec != nil {
 		authContext, err := security.ResolveAuthContextFromNetSpec(r.SecretLister, c.GetNamespace(), *c.Spec.Auth.NetSpec)
 		if err != nil {
@@ -191,6 +195,7 @@ func (r Reconciler) reconcileAuth(ctx context.Context, c *kafkainternals.Consume
 		resource.Auth = &contract.Resource_MultiAuthSecret{MultiAuthSecret: authContext.MultiSecretReference}
 		return nil
 	}
+
 	if c.Spec.Auth.AuthSpec != nil {
 		secret, err := security.Secret(ctx, &SecretLocator{Consumer: c}, r.SecretProviderFunc())
 		if err != nil {
@@ -204,15 +209,6 @@ func (r Reconciler) reconcileAuth(ctx context.Context, c *kafkainternals.Consume
 				Version:   secret.ResourceVersion,
 			},
 		}
-		ref := tracker.Reference{
-			APIVersion: c.APIVersion,
-			Kind:       c.Kind,
-			Namespace:  c.GetNamespace(),
-			Name:       c.GetName(),
-		}
-		if err := r.Tracker.TrackReference(ref, c); err != nil {
-			return fmt.Errorf("failed to track reference %v: %w", ref, err)
-		}
 		return nil
 	}
 
@@ -412,3 +408,24 @@ func podOwnerReference(p *corev1.Pod) base.ConfigMapOption {
 		})
 	}
 }
+
+func (r *Reconciler) trackAuthContext(c *kafkainternals.Consumer, auth *kafkainternals.Auth) error {
+	if auth == nil {
+		return nil
+	}
+
+	if auth.AuthSpec.HasAuth() {
+		ref := tracker.Reference{
+			APIVersion: "v1",
+			Kind:       "Secret",
+			Namespace:  c.GetNamespace(),
+			Name:       auth.AuthSpec.Secret.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)
+		}
+		return nil
+	}
+
+	return security.TrackNetSpecSecrets(r.Tracker, auth.NetSpec, c)
+}
diff --git a/control-plane/pkg/security/secrets_tracker.go b/control-plane/pkg/security/secrets_tracker.go
index 123b77f76..503975f38 100644
--- a/control-plane/pkg/security/secrets_tracker.go
+++ b/control-plane/pkg/security/secrets_tracker.go
@@ -24,7 +24,11 @@ import (
 
 // TrackNetSpecSecrets tracks all secrets referenced by a provided bindings.KafkaNetSpec.
 // parent is the object that is tracking changes to those secrets.
-func TrackNetSpecSecrets(secretsTracker tracker.Interface, netSpec bindings.KafkaNetSpec, parent metav1.Object) error {
+func TrackNetSpecSecrets(secretsTracker tracker.Interface, netSpec *bindings.KafkaNetSpec, parent metav1.Object) error {
+	if netSpec == nil {
+		return nil
+	}
+
 	secrets := []bindings.SecretValueFromSource{
 		netSpec.TLS.Key,
 		netSpec.TLS.Cert,
diff --git a/vendor/modules.txt b/vendor/modules.txt
index e11cdf718..4ad132cce 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1367,6 +1367,8 @@ knative.dev/eventing-kafka/test/lib
 knative.dev/eventing-kafka/test/lib/resources
 knative.dev/eventing-kafka/test/rekt/features/kafkachannel
 knative.dev/eventing-kafka/test/rekt/resources/kafkachannel
+knative.dev/eventing-kafka/test/rekt/resources/kafkasource
+knative.dev/eventing-kafka/test/rekt/resources/kafkatopic
 knative.dev/eventing-kafka/test/rekt/resources/resetoffset
 knative.dev/eventing-kafka/test/upgrade
 knative.dev/eventing-kafka/test/upgrade/continual


Test patch:
diff --git a/control-plane/pkg/security/secrets_tracker_test.go b/control-plane/pkg/security/secrets_tracker_test.go
index c990ac719..d5be3b293 100644
--- a/control-plane/pkg/security/secrets_tracker_test.go
+++ b/control-plane/pkg/security/secrets_tracker_test.go
@@ -131,7 +131,7 @@ func TestTrackNetSpecSecrets(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			if err := TrackNetSpecSecrets(tt.secretsTracker, tt.netSpec, tt.parent); (err != nil) != tt.wantErr {
+			if err := TrackNetSpecSecrets(tt.secretsTracker, &tt.netSpec, tt.parent); (err != nil) != tt.wantErr {
 				t.Errorf("TrackNetSpecSecrets() error = %v, wantErr %v", err, tt.wantErr)
 			}
 			if tt.secretsTracker.trackReferenceCalls != tt.expectedTrackReferenceCalls {
diff --git a/test/e2e/broker_event_trasformation_test.go b/test/e2e/broker_event_trasformation_test.go
index 28e26faf6..827889549 100644
--- a/test/e2e/broker_event_trasformation_test.go
+++ b/test/e2e/broker_event_trasformation_test.go
@@ -25,8 +25,8 @@ import (
 
 	"knative.dev/eventing/test/e2e/helpers"
 
+	pkgtesting "knative.dev/eventing-kafka-broker/test/pkg"
 	testbroker "knative.dev/eventing-kafka-broker/test/pkg/broker"
-	pkgtesting "knative.dev/eventing-kafka-broker/test/pkg/testing"
 )
 
 func TestEventTransformationForTriggerV1BrokerV1(t *testing.T) {
diff --git a/test/e2e/broker_redelivery_test.go b/test/e2e/broker_redelivery_test.go
index 937781e2d..4b3b9d73f 100644
--- a/test/e2e/broker_redelivery_test.go
+++ b/test/e2e/broker_redelivery_test.go
@@ -30,7 +30,7 @@ import (
 	"knative.dev/eventing/test/lib/resources"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
-	kafkatesting "knative.dev/eventing-kafka-broker/test/pkg/testing"
+	kafkatesting "knative.dev/eventing-kafka-broker/test/pkg"
 )
 
 const (
diff --git a/test/e2e/broker_sasl_ssl_test.go b/test/e2e/broker_sasl_ssl_test.go
index efa3ee71f..0fd539ec5 100644
--- a/test/e2e/broker_sasl_ssl_test.go
+++ b/test/e2e/broker_sasl_ssl_test.go
@@ -39,6 +39,7 @@ import (
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
+	. "knative.dev/eventing-kafka-broker/test/pkg"
 	. "knative.dev/eventing-kafka-broker/test/pkg/testing"
 )
 
diff --git a/test/e2e/broker_trigger_sink_test.go b/test/e2e/broker_trigger_sink_test.go
index 27fb01cdd..102455db1 100644
--- a/test/e2e/broker_trigger_sink_test.go
+++ b/test/e2e/broker_trigger_sink_test.go
@@ -35,9 +35,9 @@ import (
 	eventingv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
 	eventingv1alpha1clientset "knative.dev/eventing-kafka-broker/control-plane/pkg/client/clientset/versioned/typed/eventing/v1alpha1"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
+	testingpkg "knative.dev/eventing-kafka-broker/test/pkg"
 	"knative.dev/eventing-kafka-broker/test/pkg/addressable"
 	"knative.dev/eventing-kafka-broker/test/pkg/sink"
-	testingpkg "knative.dev/eventing-kafka-broker/test/pkg/testing"
 )
 
 /*
diff --git a/test/e2e/broker_trigger_test.go b/test/e2e/broker_trigger_test.go
index 9c1b3198d..2b7ca70ab 100644
--- a/test/e2e/broker_trigger_test.go
+++ b/test/e2e/broker_trigger_test.go
@@ -36,8 +36,8 @@ import (
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
+	testingpkg "knative.dev/eventing-kafka-broker/test/pkg"
 	kafkatest "knative.dev/eventing-kafka-broker/test/pkg/kafka"
-	testingpkg "knative.dev/eventing-kafka-broker/test/pkg/testing"
 )
 
 func TestBrokerTrigger(t *testing.T) {
diff --git a/test/e2e/conformance/control_plane_conformance_test.go b/test/e2e/conformance/control_plane_conformance_test.go
index c4950297c..4b39f1433 100644
--- a/test/e2e/conformance/control_plane_conformance_test.go
+++ b/test/e2e/conformance/control_plane_conformance_test.go
@@ -27,7 +27,7 @@ import (
 	"knative.dev/eventing/test/lib/resources"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
-	pkgtesting "knative.dev/eventing-kafka-broker/test/pkg/testing"
+	pkgtesting "knative.dev/eventing-kafka-broker/test/pkg"
 )
 
 func brokerCreator(client *testlib.Client, name string) {
diff --git a/test/e2e/conformance/data_plane_conformance_test.go b/test/e2e/conformance/data_plane_conformance_test.go
index 4459b818c..36dd0ab5e 100644
--- a/test/e2e/conformance/data_plane_conformance_test.go
+++ b/test/e2e/conformance/data_plane_conformance_test.go
@@ -35,7 +35,7 @@ import (
 	pkgtest "knative.dev/pkg/test"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
-	pkgtesting "knative.dev/eventing-kafka-broker/test/pkg/testing"
+	pkgtesting "knative.dev/eventing-kafka-broker/test/pkg"
 )
 
 func TestBrokerIngress(t *testing.T) {
diff --git a/test/e2e/delete_cm_test.go b/test/e2e/delete_cm_test.go
index 6fcf09c76..18e3fc7ea 100644
--- a/test/e2e/delete_cm_test.go
+++ b/test/e2e/delete_cm_test.go
@@ -34,9 +34,9 @@ import (
 
 	eventingv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
 	eventingv1alpha1clientset "knative.dev/eventing-kafka-broker/control-plane/pkg/client/clientset/versioned/typed/eventing/v1alpha1"
+	testingpkg "knative.dev/eventing-kafka-broker/test/pkg"
 	"knative.dev/eventing-kafka-broker/test/pkg/addressable"
 	"knative.dev/eventing-kafka-broker/test/pkg/sink"
-	testingpkg "knative.dev/eventing-kafka-broker/test/pkg/testing"
 )
 
 func TestDeleteSinkConfigMap(t *testing.T) {
diff --git a/test/e2e/kafka_sink.go b/test/e2e/kafka_sink.go
index 3760f60f8..b16468fdf 100644
--- a/test/e2e/kafka_sink.go
+++ b/test/e2e/kafka_sink.go
@@ -29,6 +29,7 @@ import (
 
 	eventingv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
 	eventingv1alpha1clientset "knative.dev/eventing-kafka-broker/control-plane/pkg/client/clientset/versioned/typed/eventing/v1alpha1"
+	. "knative.dev/eventing-kafka-broker/test/pkg"
 	"knative.dev/eventing-kafka-broker/test/pkg/addressable"
 	"knative.dev/eventing-kafka-broker/test/pkg/sink"
 
diff --git a/test/e2e/kafka_sink_test.go b/test/e2e/kafka_sink_test.go
index 735fc6054..0a37809ce 100644
--- a/test/e2e/kafka_sink_test.go
+++ b/test/e2e/kafka_sink_test.go
@@ -25,6 +25,7 @@ import (
 	"k8s.io/utils/pointer"
 
 	eventingv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
+	. "knative.dev/eventing-kafka-broker/test/pkg"
 	. "knative.dev/eventing-kafka-broker/test/pkg/testing"
 )
 
diff --git a/test/e2e/kafka_source_test.go b/test/e2e/kafka_source_test.go
index 06e5b9518..698749832 100644
--- a/test/e2e/kafka_source_test.go
+++ b/test/e2e/kafka_source_test.go
@@ -24,7 +24,7 @@ import (
 
 	eventingkafkahelpers "knative.dev/eventing-kafka/test/e2e/helpers"
 
-	testingpkg "knative.dev/eventing-kafka-broker/test/pkg/testing"
+	testingpkg "knative.dev/eventing-kafka-broker/test/pkg"
 )
 
 func TestKafkaSourceUpdate(t *testing.T) {
diff --git a/test/e2e/sacura_test.go b/test/e2e/sacura_test.go
index d251db478..1238ae34b 100644
--- a/test/e2e/sacura_test.go
+++ b/test/e2e/sacura_test.go
@@ -36,6 +36,7 @@ import (
 
 	kafkatest "knative.dev/eventing-kafka-broker/test/pkg/kafka"
 	pkgtesting "knative.dev/eventing-kafka-broker/test/pkg/testing"
+	pkgtest "knative.dev/eventing-kafka-broker/test/pkg"
 )
 
 const (
@@ -85,7 +86,7 @@ func TestSacuraJob(t *testing.T) {
 				Name:      sacuraVerifyCommittedOffsetJob,
 			},
 			&kafkatest.AdminConfig{
-				BootstrapServers: pkgtesting.BootstrapServersPlaintext,
+				BootstrapServers: pkgtest.BootstrapServersPlaintext,
 				Topic:            sacuraTopic,
 				Group:            string(trigger.UID),
 			},
diff --git a/test/e2e_new/features/featuressteps/copy_secret.go b/test/e2e_new/features/featuressteps/copy_secret.go
new file mode 100644
index 000000000..ee641db25
--- /dev/null
+++ b/test/e2e_new/features/featuressteps/copy_secret.go
@@ -0,0 +1,42 @@
+/*
+ * Copyright 2022 The Knative Authors
+ *
+ * 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 featuressteps
+
+import (
+	"context"
+
+	"knative.dev/eventing/pkg/utils"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/feature"
+)
+
+func CopySecretInTestNamespace(namespace, name string) feature.StepFn {
+	return func(ctx context.Context, t feature.T) {
+		tgtNs := environment.FromContext(ctx).Namespace()
+		_, err := utils.CopySecret(
+			kubeclient.Get(ctx).CoreV1(),
+			namespace,
+			name,
+			tgtNs,
+			"default",
+		)
+		if err != nil {
+			t.Fatalf("failed to copy secret %s from %s to %s: %v", name, namespace, tgtNs, err)
+		}
+	}
+}
diff --git a/test/e2e_new/features/kafka_source_create_secrets_after.go b/test/e2e_new/features/kafka_source_create_secrets_after.go
new file mode 100644
index 000000000..8b956b2d8
--- /dev/null
+++ b/test/e2e_new/features/kafka_source_create_secrets_after.go
@@ -0,0 +1,64 @@
+/*
+ * Copyright 2022 The Knative Authors
+ *
+ * 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 features
+
+import (
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/system"
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/k8s"
+	"knative.dev/reconciler-test/resources/svc"
+
+	"knative.dev/eventing-kafka/test/rekt/resources/kafkasource"
+	"knative.dev/eventing-kafka/test/rekt/resources/kafkatopic"
+
+	"knative.dev/eventing-kafka-broker/test/e2e_new/features/featuressteps"
+	testingpkg "knative.dev/eventing-kafka-broker/test/pkg"
+)
+
+func CreateSecretsAfterKafkaSource() *feature.Feature {
+	f := feature.NewFeatureNamed("Create secrets after KafkaSource")
+
+	topicName := feature.MakeRandomK8sName("topic") // A k8s name is also a valid topic name.
+	name := feature.MakeRandomK8sName("source")
+	sink := feature.MakeRandomK8sName("sink")
+
+	tlsSecretName := "strimzi-tls-secret"
+	saslSecretName := "strimzi-sasl-secret"
+
+	f.Setup("install kafka topic", kafkatopic.Install(topicName))
+	f.Setup("install a service", svc.Install(sink, "app", "rekt"))
+	f.Setup("install a KafkaSource", kafkasource.Install(name,
+		kafkasource.WithSink(&duckv1.KReference{Kind: "Service", Name: sink, APIVersion: "v1"}, ""),
+		kafkasource.WithBootstrapServers(testingpkg.BootstrapServersSslSaslScramArr),
+		kafkasource.WithTopics([]string{topicName}),
+		kafkasource.WithSASLEnabled(),
+		kafkasource.WithSASLUser(saslSecretName, "user"),
+		kafkasource.WithSASLPassword(saslSecretName, "password"),
+		kafkasource.WithSASLType(saslSecretName, "saslType"),
+		kafkasource.WithTLSEnabled(),
+		kafkasource.WithTLSCACert(tlsSecretName, "ca.crt"),
+	))
+	f.Setup("KafkaSource is not ready", k8s.IsNotReady(kafkasource.GVR(), name))
+
+	f.Requirement("Create TLS secret", featuressteps.CopySecretInTestNamespace(system.Namespace(), tlsSecretName))
+	f.Requirement("Create SASL secret", featuressteps.CopySecretInTestNamespace(system.Namespace(), saslSecretName))
+
+	f.Assert("KafkaSource is ready", kafkasource.IsReady(name))
+
+	return f
+}
diff --git a/test/e2e_new/kafka_source_test.go b/test/e2e_new/kafka_source_test.go
new file mode 100644
index 000000000..bc5dc7f20
--- /dev/null
+++ b/test/e2e_new/kafka_source_test.go
@@ -0,0 +1,46 @@
+//go:build e2e
+// +build e2e
+
+/*
+ * Copyright 2022 The Knative Authors
+ *
+ * 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 e2e_new
+
+import (
+	"testing"
+
+	"knative.dev/pkg/system"
+	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/k8s"
+	"knative.dev/reconciler-test/pkg/knative"
+
+	"knative.dev/eventing-kafka-broker/test/e2e_new/features"
+)
+
+func TestKafkaSourceCreateSecretsAfterKafkaSource(t *testing.T) {
+
+	t.Parallel()
+
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+		environment.Managed(t),
+	)
+
+	env.Test(ctx, t, features.CreateSecretsAfterKafkaSource())
+}
diff --git a/test/pkg/broker/broker.go b/test/pkg/broker/broker.go
index 6472d9d90..f9e12e892 100644
--- a/test/pkg/broker/broker.go
+++ b/test/pkg/broker/broker.go
@@ -29,7 +29,7 @@ import (
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
-	testingpkg "knative.dev/eventing-kafka-broker/test/pkg/testing"
+	testingpkg "knative.dev/eventing-kafka-broker/test/pkg"
 )
 
 func Creator(client *eventingtestlib.Client, version string) string {
diff --git a/test/pkg/testing/run.go b/test/pkg/run.go
similarity index 99%
rename from test/pkg/testing/run.go
rename to test/pkg/run.go
index 9335e3d7b..13b9553d6 100644
--- a/test/pkg/testing/run.go
+++ b/test/pkg/run.go
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package testing
+package pkg
 
 import (
 	"fmt"
diff --git a/test/pkg/sink/verify.go b/test/pkg/sink/verify.go
index 3489b19f4..352265a2f 100644
--- a/test/pkg/sink/verify.go
+++ b/test/pkg/sink/verify.go
@@ -25,8 +25,8 @@ import (
 	"k8s.io/apiserver/pkg/storage/names"
 	testlib "knative.dev/eventing/test/lib"
 
+	testingpkg "knative.dev/eventing-kafka-broker/test/pkg"
 	kafkatest "knative.dev/eventing-kafka-broker/test/pkg/kafka"
-	testingpkg "knative.dev/eventing-kafka-broker/test/pkg/testing"
 )
 
 func Verify(t *testing.T, client *testlib.Client, mode, topic string, ids []string) {
diff --git a/test/pkg/testing/auth.go b/test/pkg/testing/auth.go
index d8876b4c5..129d6e4cd 100644
--- a/test/pkg/testing/auth.go
+++ b/test/pkg/testing/auth.go
@@ -23,6 +23,8 @@ import (
 	"github.com/stretchr/testify/assert"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	testlib "knative.dev/eventing/test/lib"
+
+	"knative.dev/eventing-kafka-broker/test/pkg"
 )
 
 type SecretProvider func(t *testing.T, client *testlib.Client) map[string][]byte
@@ -34,10 +36,10 @@ func Plaintext(t *testing.T, _ *testlib.Client) map[string][]byte {
 }
 
 func Ssl(t *testing.T, client *testlib.Client) map[string][]byte {
-	caSecret, err := client.Kube.CoreV1().Secrets(KafkaClusterNamespace).Get(context.Background(), CaSecretName, metav1.GetOptions{})
+	caSecret, err := client.Kube.CoreV1().Secrets(pkg.KafkaClusterNamespace).Get(context.Background(), pkg.CaSecretName, metav1.GetOptions{})
 	assert.Nil(t, err)
 
-	tlsUserSecret, err := client.Kube.CoreV1().Secrets(KafkaClusterNamespace).Get(context.Background(), TlsUserSecretName, metav1.GetOptions{})
+	tlsUserSecret, err := client.Kube.CoreV1().Secrets(pkg.KafkaClusterNamespace).Get(context.Background(), pkg.TlsUserSecretName, metav1.GetOptions{})
 	assert.Nil(t, err)
 
 	return map[string][]byte{
@@ -50,29 +52,29 @@ func Ssl(t *testing.T, client *testlib.Client) map[string][]byte {
 
 func SaslPlaintextScram512(t *testing.T, client *testlib.Client) map[string][]byte {
 
-	saslUserSecret, err := client.Kube.CoreV1().Secrets(KafkaClusterNamespace).Get(context.Background(), SaslUserSecretName, metav1.GetOptions{})
+	saslUserSecret, err := client.Kube.CoreV1().Secrets(pkg.KafkaClusterNamespace).Get(context.Background(), pkg.SaslUserSecretName, metav1.GetOptions{})
 	assert.Nil(t, err)
 
 	return map[string][]byte{
 		"protocol":       []byte("SASL_PLAINTEXT"),
 		"sasl.mechanism": []byte("SCRAM-SHA-512"),
-		"user":           []byte(SaslUserSecretName),
+		"user":           []byte(pkg.SaslUserSecretName),
 		"password":       saslUserSecret.Data["password"],
 	}
 }
 
 func SslSaslScram512(t *testing.T, client *testlib.Client) map[string][]byte {
-	caSecret, err := client.Kube.CoreV1().Secrets(KafkaClusterNamespace).Get(context.Background(), CaSecretName, metav1.GetOptions{})
+	caSecret, err := client.Kube.CoreV1().Secrets(pkg.KafkaClusterNamespace).Get(context.Background(), pkg.CaSecretName, metav1.GetOptions{})
 	assert.Nil(t, err)
 
-	saslUserSecret, err := client.Kube.CoreV1().Secrets(KafkaClusterNamespace).Get(context.Background(), SaslUserSecretName, metav1.GetOptions{})
+	saslUserSecret, err := client.Kube.CoreV1().Secrets(pkg.KafkaClusterNamespace).Get(context.Background(), pkg.SaslUserSecretName, metav1.GetOptions{})
 	assert.Nil(t, err)
 
 	return map[string][]byte{
 		"protocol":       []byte("SASL_SSL"),
 		"sasl.mechanism": []byte("SCRAM-SHA-512"),
 		"ca.crt":         caSecret.Data["ca.crt"],
-		"user":           []byte(SaslUserSecretName),
+		"user":           []byte(pkg.SaslUserSecretName),
 		"password":       saslUserSecret.Data["password"],
 	}
 }
diff --git a/test/test_images/consumer-group-lag-provider-test/admin.go b/test/test_images/consumer-group-lag-provider-test/admin.go
index 428c8acf6..9b24e0cd2 100644
--- a/test/test_images/consumer-group-lag-provider-test/admin.go
+++ b/test/test_images/consumer-group-lag-provider-test/admin.go
@@ -25,7 +25,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
-	testingpkg "knative.dev/eventing-kafka-broker/test/pkg/testing"
+	testingpkg "knative.dev/eventing-kafka-broker/test/pkg"
 )
 
 func main() {
diff --git a/test/upgrade/continual/broker.go b/test/upgrade/continual/broker.go
index 7776a9b0f..46d27f0b1 100644
--- a/test/upgrade/continual/broker.go
+++ b/test/upgrade/continual/broker.go
@@ -36,7 +36,7 @@ import (
 	eventingkafkaupgrade "knative.dev/eventing-kafka/test/upgrade/continual"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
-	testingpkg "knative.dev/eventing-kafka-broker/test/pkg/testing"
+	testingpkg "knative.dev/eventing-kafka-broker/test/pkg"
 )
 
 const (
diff --git a/test/upgrade/continual/sink_source.go b/test/upgrade/continual/sink_source.go
index a78de876a..81aa57554 100644
--- a/test/upgrade/continual/sink_source.go
+++ b/test/upgrade/continual/sink_source.go
@@ -35,8 +35,8 @@ import (
 
 	eventing "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
 	eventingv1alpha1clientset "knative.dev/eventing-kafka-broker/control-plane/pkg/client/clientset/versioned/typed/eventing/v1alpha1"
+	testingpkg "knative.dev/eventing-kafka-broker/test/pkg"
 	"knative.dev/eventing-kafka-broker/test/pkg/sink"
-	testingpkg "knative.dev/eventing-kafka-broker/test/pkg/testing"
 )
 
 const (
diff --git a/test/upgrade/smoke.go b/test/upgrade/smoke.go
index cfb86efc6..85a5bb6a7 100644
--- a/test/upgrade/smoke.go
+++ b/test/upgrade/smoke.go
@@ -22,8 +22,8 @@ import (
 
 	"knative.dev/eventing/test/e2e/helpers"
 
+	pkgtesting "knative.dev/eventing-kafka-broker/test/pkg"
 	testbroker "knative.dev/eventing-kafka-broker/test/pkg/broker"
-	pkgtesting "knative.dev/eventing-kafka-broker/test/pkg/testing"
 )
 
 func runBrokerSmokeTest(t *testing.T) {
diff --git a/vendor/knative.dev/eventing-kafka/test/rekt/resources/kafkasource/kafkasource.go b/vendor/knative.dev/eventing-kafka/test/rekt/resources/kafkasource/kafkasource.go
new file mode 100644
index 000000000..1f9173995
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/test/rekt/resources/kafkasource/kafkasource.go
@@ -0,0 +1,240 @@
+/*
+Copyright 2021 The Knative Authors
+
+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 kafkasource
+
+import (
+	"context"
+	"embed"
+	"time"
+
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/k8s"
+	"knative.dev/reconciler-test/pkg/manifest"
+)
+
+//go:embed *.yaml
+var yaml embed.FS
+
+func GVR() schema.GroupVersionResource {
+	return schema.GroupVersionResource{Group: "sources.knative.dev", Version: "v1beta1", Resource: "kafkasources"}
+}
+
+// Install will create a KafkaSource resource, using the latest version, augmented with the config fn options.
+func Install(name string, opts ...manifest.CfgFn) feature.StepFn {
+	cfg := map[string]interface{}{
+		"name":    name,
+		"version": GVR().Version,
+	}
+	for _, fn := range opts {
+		fn(cfg)
+	}
+	return func(ctx context.Context, t feature.T) {
+		if _, err := manifest.InstallLocalYaml(ctx, cfg); err != nil {
+			t.Fatal(err, cfg)
+		}
+	}
+}
+
+// IsReady tests to see if a KafkaSource becomes ready within the time given.
+func IsReady(name string, timings ...time.Duration) feature.StepFn {
+	return k8s.IsReady(GVR(), name, timings...)
+}
+
+// WithVersion overrides the default API version
+func WithVersion(version string) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if version != "" {
+			cfg["version"] = version
+		}
+	}
+}
+
+// WithAnnotations adds annotation to a KafkaSource metadata.
+func WithAnnotations(annotations map[string]string) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if annotations != nil {
+			cfg["annotations"] = annotations
+		}
+	}
+}
+
+// WithBootstrapServers adds the bootstrapServers config to a KafkaSource spec.
+func WithBootstrapServers(bootstrapServers []string) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if bootstrapServers != nil {
+			cfg["bootstrapServers"] = bootstrapServers
+		}
+	}
+}
+
+// WithTopics adds the topics config to a KafkaSource spec.
+func WithTopics(topics []string) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if topics != nil {
+			cfg["topics"] = topics
+		}
+	}
+}
+
+// WithTLSEnabled enables TLS to a KafkaSource spec.
+func WithTLSEnabled() manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if _, ok := cfg["tls"]; !ok {
+			cfg["tls"] = map[string]interface{}{}
+		}
+	}
+}
+
+// WithTLSCert adds the TLS cert config to a KafkaSource spec.
+func WithTLSCert(name, key string) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if _, ok := cfg["tls"]; !ok {
+			cfg["tls"] = map[string]interface{}{}
+		}
+		tls := cfg["tls"].(map[string]interface{})
+		if _, ok := tls["cert"]; !ok {
+			tls["cert"] = map[string]interface{}{}
+		}
+		cert := tls["cert"].(map[string]interface{})
+		cert["name"] = name
+		cert["key"] = key
+	}
+}
+
+// WithTLSKey adds the TLS key config to a KafkaSource spec.
+func WithTLSKey(name, key string) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if _, ok := cfg["tls"]; !ok {
+			cfg["tls"] = map[string]interface{}{}
+		}
+		tls := cfg["tls"].(map[string]interface{})
+		if _, ok := tls["key"]; !ok {
+			tls["key"] = map[string]interface{}{}
+		}
+		cert := tls["key"].(map[string]interface{})
+		cert["name"] = name
+		cert["key"] = key
+	}
+}
+
+// WithTLSCACert adds the TLS caCert config to a KafkaSource spec.
+func WithTLSCACert(name, key string) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if _, ok := cfg["tls"]; !ok {
+			cfg["tls"] = map[string]interface{}{}
+		}
+		tls := cfg["tls"].(map[string]interface{})
+		if _, ok := tls["caCert"]; !ok {
+			tls["caCert"] = map[string]interface{}{}
+		}
+		cert := tls["caCert"].(map[string]interface{})
+		cert["name"] = name
+		cert["key"] = key
+	}
+}
+
+// WithSASLEnabled enables SASL to a KafkaSource spec.
+func WithSASLEnabled() manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if _, ok := cfg["sasl"]; !ok {
+			cfg["sasl"] = map[string]interface{}{}
+		}
+	}
+}
+
+// WithSASLUser adds the SASL user config to a KafkaSource spec.
+func WithSASLUser(name, key string) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if _, ok := cfg["sasl"]; !ok {
+			cfg["sasl"] = map[string]interface{}{}
+		}
+		sasl := cfg["sasl"].(map[string]interface{})
+		if _, ok := sasl["user"]; !ok {
+			sasl["user"] = map[string]interface{}{}
+		}
+		user := sasl["user"].(map[string]interface{})
+		user["name"] = name
+		user["key"] = key
+	}
+}
+
+// WithSASLPassword adds the SASL password config to a KafkaSource spec.
+func WithSASLPassword(name, key string) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if _, ok := cfg["sasl"]; !ok {
+			cfg["sasl"] = map[string]interface{}{}
+		}
+		sasl := cfg["sasl"].(map[string]interface{})
+		if _, ok := sasl["password"]; !ok {
+			sasl["password"] = map[string]interface{}{}
+		}
+		password := sasl["password"].(map[string]interface{})
+		password["name"] = name
+		password["key"] = key
+	}
+}
+
+// WithSASLType adds the SASL type config to a KafkaSource spec.
+func WithSASLType(name, key string) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if _, ok := cfg["sasl"]; !ok {
+			cfg["sasl"] = map[string]interface{}{}
+		}
+		sasl := cfg["sasl"].(map[string]interface{})
+		if _, ok := sasl["type"]; !ok {
+			sasl["type"] = map[string]interface{}{}
+		}
+		t := sasl["type"].(map[string]interface{})
+		t["name"] = name
+		t["key"] = key
+	}
+}
+
+// WithSink adds the sink related config to a KafkaSource spec.
+func WithSink(ref *duckv1.KReference, uri string) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if _, set := cfg["sink"]; !set {
+			cfg["sink"] = map[string]interface{}{}
+		}
+		sink := cfg["sink"].(map[string]interface{})
+
+		if uri != "" {
+			sink["uri"] = uri
+		}
+		if ref != nil {
+			if _, set := sink["ref"]; !set {
+				sink["ref"] = map[string]interface{}{}
+			}
+			sref := sink["ref"].(map[string]interface{})
+			sref["apiVersion"] = ref.APIVersion
+			sref["kind"] = ref.Kind
+			// skip namespace
+			sref["name"] = ref.Name
+		}
+	}
+}
+
+// WithExtensions set ceoverrides.extensions to a KafkaSource spec.
+func WithExtensions(extensions map[string]string) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if extensions != nil {
+			cfg["extensions"] = extensions
+		}
+	}
+}
diff --git a/vendor/knative.dev/eventing-kafka/test/rekt/resources/kafkasource/kafkasource.yaml b/vendor/knative.dev/eventing-kafka/test/rekt/resources/kafkasource/kafkasource.yaml
new file mode 100644
index 000000000..2c8673cfd
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/test/rekt/resources/kafkasource/kafkasource.yaml
@@ -0,0 +1,108 @@
+# Copyright 2021 The Knative Authors
+#
+# 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.
+
+apiVersion: sources.knative.dev/{{ .version }}
+kind: KafkaSource
+metadata:
+  name: {{ .name }}
+  namespace: {{ .namespace }}
+  {{ if .annotations }}
+  annotations:
+    {{ range $key, $value := .annotations }}
+    {{ $key }}: "{{ $value }}"
+    {{ end }}
+  {{ end }}
+spec:
+  {{ if .bootstrapServers }}
+  bootstrapServers:
+    {{ range $value := .bootstrapServers }}
+    - "{{ $value }}"
+    {{ end }}
+  {{ end }}
+  {{ if .consumerGroup }}
+  consumerGroup: {{ .consumerGroup }}
+  {{ end }}
+  {{ if .topics }}
+  topics:
+    {{ range $value := .topics }}
+    - "{{ $value }}"
+    {{ end }}
+  {{ end }}
+  {{ if or .tls .sasl }}
+  net:
+    {{ if .tls }}
+    tls:
+      enable: true
+      {{ if .tls.cert }}
+      cert:
+        secretKeyRef:
+          name: "{{ .tls.cert.name }}"
+          key: "{{ .tls.cert.key }}"
+      {{ end }}
+      {{ if .tls.key }}
+      key:
+        secretKeyRef:
+          name: "{{ .tls.key.name }}"
+          key: "{{ .tls.key.key }}"
+      {{ end }}
+      {{ if .tls.caCert }}
+      caCert:
+        secretKeyRef:
+          name: "{{ .tls.caCert.name }}"
+          key: "{{ .tls.caCert.key }}"
+      {{ end }}
+    {{ end }}
+    {{ if .sasl }}
+    sasl:
+      enable: true
+      {{ if .sasl.user }}
+      user:
+        secretKeyRef:
+          name: "{{ .sasl.user.name }}"
+          key: "{{ .sasl.user.key }}"
+      {{ end }}
+      {{ if .sasl.password }}
+      password:
+        secretKeyRef:
+          name: "{{ .sasl.password.name }}"
+          key: "{{ .sasl.password.key }}"
+      {{ end }}
+      {{ if .sasl.type }}
+      type:
+        secretKeyRef:
+          name: "{{ .sasl.type.name }}"
+          key: "{{ .sasl.type.key }}"
+      {{ end }}
+    {{ end }}
+  {{ end }}
+  {{ if .extensions }}
+  ceOverrides:
+    extensions:
+      {{ range $key, $value := .extensions }}
+      {{ $key }}: "{{ $value }}"
+      {{ end }}
+  {{ end }}
+  {{ if .sink }}
+  sink:
+    {{ if .sink.ref }}
+    ref:
+      kind: {{ .sink.ref.kind }}
+      namespace: {{ .namespace }}
+      name: {{ .sink.ref.name }}
+      apiVersion: {{ .sink.ref.apiVersion }}
+    {{ end }}
+    {{ if .sink.uri }}
+    uri: {{ .sink.uri }}
+    {{ end }}
+  {{ end }}
diff --git a/vendor/knative.dev/eventing-kafka/test/rekt/resources/kafkatopic/topic.go b/vendor/knative.dev/eventing-kafka/test/rekt/resources/kafkatopic/topic.go
new file mode 100644
index 000000000..3f732bf5a
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/test/rekt/resources/kafkatopic/topic.go
@@ -0,0 +1,91 @@
+/*
+Copyright 2021 The Knative Authors
+
+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 kafkatopic
+
+import (
+	"context"
+	"embed"
+	"time"
+
+	"k8s.io/apimachinery/pkg/runtime/schema"
+
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/k8s"
+	"knative.dev/reconciler-test/pkg/manifest"
+)
+
+const (
+	// kafkaNamespace is the namespace where kafka is installed
+	kafkaNamespace = "kafka"
+)
+
+//go:embed *.yaml
+var yaml embed.FS
+
+func GVR() schema.GroupVersionResource {
+	return schema.GroupVersionResource{Group: "kafka.strimzi.io", Version: "v1beta2", Resource: "kafkatopics"}
+}
+
+// Install will create a Kafka Topic via the Strimzi topic CRD,, augmented with the config fn options.
+func Install(name string, opts ...manifest.CfgFn) feature.StepFn {
+	cfg := map[string]interface{}{
+		"name":             name,
+		"clusterNamespace": "kafka",
+		"partitions":       10,
+		"clusterName":      "my-cluster",
+	}
+	for _, fn := range opts {
+		fn(cfg)
+	}
+
+	return func(ctx context.Context, t feature.T) {
+		if _, err := manifest.InstallLocalYaml(ctx, cfg); err != nil {
+			t.Fatal(err, cfg)
+		}
+	}
+}
+
+// IsReady tests to see if a KafkaTopic becomes ready within the time given.
+func IsReady(name string, timings ...time.Duration) feature.StepFn {
+	return func(ctx context.Context, t feature.T) {
+		interval, timeout := k8s.PollTimings(ctx, timings)
+		if err := k8s.WaitForResourceReady(ctx, t, kafkaNamespace, name, GVR(), interval, timeout); err != nil {
+			t.Error(GVR(), "did not become ready,", err)
+		}
+	}
+}
+
+// WithPartitions overrides the number of partitions (default: 10).
+func WithPartitions(partitions string) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		cfg["partitions"] = partitions
+	}
+}
+
+// WithClusterName overrides the Kafka cluster names where to create the topic (default: my-cluster)
+func WithClusterName(name string) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		cfg["clusterName"] = name
+	}
+}
+
+// WithClusterNamespace overrides the Kafka cluster namespace where to create the topic (default: kafka)
+func WithClusterNamespace(namespace string) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		cfg["clusterNamespace"] = namespace
+	}
+}
diff --git a/vendor/knative.dev/eventing-kafka/test/rekt/resources/kafkatopic/topic.yaml b/vendor/knative.dev/eventing-kafka/test/rekt/resources/kafkatopic/topic.yaml
new file mode 100644
index 000000000..1d6e2e543
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/test/rekt/resources/kafkatopic/topic.yaml
@@ -0,0 +1,24 @@
+# Copyright 2021 The Knative Authors
+#
+# 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.
+
+apiVersion: kafka.strimzi.io/v1beta2
+kind: KafkaTopic
+metadata:
+  name: {{ .name }}
+  namespace: {{ .clusterNamespace }}
+  labels:
+    strimzi.io/cluster: {{ .clusterName }}
+spec:
+  partitions: {{ .partitions }}
+  replicas: 1