Repository: knative-sandbox/eventing-kafka-broker
Pull request #1451 (2021-11-10 01:32:45)

[main] Update actions
Cron -knative-prow-robot

/cc knative-sandbox/channel-wg-leads
/assign knative-sandbox/channel-wg-leads

Produced by: knative-sandbox/knobots/actions/update-actions

Details:
```

```

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1451#" title="Author self-approved">knative-automation</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **rhuss** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @rhuss` in a comment when ready.

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

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

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

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

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1451/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/1451?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    #1451      +/-   ##
============================================
- Coverage     71.40%   61.71%   -9.69%     
============================================
  Files           100       36      -64     
  Lines          3941     2074    -1867     
  Branches        160        0     -160     
============================================
- Hits           2814     1280    -1534     
+ Misses          911      664     -247     
+ Partials        216      130      -86     
```

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

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

| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1451?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...roker/dispatcher/impl/filter/AttributesFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1451/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvQXR0cmlidXRlc0ZpbHRlci5qYXZh) | | |
| [.../core/reconciler/impl/ResourcesReconcilerImpl.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1451/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvcmVjb25jaWxlci9pbXBsL1Jlc291cmNlc1JlY29uY2lsZXJJbXBsLmphdmE=) | | |
| [...ispatcher/impl/http/WebClientCloudEventSender.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1451/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9odHRwL1dlYkNsaWVudENsb3VkRXZlbnRTZW5kZXIuamF2YQ==) | | |
| [...g/kafka/broker/core/security/KafkaClientsAuth.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1451/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvc2VjdXJpdHkvS2Fma2FDbGllbnRzQXV0aC5qYXZh) | | |
| [...venting/kafka/broker/core/tracing/TracingSpan.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1451/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdHJhY2luZy9UcmFjaW5nU3Bhbi5qYXZh) | | |
| [...nting/kafka/broker/dispatcher/ResponseHandler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1451/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvUmVzcG9uc2VIYW5kbGVyLmphdmE=) | | |
| [...ka/broker/dispatcher/impl/filter/SuffixFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1451/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvU3VmZml4RmlsdGVyLmphdmE=) | | |
| [.../dispatcher/impl/RecordDispatcherMutatorChain.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1451/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9SZWNvcmREaXNwYXRjaGVyTXV0YXRvckNoYWluLmphdmE=) | | |
| [...eceiver/impl/IngressProducerReconcilableStore.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1451/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL0luZ3Jlc3NQcm9kdWNlclJlY29uY2lsYWJsZVN0b3JlLmphdmE=) | | |
| [...ting/kafka/broker/core/utils/CollectionsUtils.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1451/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdXRpbHMvQ29sbGVjdGlvbnNVdGlscy5qYXZh) | | |
| ... and [54 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1451/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1451?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/1451?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [5a594f7...67d01d6](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1451?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: @knative-automation: 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-unit-tests | 67d01d6a4c42b8eda620f776ed6a3b1b736fe13f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1451/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1458246219712172032) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

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

<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: @knative-automation: 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-unit-tests | 67d01d6a4c42b8eda620f776ed6a3b1b736fe13f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1451/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1458246219712172032) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 67d01d6a4c42b8eda620f776ed6a3b1b736fe13f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1451/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1458246219741532160) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

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

<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: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1451#" title="Author self-approved">knative-automation</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1451#pullrequestreview-802149102" title="Approved">matzew</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]

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/.github/workflows/knative-boilerplate.yaml b/.github/workflows/knative-boilerplate.yaml
index b29db540a..2a0b01ba4 100644
--- a/.github/workflows/knative-boilerplate.yaml
+++ b/.github/workflows/knative-boilerplate.yaml
@@ -61,7 +61,7 @@ jobs:
           echo '::endgroup::'
 
           echo '::group:: Installing boilerplate-check ... https://github.com/mattmoor/boilerplate-check'
-          go get github.com/mattmoor/boilerplate-check/cmd/boilerplate-check
+          go install github.com/mattmoor/boilerplate-check/cmd/boilerplate-check@latest
           echo '::endgroup::'
 
           echo "${TEMP_PATH}" >> $GITHUB_PATH
diff --git a/.github/workflows/knative-releasability.yaml b/.github/workflows/knative-releasability.yaml
index 00e38a127..1d9111951 100644
--- a/.github/workflows/knative-releasability.yaml
+++ b/.github/workflows/knative-releasability.yaml
@@ -66,7 +66,7 @@ jobs:
           go-version: 1.17.x
 
       - name: Install Dependencies
-        run: GO111MODULE=on go get knative.dev/test-infra/buoy@main
+        run: go install knative.dev/test-infra/buoy@main
 
       - name: Check out code
         uses: actions/checkout@v2
diff --git a/.github/workflows/knative-release-notes.yaml b/.github/workflows/knative-release-notes.yaml
index dd8bf7afb..999231ca7 100644
--- a/.github/workflows/knative-release-notes.yaml
+++ b/.github/workflows/knative-release-notes.yaml
@@ -44,7 +44,7 @@ jobs:
           go-version: 1.17.x
 
       - name: Install Dependencies
-        run: GO111MODULE=on go get k8s.io/release/cmd/release-notes
+        run: go install k8s.io/release/cmd/release-notes@latest
 
       - name: Check out code
         uses: actions/checkout@v2
diff --git a/.github/workflows/knative-style.yaml b/.github/workflows/knative-style.yaml
index 872471099..bbd306783 100644
--- a/.github/workflows/knative-style.yaml
+++ b/.github/workflows/knative-style.yaml
@@ -37,7 +37,7 @@ jobs:
         - tool: gofmt
           options: -s
         - tool: goimports
-          importpath: golang.org/x/tools/cmd/goimports
+          package: golang.org/x/tools/cmd/goimports@latest
 
     steps:
       - name: Set up Go
@@ -50,10 +50,10 @@ jobs:
         uses: actions/checkout@v2
 
       - name: Install Dependencies
-        if: ${{ matrix.importpath != '' }}
+        if: ${{ matrix.package != '' }}
         run: |
           cd $(mktemp -d)
-          GO111MODULE=on go get ${{ matrix.importpath }}
+          go install ${{ matrix.package }}
 
       - name: ${{ matrix.tool }} ${{ matrix.options }}
         shell: bash
diff --git a/.github/workflows/knative-verify.yaml b/.github/workflows/knative-verify.yaml
index 541ca5c25..c3b0da760 100644
--- a/.github/workflows/knative-verify.yaml
+++ b/.github/workflows/knative-verify.yaml
@@ -45,15 +45,11 @@ jobs:
 
     - name: Install Dependencies
       run: |
-        curl -L https://github.com/google/ko/releases/download/v0.6.0/ko_0.6.0_Linux_x86_64.tar.gz | tar xzf - ko
-        chmod +x ./ko
-        sudo mv ko /usr/local/bin
-        go get github.com/google/go-licenses
+        go install github.com/google/ko@latest
+        go install github.com/google/go-licenses@latest
 
-    - name: Check out code onto GOPATH
+    - name: Check out code
       uses: actions/checkout@v2
-      with:
-        path: ./src/knative.dev/${{ github.event.repository.name }}
 
     - name: Update Codegen
       shell: bash
diff --git a/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index 589c3d838..3d9350728 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -23,31 +23,35 @@ import (
 	"strings"
 
 	"github.com/Shopify/sarama"
+
 	"go.uber.org/multierr"
 	"go.uber.org/zap"
+
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
 	corelisters "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/util/retry"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
-	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
-	commonconfig "knative.dev/eventing-kafka/pkg/common/config"
-	"knative.dev/eventing-kafka/pkg/common/constants"
-	"knative.dev/eventing-kafka/pkg/common/kafka/offset"
-	commonsarama "knative.dev/eventing-kafka/pkg/common/kafka/sarama"
-	v1 "knative.dev/eventing/pkg/apis/duck/v1"
+
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/reconciler"
 	"knative.dev/pkg/resolver"
 	"knative.dev/pkg/system"
 
+	v1 "knative.dev/eventing/pkg/apis/duck/v1"
+
+	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
+	commonconfig "knative.dev/eventing-kafka/pkg/common/config"
+	"knative.dev/eventing-kafka/pkg/common/constants"
+	commonsarama "knative.dev/eventing-kafka/pkg/common/kafka/sarama"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 	coreconfig "knative.dev/eventing-kafka-broker/control-plane/pkg/core/config"
 	kafkalogging "knative.dev/eventing-kafka-broker/control-plane/pkg/logging"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
 )
 
 const (
@@ -69,6 +73,11 @@ type Reconciler struct {
 	// mock the function used during the reconciliation loop.
 	NewKafkaClient kafka.NewClientFunc
 
+	// InitOffsetsFunc initialize offsets for a provided set of topics and a provided consumer group id.
+	// It's convenient to add this as Reconciler field so that we can mock the function used during the
+	// reconciliation loop.
+	InitOffsetsFunc kafka.InitOffsetsFunc
+
 	ConfigMapLister corelisters.ConfigMapLister
 }
 
@@ -439,7 +448,7 @@ func (r *Reconciler) reconcileInitialOffset(ctx context.Context, channel *messag
 
 	topicName := kafka.ChannelTopic(TopicPrefix, channel)
 	groupID := consumerGroup(channel, sub)
-	_, err := offset.InitOffsets(ctx, kafkaClient, kafkaClusterAdmin, []string{topicName}, groupID)
+	_, err := r.InitOffsetsFunc(ctx, kafkaClient, kafkaClusterAdmin, []string{topicName}, groupID)
 	return err
 }
 
diff --git a/control-plane/pkg/reconciler/channel/controller.go b/control-plane/pkg/reconciler/channel/controller.go
index c86cc490a..40e8e19c2 100644
--- a/control-plane/pkg/reconciler/channel/controller.go
+++ b/control-plane/pkg/reconciler/channel/controller.go
@@ -23,6 +23,7 @@ import (
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/tools/cache"
+	"knative.dev/eventing-kafka/pkg/common/kafka/offset"
 
 	messagingv1beta "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	kafkachannelinformer "knative.dev/eventing-kafka/pkg/client/injection/informers/messaging/v1beta1/kafkachannel"
@@ -42,6 +43,7 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 )
 
+// TODO: no need for a configmap watcher?
 func NewController(ctx context.Context, watcher configmap.Watcher, configs *config.Env) *controller.Impl {
 
 	messagingv1beta.RegisterAlternateKafkaChannelConditionSet(base.IngressConditionSet)
@@ -62,6 +64,7 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *conf
 		},
 		NewKafkaClient:             sarama.NewClient,
 		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
+		InitOffsetsFunc:            offset.InitOffsets,
 		Env:                        configs,
 		ConfigMapLister:            configmapInformer.Lister(),
 	}
diff --git a/go.mod b/go.mod
index bda791563..36e640843 100644
--- a/go.mod
+++ b/go.mod
@@ -24,9 +24,9 @@ require (
 	k8s.io/apiserver v0.21.4
 	k8s.io/client-go v0.21.4
 	k8s.io/utils v0.0.0-20210111153108-fddb29f9d009
-	knative.dev/eventing v0.27.1-0.20211108221643-a72cfc91c421
-	knative.dev/eventing-kafka v0.27.1-0.20211108161601-76680cab3d62
+	knative.dev/eventing v0.27.1-0.20211109172546-31e4e2eb26d1
+	knative.dev/eventing-kafka v0.27.1-0.20211109165044-86d36b670174
 	knative.dev/hack v0.0.0-20211108170701-96aac1c30be3
-	knative.dev/pkg v0.0.0-20211108064904-3cc697a3cb09
+	knative.dev/pkg v0.0.0-20211109100843-91d1932616a7
 	knative.dev/reconciler-test v0.0.0-20211101213739-c5f938aa9974
 )
diff --git a/go.sum b/go.sum
index 77a284b17..1f3079d5e 100644
--- a/go.sum
+++ b/go.sum
@@ -1326,26 +1326,22 @@ k8s.io/kube-openapi v0.0.0-20210305001622-591a79e4bda7/go.mod h1:wXW5VT87nVfh/iL
 k8s.io/utils v0.0.0-20201110183641-67b214c5f920/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 k8s.io/utils v0.0.0-20210111153108-fddb29f9d009 h1:0T5IaWHO3sJTEmCP6mUlBvMukxPKUQWqiI/YuiBNMiQ=
 k8s.io/utils v0.0.0-20210111153108-fddb29f9d009/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
-knative.dev/control-protocol v0.0.0-20211101205041-256f50f9b5a9/go.mod h1:BOIcHyLM9l0pHtMEOcUJo0qRe3Qvrm4p3PXoKhEpeUA=
-knative.dev/eventing v0.27.1-0.20211108073202-ee96d0b3b429/go.mod h1:WsdNr/+tT/IqPeWzZuG3O9/zla/ZFniJC8PgJJWGT58=
-knative.dev/eventing v0.27.1-0.20211108221643-a72cfc91c421 h1:RSA8U3xuGlX7S8ZF8IZKp6t7kbm4jWpBimpslvO0EI8=
+knative.dev/control-protocol v0.0.0-20211108155701-88c4c1e549c4/go.mod h1:5hOr3rdQDEBE/3LihWgyfkfowQn8V0Vur/qg+g42MXY=
 knative.dev/eventing v0.27.1-0.20211108221643-a72cfc91c421/go.mod h1:lM+1rTzjcNmrPjTGaEk+QrmQGuM3/O+e3aN96GH4V0Y=
-knative.dev/eventing-kafka v0.27.1-0.20211108161601-76680cab3d62 h1:+6iEOvrEpUUi1IdH8YjynS9n8FYlFcRGiDE312WiMKs=
-knative.dev/eventing-kafka v0.27.1-0.20211108161601-76680cab3d62/go.mod h1:y1hVCjWPPTjTktZ3x+jTAqYq838QIAMuuipYe8YaDTg=
-knative.dev/hack v0.0.0-20211028194650-b96d65a5ff5e/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
+knative.dev/eventing v0.27.1-0.20211109172546-31e4e2eb26d1 h1:vC/EpcYF1lsvaapxNGmz6vGNZfq94/1Fd770SQTZ9Es=
+knative.dev/eventing v0.27.1-0.20211109172546-31e4e2eb26d1/go.mod h1:zHDfFKGqgq9bZvUd3QC6iHRAaj2AbHercYhlxdtc9fA=
+knative.dev/eventing-kafka v0.27.1-0.20211109165044-86d36b670174 h1:mMZM9bJesUECCHg2kO5VvdOYK9cEIfAg8NNSNPgcPhY=
+knative.dev/eventing-kafka v0.27.1-0.20211109165044-86d36b670174/go.mod h1:6RjAzpeBOW7/fG+2E4d0siS7Zfy0YfxJIC2Qal+3K6M=
 knative.dev/hack v0.0.0-20211101195839-11d193bf617b/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/hack v0.0.0-20211102174540-b284d49386cc/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack v0.0.0-20211105231158-29f86c2653b5/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack v0.0.0-20211108170701-96aac1c30be3 h1:oSvRgnKoU308k7aXbPV3iL5Zh5kBGM2Ptar4hyeda+A=
 knative.dev/hack v0.0.0-20211108170701-96aac1c30be3/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/hack/schema v0.0.0-20211102174540-b284d49386cc/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
 knative.dev/hack/schema v0.0.0-20211105231158-29f86c2653b5/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
-knative.dev/pkg v0.0.0-20211028235650-5d9d300c2e40/go.mod h1:HyEqMTLzT2hGCisGaRMGzEE8b4Ym+7mhmVx5ygiX+fY=
+knative.dev/hack/schema v0.0.0-20211108170701-96aac1c30be3/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
 knative.dev/pkg v0.0.0-20211101212339-96c0204a70dc/go.mod h1:SkfDk9bWIiNZD7XtILGkG7AKVyF/M6M0bGxLgl0SYL8=
-knative.dev/pkg v0.0.0-20211103135647-464034912f7e/go.mod h1:a+9AWOb0zOaBVU6xoPgTvEA7vfMnLjPnK4SX3ciSHKE=
-knative.dev/pkg v0.0.0-20211108064904-3cc697a3cb09 h1:EuziitWfjhtDEI7vyNV79AtqFia5rACAFOy2rbinVo4=
 knative.dev/pkg v0.0.0-20211108064904-3cc697a3cb09/go.mod h1:YiSfFDMdYHZTuO5XS8UzxSeOEmocy+Kb2YOGik9C8lw=
-knative.dev/reconciler-test v0.0.0-20211029073051-cff9b538d33c/go.mod h1:PIwtS0at/rtnsNCxcZUAZUmWKKasTUYJ/w1Mk94bOdA=
+knative.dev/pkg v0.0.0-20211109100843-91d1932616a7 h1:uqo0nLez3NJf03gE9mdYpFxbqZ8iWdVElQa+s8W7VNQ=
+knative.dev/pkg v0.0.0-20211109100843-91d1932616a7/go.mod h1:plYIuKk6WNC9LyiG/9xBnOBYuxO0Ro6xN0z7iXWVUtk=
 knative.dev/reconciler-test v0.0.0-20211101213739-c5f938aa9974 h1:4eo67cEiDvEl1TgITT+3n1ivAcLFCQ6DBJnX855727E=
 knative.dev/reconciler-test v0.0.0-20211101213739-c5f938aa9974/go.mod h1:gTsbLk496j/M9xqMpx/liyCQ0X3bwDpRtcs2Zzws364=
 modernc.org/cc v1.0.0/go.mod h1:1Sk4//wdnYJiUIxnW8ddKpaOJCF37yAdqYnkxUpaYxw=
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 39e92aa45..f7d6da2d4 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1104,7 +1104,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.27.1-0.20211108221643-a72cfc91c421
+# knative.dev/eventing v0.27.1-0.20211109172546-31e4e2eb26d1
 ## explicit
 knative.dev/eventing/pkg/apis/config
 knative.dev/eventing/pkg/apis/duck
@@ -1213,7 +1213,7 @@ knative.dev/eventing/test/test_images/performance
 knative.dev/eventing/test/test_images/print
 knative.dev/eventing/test/test_images/recordevents
 knative.dev/eventing/test/test_images/request-sender
-# knative.dev/eventing-kafka v0.27.1-0.20211108161601-76680cab3d62
+# knative.dev/eventing-kafka v0.27.1-0.20211109165044-86d36b670174
 ## explicit
 knative.dev/eventing-kafka/pkg/apis/bindings
 knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1
@@ -1270,7 +1270,7 @@ knative.dev/eventing-kafka/test/lib/resources
 # knative.dev/hack v0.0.0-20211108170701-96aac1c30be3
 ## explicit
 knative.dev/hack
-# knative.dev/pkg v0.0.0-20211108064904-3cc697a3cb09
+# knative.dev/pkg v0.0.0-20211109100843-91d1932616a7
 ## explicit
 knative.dev/pkg/apis
 knative.dev/pkg/apis/duck


Test patch:
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index 0dbe6043b..4c47235cd 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -116,7 +116,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 			Name: "Reconciled normal - no DLS",
 			Objects: []runtime.Object{
 				NewBroker(),
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
@@ -345,7 +345,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
 				ConfigMapUpdate(&env, &contract.Contract{
@@ -392,7 +392,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 			Name: "Reconciled normal - config map not readable",
 			Objects: []runtime.Object{
 				NewBroker(),
-				NewConfigMap(&env, []byte(`{"hello": "world"}`)),
+				NewConfigMapWithBinaryData(&env, []byte(`{"hello": "world"}`)),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, nil),
 				BrokerDispatcherPod(env.SystemNamespace, map[string]string{
@@ -887,7 +887,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					),
 				),
 				BrokerConfig(bootstrapServers, 20, 5),
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
@@ -959,7 +959,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				),
 				NewSSLSecret(ConfigMapNamespace, "secret-1"),
 				BrokerConfig(bootstrapServers, 20, 5, BrokerAuthConfig("secret-1")),
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, map[string]string{
 					"annotation_to_preserve": "value_to_preserve",
@@ -1741,7 +1741,7 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 			},
 			Key: testKey,
 			WantCreates: []runtime.Object{
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			OtherTestData: map[string]interface{}{
diff --git a/control-plane/pkg/reconciler/channel/channel_test.go b/control-plane/pkg/reconciler/channel/channel_test.go
new file mode 100644
index 000000000..f5ee7e165
--- /dev/null
+++ b/control-plane/pkg/reconciler/channel/channel_test.go
@@ -0,0 +1,242 @@
+/*
+ * 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 channel_test
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/Shopify/sarama"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	clientgotesting "k8s.io/client-go/testing"
+
+	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/logging"
+	. "knative.dev/pkg/reconciler/testing"
+	"knative.dev/pkg/resolver"
+	"knative.dev/pkg/system"
+	"knative.dev/pkg/tracker"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
+	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/channel"
+	kafkatesting "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka/testing"
+	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/testing"
+
+	messagingv1beta "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
+	fakeeventingkafkaclient "knative.dev/eventing-kafka/pkg/client/injection/client/fake"
+	messagingv1beta1kafkachannelreconciler "knative.dev/eventing-kafka/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel"
+	"knative.dev/eventing-kafka/pkg/common/constants"
+)
+
+// TODO: are we using all of these?
+var configKafka = map[string]string{
+	"version": "1.0.0",
+	"sarama": `
+enableLogging: false
+config: |
+  Version: 2.0.0 # Kafka Version Compatibility From Sarama's Supported List (Major.Minor.Patch)
+  Admin:
+    Timeout: 10000000000  # 10 seconds
+  Net:
+    KeepAlive: 30000000000  # 30 seconds
+    MaxOpenRequests: 1 # Set to 1 for use with Idempotent Producer
+    TLS:
+      Enable: true
+    SASL:
+      Enable: true
+      Version: 1
+  Metadata:
+    RefreshFrequency: 300000000000  # 5 minutes
+`,
+	"eventing-kafka": `
+kafka:
+  authSecretName: kafka-cluster
+  authSecretNamespace: knative-eventing
+  brokers: kafka:9092
+`,
+}
+
+const (
+	finalizerName = "kafkachannels.messaging.knative.dev"
+)
+
+var finalizerUpdatedEvent = Eventf(
+	corev1.EventTypeNormal,
+	"FinalizerUpdate",
+	fmt.Sprintf(`Updated %q finalizers`, ChannelName),
+)
+
+var DefaultEnv = &config.Env{
+	DataPlaneConfigMapNamespace: "knative-eventing",
+	DataPlaneConfigMapName:      "kafka-channel-channels-subscriptions",
+	GeneralConfigMapName:        "kafka-broker-config",
+	IngressName:                 "kafka-channel-ingress",
+	SystemNamespace:             "knative-eventing",
+	DataPlaneConfigFormat:       base.Json,
+}
+
+// TODO: tests with and without subscriptions
+// TODO: tests for things from config-kafka
+// TODO: are we setting a InitialOffsetsCommitted status?
+// TODO: test if things from spec is used properly?
+
+func TestReconcileKind(t *testing.T) {
+
+	messagingv1beta.RegisterAlternateKafkaChannelConditionSet(base.IngressConditionSet)
+
+	env := *DefaultEnv
+	testKey := fmt.Sprintf("%s/%s", ChannelNamespace, ChannelName)
+
+	table := TableTest{
+		{
+			Name: "Reconciled normal - no subscription - no auth",
+			Objects: []runtime.Object{
+				NewChannel(),
+				NewService(),
+				ChannelReceiverPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				ChannelDispatcherPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, configKafka),
+			},
+			Key: testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&env, &contract.Contract{
+					Generation: 1,
+					Resources: []*contract.Resource{
+						{
+							Uid:              ChannelUUID,
+							Topics:           []string{ChannelTopic()},
+							BootstrapServers: ChannelBootstrapServers,
+							Ingress: &contract.Ingress{
+								IngressType: &contract.Ingress_Path{
+									Path: receiver.Path(ChannelNamespace, ChannelName),
+								},
+							},
+						},
+					},
+				}),
+				ChannelReceiverPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+				ChannelDispatcherPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+			},
+			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
+			WantCreates: []runtime.Object{
+				NewConfigMapWithBinaryData(&env, nil),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewChannel(
+						WithInitKafkaChannelConditions,
+						StatusConfigParsed,
+						StatusConfigMapUpdatedReady(&env),
+						StatusTopicReadyWithName(ChannelTopic()),
+						StatusDataPlaneAvailable,
+						//StatusInitialOffsetsCommitted,
+						ChannelAddressable(&env),
+					),
+				},
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+		},
+	}
+
+	useTable(t, table, env)
+}
+
+func useTable(t *testing.T, table TableTest, env config.Env) {
+	table.Test(t, NewFactory(&env, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
+
+		reconciler := &Reconciler{
+			Reconciler: &base.Reconciler{
+				KubeClient:                  kubeclient.Get(ctx),
+				PodLister:                   listers.GetPodLister(),
+				SecretLister:                listers.GetSecretLister(),
+				DataPlaneConfigMapNamespace: env.DataPlaneConfigMapNamespace,
+				DataPlaneConfigMapName:      env.DataPlaneConfigMapName,
+				DataPlaneConfigFormat:       env.DataPlaneConfigFormat,
+				SystemNamespace:             env.SystemNamespace,
+				DispatcherLabel:             base.ChannelDispatcherLabel,
+				ReceiverLabel:               base.ChannelReceiverLabel,
+			},
+			Env:             env,
+			ConfigMapLister: listers.GetConfigMapLister(),
+			InitOffsetsFunc: func(ctx context.Context, kafkaClient sarama.Client, kafkaAdminClient sarama.ClusterAdmin, topics []string, consumerGroup string) (int32, error) {
+				return 1, nil
+			},
+			NewKafkaClient: func(addrs []string, config *sarama.Config) (sarama.Client, error) {
+				return &kafkatesting.MockKafkaClient{}, nil
+			},
+			NewKafkaClusterAdminClient: func(_ []string, _ *sarama.Config) (sarama.ClusterAdmin, error) {
+				return &kafkatesting.MockKafkaClusterAdmin{
+					ExpectedTopicName: ChannelTopic(),
+					ExpectedTopicDetail: sarama.TopicDetail{
+						NumPartitions:     1,
+						ReplicationFactor: 1,
+					},
+					T: t,
+				}, nil
+			},
+		}
+
+		reconciler.ConfigMapTracker = &FakeTracker{}
+		reconciler.SecretTracker = &FakeTracker{}
+
+		reconciler.Resolver = resolver.NewURIResolverFromTracker(ctx, tracker.New(func(name types.NamespacedName) {}, 0))
+
+		r := messagingv1beta1kafkachannelreconciler.NewReconciler(
+			ctx,
+			logging.FromContext(ctx),
+			fakeeventingkafkaclient.Get(ctx),
+			listers.GetKafkaChannelLister(),
+			controller.GetEventRecorder(ctx),
+			reconciler,
+		)
+		return r
+	}))
+}
+
+func patchFinalizers() clientgotesting.PatchActionImpl {
+	action := clientgotesting.PatchActionImpl{}
+	action.Name = ChannelName
+	action.Namespace = ChannelNamespace
+	patch := `{"metadata":{"finalizers":["` + finalizerName + `"],"resourceVersion":""}}`
+	action.Patch = []byte(patch)
+	return action
+}
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink_test.go b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
index 5253258c1..f41000721 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink_test.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
@@ -111,7 +111,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 				NewSink(
 					StatusControllerOwnsTopic(reconciler.ControllerTopicOwner),
 				),
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					"annotation_to_preserve": "value_to_preserve",
 				}),
@@ -163,7 +163,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 					SinkAuthSecretRef("secret-1"),
 				),
 				NewSSLSecret(SinkNamespace, "secret-1"),
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
@@ -228,7 +228,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						s.Spec.NumPartitions = nil
 					},
 				),
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
@@ -294,7 +294,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						s.Spec.NumPartitions = nil
 					},
 				),
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 					"annotation_to_preserve":           "value_to_preserve",
@@ -344,7 +344,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						s.Spec.BootstrapServers = []string{"kafka-broker:10000"}
 					},
 				),
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
@@ -457,7 +457,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
 				ConfigMapUpdate(&env, &contract.Contract{
@@ -500,7 +500,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 					StatusControllerOwnsTopic(reconciler.ControllerTopicOwner),
 					BootstrapServers(bootstrapServersArr),
 				),
-				NewConfigMap(&env, []byte(`{"hello": "world"}`)),
+				NewConfigMapWithBinaryData(&env, []byte(`{"hello": "world"}`)),
 				SinkReceiverPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
diff --git a/control-plane/pkg/reconciler/source/source_test.go b/control-plane/pkg/reconciler/source/source_test.go
index 31fe70c7b..094312360 100644
--- a/control-plane/pkg/reconciler/source/source_test.go
+++ b/control-plane/pkg/reconciler/source/source_test.go
@@ -118,7 +118,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -182,7 +182,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -245,7 +245,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -306,7 +306,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -367,7 +367,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -428,7 +428,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
diff --git a/control-plane/pkg/reconciler/testing/listers.go b/control-plane/pkg/reconciler/testing/listers.go
index 3c1ff5d4b..3d3a0ee67 100644
--- a/control-plane/pkg/reconciler/testing/listers.go
+++ b/control-plane/pkg/reconciler/testing/listers.go
@@ -28,8 +28,10 @@ import (
 	eventinglisters "knative.dev/eventing/pkg/client/listers/eventing/v1"
 	"knative.dev/pkg/reconciler/testing"
 
+	eventingkafkachannels "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	eventingkafkasources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
 	fakeeventingkafkaclientset "knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake"
+	eventingkafkachannelslisters "knative.dev/eventing-kafka/pkg/client/listers/messaging/v1beta1"
 	eventingkafkasourceslisters "knative.dev/eventing-kafka/pkg/client/listers/sources/v1beta1"
 
 	eventingkafkabroker "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
@@ -121,6 +123,10 @@ func (l *Listers) GetKafkaSourceLister() eventingkafkasourceslisters.KafkaSource
 	return eventingkafkasourceslisters.NewKafkaSourceLister(l.indexerFor(&eventingkafkasources.KafkaSource{}))
 }
 
+func (l *Listers) GetKafkaChannelLister() eventingkafkachannelslisters.KafkaChannelLister {
+	return eventingkafkachannelslisters.NewKafkaChannelLister(l.indexerFor(&eventingkafkachannels.KafkaChannel{}))
+}
+
 func (l *Listers) indexerFor(obj runtime.Object) cache.Indexer {
 	return l.sorter.IndexerForObjectType(obj)
 }
diff --git a/control-plane/pkg/reconciler/testing/objects_channel.go b/control-plane/pkg/reconciler/testing/objects_channel.go
new file mode 100644
index 000000000..1447f43dc
--- /dev/null
+++ b/control-plane/pkg/reconciler/testing/objects_channel.go
@@ -0,0 +1,149 @@
+/*
+ * 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 testing
+
+import (
+	"context"
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	clientgotesting "k8s.io/client-go/testing"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
+	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/channel"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
+	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/network"
+)
+
+const (
+	ChannelName             = "kc"
+	ChannelNamespace        = "test-nc"
+	ChannelUUID             = "c1234567-8901-2345-6789-123456789101"
+	ChannelBootstrapServers = "kafka:9092"
+)
+
+func ChannelTopic() string {
+	c := NewChannel().(metav1.Object)
+	return kafka.ChannelTopic(TopicPrefix, c)
+}
+
+func NewChannel(options ...KRShapedOption) runtime.Object {
+	c := &messagingv1beta1.KafkaChannel{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: ChannelNamespace,
+			Name:      ChannelName,
+			UID:       ChannelUUID,
+		},
+	}
+	for _, opt := range options {
+		opt(c)
+	}
+	c.SetDefaults(context.Background())
+	return c
+}
+
+func ChannelReceiverPod(namespace string, annotations map[string]string) runtime.Object {
+	return &corev1.Pod{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Pod",
+			APIVersion: "v1",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:        "kafka-channel-receiver",
+			Namespace:   namespace,
+			Annotations: annotations,
+			Labels: map[string]string{
+				"app": base.ChannelReceiverLabel,
+			},
+		},
+		Status: corev1.PodStatus{
+			Phase: corev1.PodRunning,
+		},
+	}
+}
+
+func ChannelDispatcherPod(namespace string, annotations map[string]string) runtime.Object {
+	return &corev1.Pod{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Pod",
+			APIVersion: "v1",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:        "kafka-channel-dispatcher",
+			Namespace:   namespace,
+			Annotations: annotations,
+			Labels: map[string]string{
+				"app": base.ChannelDispatcherLabel,
+			},
+		},
+		Status: corev1.PodStatus{
+			Phase: corev1.PodRunning,
+		},
+	}
+}
+
+func ChannelDispatcherPodUpdate(namespace string, annotations map[string]string) clientgotesting.UpdateActionImpl {
+	return clientgotesting.NewUpdateAction(
+		schema.GroupVersionResource{
+			Group:    "*",
+			Version:  "v1",
+			Resource: "Pod",
+		},
+		namespace,
+		ChannelDispatcherPod(namespace, annotations),
+	)
+}
+
+func ChannelReceiverPodUpdate(namespace string, annotations map[string]string) clientgotesting.UpdateActionImpl {
+	return clientgotesting.NewUpdateAction(
+		schema.GroupVersionResource{
+			Group:    "*",
+			Version:  "v1",
+			Resource: "Pod",
+		},
+		namespace,
+		ChannelReceiverPod(namespace, annotations),
+	)
+}
+
+func ChannelAddressable(env *config.Env) func(obj duckv1.KRShaped) {
+
+	return func(obj duckv1.KRShaped) {
+		channel := obj.(*messagingv1beta1.KafkaChannel)
+
+		channel.Status.Address = &duckv1.Addressable{}
+
+		channel.Status.Address.URL = &apis.URL{
+			Scheme: "http",
+			Host:   network.GetServiceHostname(env.IngressName, env.SystemNamespace),
+			Path:   fmt.Sprintf("/%s/%s", channel.Namespace, channel.Name),
+		}
+
+		channel.GetConditionSet().Manage(&channel.Status).MarkTrue(base.ConditionAddressable)
+	}
+}
+
+func WithInitKafkaChannelConditions(obj duckv1.KRShaped) {
+	channel := obj.(*messagingv1beta1.KafkaChannel)
+	channel.Status.InitializeConditions()
+}
diff --git a/control-plane/pkg/reconciler/testing/objects_common.go b/control-plane/pkg/reconciler/testing/objects_common.go
index c4626e39e..ce0f8af11 100644
--- a/control-plane/pkg/reconciler/testing/objects_common.go
+++ b/control-plane/pkg/reconciler/testing/objects_common.go
@@ -83,7 +83,7 @@ func ServiceURLFrom(ns, name string) string {
 	return fmt.Sprintf("http://%s.%s.svc.cluster.local", name, ns)
 }
 
-func NewConfigMap(env *config.Env, data []byte) runtime.Object {
+func NewConfigMapWithBinaryData(env *config.Env, data []byte) runtime.Object {
 	return reconcilertesting.NewConfigMap(
 		env.DataPlaneConfigMapName,
 		env.DataPlaneConfigMapNamespace,
@@ -99,6 +99,16 @@ func NewConfigMap(env *config.Env, data []byte) runtime.Object {
 	)
 }
 
+func NewConfigMapWithTextData(namespace, name string, data map[string]string) runtime.Object {
+	return reconcilertesting.NewConfigMap(
+		name,
+		namespace,
+		func(configMap *corev1.ConfigMap) {
+			configMap.Data = data
+		},
+	)
+}
+
 func NewConfigMapFromContract(contract *contract.Contract, env *config.Env) runtime.Object {
 	var data []byte
 	var err error
@@ -111,7 +121,7 @@ func NewConfigMapFromContract(contract *contract.Contract, env *config.Env) runt
 		panic(err)
 	}
 
-	return NewConfigMap(env, data)
+	return NewConfigMapWithBinaryData(env, data)
 }
 
 func ConfigMapUpdate(env *config.Env, contract *contract.Contract) clientgotesting.UpdateActionImpl {
diff --git a/control-plane/pkg/reconciler/trigger/trigger_test.go b/control-plane/pkg/reconciler/trigger/trigger_test.go
index abaacb823..45ad6dbdf 100644
--- a/control-plane/pkg/reconciler/trigger/trigger_test.go
+++ b/control-plane/pkg/reconciler/trigger/trigger_test.go
@@ -627,7 +627,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 			WantErr:                 true,
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 			},
 			WantEvents: []string{
 				finalizerUpdatedEvent,
@@ -658,7 +658,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 				NewBroker(BrokerReady),
 				newTrigger(),
 				NewService(),
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
diff --git a/vendor/knative.dev/eventing-kafka/test/lib/creation.go b/vendor/knative.dev/eventing-kafka/test/lib/creation.go
index d667b3686..bb796517d 100644
--- a/vendor/knative.dev/eventing-kafka/test/lib/creation.go
+++ b/vendor/knative.dev/eventing-kafka/test/lib/creation.go
@@ -31,75 +31,91 @@ import (
 )
 
 func CreateKafkaChannelV1Beta1OrFail(c *testlib.Client, kafkaChannel *channelsv1beta1.KafkaChannel) {
-	kafkaChannelClientSet, err := kafkaclientset.NewForConfig(c.Config)
+	client, err := kafkaclientset.NewForConfig(c.Config)
 	if err != nil {
 		c.T.Fatalf("Failed to create v1beta1 KafkaChannel client: %v", err)
 	}
 
-	kChannels := kafkaChannelClientSet.MessagingV1beta1().KafkaChannels(c.Namespace)
-	if createdKafkaChannel, err := kChannels.Create(context.Background(), kafkaChannel, metav1.CreateOptions{}); err != nil {
-		c.T.Fatalf("Failed to create v1beta1 KafkaChannel %q: %v", kafkaChannel.Name, err)
-	} else {
+	err = c.RetryWebhookErrors(func(i int) error {
+		createdKafkaChannel, err := client.MessagingV1beta1().KafkaChannels(c.Namespace).Create(context.Background(), kafkaChannel, metav1.CreateOptions{})
+		if err != nil {
+			return err
+		}
 		c.Tracker.AddObj(createdKafkaChannel)
+		return nil
+	})
+	if err != nil {
+		c.T.Fatalf("Failed to create v1beta1 KafkaChannel %q: %v", kafkaChannel.Name, err)
 	}
 }
 
 func GetKafkaChannelV1Beta1OrFail(c *testlib.Client, kafkaChannel string) *channelsv1beta1.KafkaChannel {
-	kafkaChannelClientSet, err := kafkaclientset.NewForConfig(c.Config)
+	client, err := kafkaclientset.NewForConfig(c.Config)
 	if err != nil {
 		c.T.Fatalf("Failed to create v1beta1 KafkaChannel client: %v", err)
 	}
 
-	kChannels := kafkaChannelClientSet.MessagingV1beta1().KafkaChannels(c.Namespace)
-	if kcObj, err := kChannels.Get(context.Background(), kafkaChannel, metav1.GetOptions{}); err != nil {
+	var kcObj *channelsv1beta1.KafkaChannel
+	err = c.RetryWebhookErrors(func(i int) error {
+		kcObj, err = client.MessagingV1beta1().KafkaChannels(c.Namespace).Get(context.Background(), kafkaChannel, metav1.GetOptions{})
+		return err
+	})
+	if err != nil {
 		c.T.Fatalf("Failed to get v1beta1 KafkaChannel %q: %v", kafkaChannel, err)
-	} else {
-		return kcObj
 	}
-	return nil
+	return kcObj
 }
 
 func CreateKafkaSourceV1Beta1OrFail(c *testlib.Client, kafkaSource *sourcesv1beta1.KafkaSource) {
-	kafkaSourceClientSet, err := kafkaclientset.NewForConfig(c.Config)
+	client, err := kafkaclientset.NewForConfig(c.Config)
 	if err != nil {
 		c.T.Fatalf("Failed to create v1beta1 KafkaSource client: %v", err)
 	}
 
-	kSources := kafkaSourceClientSet.SourcesV1beta1().KafkaSources(c.Namespace)
-	if createdKafkaSource, err := kSources.Create(context.Background(), kafkaSource, metav1.CreateOptions{}); err != nil {
-		c.T.Fatalf("Failed to create v1beta1 KafkaSource %q: %v", kafkaSource.Name, err)
-	} else {
+	err = c.RetryWebhookErrors(func(i int) error {
+		createdKafkaSource, err := client.SourcesV1beta1().KafkaSources(c.Namespace).Create(context.Background(), kafkaSource, metav1.CreateOptions{})
+		if err != nil {
+			return err
+		}
 		c.Tracker.AddObj(createdKafkaSource)
+		return nil
+	})
+	if err != nil {
+		c.T.Fatalf("Failed to create v1beta1 KafkaSource %q: %v", kafkaSource.Name, err)
 	}
 }
 
 func GetKafkaSourceV1Beta1OrFail(c *testlib.Client, kafkaSource string) *sourcesv1beta1.KafkaSource {
-	kafkaSourceClientSet, err := kafkaclientset.NewForConfig(c.Config)
+	client, err := kafkaclientset.NewForConfig(c.Config)
 	if err != nil {
 		c.T.Fatalf("Failed to create v1beta1 KafkaSource client: %v", err)
 	}
 
-	kSources := kafkaSourceClientSet.SourcesV1beta1().KafkaSources(c.Namespace)
-	if ksObj, err := kSources.Get(context.Background(), kafkaSource, metav1.GetOptions{}); err != nil {
+	var ksObj *sourcesv1beta1.KafkaSource
+	err = c.RetryWebhookErrors(func(i int) error {
+		ksObj, err = client.SourcesV1beta1().KafkaSources(c.Namespace).Get(context.Background(), kafkaSource, metav1.GetOptions{})
+		return err
+	})
+	if err != nil {
 		c.T.Fatalf("Failed to get v1beta1 KafkaSource %q: %v", kafkaSource, err)
-	} else {
-		return ksObj
 	}
-	return nil
+	return ksObj
 }
 
 func UpdateKafkaSourceV1Beta1OrFail(c *testlib.Client, kafkaSource *sourcesv1beta1.KafkaSource) {
-	err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
-		latestKafkaSource := GetKafkaSourceV1Beta1OrFail(c, kafkaSource.Name)
-		kafkaSource.Spec.DeepCopyInto(&latestKafkaSource.Spec)
-		kafkaSourceClientSet, err := kafkaclientset.NewForConfig(c.Config)
-		if err != nil {
-			c.T.Fatalf("Failed to create v1beta1 KafkaSource client: %v", err)
-		}
-
-		kSources := kafkaSourceClientSet.SourcesV1beta1().KafkaSources(c.Namespace)
-		_, err = kSources.Update(context.Background(), latestKafkaSource, metav1.UpdateOptions{})
-		return err
+	err := c.RetryWebhookErrors(func(i int) error {
+		return retry.RetryOnConflict(retry.DefaultRetry, func() error {
+			latestKafkaSource := GetKafkaSourceV1Beta1OrFail(c, kafkaSource.Name)
+			kafkaSource.Spec.DeepCopyInto(&latestKafkaSource.Spec)
+			kafkaSourceClientSet, err := kafkaclientset.NewForConfig(c.Config)
+			if err != nil {
+				c.T.Fatalf("Failed to create v1beta1 KafkaSource client: %v", err)
+			}
+
+			kSources := kafkaSourceClientSet.SourcesV1beta1().KafkaSources(c.Namespace)
+			_, err = kSources.Update(context.Background(), latestKafkaSource, metav1.UpdateOptions{})
+			return err
+		})
 	})
 	if err != nil {
 		c.T.Fatalf("Failed to update v1beta1 KafkaSource %q: %v", kafkaSource.Name, err)
diff --git a/vendor/knative.dev/eventing/test/rekt/features/broker/readyness.go b/vendor/knative.dev/eventing/test/rekt/features/broker/readyness.go
index e97c38adf..bc5811242 100644
--- a/vendor/knative.dev/eventing/test/rekt/features/broker/readyness.go
+++ b/vendor/knative.dev/eventing/test/rekt/features/broker/readyness.go
@@ -19,10 +19,12 @@ package broker
 import (
 	"fmt"
 
+	"knative.dev/reconciler-test/pkg/eventshub"
 	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/manifest"
 
 	"knative.dev/eventing/test/rekt/resources/broker"
+	"knative.dev/eventing/test/rekt/resources/delivery"
 	"knative.dev/eventing/test/rekt/resources/trigger"
 	"knative.dev/reconciler-test/resources/svc"
 )
@@ -64,3 +66,22 @@ func GoesReady(name string, cfg ...manifest.CfgFn) *feature.Feature {
 
 	return f
 }
+
+// GoesReady returns a feature that will create a Broker of the given
+// name and class, with an event receiver as its DLS
+// and confirm it becomes ready with an address.
+func GoesReadyWithProbeReceiver(name, sinkName string, prober *eventshub.EventProber, cfg ...manifest.CfgFn) *feature.Feature {
+	f := new(feature.Feature)
+
+	f.Setup("install probe", prober.ReceiverInstall(sinkName))
+
+	brokerConfig := append(cfg, delivery.WithDeadLetterSink(prober.AsKReference(sinkName), ""))
+	f.Setup(fmt.Sprintf("install broker %q", name), broker.Install(name, brokerConfig...))
+
+	f.Requirement("Broker is ready", broker.IsReady(name))
+
+	f.Stable("broker").
+		Must("be addressable", broker.IsAddressable(name))
+
+	return f
+}
diff --git a/vendor/knative.dev/eventing/test/rekt/features/broker/source_to_sink.go b/vendor/knative.dev/eventing/test/rekt/features/broker/source_to_sink.go
index e287cba97..40907d43e 100644
--- a/vendor/knative.dev/eventing/test/rekt/features/broker/source_to_sink.go
+++ b/vendor/knative.dev/eventing/test/rekt/features/broker/source_to_sink.go
@@ -22,7 +22,6 @@ import (
 
 	"github.com/google/uuid"
 	"knative.dev/eventing/test/rekt/resources/broker"
-	"knative.dev/eventing/test/rekt/resources/delivery"
 	"knative.dev/eventing/test/rekt/resources/eventlibrary"
 	"knative.dev/eventing/test/rekt/resources/flaker"
 	"knative.dev/eventing/test/rekt/resources/trigger"
@@ -32,8 +31,6 @@ import (
 	"knative.dev/reconciler-test/pkg/manifest"
 	"knative.dev/reconciler-test/resources/svc"
 
-	b "knative.dev/eventing/test/rekt/resources/broker"
-
 	. "github.com/cloudevents/sdk-go/v2/test"
 	. "knative.dev/reconciler-test/pkg/eventshub/assert"
 )
@@ -79,8 +76,7 @@ func SourceToSink(brokerName string) *feature.Feature {
 //                |
 //                +--[DLQ]--> sink
 //
-func SourceToSinkWithDLQ(brokerName string) *feature.Feature {
-	prober := eventshub.NewProber()
+func SourceToSinkWithDLQ(brokerName, sinkName string, prober *eventshub.EventProber) *feature.Feature {
 	prober.SetTargetResource(broker.GVR(), brokerName)
 
 	via := feature.MakeRandomK8sName("via")
@@ -95,15 +91,6 @@ func SourceToSinkWithDLQ(brokerName string) *feature.Feature {
 		panic(fmt.Errorf("can not find event files: %s", err))
 	}
 
-	// Setup Probes
-	f.Setup("install recorder", prober.ReceiverInstall("sink"))
-
-	// Setup data plane
-	brokerConfig := append(b.WithEnvConfig(), delivery.WithDeadLetterSink(prober.AsKReference("sink"), ""))
-	f.Setup("update broker with DLQ", broker.Install(
-		brokerName,
-		brokerConfig...,
-	))
 	f.Setup("install trigger", trigger.Install(via, brokerName, trigger.WithSubscriber(nil, "bad://uri")))
 
 	// Resources ready.
@@ -118,7 +105,7 @@ func SourceToSinkWithDLQ(brokerName string) *feature.Feature {
 	// Assert events ended up where we expected.
 	f.Stable("broker with DLQ").
 		Must("accepted all events", prober.AssertSentAll("source")).
-		Must("deliver event to DLQ", prober.AssertReceivedAll("source", "sink"))
+		Must("deliver event to DLQ", prober.AssertReceivedAll("source", sinkName))
 
 	return f
 }
@@ -131,8 +118,7 @@ func SourceToSinkWithDLQ(brokerName string) *feature.Feature {
 //                |
 //                +--[DLQ]--> sink1
 //
-func SourceToTwoSinksWithDLQ(brokerName string) *feature.Feature {
-	prober := eventshub.NewProber()
+func SourceToTwoSinksWithDLQ(brokerName, sink1, sink2 string, prober *eventshub.EventProber) *feature.Feature {
 	prober.SetTargetResource(broker.GVR(), brokerName)
 
 	via1 := feature.MakeRandomK8sName("via")
@@ -149,13 +135,11 @@ func SourceToTwoSinksWithDLQ(brokerName string) *feature.Feature {
 	}
 
 	// Setup Probes
-	f.Setup("install recorder1", prober.ReceiverInstall("sink1"))
-	f.Setup("install recorder2", prober.ReceiverInstall("sink2"))
+	f.Setup("install recorder2", prober.ReceiverInstall(sink2))
 
 	// Setup data plane
-	f.Setup("update broker with DLQ", broker.Install(brokerName, delivery.WithDeadLetterSink(prober.AsKReference("sink1"), "")))
 	f.Setup("install trigger via1", trigger.Install(via1, brokerName, trigger.WithSubscriber(nil, "bad://uri")))
-	f.Setup("install trigger via2", trigger.Install(via2, brokerName, trigger.WithSubscriber(prober.AsKReference("sink2"), "")))
+	f.Setup("install trigger via2", trigger.Install(via2, brokerName, trigger.WithSubscriber(prober.AsKReference(sink2), "")))
 
 	// Resources ready.
 	f.Setup("trigger1 goes ready", trigger.IsReady(via1))
@@ -166,14 +150,14 @@ func SourceToTwoSinksWithDLQ(brokerName string) *feature.Feature {
 
 	// After we have finished sending.
 	f.Requirement("sender is finished", prober.SenderDone("source"))
-	f.Requirement("receiver 1 is finished", prober.ReceiverDone("source", "sink1"))
-	f.Requirement("receiver 2 is finished", prober.ReceiverDone("source", "sink2"))
+	f.Requirement("receiver 1 is finished", prober.ReceiverDone("source", sink1))
+	f.Requirement("receiver 2 is finished", prober.ReceiverDone("source", sink2))
 
 	// Assert events ended up where we expected.
 	f.Stable("broker with DLQ").
 		Must("accepted all events", prober.AssertSentAll("source")).
-		Must("deliver event to DLQ (via1)", prober.AssertReceivedAll("source", "sink1")).
-		Must("deliver event to sink (via2)", prober.AssertReceivedAll("source", "sink2"))
+		Must("deliver event to DLQ (via1)", prober.AssertReceivedAll("source", sink1)).
+		Must("deliver event to sink (via2)", prober.AssertReceivedAll("source", sink2))
 
 	return f
 }
diff --git a/vendor/knative.dev/eventing/test/rekt/features/trigger/trigger_sink_resolution.go b/vendor/knative.dev/eventing/test/rekt/features/trigger/trigger_sink_resolution.go
new file mode 100644
index 000000000..6a5e3677d
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/rekt/features/trigger/trigger_sink_resolution.go
@@ -0,0 +1,143 @@
+/*
+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 trigger
+
+import (
+	"context"
+	"fmt"
+
+	"knative.dev/eventing/test/rekt/resources/broker"
+	"knative.dev/eventing/test/rekt/resources/delivery"
+	"knative.dev/eventing/test/rekt/resources/eventlibrary"
+	"knative.dev/eventing/test/rekt/resources/trigger"
+	"knative.dev/reconciler-test/pkg/eventshub"
+	"knative.dev/reconciler-test/pkg/feature"
+)
+
+// SourceToTriggerSinkWithDLS tests to see if a Ready Trigger with a DLS defined send
+// failing events to it's DLS.
+//
+// source ---> broker --[trigger]--> bad uri
+//                          |
+//                          +--[DLS]--> sink
+//
+func SourceToTriggerSinkWithDLS(triggerName, brokerName string, prober *eventshub.EventProber) *feature.Feature {
+	prober.SetTargetResource(broker.GVR(), brokerName)
+	f := feature.NewFeature()
+
+	lib := feature.MakeRandomK8sName("lib")
+	f.Setup("install events", eventlibrary.Install(lib))
+	f.Setup("event cache is ready", eventlibrary.IsReady(lib))
+	f.Setup("use events cache", prober.SenderEventsFromSVC(lib, "events/three.ce"))
+	if err := prober.ExpectYAMLEvents(eventlibrary.PathFor("events/three.ce")); err != nil {
+		panic(fmt.Errorf("can not find event files: %s", err))
+	}
+
+	// Setup Probes
+	f.Setup("install recorder", prober.ReceiverInstall("sink"))
+
+	// Setup trigger
+	f.Setup("install trigger", trigger.Install(
+		triggerName,
+		brokerName,
+		trigger.WithSubscriber(nil, "bad://uri"),
+		delivery.WithDeadLetterSink(prober.AsKReference("sink"), "")))
+
+	// Resources ready.
+	f.Setup("trigger goes ready", trigger.IsReady(triggerName))
+
+	// Install sender.
+	f.Setup("install source", prober.SenderInstall("source"))
+
+	// After we have finished sending.
+	f.Requirement("sender is finished", prober.SenderDone("source"))
+
+	// Assert events ended up where we expected.
+	f.Stable("trigger with DLS").
+		Must("accepted all events", prober.AssertSentAll("source")).
+		Must("deliver event to DLS", prober.AssertReceivedAll("source", "sink"))
+
+	return f
+}
+
+// SourceToTriggerSinkWithDLSDontUseBrokers tests to see if a Ready Trigger sends
+// failing events to it's DLS even when it's corresponding Ready Broker also have a DLS defined.
+//
+// source ---> broker --[trigger]--> bad uri
+//               |          |
+//               +--[DLS]   +--[DLS]--> sink
+//
+func SourceToTriggerSinkWithDLSDontUseBrokers(triggerName, brokerName, brokerSinkName string, prober *eventshub.EventProber) *feature.Feature {
+	triggerSinkName := "trigger-sink"
+	prober.SetTargetResource(broker.GVR(), brokerName)
+
+	f := feature.NewFeature()
+
+	lib := feature.MakeRandomK8sName("lib")
+	f.Setup("install events", eventlibrary.Install(lib))
+	f.Setup("event cache is ready", eventlibrary.IsReady(lib))
+	f.Setup("use events cache", prober.SenderEventsFromSVC(lib, "events/three.ce"))
+	if err := prober.ExpectYAMLEvents(eventlibrary.PathFor("events/three.ce")); err != nil {
+		panic(fmt.Errorf("can not find event files: %s", err))
+	}
+
+	// Setup Probes
+	f.Setup("install trigger recorder", prober.ReceiverInstall(triggerSinkName))
+	f.Setup("install brokers recorder", prober.ReceiverInstall(brokerSinkName))
+
+	// Setup topology
+	brokerConfig := append(
+		broker.WithEnvConfig(),
+		delivery.WithDeadLetterSink(prober.AsKReference(brokerSinkName), ""))
+	f.Setup("install broker with DLS", broker.Install(
+		brokerName,
+		brokerConfig...,
+	))
+
+	f.Setup("install trigger", trigger.Install(
+		triggerName,
+		brokerName,
+		trigger.WithSubscriber(nil, "bad://uri"),
+		delivery.WithDeadLetterSink(prober.AsKReference(triggerSinkName), "")))
+
+	// Resources ready.
+	f.Setup("trigger goes ready", trigger.IsReady(triggerName))
+
+	// Install events after topology is ready.
+	f.Setup("install source", prober.SenderInstall("source"))
+
+	// After we have finished sending.
+	f.Requirement("sender is finished", prober.SenderDone("source"))
+
+	// Assert events ended up where we expected.
+	f.Stable("trigger with a valid DLS ref").
+		Must("accept all events", prober.AssertSentAll("source")).
+		Must("deliver events to trigger DLS", prober.AssertReceivedAll("source", triggerSinkName)).
+		Must("not deliver events to its broker DLS", noEventsToDLS(prober, brokerSinkName))
+
+	return f
+}
+
+func noEventsToDLS(prober *eventshub.EventProber, sinkName string) feature.StepFn {
+	return func(ctx context.Context, t feature.T) {
+		if len(prober.ReceivedBy(ctx, sinkName)) == 0 {
+			t.Log("no events were sent to %s DLS", sinkName)
+		} else {
+			t.Errorf("events were received by %s DLS", sinkName)
+		}
+	}
+}