Repository: knative-sandbox/eventing-kafka-broker
Pull request #2484 (2022-08-08 12:47:21)

Retry webhook EOF errors
This test sometimes fails due to [1].

[1] https://github.com/knative/pkg/issues/1509

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

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/2484#" 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/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":[]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2484?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2484](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2484?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (d204560) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/c7e7f5ecd5f490b1d7e535024af348974ddf77ac?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (c7e7f5e) will **increase** coverage by `16.59%`.
> The diff coverage is `n/a`.

```diff
@@              Coverage Diff              @@
##               main    #2484       +/-   ##
=============================================
+ Coverage     64.62%   81.22%   +16.59%     
+ Complexity      711      710        -1     
=============================================
  Files           146       72       -74     
  Lines          9807     2466     -7341     
  Branches        223      223               
=============================================
- Hits           6338     2003     -4335     
+ Misses         3037      338     -2699     
+ Partials        432      125      -307     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `81.22% <ø> (-0.21%)` | :arrow_down: |

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/2484?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...ive/eventing/kafka/broker/core/AsyncCloseable.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2484/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvQXN5bmNDbG9zZWFibGUuamF2YQ==) | `66.66% <0.00%> (-13.34%)` | :arrow_down: |
| [...dispatcher/impl/consumer/BaseConsumerVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2484/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9CYXNlQ29uc3VtZXJWZXJ0aWNsZS5qYXZh) | `87.09% <0.00%> (-6.46%)` | :arrow_down: |
| [...broker/dispatcher/impl/consumer/OffsetManager.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2484/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9PZmZzZXRNYW5hZ2VyLmphdmE=) | `94.62% <0.00%> (-1.08%)` | :arrow_down: |
| [...ntrol-plane/pkg/reconciler/consumergroup/filter.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2484/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2ZpbHRlci5nbw==) | | |
| [...ol-plane/pkg/apis/eventing/v1/broker\_validation.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2484/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MS9icm9rZXJfdmFsaWRhdGlvbi5nbw==) | | |
| [control-plane/pkg/apis/config/features.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2484/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9jb25maWcvZmVhdHVyZXMuZ28=) | | |
| [...ls/kafka/eventing/v1alpha1/consumer\_group\_types.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2484/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9pbnRlcm5hbHMva2Fma2EvZXZlbnRpbmcvdjFhbHBoYTEvY29uc3VtZXJfZ3JvdXBfdHlwZXMuZ28=) | | |
| [control-plane/pkg/apis/core/pod\_defaulting.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2484/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9jb3JlL3BvZF9kZWZhdWx0aW5nLmdv) | | |
| [control-plane/pkg/reconciler/sink/kafka\_sink.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2484/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zaW5rL2thZmthX3NpbmsuZ28=) | | |
| [control-plane/pkg/reconciler/trigger/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2484/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci90cmlnZ2VyL2NvbnRyb2xsZXIuZ28=) | | |
| ... and [67 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2484/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]: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2484#pullrequestreview-1065231974" title="Approved">matzew</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2484#" 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/main/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 bc521594b..7c791b71a 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
 	}
 
@@ -413,3 +409,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/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/OrderedAsyncExecutor.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/OrderedAsyncExecutor.java
index 95d7426cf..c8e060bb7 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/OrderedAsyncExecutor.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/OrderedAsyncExecutor.java
@@ -26,6 +26,7 @@ import io.vertx.kafka.client.common.TopicPartition;
 
 import java.util.ArrayDeque;
 import java.util.Queue;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Supplier;
 
 /**
@@ -37,8 +38,8 @@ public class OrderedAsyncExecutor {
 
   private final Queue<Task> queue;
 
-  private boolean isStopped;
-  private boolean inFlight;
+  private final AtomicBoolean isStopped;
+  private final AtomicBoolean inFlight;
   private final MeterRegistry meterRegistry;
   private final DistributionSummary executorLatency;
   private final Gauge executorQueueLength;
@@ -49,8 +50,8 @@ public class OrderedAsyncExecutor {
                               final DataPlaneContract.Egress egress) {
     this.meterRegistry = meterRegistry;
     this.queue = new ArrayDeque<>();
-    this.isStopped = false;
-    this.inFlight = false;
+    this.isStopped = new AtomicBoolean(false);
+    this.inFlight = new AtomicBoolean(false);
     this.egress = egress;
 
     if (meterRegistry != null && egress != null && egress.getFeatureFlags().getEnableOrderedExecutorMetrics()) {
@@ -78,7 +79,7 @@ public class OrderedAsyncExecutor {
    * @param task the task to offer
    */
   public void offer(Supplier<Future<?>> task) {
-    if (this.isStopped) {
+    if (this.isStopped.get()) {
       // Executor is stopped, return without adding the task to the queue.
       return;
     }
@@ -93,22 +94,23 @@ public class OrderedAsyncExecutor {
   }
 
   private void consume() {
-    if (queue.isEmpty() || this.inFlight || this.isStopped) {
+    if (queue.isEmpty() || this.inFlight.get() || this.isStopped.get()) {
       return;
     }
-    this.inFlight = true;
+    if (this.inFlight.compareAndSet(false, true)) {
+      final var task = this.queue.remove();
+      task.task
+        .get()
+        .onComplete(ar -> {
+          this.inFlight.set(false);
+          if (egress != null && egress.getFeatureFlags().getEnableOrderedExecutorMetrics() && !this.isStopped.get()) {
+            this.executorLatency.record(System.currentTimeMillis() - task.queueTimestamp);
+            this.executorQueueLength.value();
+          }
+          consume();
+        });
+    }
 
-    final var task = this.queue.remove();
-    task.task
-      .get()
-      .onComplete(ar -> {
-        this.inFlight = false;
-        if (egress != null && egress.getFeatureFlags().getEnableOrderedExecutorMetrics() && !this.isStopped) {
-          this.executorLatency.record(System.currentTimeMillis() - task.queueTimestamp);
-          this.executorQueueLength.value();
-        }
-        consume();
-      });
   }
 
   public boolean isWaitingForTasks() {
@@ -121,7 +123,7 @@ public class OrderedAsyncExecutor {
    * Stop the executor. This won't stop the actual in-flight task, but it will prevent queued tasks to be executed.
    */
   public void stop() {
-    this.isStopped = true;
+    this.isStopped.set(true);
     this.queue.clear();
     if (meterRegistry != null && egress != null && egress.getFeatureFlags().getEnableOrderedExecutorMetrics()) {
       this.meterRegistry.remove(executorLatency);
diff --git a/hack/run.sh b/hack/run.sh
index e25bc1c36..d703cfea4 100755
--- a/hack/run.sh
+++ b/hack/run.sh
@@ -64,7 +64,9 @@ elif [[ "${action}" == "unit-test-control-plane" || "${action}" == "unit-tests-c
 elif [[ "${action}" == "build-test" || "${action}" == "build-tests" ]]; then
   "${ROOT_DIR}"/test/presubmit-tests.sh --build-tests
 elif [[ "${action}" == "deploy-sacura" ]]; then
-  source "${ROOT_DIR}"/test/e2e-common.sh && apply_sacura
+  source "${ROOT_DIR}"/test/e2e-common.sh &&
+    apply_sacura &&
+    apply_sacura_sink_source
 elif [[ "${action}" == "sacura-test" || "${action}" == "sacura-tests" ]]; then
   source "${ROOT_DIR}"/test/e2e-common.sh && go_test_e2e -tags=sacura -timeout=40m ./test/e2e/...
 elif [[ "${action}" == "teardown-sacura" ]]; then
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 1c9b175b3..e51fec1b9 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1261,6 +1261,7 @@ knative.dev/eventing/test
 knative.dev/eventing/test/conformance/helpers
 knative.dev/eventing/test/conformance/helpers/tracing
 knative.dev/eventing/test/e2e/helpers
+knative.dev/eventing/test/experimental/features/delivery_timeout
 knative.dev/eventing/test/experimental/features/new_trigger_filters
 knative.dev/eventing/test/flags
 knative.dev/eventing/test/lib
@@ -1384,6 +1385,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/config/100-config-experimental-features.yaml b/test/config/100-config-experimental-features.yaml
new file mode 120000
index 000000000..bfbbc3e52
--- /dev/null
+++ b/test/config/100-config-experimental-features.yaml
@@ -0,0 +1,1 @@
+../experimental/features_config/features.yaml
\ No newline at end of file
diff --git a/test/config/sacura-sink-source/300-job.yaml b/test/config/sacura-sink-source/300-job.yaml
new file mode 100644
index 000000000..2cefa2d68
--- /dev/null
+++ b/test/config/sacura-sink-source/300-job.yaml
@@ -0,0 +1,72 @@
+# 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
+#
+#     https://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: batch/v1
+kind: Job
+metadata:
+  name: sacura
+  namespace: sacura-sink-source
+  labels:
+    app: sacura
+spec:
+  backoffLimit: 0
+  completions: 1
+  parallelism: 1
+  template:
+    metadata:
+      labels:
+        app: sacura
+    spec:
+
+      securityContext:
+        runAsNonRoot: true
+
+      # To avoid scheduling the Job on the same node of the receiver
+      affinity:
+        podAntiAffinity:
+          preferredDuringSchedulingIgnoredDuringExecution:
+            - podAffinityTerm:
+                labelSelector:
+                  matchLabels:
+                    app: kafka-sink-receiver
+                topologyKey: kubernetes.io/hostname
+              weight: 100
+
+      containers:
+        - name: sacura
+          image: ghcr.io/pierdipi/sacura/sacura-7befbbbc92911c6727467cfbf23af88f
+          args:
+            - "--config"
+            - "/etc/sacura/sacura.yaml"
+          securityContext:
+            privileged: false
+            readOnlyRootFilesystem: true
+            runAsNonRoot: true
+            allowPrivilegeEscalation: false
+          volumeMounts:
+            - mountPath: /etc/sacura
+              name: config
+          resources:
+            requests:
+              cpu: 1000m
+              memory: 4Gi
+          ports:
+            - containerPort: 8080
+              protocol: TCP
+              name: receiver
+      volumes:
+        - name: config
+          configMap:
+            name: config-sacura
+      restartPolicy: Never
diff --git a/test/config/sacura-sink-source/README.md b/test/config/sacura-sink-source/README.md
new file mode 100644
index 000000000..eaffc1e99
--- /dev/null
+++ b/test/config/sacura-sink-source/README.md
@@ -0,0 +1,9 @@
+### Test Topology
+
+```
++----------+    +------+     +--------+
+|  Sacura  +--->| Sink +---->| Source +
++----------+    +------+     +--------+
+      ^                          |
+      +---------------------------
+```
diff --git a/test/config/sacura-sink-source/resources/0-kafka-topic.yaml b/test/config/sacura-sink-source/resources/0-kafka-topic.yaml
new file mode 100644
index 000000000..5f3edb6a9
--- /dev/null
+++ b/test/config/sacura-sink-source/resources/0-kafka-topic.yaml
@@ -0,0 +1,27 @@
+# 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
+#
+#     https://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: sacura-sink-source-topic
+  namespace: kafka
+  labels:
+    strimzi.io/cluster: my-cluster
+spec:
+  partitions: 50
+  replicas: 3
+  config:
+    retention.ms: 7200000
+    segment.bytes: 1073741824
diff --git a/test/config/sacura-sink-source/resources/0-namespace.yaml b/test/config/sacura-sink-source/resources/0-namespace.yaml
new file mode 100644
index 000000000..cefe413c8
--- /dev/null
+++ b/test/config/sacura-sink-source/resources/0-namespace.yaml
@@ -0,0 +1,18 @@
+# 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
+#
+#     https://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: v1
+kind: Namespace
+metadata:
+  name: sacura-sink-source
diff --git a/test/config/sacura-sink-source/resources/100-sink.yaml b/test/config/sacura-sink-source/resources/100-sink.yaml
new file mode 100644
index 000000000..c6be08ecc
--- /dev/null
+++ b/test/config/sacura-sink-source/resources/100-sink.yaml
@@ -0,0 +1,23 @@
+# 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
+#
+#     https://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: eventing.knative.dev/v1alpha1
+kind: KafkaSink
+metadata:
+  name: sink
+  namespace: sacura-sink-source
+spec:
+  topic: sacura-sink-source-topic
+  bootstrapServers:
+    - my-cluster-kafka-bootstrap.kafka:9092
diff --git a/test/config/sacura-sink-source/resources/100-source.yaml b/test/config/sacura-sink-source/resources/100-source.yaml
new file mode 100644
index 000000000..cbfdfedf0
--- /dev/null
+++ b/test/config/sacura-sink-source/resources/100-source.yaml
@@ -0,0 +1,31 @@
+# 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
+#
+#     https://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/v1beta1
+kind: KafkaSource
+metadata:
+  name: source
+  namespace: sacura-sink-source
+spec:
+  consumers: 50
+  topics:
+    - sacura-sink-source-topic
+  bootstrapServers:
+    - my-cluster-kafka-bootstrap.kafka:9092
+  consumerGroup: sacura-sink-source-group
+  sink:
+    ref:
+      apiVersion: v1
+      kind: Service
+      name: sacura
diff --git a/test/config/sacura-sink-source/resources/200-config.yaml b/test/config/sacura-sink-source/resources/200-config.yaml
new file mode 100644
index 000000000..eac1988f0
--- /dev/null
+++ b/test/config/sacura-sink-source/resources/200-config.yaml
@@ -0,0 +1,34 @@
+# 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
+#
+#     https://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: v1
+kind: ConfigMap
+metadata:
+  name: config-sacura
+  namespace: sacura-sink-source
+data:
+  sacura.yaml: |
+    sender:
+      target: http://kafka-sink-ingress.knative-eventing.svc.cluster.local/sacura-sink-source/sink
+      frequency: 100
+      workers: 4
+      keepAlive: true
+    receiver:
+      port: 8080
+      timeout: 30m
+      maxDuplicatesPercentage: 0.1
+      fault:
+        minSleepDuration: 4s
+        maxSleepDuration: 5s
+    duration: 3m
diff --git a/test/config/sacura-sink-source/resources/400-service.yaml b/test/config/sacura-sink-source/resources/400-service.yaml
new file mode 100644
index 000000000..5deb6fcf0
--- /dev/null
+++ b/test/config/sacura-sink-source/resources/400-service.yaml
@@ -0,0 +1,27 @@
+# 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
+#
+#     https://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: v1
+kind: Service
+metadata:
+  name: sacura
+  namespace: sacura-sink-source
+spec:
+  selector:
+    app: sacura
+  ports:
+    - port: 80
+      protocol: TCP
+      targetPort: receiver
+      name: http
diff --git a/test/config/sacura/300-job.yaml b/test/config/sacura/300-job.yaml
index b4d787520..e72723bcd 100644
--- a/test/config/sacura/300-job.yaml
+++ b/test/config/sacura/300-job.yaml
@@ -50,8 +50,8 @@ spec:
             - mountPath: /etc/sacura
               name: config
           resources:
-            limits:
-              cpu: 2000m
+            requests:
+              cpu: 1000m
               memory: 4Gi
           ports:
             - containerPort: 8080
diff --git a/test/config/sacura/0-namespace.yaml b/test/config/sacura/resources/0-namespace.yaml
similarity index 100%
rename from test/config/sacura/0-namespace.yaml
rename to test/config/sacura/resources/0-namespace.yaml
diff --git a/test/config/sacura/100-broker-config.yaml b/test/config/sacura/resources/100-broker-config.yaml
similarity index 100%
rename from test/config/sacura/100-broker-config.yaml
rename to test/config/sacura/resources/100-broker-config.yaml
diff --git a/test/config/sacura/101-broker.yaml b/test/config/sacura/resources/101-broker.yaml
similarity index 80%
rename from test/config/sacura/101-broker.yaml
rename to test/config/sacura/resources/101-broker.yaml
index 4f35b32da..8373019af 100644
--- a/test/config/sacura/101-broker.yaml
+++ b/test/config/sacura/resources/101-broker.yaml
@@ -3,6 +3,8 @@ kind: Broker
 metadata:
   name: broker
   namespace: sacura
+  annotations:
+    eventing.knative.dev/broker.class: Kafka
 spec:
 
   config:
diff --git a/test/config/sacura/200-config.yaml b/test/config/sacura/resources/200-config.yaml
similarity index 100%
rename from test/config/sacura/200-config.yaml
rename to test/config/sacura/resources/200-config.yaml
diff --git a/test/config/sacura/300-trigger.yaml b/test/config/sacura/resources/300-trigger.yaml
similarity index 100%
rename from test/config/sacura/300-trigger.yaml
rename to test/config/sacura/resources/300-trigger.yaml
diff --git a/test/config/sacura/400-service.yaml b/test/config/sacura/resources/400-service.yaml
similarity index 100%
rename from test/config/sacura/400-service.yaml
rename to test/config/sacura/resources/400-service.yaml
diff --git a/test/e2e-common.sh b/test/e2e-common.sh
index 7c04d0433..d0d0d1d50 100644
--- a/test/e2e-common.sh
+++ b/test/e2e-common.sh
@@ -258,23 +258,26 @@ function delete_chaos() {
 }
 
 function apply_sacura() {
-  ko apply ${KO_FLAGS} -f ./test/config/sacura/0-namespace.yaml || return $?
-  ko apply ${KO_FLAGS} -f ./test/config/sacura/100-broker-config.yaml || return $?
-  ko apply ${KO_FLAGS} -f ./test/config/sacura/101-broker.yaml || return $?
+  ko apply ${KO_FLAGS} -Rf ./test/config/sacura/resources || return $?
 
-  kubectl wait --for=condition=ready --timeout=3m -n sacura broker/broker || {
-    local ret=$?
-    kubectl get -n sacura broker/broker -oyaml
-    return ${ret}
-  }
+  kubectl wait --for=condition=ready --timeout=3m -n sacura broker/broker || return $?
+  kubectl wait --for=condition=ready --timeout=3m -n sacura trigger/trigger || return $?
 
-  ko apply ${KO_FLAGS} -f ./test/config/sacura || return $?
+  ko apply ${KO_FLAGS} -Rf ./test/config/sacura || return $?
+}
+
+function apply_sacura_sink_source() {
+  ko apply ${KO_FLAGS} -Rf ./test/config/sacura-sink-source/resources || return $?
+
+  kubectl wait --for=condition=ready --timeout=3m -n sacura-sink-source kafkasink/sink || return $?
+  kubectl wait --for=condition=ready --timeout=3m -n sacura-sink-source kafkasource/source || return $?
+
+  ko apply ${KO_FLAGS} -Rf ./test/config/sacura-sink-source || return $?
 }
 
 function delete_sacura() {
-  kubectl delete --ignore-not-found -f ./test/config/sacura/101-broker.yaml || return $?
-  kubectl delete --ignore-not-found -f ./test/config/sacura/100-broker-config.yaml || return $?
-  kubectl delete --ignore-not-found -f ./test/config/sacura/0-namespace.yaml || return $?
+  kubectl delete --ignore-not-found ns sacura || return $?
+  kubectl delete --ignore-not-found ns sacura-sink-source || return $?
 }
 
 function export_logs_continuously() {
diff --git a/test/e2e-tests.sh b/test/e2e-tests.sh
index 1ae285108..0eaab8313 100755
--- a/test/e2e-tests.sh
+++ b/test/e2e-tests.sh
@@ -9,8 +9,9 @@ fi
 
 if ! ${LOCAL_DEVELOPMENT}; then
   scale_controlplane kafka-controller kafka-webhook-eventing eventing-webhook eventing-controller
-  apply_sacura || fail_test "Failed to apply Sacura"
-  apply_chaos || fail_test "Failed to apply chaos"
+  apply_sacura                            || fail_test "Failed to apply Sacura"
+  apply_sacura_sink_source || fail_test "Failed to apply Sacura (Source, Sink, Broker, Channel)"
+  apply_chaos                             || fail_test "Failed to apply chaos"
 fi
 
 header "Waiting Knative eventing to come up"
diff --git a/test/e2e/conformance/data_plane_conformance_test.go b/test/e2e/conformance/data_plane_conformance_test.go
index 481539df2..35140405b 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/sacura_test.go b/test/e2e/sacura_test.go
index d251db478..071d2e3d8 100644
--- a/test/e2e/sacura_test.go
+++ b/test/e2e/sacura_test.go
@@ -30,26 +30,67 @@ import (
 	"github.com/stretchr/testify/require"
 	batchv1 "k8s.io/api/batch/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/apiserver/pkg/storage/names"
+	"k8s.io/client-go/dynamic"
+	"k8s.io/utils/pointer"
+	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	messaging "knative.dev/eventing/pkg/apis/messaging/v1"
+	eventingclientset "knative.dev/eventing/pkg/client/clientset/versioned"
 	testlib "knative.dev/eventing/test/lib"
 
 	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 (
 	app                            = "sacura"
-	namespace                      = "sacura"
 	sacuraVerifyCommittedOffsetJob = "verify-committed-offset"
-	sacuraTriggerName              = "trigger"
-	sacuraTopic                    = "knative-broker-sacura-sacura"
 
-	pollTimeout  = 30 * time.Minute
+	sacuraChannelName      = "channel"
+	sacuraTriggerName      = "trigger"
+	sacuraSubscriptionName = "subscription"
+	sacuraSourceName       = "source"
+
+	pollTimeout  = 40 * time.Minute
 	pollInterval = 10 * time.Second
 )
 
-func TestSacuraJob(t *testing.T) {
+type SacuraTestConfig struct {
+	// Namespace is the test namespace.
+	Namespace string
+
+	// BrokerTopic is the expected Broker topic.
+	// It's used to verify the committed offset.
+	BrokerTopic *string
+
+	// ChannelTopic is the expected Channel topic.
+	// It's used to verify the committed offset.
+	ChannelTopic *string
+
+	// SourceTopic is the Source topic.
+	// It's used to verify the committed offset.
+	SourceTopic *string
+}
+
+func TestSacuraSinkSourceJob(t *testing.T) {
+	runSacuraTest(t, SacuraTestConfig{
+		Namespace:   "sacura-sink-source",
+		SourceTopic: pointer.StringPtr("sacura-sink-source-topic"),
+	})
+}
+
+func TestSacuraBrokerJob(t *testing.T) {
+	runSacuraTest(t, SacuraTestConfig{
+		Namespace:   "sacura",
+		BrokerTopic: pointer.StringPtr("knative-broker-sacura-sink-source-broker"),
+	})
+}
+
+func runSacuraTest(t *testing.T, config SacuraTestConfig) {
 
 	c := testlib.Setup(t, false)
 	defer testlib.TearDown(c)
@@ -57,40 +98,53 @@ func TestSacuraJob(t *testing.T) {
 	ctx := context.Background()
 
 	jobPollError := wait.Poll(pollInterval, pollTimeout, func() (done bool, err error) {
-		job, err := c.Kube.BatchV1().Jobs(namespace).Get(ctx, app, metav1.GetOptions{})
+		job, err := c.Kube.BatchV1().Jobs(config.Namespace).Get(ctx, app, metav1.GetOptions{})
 		assert.Nil(t, err)
 
 		return isJobSucceeded(job)
 	})
 
-	pkgtesting.LogJobOutput(t, ctx, c.Kube, namespace, app)
+	pkgtesting.LogJobOutput(t, ctx, c.Kube, config.Namespace, app)
 
 	if jobPollError != nil {
 		t.Fatal(jobPollError)
 	}
 
-	t.Log(strings.Repeat("-", 30))
-	t.Log("Verify committed offset")
-	t.Log(strings.Repeat("-", 30))
-
-	trigger, err := c.Eventing.EventingV1().Triggers(namespace).Get(ctx, sacuraTriggerName, metav1.GetOptions{})
-	require.Nil(t, err, "Failed to get trigger %s/%s: %v", namespace, sacuraTriggerName)
+	topics := map[*string]func(t *testing.T) string{
+		config.BrokerTopic:  getKafkaTriggerConsumerGroup(ctx, c.Eventing, config.Namespace),
+		config.ChannelTopic: getKafkaSubscriptionConsumerGroup(ctx, c.Dynamic, config.Namespace),
+		config.SourceTopic:  getKafkaSourceConsumerGroup(ctx, c.Dynamic, config.Namespace),
+	}
 
 	t.Run("verify committed offset", func(t *testing.T) {
-		err = kafkatest.VerifyCommittedOffset(
-			c.Kube,
-			c.Tracker,
-			types.NamespacedName{
-				Namespace: namespace,
-				Name:      sacuraVerifyCommittedOffsetJob,
-			},
-			&kafkatest.AdminConfig{
-				BootstrapServers: pkgtesting.BootstrapServersPlaintext,
-				Topic:            sacuraTopic,
-				Group:            string(trigger.UID),
-			},
-		)
-		require.Nil(t, err, "Failed to verify committed offset")
+		for topic, groupFunc := range topics {
+			if topic == nil {
+				continue
+			}
+
+			t.Run(*topic, func(t *testing.T) {
+				t.Log(strings.Repeat("-", 30))
+				t.Log("Verify committed offset")
+				t.Log(strings.Repeat("-", 30))
+
+				consumerGroup := groupFunc(t)
+
+				err := kafkatest.VerifyCommittedOffset(
+					c.Kube,
+					c.Tracker,
+					types.NamespacedName{
+						Namespace: config.Namespace,
+						Name:      names.SimpleNameGenerator.GenerateName(sacuraVerifyCommittedOffsetJob + "-" + *topic),
+					},
+					&kafkatest.AdminConfig{
+						BootstrapServers: pkgtest.BootstrapServersPlaintext,
+						Topic:            *topic,
+						Group:            consumerGroup,
+					},
+				)
+				require.Nil(t, err, "Failed to verify committed offset")
+			})
+		}
 	})
 }
 
@@ -101,3 +155,40 @@ func isJobSucceeded(job *batchv1.Job) (bool, error) {
 
 	return job.Status.Succeeded > 0, nil
 }
+
+func getKafkaTriggerConsumerGroup(ctx context.Context, c *eventingclientset.Clientset, ns string) func(t *testing.T) string {
+	return func(t *testing.T) string {
+		trigger, err := c.EventingV1().Triggers(ns).Get(ctx, sacuraTriggerName, metav1.GetOptions{})
+		require.Nil(t, err, "Failed to get trigger %s/%s: %v", ns, sacuraTriggerName)
+		return string(trigger.UID)
+	}
+}
+
+func getKafkaSourceConsumerGroup(ctx context.Context, c dynamic.Interface, ns string) func(t *testing.T) string {
+	return func(t *testing.T) string {
+		gvr := sources.SchemeGroupVersion.WithResource("kafkasources")
+		ksUnstr, err := c.Resource(gvr).Namespace(ns).Get(ctx, sacuraSourceName, metav1.GetOptions{})
+		require.Nilf(t, err, "%+v %s/%s", gvr, ns, sacuraSourceName)
+
+		ks := sources.KafkaSource{}
+		err = runtime.DefaultUnstructuredConverter.FromUnstructured(ksUnstr.UnstructuredContent(), &ks)
+		require.Nil(t, err)
+		require.NotEmpty(t, ks.Spec.ConsumerGroup)
+
+		return ks.Spec.ConsumerGroup
+	}
+}
+
+func getKafkaSubscriptionConsumerGroup(ctx context.Context, c dynamic.Interface, ns string) func(t *testing.T) string {
+	return func(t *testing.T) string {
+		gvr := messaging.SchemeGroupVersion.WithResource("subscriptions")
+		subUnstr, err := c.Resource(gvr).Namespace(ns).Get(ctx, sacuraSubscriptionName, metav1.GetOptions{})
+		require.Nil(t, err)
+
+		sub := messaging.Subscription{}
+		err = runtime.DefaultUnstructuredConverter.FromUnstructured(subUnstr.UnstructuredContent(), &sub)
+		require.Nil(t, err)
+
+		return fmt.Sprintf("kafka.%s.%s.%s", c, sacuraChannelName, string(sub.UID))
+	}
+}
diff --git a/test/e2e_broker/broker_event_trasformation_test.go b/test/e2e_broker/broker_event_trasformation_test.go
index 9710b185a..2927e54ad 100644
--- a/test/e2e_broker/broker_event_trasformation_test.go
+++ b/test/e2e_broker/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/broker_redelivery_test.go b/test/e2e_broker/broker_redelivery_test.go
index f5d7c0b91..3a73f61f5 100644
--- a/test/e2e_broker/broker_redelivery_test.go
+++ b/test/e2e_broker/broker_redelivery_test.go
@@ -29,8 +29,8 @@ import (
 	testlib "knative.dev/eventing/test/lib"
 	"knative.dev/eventing/test/lib/resources"
 
+	kafkatesting "knative.dev/eventing-kafka-broker/test/pkg"
 	"knative.dev/eventing-kafka-broker/test/pkg/broker"
-	kafkatesting "knative.dev/eventing-kafka-broker/test/pkg/testing"
 )
 
 const (
diff --git a/test/e2e_broker/broker_sasl_ssl_test.go b/test/e2e_broker/broker_sasl_ssl_test.go
index 9ae083951..baa7c9f18 100644
--- a/test/e2e_broker/broker_sasl_ssl_test.go
+++ b/test/e2e_broker/broker_sasl_ssl_test.go
@@ -38,6 +38,7 @@ import (
 	"knative.dev/eventing/test/lib/sender"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
+	. "knative.dev/eventing-kafka-broker/test/pkg"
 	"knative.dev/eventing-kafka-broker/test/pkg/broker"
 	. "knative.dev/eventing-kafka-broker/test/pkg/testing"
 )
diff --git a/test/e2e_broker/broker_trigger_sink_test.go b/test/e2e_broker/broker_trigger_sink_test.go
index 1f2c43db6..b7226ebf4 100644
--- a/test/e2e_broker/broker_trigger_sink_test.go
+++ b/test/e2e_broker/broker_trigger_sink_test.go
@@ -34,10 +34,10 @@ 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/broker"
 	"knative.dev/eventing-kafka-broker/test/pkg/sink"
-	testingpkg "knative.dev/eventing-kafka-broker/test/pkg/testing"
 )
 
 /*
diff --git a/test/e2e_broker/broker_trigger_test.go b/test/e2e_broker/broker_trigger_test.go
index 0e0f2eef4..ec04237af 100644
--- a/test/e2e_broker/broker_trigger_test.go
+++ b/test/e2e_broker/broker_trigger_test.go
@@ -36,9 +36,9 @@ 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"
 	brokertest "knative.dev/eventing-kafka-broker/test/pkg/broker"
 	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_broker/conformance/control_plane_conformance_test.go b/test/e2e_broker/conformance/control_plane_conformance_test.go
index f436a0346..deaed366e 100644
--- a/test/e2e_broker/conformance/control_plane_conformance_test.go
+++ b/test/e2e_broker/conformance/control_plane_conformance_test.go
@@ -27,8 +27,8 @@ import (
 	testlib "knative.dev/eventing/test/lib"
 	"knative.dev/eventing/test/lib/resources"
 
+	pkgtesting "knative.dev/eventing-kafka-broker/test/pkg"
 	"knative.dev/eventing-kafka-broker/test/pkg/broker"
-	pkgtesting "knative.dev/eventing-kafka-broker/test/pkg/testing"
 )
 
 func brokerCreator(client *testlib.Client, name string) {
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/e2e_sink/kafka_sink.go b/test/e2e_sink/kafka_sink.go
index 5c607f68b..ffd4ce156 100644
--- a/test/e2e_sink/kafka_sink.go
+++ b/test/e2e_sink/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_sink/kafka_sink_test.go b/test/e2e_sink/kafka_sink_test.go
index 0284f9090..b48c2cc31 100644
--- a/test/e2e_sink/kafka_sink_test.go
+++ b/test/e2e_sink/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_source/kafka_source_test.go b/test/e2e_source/kafka_source_test.go
index 99f29b8fa..a6ee020f1 100644
--- a/test/e2e_source/kafka_source_test.go
+++ b/test/e2e_source/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/experimental/delivery_timeout_test.go b/test/experimental/delivery_timeout_test.go
new file mode 100644
index 000000000..8bb2ebe00
--- /dev/null
+++ b/test/experimental/delivery_timeout_test.go
@@ -0,0 +1,65 @@
+//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 experimental
+
+import (
+	"testing"
+
+	"knative.dev/pkg/system"
+	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/k8s"
+	"knative.dev/reconciler-test/pkg/knative"
+
+	"knative.dev/eventing/test/experimental/features/delivery_timeout"
+	"knative.dev/eventing/test/rekt/features/broker"
+	b "knative.dev/eventing/test/rekt/resources/broker"
+	rt "knative.dev/eventing/test/rekt/resources/trigger"
+)
+
+func TestDeliveryTimeout(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, delivery_timeout.ChannelToSink())
+}
+
+func TestBrokerTriggerWithDeliveryTimeout(t *testing.T) {
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+		environment.Managed(t),
+	)
+
+	brokerName := feature.MakeRandomK8sName("broker")
+	triggerName := feature.MakeRandomK8sName("trigger")
+
+	env.Test(ctx, t, broker.GoesReady(brokerName, append(b.WithEnvConfig(), b.WithTimeout("PT1S"))...))
+	env.Test(ctx, t, broker.TriggerGoesReady(triggerName, brokerName, rt.WithTimeout("PT10S")))
+}
diff --git a/test/experimental/features_config/features.yaml b/test/experimental/features_config/features.yaml
index dd4b493d4..cf6829924 100644
--- a/test/experimental/features_config/features.yaml
+++ b/test/experimental/features_config/features.yaml
@@ -24,6 +24,6 @@ metadata:
 data:
   kreference-group: "disabled"
   delivery-retryafter: "disabled"
-  delivery-timeout: "disabled"
+  delivery-timeout: "enabled"
   strict-subscriber: "disabled"
   new-trigger-filters: "enabled"
diff --git a/test/experimental/main_test.go b/test/experimental/main_test.go
index f9d653c99..9ab5fdab0 100644
--- a/test/experimental/main_test.go
+++ b/test/experimental/main_test.go
@@ -26,6 +26,7 @@ import (
 
 	// Uncomment the following line to load the gcp plugin (only required to authenticate against GKE clusters).
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
+	"knative.dev/eventing/test/rekt/resources/channel_impl"
 	"knative.dev/pkg/injection"
 	_ "knative.dev/pkg/system/testing"
 
@@ -40,6 +41,10 @@ var global environment.GlobalEnvironment
 func init() {
 	// environment.InitFlags registers state and level filter flags.
 	environment.InitFlags(flag.CommandLine)
+
+	// CHANNEL_GROUP_KIND=KafkaChannel.messaging.knative.dev;CHANNEL_VERSION=v1beta1
+	channel_impl.EnvCfg.ChannelGK = "KafkaChannel.messaging.knative.dev"
+	channel_impl.EnvCfg.ChannelV = "v1beta1"
 }
 
 // TestMain is the first entry point for `go test`.
diff --git a/test/pkg/broker/broker.go b/test/pkg/broker/broker.go
index 8248c92bc..6b1eb108c 100644
--- a/test/pkg/broker/broker.go
+++ b/test/pkg/broker/broker.go
@@ -30,7 +30,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"
 )
 
 const BrokerClassEnvVarKey = "BROKER_CLASS"
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/reconciler-tests.sh b/test/reconciler-tests.sh
index c44fefa61..3c208f671 100755
--- a/test/reconciler-tests.sh
+++ b/test/reconciler-tests.sh
@@ -9,8 +9,9 @@ fi
 
 if ! ${LOCAL_DEVELOPMENT}; then
   scale_controlplane kafka-controller kafka-webhook-eventing eventing-webhook eventing-controller
-  apply_sacura || fail_test "Failed to apply Sacura"
-  apply_chaos || fail_test "Failed to apply chaos"
+  apply_sacura                            || fail_test "Failed to apply Sacura"
+  apply_sacura_sink_source || fail_test "Failed to apply Sacura (Source, Sink, Broker, Channel)"
+  apply_chaos                             || fail_test "Failed to apply chaos"
 fi
 
 header "Waiting Knative eventing to come up"
diff --git a/test/scripts/first-event-delay.sh b/test/scripts/first-event-delay.sh
index f4d4f3be3..f43cf83f7 100755
--- a/test/scripts/first-event-delay.sh
+++ b/test/scripts/first-event-delay.sh
@@ -154,11 +154,16 @@ function wait_for_cloudevent {
   done
 }
 
+function create {
+  app foo$i | ko resolve ${KO_FLAGS} -Bf - | kubectl apply -f -
+}
+
 function run {
   i=$1
 
+  # Retry 3 times, workaround for https://github.com/knative/pkg/issues/1509
+  create "$i" || create "$i" || create "$i" || return $?
   failed=false
-  app foo$i | ko resolve ${KO_FLAGS} -Bf - | kubectl apply -f -
   kubectl wait kafkachannel --timeout=60s -n foo$i channel --for=condition=Ready=True || failed=true
   kubectl wait subscription --timeout=60s -n foo$i event-display --for=condition=Ready=True || failed=true
 
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/channel-config.toml b/test/upgrade/continual/channel-config.toml
index b825ddcad..1a7ef6881 100644
--- a/test/upgrade/continual/channel-config.toml
+++ b/test/upgrade/continual/channel-config.toml
@@ -6,3 +6,8 @@ interval = {{ .Config.Interval.Nanoseconds }}
 
 [forwarder]
 target = '{{- .ForwarderTarget -}}'
+
+[receiver]
+[receiver.teardown]
+duration = 10000000000
+
diff --git a/test/upgrade/continual/kafka-broker-config.toml b/test/upgrade/continual/kafka-broker-config.toml
index b825ddcad..1a7ef6881 100644
--- a/test/upgrade/continual/kafka-broker-config.toml
+++ b/test/upgrade/continual/kafka-broker-config.toml
@@ -6,3 +6,8 @@ interval = {{ .Config.Interval.Nanoseconds }}
 
 [forwarder]
 target = '{{- .ForwarderTarget -}}'
+
+[receiver]
+[receiver.teardown]
+duration = 10000000000
+
diff --git a/test/upgrade/continual/kafka-sink-source-config.toml b/test/upgrade/continual/kafka-sink-source-config.toml
index b825ddcad..1a7ef6881 100644
--- a/test/upgrade/continual/kafka-sink-source-config.toml
+++ b/test/upgrade/continual/kafka-sink-source-config.toml
@@ -6,3 +6,8 @@ interval = {{ .Config.Interval.Nanoseconds }}
 
 [forwarder]
 target = '{{- .ForwarderTarget -}}'
+
+[receiver]
+[receiver.teardown]
+duration = 10000000000
+
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
diff --git a/vendor/knative.dev/eventing/test/experimental/features/delivery_timeout/channel.go b/vendor/knative.dev/eventing/test/experimental/features/delivery_timeout/channel.go
new file mode 100644
index 000000000..ed8903ff0
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/experimental/features/delivery_timeout/channel.go
@@ -0,0 +1,119 @@
+/*
+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 delivery_timeout
+
+import (
+	"context"
+	"time"
+
+	cetest "github.com/cloudevents/sdk-go/v2/test"
+	"github.com/rickb777/date/period"
+	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
+	messagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
+	eventingclient "knative.dev/eventing/pkg/client/injection/client"
+	"knative.dev/eventing/test/rekt/resources/channel"
+	"knative.dev/eventing/test/rekt/resources/subscription"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/eventshub"
+	"knative.dev/reconciler-test/pkg/eventshub/assert"
+	"knative.dev/reconciler-test/pkg/feature"
+)
+
+// ChannelToSink tests a scenario where the flow is source -> channel -> sink (timeout) -- fallback to -> dead letter sink
+func ChannelToSink() *feature.Feature {
+	f := feature.NewFeature()
+
+	timeout := 6 * time.Second
+	// Clocks are funny, let's add 1 second to take in account eventual clock skews
+	timeoutPeriod, _ := period.NewOf(timeout - time.Second)
+	timeoutString := timeoutPeriod.String()
+
+	channelAPIVersion, imcKind := channel.GVK().ToAPIVersionAndKind()
+
+	channelName := feature.MakeRandomK8sName("channel")
+	subName := feature.MakeRandomK8sName("sub-sink")
+	sinkName := feature.MakeRandomK8sName("sink")
+	deadLetterSinkName := feature.MakeRandomK8sName("dead-letter")
+	sourceName := feature.MakeRandomK8sName("source")
+
+	ev := cetest.FullEvent()
+
+	f.Setup("install sink", eventshub.Install(
+		sinkName,
+		eventshub.StartReceiver,
+		eventshub.ResponseWaitTime(timeout),
+	))
+	f.Setup("install dead letter sink", eventshub.Install(
+		deadLetterSinkName,
+		eventshub.StartReceiver,
+	))
+
+	f.Setup("Install channel", channel.Install(channelName))
+	f.Setup("Channel is ready", channel.IsReady(channelName))
+
+	f.Setup("Install channel -> sink subscription", func(ctx context.Context, t feature.T) {
+		namespace := environment.FromContext(ctx).Namespace()
+		_, err := eventingclient.Get(ctx).MessagingV1().Subscriptions(namespace).Create(ctx,
+			&messagingv1.Subscription{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      subName,
+					Namespace: namespace,
+				},
+				Spec: messagingv1.SubscriptionSpec{
+					Channel: duckv1.KReference{
+						APIVersion: channelAPIVersion,
+						Kind:       imcKind,
+						Name:       channelName,
+					},
+					Subscriber: &duckv1.Destination{
+						Ref: &duckv1.KReference{
+							APIVersion: "v1",
+							Kind:       "Service",
+							Name:       sinkName,
+						},
+					},
+					Delivery: &eventingduckv1.DeliverySpec{
+						DeadLetterSink: &duckv1.Destination{
+							Ref: &duckv1.KReference{
+								APIVersion: "v1",
+								Kind:       "Service",
+								Name:       deadLetterSinkName,
+							},
+						},
+						Timeout: &timeoutString,
+					},
+				},
+			}, metav1.CreateOptions{})
+		require.NoError(t, err)
+	})
+
+	f.Setup("subscription channel -> Sink is ready", subscription.IsReady(subName))
+
+	f.Setup("install source", eventshub.Install(
+		sourceName,
+		eventshub.StartSenderToResource(channel.GVR(), channelName),
+		eventshub.InputEvent(ev),
+	))
+
+	f.Assert("receive event on sink", assert.OnStore(sinkName).MatchEvent(cetest.HasId(ev.ID())).Exact(1))
+	f.Assert("receive event on dead letter sink", assert.OnStore(deadLetterSinkName).MatchEvent(cetest.HasId(ev.ID())).Exact(1))
+
+	return f
+}