Repository: knative-sandbox/eventing-kafka-broker
Pull request #1858 (2022-02-03 09:12:18)

Fix KafkaChannel tests 

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1858#" 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/1858?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1858](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1858?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (aa9272b) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/757f781d06e535493a6c3e99f41e96867c2f1d45?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (757f781) will **increase** coverage by `0.08%`.
> The diff coverage is `n/a`.

> :exclamation: Current head aa9272b differs from pull request most recent head c1c80dc. Consider uploading reports for the commit c1c80dc to get more accurate results

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1858/graphs/tree.svg?width=650&height=150&src=pr&token=xyPIgGOiwy&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1858?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@             Coverage Diff              @@
##               main    #1858      +/-   ##
============================================
+ Coverage     66.60%   66.68%   +0.08%     
+ Complexity      617      614       -3     
============================================
  Files           134      135       +1     
  Lines          5686     5722      +36     
  Branches        178      178              
============================================
+ Hits           3787     3816      +29     
- Misses         1595     1602       +7     
  Partials        304      304              
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `82.01% <ø> (-0.15%)` | :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/1858?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...dispatcher/impl/consumer/BaseConsumerVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1858/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.50% <0.00%> (-6.25%)` | :arrow_down: |
| [...tcher/impl/consumer/UnorderedConsumerVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1858/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9Vbm9yZGVyZWRDb25zdW1lclZlcnRpY2xlLmphdmE=) | `86.95% <0.00%> (-4.35%)` | :arrow_down: |
| [...lane/pkg/reconciler/consumergroup/consumergroup.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1858/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2NvbnN1bWVyZ3JvdXAuZ28=) | `76.82% <0.00%> (ø)` | |
| [...nternals/kafka/eventing/v1alpha1/consumer\_types.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1858/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9pbnRlcm5hbHMva2Fma2EvZXZlbnRpbmcvdjFhbHBoYTEvY29uc3VtZXJfdHlwZXMuZ28=) | `23.07% <0.00%> (ø)` | |
| [control-plane/pkg/apis/core/pod\_defaulting.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1858/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9jb3JlL3BvZF9kZWZhdWx0aW5nLmdv) | `88.88% <0.00%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1858?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1858?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [757f781...c1c80dc](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1858?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).

knative-prow-robot: @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
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain | 43fae3e3dc88baf3f0f766834ca6ab6cf294fe68 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1858/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain/1489171421316780032) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1858). [Your PR dashboard](https://gubernator.knative.dev/pr/pierDipi).

<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-robot: @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
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain | 43fae3e3dc88baf3f0f766834ca6ab6cf294fe68 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1858/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain/1489171421316780032) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain`
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl | 43fae3e3dc88baf3f0f766834ca6ab6cf294fe68 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1858/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl/1489171418804391936) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1858). [Your PR dashboard](https://gubernator.knative.dev/pr/pierDipi).

<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 -->

Patch:
diff --git a/control-plane/config/200-controller/200-controller-cluster-role.yaml b/control-plane/config/200-controller/200-controller-cluster-role.yaml
index dfb85a5d7..c77a7e915 100644
--- a/control-plane/config/200-controller/200-controller-cluster-role.yaml
+++ b/control-plane/config/200-controller/200-controller-cluster-role.yaml
@@ -69,6 +69,50 @@ rules:
       - get
       - watch
 
+  # Scheduler permissions
+  - apiGroups:
+      - "*"
+    resources:
+      - nodes
+    verbs:
+      - get
+      - list
+      - watch
+  - apiGroups:
+      - "apps"
+    resources:
+      - statefulsets
+      - statefulsets/scale
+    verbs:
+      - get
+      - list
+      - watch
+      - update
+      - patch
+
+  # Internal APIs
+  - apiGroups:
+      - "internal.kafka.eventing.knative.dev"
+    resources:
+      - "consumers"
+      - "consumers/status"
+      - "consumergroups"
+      - "consumergroups/status"
+    verbs:
+      - create
+      - get
+      - list
+      - watch
+      - patch
+      - update
+  - apiGroups:
+      - "internal.kafka.eventing.knative.dev"
+    resources:
+      - "consumers/finalizers"
+      - "consumergroups/finalizers"
+    verbs:
+      - update
+
   # Eventing resources and statuses we care about
   - apiGroups:
       - "eventing.knative.dev"
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go
index e6078edab..02b7b1964 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go
@@ -191,3 +191,13 @@ func (cg *ConsumerGroup) IsNotScheduled() bool {
 	cond := cg.Status.GetCondition(ConditionConsumerGroupConsumersScheduled)
 	return cond.IsFalse() || cond.IsUnknown()
 }
+
+func (cg *ConsumerGroup) HasDeadLetterSink() bool {
+	return hasDeadLetterSink(cg.Spec.Template.Spec.Delivery)
+}
+
+func hasDeadLetterSink(d *DeliverySpec) bool {
+	return d != nil && d.DeliverySpec != nil &&
+		d.DeliverySpec.DeadLetterSink != nil &&
+		(d.DeliverySpec.DeadLetterSink.Ref != nil || d.DeliverySpec.DeadLetterSink.URI != nil)
+}
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
index 92d17daca..612d55d57 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
@@ -230,3 +230,7 @@ func (c *Consumer) GetConsumerGroup() *metav1.OwnerReference {
 	}
 	return nil
 }
+
+func (c Consumer) HasDeadLetterSink() bool {
+	return hasDeadLetterSink(c.Spec.Delivery)
+}
diff --git a/control-plane/pkg/reconciler/base/reconciler.go b/control-plane/pkg/reconciler/base/reconciler.go
index 5b4b8f154..2fe24b639 100644
--- a/control-plane/pkg/reconciler/base/reconciler.go
+++ b/control-plane/pkg/reconciler/base/reconciler.go
@@ -199,7 +199,10 @@ func (r *Reconciler) UpdateDataPlaneConfigMap(ctx context.Context, contract *con
 		return fmt.Errorf("failed to marshal contract: %w", err)
 	}
 
-	// Update config map data. TODO is it safe to update this config map? do we need to copy it?
+	// Update config map data.
+	if configMap.BinaryData == nil {
+		configMap.BinaryData = make(map[string][]byte, 1)
+	}
 	configMap.BinaryData[ConfigMapDataKey] = data
 
 	_, err = r.KubeClient.CoreV1().ConfigMaps(configMap.Namespace).Update(ctx, configMap, metav1.UpdateOptions{})
diff --git a/control-plane/pkg/reconciler/consumer/consumer.go b/control-plane/pkg/reconciler/consumer/consumer.go
index 062e08a2f..65b7a6078 100644
--- a/control-plane/pkg/reconciler/consumer/consumer.go
+++ b/control-plane/pkg/reconciler/consumer/consumer.go
@@ -320,11 +320,6 @@ func (r Reconciler) schedule(ctx context.Context, logger *zap.Logger, c *kafkain
 		return false, err
 	}
 
-	// Check if the pod is running.
-	if p.Status.Phase != corev1.PodRunning {
-		return false, nil
-	}
-
 	// Get contract associated with the pod.
 	cmName, err := cmNameFromPod(p, c)
 	if err != nil {
@@ -338,6 +333,14 @@ func (r Reconciler) schedule(ctx context.Context, logger *zap.Logger, c *kafkain
 		return false, fmt.Errorf("failed to get or create data plane ConfigMap %s/%s: %w", p.GetNamespace(), cmName, err)
 	}
 
+	// Check if the pod is running after trying to
+	// get or create the associated ConfigMap, since
+	// it won't become ready until we have created the
+	// ConfigMap
+	if p.Status.Phase != corev1.PodRunning {
+		return false, nil
+	}
+
 	ct, err := b.GetDataPlaneConfigMapData(logger, cm)
 	if err != nil {
 		return false, fmt.Errorf("failed to get contract from ConfigMap %s/%s: %w", p.GetNamespace(), cmName, err)
@@ -375,7 +378,7 @@ func (r Reconciler) commonReconciler(p *corev1.Pod, cmName string) base.Reconcil
 func cmNameFromPod(p *corev1.Pod, c *kafkainternals.Consumer) (string, error) {
 	var vDp *corev1.Volume
 	for i, v := range p.Spec.Volumes {
-		if v.Name == eventing.ConfigMapVolumeName && v.ConfigMap != nil && v.ConfigMap.Name != "" {
+		if v.Name == kafkainternals.DispatcherVolumeName && v.ConfigMap != nil && v.ConfigMap.Name != "" {
 			vDp = &p.Spec.Volumes[i]
 			break
 		}
diff --git a/control-plane/pkg/reconciler/consumergroup/consumergroup.go b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
index 6445c606f..401de505b 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
@@ -18,6 +18,7 @@ package consumergroup
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"math"
 	"sort"
@@ -39,6 +40,11 @@ import (
 	kafkainternalslisters "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/listers/eventing/v1alpha1"
 )
 
+var (
+	ErrNoSubscriberURI     = errors.New("no subscriber URI resolved")
+	ErrNoDeadLetterSinkURI = errors.New("no dead letter sink URI resolved")
+)
+
 type schedulerFunc func(s string) scheduler.Scheduler
 
 type Reconciler struct {
@@ -60,6 +66,20 @@ func (r Reconciler) ReconcileKind(ctx context.Context, cg *kafkainternals.Consum
 	if err := r.reconcileConsumers(ctx, cg); err != nil {
 		return err
 	}
+
+	if err := r.propagateStatus(cg); err != nil {
+		return cg.MarkReconcileConsumersFailed("PropagateConsumerStatus", err)
+	}
+
+	if cg.Status.SubscriberURI == nil {
+		_ = cg.MarkReconcileConsumersFailed("PropagateSubscriberURI", ErrNoSubscriberURI)
+		return nil
+	}
+	if cg.HasDeadLetterSink() && cg.Status.DeadLetterSinkURI == nil {
+		_ = cg.MarkReconcileConsumersFailed("PropagateDeadLetterSinkURI", ErrNoDeadLetterSinkURI)
+		return nil
+	}
+
 	cg.MarkReconcileConsumersSucceeded()
 
 	return nil
@@ -241,6 +261,30 @@ func (r Reconciler) joinConsumersByPlacement(placements []eventingduckv1alpha1.P
 	return placementConsumers
 }
 
+func (r Reconciler) propagateStatus(cg *kafkainternals.ConsumerGroup) error {
+	consumers, err := r.ConsumerLister.Consumers(cg.GetNamespace()).List(labels.SelectorFromSet(cg.Spec.Selector))
+	if err != nil {
+		return fmt.Errorf("failed to list consumers for selector %+v: %w", cg.Spec.Selector, err)
+	}
+	count := int32(0)
+	for _, c := range consumers {
+		if c.IsReady() {
+			if c.Spec.VReplicas != nil {
+				count += *c.Spec.VReplicas
+			}
+			if c.Status.SubscriberURI != nil {
+				cg.Status.SubscriberURI = c.Status.SubscriberURI
+			}
+			if c.Status.DeliveryStatus.DeadLetterSinkURI != nil {
+				cg.Status.DeliveryStatus.DeadLetterSinkURI = c.Status.DeadLetterSinkURI
+			}
+		}
+	}
+	cg.Status.Replicas = pointer.Int32(count)
+
+	return nil
+}
+
 var (
 	_ consumergroup.Interface = &Reconciler{}
 )
diff --git a/control-plane/pkg/reconciler/consumergroup/evictor.go b/control-plane/pkg/reconciler/consumergroup/evictor.go
index 28b2149d6..8e5d10675 100644
--- a/control-plane/pkg/reconciler/consumergroup/evictor.go
+++ b/control-plane/pkg/reconciler/consumergroup/evictor.go
@@ -58,7 +58,7 @@ func newEvictor(ctx context.Context, fields ...zap.Field) evictor {
 	}
 }
 
-func (e *evictor) evict(pod *corev1.Pod, vpod scheduler.VPod, from *eventingduckv1alpha1.Placement) error {
+func (e evictor) evict(pod *corev1.Pod, vpod scheduler.VPod, from *eventingduckv1alpha1.Placement) error {
 	key := vpod.GetKey()
 
 	logger := e.logger.
diff --git a/control-plane/pkg/reconciler/source/v2/controllerv2.go b/control-plane/pkg/reconciler/source/v2/controllerv2.go
index b9b072de1..7bb4b69f3 100644
--- a/control-plane/pkg/reconciler/source/v2/controllerv2.go
+++ b/control-plane/pkg/reconciler/source/v2/controllerv2.go
@@ -20,9 +20,10 @@ import (
 	"context"
 
 	"k8s.io/client-go/tools/cache"
+	"knative.dev/pkg/controller"
+
 	consumergroupclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client"
 	consumergroupinformer "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumergroup"
-	"knative.dev/pkg/controller"
 
 	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
 
@@ -47,9 +48,7 @@ func NewController(ctx context.Context, configs *config.Env) *controller.Impl {
 
 	impl := kafkasource.NewImpl(ctx, r)
 
-	kafkaInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
-		Handler: controller.HandleAll(impl.Enqueue),
-	})
+	kafkaInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
 
 	// ConsumerGroup changes and enqueue associated KafkaSource
 	consumerGroupInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{


Test patch:
diff --git a/.github/workflows/kind-e2e.yaml b/.github/workflows/kind-e2e.yaml
index 86c037c7d..fa8218691 100644
--- a/.github/workflows/kind-e2e.yaml
+++ b/.github/workflows/kind-e2e.yaml
@@ -24,6 +24,12 @@ jobs:
         eventing-config:
           - "./third_party/eventing-latest/"
 
+        channel-auth-scenario:
+          - ""
+          - "SSL"
+          - "SASL_SSL"
+          - "SASL_PLAIN"
+
         test-suite:
           - ./test/e2e
           - ./test/e2e_new
@@ -43,6 +49,21 @@ jobs:
           - test-suite: ./test/e2e_channel
             extra-test-flags: -channels=messaging.knative.dev/v1beta1:KafkaChannel
 
+        # don't run e2e for tests other than KafkaChannel when channel-auth-scenario is set
+        exclude:
+          - test-suite: ./test/e2e
+            channel-auth-scenario: "SSL"
+          - test-suite: ./test/e2e
+            channel-auth-scenario: "SASL_SSL"
+          - test-suite: ./test/e2e
+            channel-auth-scenario: "SASL_PLAIN"
+          - test-suite: ./test/e2e_new
+            channel-auth-scenario: "SSL"
+          - test-suite: ./test/e2e_new
+            channel-auth-scenario: "SASL_SSL"
+          - test-suite: ./test/e2e_new
+            channel-auth-scenario: "SASL_PLAIN"
+
     env:
       GOPATH: ${{ github.workspace }}
       KO_DOCKER_REPO: kind.local
@@ -57,6 +78,8 @@ jobs:
       NODE_VERSION: ${{ matrix.k8s-version }}
       NODE_SHA: ${{ matrix.kind-image-sha }}
 
+      EVENTING_KAFKA_BROKER_CHANNEL_AUTH_SCENARIO: ${{ matrix.channel-auth-scenario }}
+
     steps:
       - name: Install Dependencies
         working-directory: ./
@@ -120,6 +143,8 @@ jobs:
 
       - name: Run e2e Tests
         run: |
+          source test/e2e-common.sh
+
           export KO_FLAGS="--platform=linux/amd64"
           export SYSTEM_NAMESPACE=knative-eventing
           export CLUSTER_DOMAIN=${CLUSTER_SUFFIX}
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types_test.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types_test.go
index f44f42c35..8a0e88a4d 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types_test.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types_test.go
@@ -17,10 +17,14 @@
 package v1alpha1
 
 import (
+	"context"
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
 )
 
 func TestConsumerGroup_GetUserFacingResourceRef(t *testing.T) {
@@ -124,3 +128,58 @@ func TestConsumerGroup_GetUserFacingResourceRef(t *testing.T) {
 		})
 	}
 }
+
+func TestHasDeadLetterSink(t *testing.T) {
+
+	type HasDeadLetterSink interface {
+		HasDeadLetterSink() bool
+		apis.Defaultable
+	}
+
+	tt := []struct {
+		name              string
+		object            HasDeadLetterSink
+		hasDeadLetterSink bool
+	}{
+		{
+			name: "consumer group has dead letter sink",
+			object: &ConsumerGroup{Spec: ConsumerGroupSpec{Template: ConsumerTemplateSpec{Spec: ConsumerSpec{
+				Delivery: &DeliverySpec{DeliverySpec: &eventingduck.DeliverySpec{
+					DeadLetterSink: &duckv1.Destination{URI: apis.HTTP("localhost")},
+				}},
+			}}}},
+			hasDeadLetterSink: true,
+		},
+		{
+			name:              "consumer group has no dead letter sink",
+			object:            &ConsumerGroup{},
+			hasDeadLetterSink: false,
+		},
+		{
+			name: "consumer has dead letter sink",
+			object: &Consumer{Spec: ConsumerSpec{
+				Delivery: &DeliverySpec{DeliverySpec: &eventingduck.DeliverySpec{
+					DeadLetterSink: &duckv1.Destination{URI: apis.HTTP("localhost")},
+				}},
+			}},
+			hasDeadLetterSink: true,
+		},
+		{
+			name:              "consumer has no dead letter sink",
+			object:            &Consumer{},
+			hasDeadLetterSink: false,
+		},
+	}
+
+	for _, tc := range tt {
+		t.Run(tc.name, func(t *testing.T) {
+			tc.object.SetDefaults(context.Background())
+			got := tc.object.HasDeadLetterSink()
+
+			if tc.hasDeadLetterSink != got {
+				t.Errorf("want %v got %v", tc.hasDeadLetterSink, got)
+			}
+		})
+	}
+
+}
diff --git a/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go b/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
index 81f9e3ec3..9e9ef95f3 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
@@ -25,6 +25,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	clientgotesting "k8s.io/client-go/testing"
+	"k8s.io/utils/pointer"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
 	. "knative.dev/pkg/reconciler/testing"
@@ -32,6 +33,7 @@ import (
 	eventingduckv1alpha1 "knative.dev/eventing/pkg/apis/duck/v1alpha1"
 	"knative.dev/eventing/pkg/scheduler"
 
+	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 	fakekafkainternalsclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client/fake"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/reconciler/eventing/v1alpha1/consumergroup"
@@ -115,13 +117,91 @@ func TestReconcileKind(t *testing.T) {
 								),
 							)),
 							ConsumerGroupReplicas(2),
+							ConsumerGroupStatusReplicas(0),
 							ConsumerForTrigger(),
 						)
 						cg.Status.Placements = []eventingduckv1alpha1.Placement{
 							{PodName: "p1", VReplicas: 1},
 							{PodName: "p2", VReplicas: 1},
 						}
-						cg.MarkReconcileConsumersSucceeded()
+						_ = cg.MarkReconcileConsumersFailed("PropagateSubscriberURI", ErrNoSubscriberURI)
+						cg.MarkScheduleSucceeded()
+						return cg
+					}(),
+				},
+			},
+		},
+		{
+			Name: "Consumers in multiple pods, one exists - not ready",
+			Objects: []runtime.Object{
+				NewConsumer(2,
+					ConsumerSpec(NewConsumerSpec(
+						ConsumerTopics("t1", "t2"),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+						ConsumerVReplicas(1),
+						ConsumerPlacement(kafkainternals.PodBind{
+							PodName:      "p2",
+							PodNamespace: systemNamespace,
+						}),
+					)),
+				),
+				NewConsumerGroup(
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics("t1", "t2"),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+					)),
+					ConsumerForTrigger(),
+					ConsumerGroupReplicas(2),
+				),
+			},
+			Key: ConsumerGroupTestKey,
+			OtherTestData: map[string]interface{}{
+				testSchedulerKey: SchedulerFunc(func(vpod scheduler.VPod) ([]eventingduckv1alpha1.Placement, error) {
+					return []eventingduckv1alpha1.Placement{
+						{PodName: "p1", VReplicas: 1},
+						{PodName: "p2", VReplicas: 1},
+					}, nil
+				}),
+			},
+			WantCreates: []runtime.Object{
+				NewConsumer(1,
+					ConsumerSpec(NewConsumerSpec(
+						ConsumerTopics("t1", "t2"),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+						ConsumerVReplicas(1),
+						ConsumerPlacement(kafkainternals.PodBind{PodName: "p1", PodNamespace: systemNamespace}),
+					)),
+				),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: func() runtime.Object {
+						cg := NewConsumerGroup(
+							ConsumerGroupConsumerSpec(NewConsumerSpec(
+								ConsumerTopics("t1", "t2"),
+								ConsumerConfigs(
+									ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+									ConsumerGroupIdConfig("my.group.id"),
+								),
+							)),
+							ConsumerGroupReplicas(2),
+							ConsumerGroupStatusReplicas(0),
+							ConsumerForTrigger(),
+						)
+						cg.Status.Placements = []eventingduckv1alpha1.Placement{
+							{PodName: "p1", VReplicas: 1},
+							{PodName: "p2", VReplicas: 1},
+						}
+						_ = cg.MarkReconcileConsumersFailed("PropagateSubscriberURI", ErrNoSubscriberURI)
 						cg.MarkScheduleSucceeded()
 						return cg
 					}(),
@@ -129,7 +209,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 		},
 		{
-			Name: "Consumers in multiple pods, one exists",
+			Name: "Consumers in multiple pods, one exists - ready",
 			Objects: []runtime.Object{
 				NewConsumer(2,
 					ConsumerSpec(NewConsumerSpec(
@@ -144,6 +224,7 @@ func TestReconcileKind(t *testing.T) {
 							PodNamespace: systemNamespace,
 						}),
 					)),
+					ConsumerReady(),
 				),
 				NewConsumerGroup(
 					ConsumerGroupConsumerSpec(NewConsumerSpec(
@@ -191,6 +272,7 @@ func TestReconcileKind(t *testing.T) {
 								),
 							)),
 							ConsumerGroupReplicas(2),
+							ConsumerGroupStatusReplicas(1),
 							ConsumerForTrigger(),
 						)
 						cg.Status.Placements = []eventingduckv1alpha1.Placement{
@@ -199,13 +281,131 @@ func TestReconcileKind(t *testing.T) {
 						}
 						cg.MarkReconcileConsumersSucceeded()
 						cg.MarkScheduleSucceeded()
+						cg.Status.SubscriberURI = ConsumerSubscriberURI
+						return cg
+					}(),
+				},
+			},
+		},
+		{
+			Name: "Consumers in multiple pods, one exists - ready, increase replicas, propagate dead letter sink URI",
+			Objects: []runtime.Object{
+				NewConsumer(2,
+					ConsumerSpec(NewConsumerSpec(
+						ConsumerTopics("t1", "t2"),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered,
+							NewConsumerSpecDeliveryDeadLetterSink(),
+						)),
+						ConsumerVReplicas(1),
+						ConsumerPlacement(kafkainternals.PodBind{
+							PodName:      "p2",
+							PodNamespace: systemNamespace,
+						}),
+					)),
+					ConsumerReady(),
+				),
+				NewConsumerGroup(
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics("t1", "t2"),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered,
+							NewConsumerSpecDeliveryDeadLetterSink(),
+						)),
+					)),
+					ConsumerGroupReplicas(3),
+					ConsumerForTrigger(),
+				),
+			},
+			Key: ConsumerGroupTestKey,
+			OtherTestData: map[string]interface{}{
+				testSchedulerKey: SchedulerFunc(func(vpod scheduler.VPod) ([]eventingduckv1alpha1.Placement, error) {
+					return []eventingduckv1alpha1.Placement{
+						{PodName: "p1", VReplicas: 1},
+						{PodName: "p2", VReplicas: 2},
+					}, nil
+				}),
+			},
+			WantCreates: []runtime.Object{
+				NewConsumer(1,
+					ConsumerSpec(NewConsumerSpec(
+						ConsumerTopics("t1", "t2"),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered,
+							NewConsumerSpecDeliveryDeadLetterSink(),
+						)),
+						ConsumerVReplicas(1),
+						ConsumerPlacement(kafkainternals.PodBind{PodName: "p1", PodNamespace: systemNamespace}),
+					)),
+				),
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				clientgotesting.NewUpdateAction(
+					schema.GroupVersionResource{
+						Group:    kafkainternals.SchemeGroupVersion.Group,
+						Version:  kafkainternals.SchemeGroupVersion.Version,
+						Resource: "consumers",
+					},
+					ConsumerNamespace,
+					NewConsumer(2,
+						ConsumerSpec(NewConsumerSpec(
+							ConsumerTopics("t1", "t2"),
+							ConsumerConfigs(
+								ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+								ConsumerGroupIdConfig("my.group.id"),
+							),
+							ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered,
+								NewConsumerSpecDeliveryDeadLetterSink(),
+							)),
+							ConsumerVReplicas(2),
+							ConsumerPlacement(kafkainternals.PodBind{PodName: "p2", PodNamespace: systemNamespace}),
+						)),
+						ConsumerReady(),
+					),
+				),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: func() runtime.Object {
+						cg := NewConsumerGroup(
+							ConsumerGroupConsumerSpec(NewConsumerSpec(
+								ConsumerTopics("t1", "t2"),
+								ConsumerConfigs(
+									ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+									ConsumerGroupIdConfig("my.group.id"),
+								),
+								ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered,
+									NewConsumerSpecDeliveryDeadLetterSink(),
+								)),
+							)),
+							ConsumerGroupReplicas(3),
+							ConsumerForTrigger(),
+						)
+						cg.Status.Placements = []eventingduckv1alpha1.Placement{
+							{PodName: "p1", VReplicas: 1},
+							{PodName: "p2", VReplicas: 2},
+						}
+						cg.MarkReconcileConsumersSucceeded()
+						cg.MarkScheduleSucceeded()
+						cg.Status.SubscriberURI = ConsumerSubscriberURI
+						cg.Status.DeadLetterSinkURI = ConsumerDeadLetterSinkURI
+						cg.Status.Replicas = pointer.Int32(1)
 						return cg
 					}(),
 				},
 			},
 		},
 		{
-			Name: "Consumers in multiple pods, one exists, increase replicas",
+			Name: "Consumers in multiple pods, one exists - ready, increase replicas",
 			Objects: []runtime.Object{
 				NewConsumer(2,
 					ConsumerSpec(NewConsumerSpec(
@@ -220,6 +420,7 @@ func TestReconcileKind(t *testing.T) {
 							PodNamespace: systemNamespace,
 						}),
 					)),
+					ConsumerReady(),
 				),
 				NewConsumerGroup(
 					ConsumerGroupConsumerSpec(NewConsumerSpec(
@@ -273,6 +474,7 @@ func TestReconcileKind(t *testing.T) {
 							ConsumerVReplicas(2),
 							ConsumerPlacement(kafkainternals.PodBind{PodName: "p2", PodNamespace: systemNamespace}),
 						)),
+						ConsumerReady(),
 					),
 				),
 			},
@@ -296,6 +498,8 @@ func TestReconcileKind(t *testing.T) {
 						}
 						cg.MarkReconcileConsumersSucceeded()
 						cg.MarkScheduleSucceeded()
+						cg.Status.SubscriberURI = ConsumerSubscriberURI
+						cg.Status.Replicas = pointer.Int32(1)
 						return cg
 					}(),
 				},
@@ -392,8 +596,9 @@ func TestReconcileKind(t *testing.T) {
 							{PodName: "p1", VReplicas: 1},
 							{PodName: "p2", VReplicas: 1},
 						}
-						cg.MarkReconcileConsumersSucceeded()
+						_ = cg.MarkReconcileConsumersFailed("PropagateSubscriberURI", ErrNoSubscriberURI)
 						cg.MarkScheduleSucceeded()
+						cg.Status.Replicas = pointer.Int32(0)
 						return cg
 					}(),
 				},
diff --git a/control-plane/pkg/reconciler/testing/objects_consumer.go b/control-plane/pkg/reconciler/testing/objects_consumer.go
index 4797e71f5..bb2a6ca3b 100644
--- a/control-plane/pkg/reconciler/testing/objects_consumer.go
+++ b/control-plane/pkg/reconciler/testing/objects_consumer.go
@@ -22,7 +22,9 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/pointer"
 	"knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
+	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	v1 "knative.dev/eventing/pkg/apis/eventing/v1"
+	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
 	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
@@ -34,6 +36,11 @@ const (
 	ConsumerNamespace  = "test-cg-ns"
 )
 
+var (
+	ConsumerSubscriberURI     = apis.HTTP("localhost")
+	ConsumerDeadLetterSinkURI = apis.HTTP("dls.com")
+)
+
 type ConsumerOption func(cg *kafkainternals.Consumer)
 
 type ConsumerSpecOption func(c *kafkainternals.ConsumerSpec)
@@ -84,9 +91,24 @@ func NewConsumerSpec(opts ...ConsumerSpecOption) kafkainternals.ConsumerSpec {
 	return *spec
 }
 
-func NewConsumerSpecDelivery(order internals.DeliveryOrdering) *kafkainternals.DeliverySpec {
-	return &kafkainternals.DeliverySpec{
-		Ordering: order,
+type DeliverySpecOption func(spec *kafkainternals.DeliverySpec)
+
+func NewConsumerSpecDelivery(order internals.DeliveryOrdering, options ...DeliverySpecOption) *kafkainternals.DeliverySpec {
+	d := &kafkainternals.DeliverySpec{Ordering: order}
+
+	for _, opt := range options {
+		opt(d)
+	}
+
+	return d
+}
+
+func NewConsumerSpecDeliveryDeadLetterSink() DeliverySpecOption {
+	return func(spec *kafkainternals.DeliverySpec) {
+		if spec.DeliverySpec == nil {
+			spec.DeliverySpec = &eventingduck.DeliverySpec{}
+		}
+		spec.DeliverySpec.DeadLetterSink = &duckv1.Destination{URI: ConsumerDeadLetterSinkURI}
 	}
 }
 
@@ -192,3 +214,15 @@ func ConsumerForTrigger() ConsumerGroupOption {
 		})
 	}
 }
+
+func ConsumerReady() ConsumerOption {
+	return func(c *kafkainternals.Consumer) {
+		c.MarkBindSucceeded()
+		c.MarkReconcileContractSucceeded()
+		c.Status.SubscriberURI = ConsumerSubscriberURI
+
+		if c.HasDeadLetterSink() {
+			c.Status.DeadLetterSinkURI = ConsumerDeadLetterSinkURI
+		}
+	}
+}
diff --git a/control-plane/pkg/reconciler/testing/objects_consumergroup.go b/control-plane/pkg/reconciler/testing/objects_consumergroup.go
index a569f4f9a..79cca14d2 100644
--- a/control-plane/pkg/reconciler/testing/objects_consumergroup.go
+++ b/control-plane/pkg/reconciler/testing/objects_consumergroup.go
@@ -115,6 +115,12 @@ func ConsumerGroupReplicas(replicas int32) ConsumerGroupOption {
 	}
 }
 
+func ConsumerGroupStatusReplicas(replicas int32) ConsumerGroupOption {
+	return func(cg *kafkainternals.ConsumerGroup) {
+		cg.Status.Replicas = pointer.Int32Ptr(replicas)
+	}
+}
+
 func ConsumerGroupReplicasStatus(replicas int32) ConsumerGroupOption {
 	return func(cg *kafkainternals.ConsumerGroup) {
 		cg.Status.Replicas = pointer.Int32Ptr(replicas)
diff --git a/test/channel-tests.sh b/test/channel-tests.sh
new file mode 100755
index 000000000..b774742f9
--- /dev/null
+++ b/test/channel-tests.sh
@@ -0,0 +1,5 @@
+#!/usr/bin/env bash
+
+source $(dirname $0)/e2e-common.sh
+
+go_test_e2e -timeout=1h ./test/e2e_channel/... -channels=messaging.knative.dev/v1beta1:KafkaChannel || fail_test "E2E suite (KafkaChannel) failed"
diff --git a/test/e2e-common.sh b/test/e2e-common.sh
index 24509c2f5..5e3c60d04 100644
--- a/test/e2e-common.sh
+++ b/test/e2e-common.sh
@@ -154,6 +154,8 @@ function test_setup() {
   # Apply test configurations, and restart data plane components (we don't have hot reload)
   ko apply -f ./test/config/ || fail_test "Failed to apply test configurations"
 
+  setup_kafka_channel_auth || fail_test "Failed to apply channel auth configuration ${EVENTING_KAFKA_BROKER_CHANNEL_AUTH_SCENARIO}"
+
   kubectl rollout restart deployment -n knative-eventing kafka-source-dispatcher
   kubectl rollout restart deployment -n knative-eventing kafka-broker-receiver
   kubectl rollout restart deployment -n knative-eventing kafka-broker-dispatcher
@@ -257,3 +259,37 @@ function build_monitoring_artifacts() {
   ko resolve ${KO_FLAGS} -Rf "${EVENTING_KAFKA_CHANNEL_PROMETHEUS_OPERATOR_ARTIFACT_PATH}" |
     "${LABEL_YAML_CMD[@]}" >"${EVENTING_KAFKA_CHANNEL_PROMETHEUS_OPERATOR_ARTIFACT}" || return $?
 }
+
+function setup_kafka_channel_auth() {
+  echo "Apply channel auth config ${EVENTING_KAFKA_BROKER_CHANNEL_AUTH_SCENARIO}"
+
+  if [ "$EVENTING_KAFKA_BROKER_CHANNEL_AUTH_SCENARIO" == "SSL" ]; then
+    echo "Setting up SSL configuration for KafkaChannel"
+    kubectl patch configmap/kafka-channel-config \
+      -n knative-eventing \
+      --type merge \
+      -p '{"data":{"bootstrap.servers":"my-cluster-kafka-bootstrap.kafka:9093", "auth.secret.ref.name": "strimzi-tls-secret"}}'
+  elif [ "$EVENTING_KAFKA_BROKER_CHANNEL_AUTH_SCENARIO" == "SASL_SSL" ]; then
+    echo "Setting up SASL_SSL configuration for KafkaChannel"
+    kubectl patch configmap/kafka-channel-config \
+      -n knative-eventing \
+      --type merge \
+      -p '{"data":{"bootstrap.servers":"my-cluster-kafka-bootstrap.kafka:9094", "auth.secret.ref.name": "strimzi-sasl-secret"}}'
+  elif [ "$EVENTING_KAFKA_BROKER_CHANNEL_AUTH_SCENARIO" == "SASL_PLAIN" ]; then
+    echo "Setting up SASL_PLAIN configuration for KafkaChannel"
+    kubectl patch configmap/kafka-channel-config \
+      -n knative-eventing \
+      --type merge \
+      -p '{"data":{"bootstrap.servers":"my-cluster-kafka-bootstrap.kafka:9095", "auth.secret.ref.name": "strimzi-sasl-plain-secret"}}'
+  else
+    echo "Setting up no auth configuration for KafkaChannel"
+    kubectl patch configmap/kafka-channel-config \
+      -n knative-eventing \
+      --type merge \
+      -p '{"data":{"bootstrap.servers":"my-cluster-kafka-bootstrap.kafka:9092"}}'
+    kubectl patch configmap/kafka-channel-config \
+      -n knative-eventing \
+      --type=json \
+      -p='[{"op": "remove", "path": "/data/auth.secret.ref.name"}]' || true
+  fi
+}
diff --git a/test/e2e-tests.sh b/test/e2e-tests.sh
index eeedc53ae..35cf47006 100755
--- a/test/e2e-tests.sh
+++ b/test/e2e-tests.sh
@@ -17,9 +17,15 @@ header "Waiting Knative eventing to come up"
 
 wait_until_pods_running knative-eventing || fail_test "Pods in knative-eventing didn't come up"
 
+export_logs_continuously
+
 header "Running tests"
 
-export_logs_continuously
+if [ "${EVENTING_KAFKA_BROKER_CHANNEL_AUTH_SCENARIO:-""}" != "" ]; then
+  # if this flag exists, only test Kafka channel scenarios with auth
+  $(dirname $0)/channel-tests.sh || fail_test "Failed to execute KafkaChannel tests"
+  success
+fi
 
 go_test_e2e -timeout=1h ./test/e2e/... || fail_test "E2E suite failed"
 
diff --git a/test/kafka/kafka_setup.sh b/test/kafka/kafka_setup.sh
index 7c4936284..a4ed609d0 100755
--- a/test/kafka/kafka_setup.sh
+++ b/test/kafka/kafka_setup.sh
@@ -66,7 +66,7 @@ function create_tls_secrets() {
   ca_cert_secret="my-cluster-cluster-ca-cert"
   tls_user="my-tls-user"
 
-  echo "Waiting until secrets: ${ca_cert_secret}, ${tls_user} don't exist"
+  echo "Waiting until secrets: ${ca_cert_secret}, ${tls_user} exist"
   wait_until_object_exists secret ${ca_cert_secret} kafka
   wait_until_object_exists secret ${tls_user} kafka
 
@@ -80,6 +80,7 @@ function create_tls_secrets() {
     --from-literal=ca.crt="$STRIMZI_CRT" \
     --from-literal=user.crt="$TLSUSER_CRT" \
     --from-literal=user.key="$TLSUSER_KEY" \
+    --from-literal=protocol="SSL" \
     --dry-run=client -o yaml | kubectl apply -n "${SYSTEM_NAMESPACE}" -f -
 }
 
@@ -87,11 +88,11 @@ function create_sasl_secrets() {
   ca_cert_secret="my-cluster-cluster-ca-cert"
   sasl_user="my-sasl-user"
 
-  echo "Waiting until secrets: ${ca_cert_secret}, ${sasl_user} don't exist"
+  echo "Waiting until secrets: ${ca_cert_secret}, ${sasl_user} exist"
   wait_until_object_exists secret ${ca_cert_secret} kafka
   wait_until_object_exists secret ${sasl_user} kafka
 
-  echo "Creating SASL Kafka secret"
+  echo "Creating SASL_SSL and SASL_PLAINTEXT Kafka secrets"
 
   STRIMZI_CRT=$(kubectl -n kafka get secret ${ca_cert_secret} --template='{{index .data "ca.crt"}}' | base64 --decode )
   SASL_PASSWD=$(kubectl -n kafka get secret ${sasl_user} --template='{{index .data "password"}}' | base64 --decode )
@@ -99,8 +100,18 @@ function create_sasl_secrets() {
   kubectl create secret --namespace "${SYSTEM_NAMESPACE}" generic strimzi-sasl-secret \
     --from-literal=ca.crt="$STRIMZI_CRT" \
     --from-literal=password="$SASL_PASSWD" \
+    --from-literal=user="my-sasl-user" \
+    --from-literal=protocol="SASL_SSL" \
+    --from-literal=sasl.mechanism="SCRAM-SHA-512" \
     --from-literal=saslType="SCRAM-SHA-512" \
+    --dry-run=client -o yaml | kubectl apply -n "${SYSTEM_NAMESPACE}" -f -
+
+  kubectl create secret --namespace "${SYSTEM_NAMESPACE}" generic strimzi-sasl-plain-secret \
+    --from-literal=password="$SASL_PASSWD" \
     --from-literal=user="my-sasl-user" \
+    --from-literal=protocol="SASL_PLAINTEXT" \
+    --from-literal=sasl.mechanism="SCRAM-SHA-512" \
+    --from-literal=saslType="SCRAM-SHA-512" \
     --dry-run=client -o yaml | kubectl apply -n "${SYSTEM_NAMESPACE}" -f -
 }
 
diff --git a/test/reconciler-tests.sh b/test/reconciler-tests.sh
index 0f823dfe1..47cb42619 100755
--- a/test/reconciler-tests.sh
+++ b/test/reconciler-tests.sh
@@ -17,9 +17,15 @@ header "Waiting Knative eventing to come up"
 
 wait_until_pods_running knative-eventing || fail_test "Pods in knative-eventing didn't come up"
 
+export_logs_continuously
+
 header "Running tests"
 
-export_logs_continuously
+if [ "${EVENTING_KAFKA_BROKER_CHANNEL_AUTH_SCENARIO:-""}" != "" ]; then
+  # if this flag exists, only test Kafka channel scenarios with auth
+  $(dirname $0)/channel-tests.sh || fail_test "Failed to execute KafkaChannel tests"
+  success
+fi
 
 go_test_e2e -timeout=1h ./test/e2e_new/... || fail_test "E2E (new) suite failed"