Repository: knative-sandbox/eventing-kafka
Pull request #319 (2021-01-20 03:20:48)

[] Upgrade to latest dependencies
Cron. -knative-prow-robot

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

Produced via:
```shell
./hack/update-deps.sh --upgrade --release v0.21
./hack/update-codegen.sh
```

Details:
```
bumping knative.dev/pkg f0ea5e6...1bbf0a6:
  > 1bbf0a6 add LabelSelector to InformerFactory so that the informers only list and watch resources with the given labels (# 1940)
  > 75d66b5 preserve the stable webhook logger names we had before (# 1994)
  > 22f4898 upgrade to latest dependencies (# 1990)
  > 5bb97df fix duration logging (# 1992)
  > a74906c Use structured logging to augment our logger vs. naming (# 1991)
bumping knative.dev/eventing 1c94ef8...5dc363d:
  > 5dc363d [# 4744] Update conformance test helper (# 4756)
  > 764685c upgrade to latest dependencies (# 4761)
  > b10d720 [# 4744] Bump test zipkin version (# 4755)
  > 2b95f45 upgrade to latest dependencies (# 4747)
  > d395f6f Cleanup channel template internal types (# 4749)
  > 8e0e286 :minidisc: Change KIND's k8s to 1.19.4 and add 1.20.0 (# 4729)
  > 0f2f567 Format markdown (# 4746)
```

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/pull/319#" 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 **slinkydeveloper** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @slinkydeveloper` 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).

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

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka/blob/master/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":["slinkydeveloper"]} -->
knative-prow-robot: @knative-automation: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | f699a313f82a6b08b45323100a43ea6905589ebb | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/319/pull-knative-sandbox-eventing-kafka-unit-tests/1351731393314426880) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/319). [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:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | f699a313f82a6b08b45323100a43ea6905589ebb | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/319/pull-knative-sandbox-eventing-kafka-unit-tests/1351731393314426880) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | f699a313f82a6b08b45323100a43ea6905589ebb | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/319/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1351731393213763584) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/319). [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:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | f699a313f82a6b08b45323100a43ea6905589ebb | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/319/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1351731393213763584) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-unit-tests | 41c94bde5fcc1db28b02c2f1607cd95574d9ee74 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/319/pull-knative-sandbox-eventing-kafka-unit-tests/1351898122489434112) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/319). [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 -->

Patch:
diff --git a/.github/workflows/knative-style.yaml b/.github/workflows/knative-style.yaml
index 3c7f28dde..b60a1d861 100644
--- a/.github/workflows/knative-style.yaml
+++ b/.github/workflows/knative-style.yaml
@@ -141,7 +141,12 @@ jobs:
           echo '::group:: Running github.com/client9/misspell with reviewdog 🐶 ...'
           # Don't fail because of misspell
           set +o pipefail
-          find . -type f -not -path './vendor/*' -not -path './third_party/*' -not -path './.git/*' |
+          # Exclude generated and vendored files, plus some legacy
+          # paths until we update all .gitattributes
+          git ls-files |
+          git check-attr --stdin linguist-generated | grep -Ev ': (set|true)$' | cut -d: -f1 |
+          git check-attr --stdin linguist-vendored | grep -Ev ': (set|true)$' | cut -d: -f1 |
+          grep -Ev '^(vendor/|third_party/|.git)' |
           xargs misspell -error |
           reviewdog -efm="%f:%l:%c: %m" \
                 -name="github.com/client9/misspell" \
@@ -164,7 +169,13 @@ jobs:
           echo '::group:: Flagging trailing whitespace with reviewdog 🐶 ...'
           # Don't fail because of grep
           set +o pipefail
-          find . -type f -not -path './vendor/*' -not -path './third_party/*' -not -path './.git/*' |
+          
+          # Exclude generated and vendored files, plus some legacy
+          # paths until we update all .gitattributes
+          git ls-files |
+          git check-attr --stdin linguist-generated | grep -Ev ': (set|true)$' | cut -d: -f1 |
+          git check-attr --stdin linguist-vendored | grep -Ev ': (set|true)$' | cut -d: -f1 |
+          grep -Ev '^(vendor/|third_party/|.git)' |
           xargs grep -nE " +$" |
           reviewdog -efm="%f:%l:%m" \
                 -name="trailing whitespace" \
@@ -192,7 +203,13 @@ jobs:
           #  - nothing in third_party
           #  - nothing in .git/
           #  - no *.ai (Adobe Illustrator) files.
-          for x in $(find . -type f -not -name '*.ai' -not -path './vendor/*' -not -path './third_party/*' -not -path './.git/*'); do
+          LINT_FILES=$(git ls-files |
+          git check-attr --stdin linguist-generated | grep -Ev ': (set|true)$' | cut -d: -f1 |
+          git check-attr --stdin linguist-vendored | grep -Ev ': (set|true)$' | cut -d: -f1 |
+          grep -Ev '^(vendor/|third_party/|.git)' |
+          grep -v '\.ai$')
+
+          for x in $LINT_FILES; do
             # Based on https://stackoverflow.com/questions/34943632/linux-check-if-there-is-an-empty-line-at-the-end-of-a-file
             if [[ -f $x && ! ( -s "$x" && -z "$(tail -c 1 $x)" ) ]]; then
               # We add 1 to `wc -l` here because of this limitation (from the man page):
diff --git a/cmd/channel/distributed/dispatcher/main.go b/cmd/channel/distributed/dispatcher/main.go
index 414ef2889..145a97456 100644
--- a/cmd/channel/distributed/dispatcher/main.go
+++ b/cmd/channel/distributed/dispatcher/main.go
@@ -17,6 +17,7 @@ limitations under the License.
 package main
 
 import (
+	"context"
 	"flag"
 	"strconv"
 	"strings"
@@ -193,7 +194,8 @@ func flush(logger *zap.Logger) {
 }
 
 // configMapObserver is the callback function that handles changes to our ConfigMap
-func configMapObserver(logger *zap.SugaredLogger, configMap *corev1.ConfigMap) {
+func configMapObserver(ctx context.Context, configMap *corev1.ConfigMap) {
+	logger := logging.FromContext(ctx)
 
 	if configMap == nil {
 		logger.Warn("Nil ConfigMap passed to configMapObserver; ignoring")
@@ -207,7 +209,7 @@ func configMapObserver(logger *zap.SugaredLogger, configMap *corev1.ConfigMap) {
 	}
 
 	// Toss the new config map to the dispatcher for inspection and action
-	newDispatcher := dispatcher.ConfigChanged(configMap)
+	newDispatcher := dispatcher.ConfigChanged(ctx, configMap)
 	if newDispatcher != nil {
 		// The configuration change caused a new dispatcher to be created, so switch to that one
 		dispatcher = newDispatcher
diff --git a/cmd/channel/distributed/receiver/main.go b/cmd/channel/distributed/receiver/main.go
index 64265ffd9..1243141cf 100644
--- a/cmd/channel/distributed/receiver/main.go
+++ b/cmd/channel/distributed/receiver/main.go
@@ -198,7 +198,8 @@ func handleMessage(ctx context.Context, channelReference eventingchannel.Channel
 }
 
 // configMapObserver is the callback function that handles changes to our ConfigMap
-func configMapObserver(logger *zap.SugaredLogger, configMap *corev1.ConfigMap) {
+func configMapObserver(ctx context.Context, configMap *corev1.ConfigMap) {
+	logger := logging.FromContext(ctx)
 
 	if configMap == nil {
 		logger.Warn("Nil ConfigMap passed to configMapObserver; ignoring")
@@ -211,7 +212,7 @@ func configMapObserver(logger *zap.SugaredLogger, configMap *corev1.ConfigMap) {
 	}
 
 	// Toss the new config map to the producer for inspection and action
-	newProducer := kafkaProducer.ConfigChanged(configMap)
+	newProducer := kafkaProducer.ConfigChanged(ctx, configMap)
 	if newProducer != nil {
 		// The configuration change caused a new producer to be created, so switch to that one
 		logger.Info("Producer Reconfigured; Switching To New Producer")
diff --git a/go.mod b/go.mod
index 008ec1854..d29100ff8 100644
--- a/go.mod
+++ b/go.mod
@@ -32,9 +32,9 @@ require (
 	k8s.io/apimachinery v0.18.12
 	k8s.io/client-go v11.0.1-0.20190805182717-6502b5e7b1b5+incompatible
 	k8s.io/utils v0.0.0-20200603063816-c1c6865ac451
-	knative.dev/eventing v0.20.1-0.20210115015220-1c94ef83b6b5
+	knative.dev/eventing v0.20.1-0.20210120133253-5dc363dcb5be
 	knative.dev/hack v0.0.0-20210114150620-4422dcadb3c8
-	knative.dev/pkg v0.0.0-20210114223020-f0ea5e6b9c4e
+	knative.dev/pkg v0.0.0-20210119162123-1bbf0a6436c3
 )
 
 replace (
diff --git a/go.sum b/go.sum
index 801e16600..ff37f5edb 100644
--- a/go.sum
+++ b/go.sum
@@ -1244,21 +1244,17 @@ k8s.io/kube-openapi v0.0.0-20200410145947-bcb3869e6f29/go.mod h1:F+5wygcW0wmRTnM
 k8s.io/utils v0.0.0-20200324210504-a9aa75ae1b89/go.mod h1:sZAwmy6armz5eXlNoLmJcl4F1QuKu7sr+mFQ0byX7Ew=
 k8s.io/utils v0.0.0-20200603063816-c1c6865ac451 h1:v8ud2Up6QK1lNOKFgiIVrZdMg7MpmSnvtrOieolJKoE=
 k8s.io/utils v0.0.0-20200603063816-c1c6865ac451/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
-knative.dev/eventing v0.20.1-0.20210115015220-1c94ef83b6b5 h1:Gb18mayz8CukHpC6SFXUVDbd2YVdgrkUbJ50zdNX140=
-knative.dev/eventing v0.20.1-0.20210115015220-1c94ef83b6b5/go.mod h1:HdS7UgtDbKQuCuEwgS3+jxtHvpm+PPp/eEsjaQwfWl8=
+knative.dev/eventing v0.20.1-0.20210120133253-5dc363dcb5be h1:djUnZoRaf177Bl4Hyst7H6tz3bDW02cwIi8CNZjDdvg=
+knative.dev/eventing v0.20.1-0.20210120133253-5dc363dcb5be/go.mod h1:oOTcsXCXgKH2YbsPuF+eFdxXfNI3H7jSOaUKsbdyTSs=
 knative.dev/hack v0.0.0-20201214230143-4ed1ecb8db24 h1:kIztWfvnIFV8Lhlea02K3YO2mIzcDyQNzrBLn0Oq9sA=
 knative.dev/hack v0.0.0-20201214230143-4ed1ecb8db24/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/hack v0.0.0-20210112093330-d946d2557383 h1:YDYKfHaplhMelgOVP8eRvHlYf5elgZveo36GBDFfDqA=
-knative.dev/hack v0.0.0-20210112093330-d946d2557383/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack v0.0.0-20210114150620-4422dcadb3c8 h1:N/Kt6C5zZzTtFcTntRThB+vEJ7WcLKaZ5Q2Jm7b23vw=
 knative.dev/hack v0.0.0-20210114150620-4422dcadb3c8/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/pkg v0.0.0-20210107022335-51c72e24c179 h1:lkrgrv69iUk2qhOG9symy15kJUaJZmMybSloi7C3gIw=
-knative.dev/pkg v0.0.0-20210107022335-51c72e24c179/go.mod h1:hckgW978SdzPA2H5EDvRPY8xsnPuDZLJLbPf8Jte7Q0=
-knative.dev/pkg v0.0.0-20210112143930-acbf2af596cf h1:6/VyHMk6pa395xWho0KLfVj+7NNrmlO7Ki9jLbVjoaI=
-knative.dev/pkg v0.0.0-20210112143930-acbf2af596cf/go.mod h1:hckgW978SdzPA2H5EDvRPY8xsnPuDZLJLbPf8Jte7Q0=
 knative.dev/pkg v0.0.0-20210114223020-f0ea5e6b9c4e h1:3k5tzvlM9VGZFiXRj8UKc3CUpMGpqBlEbIY0Dp3F3NU=
 knative.dev/pkg v0.0.0-20210114223020-f0ea5e6b9c4e/go.mod h1:hckgW978SdzPA2H5EDvRPY8xsnPuDZLJLbPf8Jte7Q0=
-knative.dev/reconciler-test v0.0.0-20210111135237-9b251467ccb6/go.mod h1:rmQpZseeqDpg6/ToFzIeV5hTRkOJujaXBCK7iYL7M4E=
+knative.dev/pkg v0.0.0-20210119162123-1bbf0a6436c3 h1:kvBQKadSBqhTMveExPsvlkTWNS8yQcvxes6PS5RxDQs=
+knative.dev/pkg v0.0.0-20210119162123-1bbf0a6436c3/go.mod h1:cZdMjcJE6JGSNaEypgbUigX1TjteMIwQsW2woNBPVCA=
+knative.dev/reconciler-test v0.0.0-20210115075620-6fe6ef693370/go.mod h1:A5ZaQo+1lWGw2OoSc06wSyReX6huzBPJ2ra/R48s1yo=
 pgregory.net/rapid v0.3.3 h1:jCjBsY4ln4Atz78QoBWxUEvAHaFyNDQg9+WU62aCn1U=
 pgregory.net/rapid v0.3.3/go.mod h1:UYpPVyjFHzYBGHIxLFoupi8vwk6rXNzRY9OMvVxFIOU=
 rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8=
diff --git a/pkg/channel/consolidated/README.md b/pkg/channel/consolidated/README.md
index 46a13257b..da453612d 100644
--- a/pkg/channel/consolidated/README.md
+++ b/pkg/channel/consolidated/README.md
@@ -140,3 +140,73 @@ kubectl get deployment -n <YOUR_NAMESPACE> kafka-ch-dispatcher
 
 Both cluster-scoped and namespace-scoped dispatcher can coexist. However once
 the annotation is set (or not set), its value is immutable.
+
+### Configuring Kafka client, Sarama
+
+You can configure the Sarama instance used in the KafkaChannel by defining a
+`sarama` field inside the `config-kafka` configmap.
+
+```yaml
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: config-kafka
+  namespace: knative-eventing
+data:
+  bootstrapServers: ...
+  ...
+  sarama: |
+    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
+    Metadata:
+      RefreshFrequency: 300000000000  # 5 minutes
+    Consumer:
+      Offsets:
+        AutoCommit:
+          Interval: 5000000000  # 5 seconds
+        Retention: 604800000000000  # 1 week
+    Producer:
+      Idempotent: true  # Must be false for Azure EventHubs
+      RequiredAcks: -1  # -1 = WaitForAll, Most stringent option for "at-least-once" delivery.
+```
+
+Settings defined here are used as the defaults by the KafkaChannel. The additional settings defined in the
+channel CR, such as authentication, are applied on top of these defaults.
+
+Also, some Sarama settings are required for the channel to work, such as `Consumer.Return.Errors`
+and `Producer.Return.Successes`, so the value for these in the `config-kafka` is ignored.
+
+Value of the `sarama` key must be valid YAML string. The string is marshalled into a
+[Sarama config struct](https://github.com/Shopify/sarama/blob/master/config.go), with a few
+exceptions (`Version` and certificates).
+
+To specify a certificate, you can use the following format
+where you should make sure to use the YAML string syntax of "|-" in order to
+prevent trailing linefeed. The indentation of the PEM content is also important
+and must be aligned as shown.
+
+```yaml
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: config-kafka
+  namespace: knative-eventing
+data:
+  sarama: |
+    ...
+    Net:
+      TLS:
+        ...
+        Config:
+          RootCaPems:
+          - |-
+            -----BEGIN CERTIFICATE-----
+            MIIGBDCCA+ygAwIBAgIJAKi1aEV58cQ1MA0GCSqGSIb3DQEBCwUAMIGOMQswCQYD
+            ...
+            2wk9rLRZaQnhspt6MhlmU0qkaEZpYND3emR2XZ07m51jXqDUgTjXYCSggImUsARs
+            NAehp9bMeco=
+            -----END CERTIFICATE-----
+```
diff --git a/pkg/channel/consolidated/dispatcher/dispatcher.go b/pkg/channel/consolidated/dispatcher/dispatcher.go
index 011d9a914..7298c3529 100644
--- a/pkg/channel/consolidated/dispatcher/dispatcher.go
+++ b/pkg/channel/consolidated/dispatcher/dispatcher.go
@@ -90,10 +90,10 @@ func (sub Subscription) String() string {
 func NewDispatcher(ctx context.Context, args *KafkaDispatcherArgs) (*KafkaDispatcher, error) {
 	conf, err := client.NewConfigBuilder().
 		WithClientId(args.ClientID).
-		WithVersion(&sarama.V2_0_0_0).
 		WithDefaults().
+		FromYaml(args.SaramaSettingsYamlString).
 		WithAuth(args.KafkaAuthConfig).
-		Build()
+		Build(ctx)
 
 	if err != nil {
 		return nil, fmt.Errorf("Error updating the Sarama Auth config: %w", err)
@@ -163,12 +163,13 @@ func NewDispatcher(ctx context.Context, args *KafkaDispatcherArgs) (*KafkaDispat
 type TopicFunc func(separator, namespace, name string) string
 
 type KafkaDispatcherArgs struct {
-	KnCEConnectionArgs *kncloudevents.ConnectionArgs
-	ClientID           string
-	Brokers            []string
-	KafkaAuthConfig    *client.KafkaAuthConfig
-	TopicFunc          TopicFunc
-	Logger             *zap.SugaredLogger
+	KnCEConnectionArgs       *kncloudevents.ConnectionArgs
+	ClientID                 string
+	Brokers                  []string
+	KafkaAuthConfig          *client.KafkaAuthConfig
+	SaramaSettingsYamlString string
+	TopicFunc                TopicFunc
+	Logger                   *zap.SugaredLogger
 }
 
 type consumerMessageHandler struct {
diff --git a/pkg/channel/consolidated/reconciler/controller/kafkachannel.go b/pkg/channel/consolidated/reconciler/controller/kafkachannel.go
index acf252b11..93ff5206d 100644
--- a/pkg/channel/consolidated/reconciler/controller/kafkachannel.go
+++ b/pkg/channel/consolidated/reconciler/controller/kafkachannel.go
@@ -165,7 +165,7 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, kc *v1beta1.KafkaChannel
 		return r.kafkaConfigError
 	}
 
-	kafkaClusterAdmin, err := r.createClient()
+	kafkaClusterAdmin, err := r.createClient(ctx)
 	if err != nil {
 		kc.Status.MarkConfigFailed("InvalidConfiguration", "Unable to build Kafka admin client for channel %s: %v", kc.Name, err)
 		return err
@@ -524,7 +524,7 @@ func (r *Reconciler) reconcileChannelService(ctx context.Context, dispatcherName
 	return svc, nil
 }
 
-func (r *Reconciler) createClient() (sarama.ClusterAdmin, error) {
+func (r *Reconciler) createClient(ctx context.Context) (sarama.ClusterAdmin, error) {
 	// We don't currently initialize r.kafkaClusterAdmin, hence we end up creating the cluster admin client every time.
 	// This is because of an issue with Shopify/sarama. See https://github.com/Shopify/sarama/issues/1162.
 	// Once the issue is fixed we should use a shared cluster admin client. Also, r.kafkaClusterAdmin is currently
@@ -532,7 +532,7 @@ func (r *Reconciler) createClient() (sarama.ClusterAdmin, error) {
 	kafkaClusterAdmin := r.kafkaClusterAdmin
 	if kafkaClusterAdmin == nil {
 		var err error
-		kafkaClusterAdmin, err = source.MakeAdminClient(controllerAgentName, r.kafkaAuthConfig, r.kafkaConfig.Brokers)
+		kafkaClusterAdmin, err = source.MakeAdminClient(ctx, controllerAgentName, r.kafkaAuthConfig, r.kafkaConfig)
 		if err != nil {
 			return nil, err
 		}
@@ -592,7 +592,7 @@ func (r *Reconciler) updateKafkaConfig(ctx context.Context, configMap *corev1.Co
 	r.kafkaConfig = kafkaConfig
 	r.kafkaConfigError = err
 	ac, err := kafka.NewAdminClient(ctx, func() (sarama.ClusterAdmin, error) {
-		return source.MakeAdminClient(controllerAgentName, r.kafkaAuthConfig, kafkaConfig.Brokers)
+		return source.MakeAdminClient(ctx, controllerAgentName, r.kafkaAuthConfig, kafkaConfig)
 	})
 
 	if err != nil {
@@ -612,7 +612,7 @@ func (r *Reconciler) updateKafkaConfig(ctx context.Context, configMap *corev1.Co
 func (r *Reconciler) FinalizeKind(ctx context.Context, kc *v1beta1.KafkaChannel) pkgreconciler.Event {
 	// Do not attempt retrying creating the client because it might be a permanent error
 	// in which case the finalizer will never get removed.
-	if kafkaClusterAdmin, err := r.createClient(); err == nil && r.kafkaConfig != nil {
+	if kafkaClusterAdmin, err := r.createClient(ctx); err == nil && r.kafkaConfig != nil {
 		if err := r.deleteTopic(ctx, kc, kafkaClusterAdmin); err != nil {
 			return err
 		}
diff --git a/pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go b/pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go
index 96ea3e58e..c34b6d06f 100644
--- a/pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go
+++ b/pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go
@@ -95,12 +95,13 @@ func NewController(ctx context.Context, cmw configmap.Watcher) *controller.Impl
 
 	kafkaChannelInformer := kafkachannel.Get(ctx)
 	args := &dispatcher.KafkaDispatcherArgs{
-		KnCEConnectionArgs: connectionArgs,
-		ClientID:           "kafka-ch-dispatcher",
-		Brokers:            kafkaConfig.Brokers,
-		KafkaAuthConfig:    kafkaAuthCfg,
-		TopicFunc:          utils.TopicName,
-		Logger:             logger,
+		KnCEConnectionArgs:       connectionArgs,
+		ClientID:                 "kafka-ch-dispatcher",
+		Brokers:                  kafkaConfig.Brokers,
+		KafkaAuthConfig:          kafkaAuthCfg,
+		SaramaSettingsYamlString: kafkaConfig.SaramaSettingsYamlString,
+		TopicFunc:                utils.TopicName,
+		Logger:                   logger,
 	}
 	kafkaDispatcher, err := dispatcher.NewDispatcher(ctx, args)
 	if err != nil {
diff --git a/pkg/channel/consolidated/utils/util.go b/pkg/channel/consolidated/utils/util.go
index 311da3d86..a24acc7f3 100644
--- a/pkg/channel/consolidated/utils/util.go
+++ b/pkg/channel/consolidated/utils/util.go
@@ -24,6 +24,7 @@ import (
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"knative.dev/eventing-kafka/pkg/common/client"
+	"knative.dev/eventing-kafka/pkg/common/constants"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/logging"
 
@@ -56,11 +57,12 @@ const (
 )
 
 type KafkaConfig struct {
-	Brokers             []string
-	MaxIdleConns        int32
-	MaxIdleConnsPerHost int32
-	AuthSecretName      string
-	AuthSecretNamespace string
+	Brokers                  []string
+	MaxIdleConns             int32
+	MaxIdleConnsPerHost      int32
+	AuthSecretName           string
+	AuthSecretNamespace      string
+	SaramaSettingsYamlString string
 }
 
 func GetKafkaAuthData(ctx context.Context, secretname string, secretNS string) *client.KafkaAuthConfig {
@@ -116,6 +118,7 @@ func GetKafkaConfig(configMap map[string]string) (*KafkaConfig, error) {
 		configmap.AsString(AuthSecretNamespace, &authSecretNamespace),
 		configmap.AsInt32(MaxIdleConnectionsKey, &config.MaxIdleConns),
 		configmap.AsInt32(MaxIdleConnectionsPerHostKey, &config.MaxIdleConnsPerHost),
+		configmap.AsString(constants.SaramaSettingsConfigKey, &config.SaramaSettingsYamlString),
 	)
 	if err != nil {
 		return nil, err
diff --git a/pkg/channel/distributed/common/config/configwatcher.go b/pkg/channel/distributed/common/config/configwatcher.go
index 1ce5d844c..fd9e43c32 100644
--- a/pkg/channel/distributed/common/config/configwatcher.go
+++ b/pkg/channel/distributed/common/config/configwatcher.go
@@ -24,13 +24,13 @@ import (
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"knative.dev/eventing-kafka/pkg/channel/distributed/common/config/constants"
+	"knative.dev/eventing-kafka/pkg/common/constants"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/injection/sharedmain"
 )
 
 // This function type is for a shim so that we can pass our own logger to the Observer function
-type LoggingObserver func(*zap.SugaredLogger, *corev1.ConfigMap)
+type LoggingObserver func(ctx context.Context, configMap *corev1.ConfigMap)
 
 // The EventingKafkaConfig and these EK sub-structs contain our custom configuration settings,
 // stored in the config-kafka configmap.  The sub-structs are explicitly declared so that they
@@ -88,7 +88,7 @@ func InitializeConfigWatcher(ctx context.Context, logger *zap.SugaredLogger, han
 	// Start The ConfigMap Watcher
 	// Taken from knative.dev/pkg/injection/sharedmain/main.go::WatchObservabilityConfigOrDie
 	if _, err := kubeclient.Get(ctx).CoreV1().ConfigMaps(namespace).Get(ctx, constants.SettingsConfigMapName, metav1.GetOptions{}); err == nil {
-		watcher.Watch(constants.SettingsConfigMapName, func(configmap *corev1.ConfigMap) { handler(logger, configmap) })
+		watcher.Watch(constants.SettingsConfigMapName, func(configmap *corev1.ConfigMap) { handler(ctx, configmap) })
 	} else if !apierrors.IsNotFound(err) {
 		logger.Error("Error reading ConfigMap "+constants.SettingsConfigMapName, zap.Error(err))
 		return err
diff --git a/pkg/channel/distributed/common/config/constants/constants.go b/pkg/channel/distributed/common/config/constants/constants.go
deleted file mode 100644
index c4f527f9c..000000000
--- a/pkg/channel/distributed/common/config/constants/constants.go
+++ /dev/null
@@ -1,26 +0,0 @@
-/*
-Copyright 2020 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 constants
-
-// Package Constants
-const (
-	// The name of the configmap used to hold eventing-kafka settings
-	SettingsConfigMapName = "config-kafka"
-	// The name of the keys in the Data section of the eventing-kafka configmap that holds Sarama and Eventing-Kafka configuration YAML
-	SaramaSettingsConfigKey        = "sarama"
-	EventingKafkaSettingsConfigKey = "eventing-kafka"
-)
diff --git a/pkg/channel/distributed/common/kafka/sarama/sarama.go b/pkg/channel/distributed/common/kafka/sarama/sarama.go
index fb7b4786f..4328e7ee0 100644
--- a/pkg/channel/distributed/common/kafka/sarama/sarama.go
+++ b/pkg/channel/distributed/common/kafka/sarama/sarama.go
@@ -28,8 +28,8 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	commonconfig "knative.dev/eventing-kafka/pkg/channel/distributed/common/config"
-	"knative.dev/eventing-kafka/pkg/channel/distributed/common/config/constants"
 	"knative.dev/eventing-kafka/pkg/common/client"
+	"knative.dev/eventing-kafka/pkg/common/constants"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/system"
 )
@@ -74,7 +74,7 @@ func LoadSettings(ctx context.Context, clientId string, kafkaAuthConfig *client.
 		FromYaml(saramaSettingsYamlString).
 		WithAuth(kafkaAuthConfig).
 		WithClientId(clientId).
-		Build()
+		Build(ctx)
 
 	return saramaConfig, eventingKafkaConfig, err
 }
diff --git a/pkg/channel/distributed/controller/kafkachannel/reconciler.go b/pkg/channel/distributed/controller/kafkachannel/reconciler.go
index 2cab5acfd..b445e90c8 100644
--- a/pkg/channel/distributed/controller/kafkachannel/reconciler.go
+++ b/pkg/channel/distributed/controller/kafkachannel/reconciler.go
@@ -31,7 +31,6 @@ import (
 	"k8s.io/client-go/tools/cache"
 	kafkav1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/config"
-	configconstants "knative.dev/eventing-kafka/pkg/channel/distributed/common/config/constants"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/admin"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/admin/types"
 	kafkasarama "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/sarama"
@@ -43,6 +42,7 @@ import (
 	"knative.dev/eventing-kafka/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel"
 	kafkalisters "knative.dev/eventing-kafka/pkg/client/listers/messaging/v1beta1"
 	"knative.dev/eventing-kafka/pkg/common/client"
+	commonconstants "knative.dev/eventing-kafka/pkg/common/constants"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/reconciler"
@@ -230,7 +230,8 @@ func (r *Reconciler) reconcile(ctx context.Context, channel *kafkav1beta1.KafkaC
 }
 
 // configMapObserver is the callback function that handles changes to our ConfigMap
-func (r *Reconciler) configMapObserver(logger *zap.SugaredLogger, configMap *corev1.ConfigMap) {
+func (r *Reconciler) configMapObserver(ctx context.Context, configMap *corev1.ConfigMap) {
+	logger := logging.FromContext(ctx)
 
 	if r == nil {
 		// This typically happens during startup and can be ignored
@@ -265,7 +266,7 @@ func (r *Reconciler) configMapObserver(logger *zap.SugaredLogger, configMap *cor
 	}
 
 	// Get The Sarama Config Yaml From The ConfigMap
-	saramaSettingsYamlString := configMap.Data[configconstants.SaramaSettingsConfigKey]
+	saramaSettingsYamlString := configMap.Data[commonconstants.SaramaSettingsConfigKey]
 
 	// Create A Kafka Auth Config From Current Credentials (Secret Data Takes Precedence Over ConfigMap)
 	var kafkaAuthCfg *client.KafkaAuthConfig
@@ -283,7 +284,7 @@ func (r *Reconciler) configMapObserver(logger *zap.SugaredLogger, configMap *cor
 		WithDefaults().
 		WithAuth(kafkaAuthCfg).
 		FromYaml(saramaSettingsYamlString).
-		Build()
+		Build(ctx)
 	if err != nil {
 		logger.Fatal("Failed To Load Eventing-Kafka Settings", zap.Any("configMap", configMap), zap.Error(err))
 	}
diff --git a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
index 8680f6572..186975258 100644
--- a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
+++ b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
@@ -25,11 +25,11 @@ import (
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
-	"knative.dev/eventing-kafka/pkg/channel/distributed/common/config/constants"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/consumer"
 	kafkasarama "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/sarama"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/metrics"
 	"knative.dev/eventing-kafka/pkg/common/client"
+	"knative.dev/eventing-kafka/pkg/common/constants"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/eventing/pkg/channel"
 )
@@ -63,7 +63,7 @@ func NewSubscriberWrapper(subscriberSpec eventingduck.SubscriberSpec, groupId st
 
 //  Dispatcher Interface
 type Dispatcher interface {
-	ConfigChanged(*corev1.ConfigMap) Dispatcher
+	ConfigChanged(ctx context.Context, configMap *corev1.ConfigMap) Dispatcher
 	Shutdown()
 	UpdateSubscriptions(subscriberSpecs []eventingduck.SubscriberSpec) map[eventingduck.SubscriberSpec]error
 }
@@ -269,7 +269,7 @@ func (d *DispatcherImpl) closeConsumerGroup(subscriber *SubscriberWrapper) {
 // are needed in the future, the environment will also need to be re-parsed here.
 // If there aren't any consumer-specific differences between the current config and the new one,
 // then just log that and move on; do not restart the ConsumerGroups unnecessarily.
-func (d *DispatcherImpl) ConfigChanged(configMap *corev1.ConfigMap) Dispatcher {
+func (d *DispatcherImpl) ConfigChanged(ctx context.Context, configMap *corev1.ConfigMap) Dispatcher {
 
 	d.Logger.Debug("New ConfigMap Received", zap.String("configMap.Name", configMap.ObjectMeta.Name))
 
@@ -303,7 +303,7 @@ func (d *DispatcherImpl) ConfigChanged(configMap *corev1.ConfigMap) Dispatcher {
 		}
 	}
 
-	newConfig, err := configBuilder.Build()
+	newConfig, err := configBuilder.Build(ctx)
 	if err != nil {
 		d.Logger.Error("Unable to build sarama settings", zap.Error(err))
 		return nil
diff --git a/pkg/channel/distributed/receiver/producer/producer.go b/pkg/channel/distributed/receiver/producer/producer.go
index 3da38cbb9..4a4ec5406 100644
--- a/pkg/channel/distributed/receiver/producer/producer.go
+++ b/pkg/channel/distributed/receiver/producer/producer.go
@@ -28,7 +28,6 @@ import (
 	"go.opencensus.io/trace"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
-	constants2 "knative.dev/eventing-kafka/pkg/channel/distributed/common/config/constants"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/producer"
 	kafkasarama "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/sarama"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/metrics"
@@ -36,6 +35,7 @@ import (
 	"knative.dev/eventing-kafka/pkg/channel/distributed/receiver/health"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/receiver/util"
 	"knative.dev/eventing-kafka/pkg/common/client"
+	commonconstants "knative.dev/eventing-kafka/pkg/common/constants"
 	"knative.dev/eventing-kafka/pkg/common/tracing"
 	eventingChannel "knative.dev/eventing/pkg/channel"
 )
@@ -170,7 +170,7 @@ func (p *Producer) ObserveMetrics(interval time.Duration) {
 // are needed in the future, the environment will also need to be re-parsed here.
 // If there aren't any producer-specific differences between the current config and the new one,
 // then just log that and move on; do not restart the Producer unnecessarily.
-func (p *Producer) ConfigChanged(configMap *corev1.ConfigMap) *Producer {
+func (p *Producer) ConfigChanged(ctx context.Context, configMap *corev1.ConfigMap) *Producer {
 
 	// Debug Log The ConfigMap Change
 	p.logger.Debug("New ConfigMap Received", zap.String("configMap.Name", configMap.ObjectMeta.Name))
@@ -182,7 +182,7 @@ func (p *Producer) ConfigChanged(configMap *corev1.ConfigMap) *Producer {
 	}
 
 	// Merge The ConfigMap Settings Into The Provided Config
-	saramaSettingsYamlString := configMap.Data[constants2.SaramaSettingsConfigKey]
+	saramaSettingsYamlString := configMap.Data[commonconstants.SaramaSettingsConfigKey]
 
 	// Merge The Sarama Config From ConfigMap Into New Sarama Config
 	configBuilder := client.NewConfigBuilder().
@@ -204,7 +204,7 @@ func (p *Producer) ConfigChanged(configMap *corev1.ConfigMap) *Producer {
 		}
 	}
 
-	newConfig, err := configBuilder.Build()
+	newConfig, err := configBuilder.Build(ctx)
 	if err != nil {
 		p.logger.Error("Unable to merge sarama settings", zap.Error(err))
 		return nil
diff --git a/pkg/client/injection/informers/bindings/v1alpha1/kafkabinding/filtered/fake/fake.go b/pkg/client/injection/informers/bindings/v1alpha1/kafkabinding/filtered/fake/fake.go
new file mode 100644
index 000000000..a5f553e7b
--- /dev/null
+++ b/pkg/client/injection/informers/bindings/v1alpha1/kafkabinding/filtered/fake/fake.go
@@ -0,0 +1,52 @@
+/*
+Copyright 2020 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.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	context "context"
+
+	filtered "knative.dev/eventing-kafka/pkg/client/injection/informers/bindings/v1alpha1/kafkabinding/filtered"
+	factoryfiltered "knative.dev/eventing-kafka/pkg/client/injection/informers/factory/filtered"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+var Get = filtered.Get
+
+func init() {
+	injection.Fake.RegisterFilteredInformers(withInformer)
+}
+
+func withInformer(ctx context.Context) (context.Context, []controller.Informer) {
+	untyped := ctx.Value(factoryfiltered.LabelKey{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch labelkey from context.")
+	}
+	labelSelectors := untyped.([]string)
+	infs := []controller.Informer{}
+	for _, selector := range labelSelectors {
+		f := factoryfiltered.Get(ctx, selector)
+		inf := f.Bindings().V1alpha1().KafkaBindings()
+		ctx = context.WithValue(ctx, filtered.Key{Selector: selector}, inf)
+		infs = append(infs, inf.Informer())
+	}
+	return ctx, infs
+}
diff --git a/pkg/client/injection/informers/bindings/v1alpha1/kafkabinding/filtered/kafkabinding.go b/pkg/client/injection/informers/bindings/v1alpha1/kafkabinding/filtered/kafkabinding.go
new file mode 100644
index 000000000..7971f68b4
--- /dev/null
+++ b/pkg/client/injection/informers/bindings/v1alpha1/kafkabinding/filtered/kafkabinding.go
@@ -0,0 +1,65 @@
+/*
+Copyright 2020 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.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package filtered
+
+import (
+	context "context"
+
+	v1alpha1 "knative.dev/eventing-kafka/pkg/client/informers/externalversions/bindings/v1alpha1"
+	filtered "knative.dev/eventing-kafka/pkg/client/injection/informers/factory/filtered"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+func init() {
+	injection.Default.RegisterFilteredInformers(withInformer)
+}
+
+// Key is used for associating the Informer inside the context.Context.
+type Key struct {
+	Selector string
+}
+
+func withInformer(ctx context.Context) (context.Context, []controller.Informer) {
+	untyped := ctx.Value(filtered.LabelKey{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch labelkey from context.")
+	}
+	labelSelectors := untyped.([]string)
+	infs := []controller.Informer{}
+	for _, selector := range labelSelectors {
+		f := filtered.Get(ctx, selector)
+		inf := f.Bindings().V1alpha1().KafkaBindings()
+		ctx = context.WithValue(ctx, Key{Selector: selector}, inf)
+		infs = append(infs, inf.Informer())
+	}
+	return ctx, infs
+}
+
+// Get extracts the typed informer from the context.
+func Get(ctx context.Context, selector string) v1alpha1.KafkaBindingInformer {
+	untyped := ctx.Value(Key{Selector: selector})
+	if untyped == nil {
+		logging.FromContext(ctx).Panicf(
+			"Unable to fetch knative.dev/eventing-kafka/pkg/client/informers/externalversions/bindings/v1alpha1.KafkaBindingInformer with selector %s from context.", selector)
+	}
+	return untyped.(v1alpha1.KafkaBindingInformer)
+}
diff --git a/pkg/client/injection/informers/bindings/v1beta1/kafkabinding/filtered/fake/fake.go b/pkg/client/injection/informers/bindings/v1beta1/kafkabinding/filtered/fake/fake.go
new file mode 100644
index 000000000..806d684b9
--- /dev/null
+++ b/pkg/client/injection/informers/bindings/v1beta1/kafkabinding/filtered/fake/fake.go
@@ -0,0 +1,52 @@
+/*
+Copyright 2020 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.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	context "context"
+
+	filtered "knative.dev/eventing-kafka/pkg/client/injection/informers/bindings/v1beta1/kafkabinding/filtered"
+	factoryfiltered "knative.dev/eventing-kafka/pkg/client/injection/informers/factory/filtered"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+var Get = filtered.Get
+
+func init() {
+	injection.Fake.RegisterFilteredInformers(withInformer)
+}
+
+func withInformer(ctx context.Context) (context.Context, []controller.Informer) {
+	untyped := ctx.Value(factoryfiltered.LabelKey{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch labelkey from context.")
+	}
+	labelSelectors := untyped.([]string)
+	infs := []controller.Informer{}
+	for _, selector := range labelSelectors {
+		f := factoryfiltered.Get(ctx, selector)
+		inf := f.Bindings().V1beta1().KafkaBindings()
+		ctx = context.WithValue(ctx, filtered.Key{Selector: selector}, inf)
+		infs = append(infs, inf.Informer())
+	}
+	return ctx, infs
+}
diff --git a/pkg/client/injection/informers/bindings/v1beta1/kafkabinding/filtered/kafkabinding.go b/pkg/client/injection/informers/bindings/v1beta1/kafkabinding/filtered/kafkabinding.go
new file mode 100644
index 000000000..89791ec60
--- /dev/null
+++ b/pkg/client/injection/informers/bindings/v1beta1/kafkabinding/filtered/kafkabinding.go
@@ -0,0 +1,65 @@
+/*
+Copyright 2020 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.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package filtered
+
+import (
+	context "context"
+
+	v1beta1 "knative.dev/eventing-kafka/pkg/client/informers/externalversions/bindings/v1beta1"
+	filtered "knative.dev/eventing-kafka/pkg/client/injection/informers/factory/filtered"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+func init() {
+	injection.Default.RegisterFilteredInformers(withInformer)
+}
+
+// Key is used for associating the Informer inside the context.Context.
+type Key struct {
+	Selector string
+}
+
+func withInformer(ctx context.Context) (context.Context, []controller.Informer) {
+	untyped := ctx.Value(filtered.LabelKey{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch labelkey from context.")
+	}
+	labelSelectors := untyped.([]string)
+	infs := []controller.Informer{}
+	for _, selector := range labelSelectors {
+		f := filtered.Get(ctx, selector)
+		inf := f.Bindings().V1beta1().KafkaBindings()
+		ctx = context.WithValue(ctx, Key{Selector: selector}, inf)
+		infs = append(infs, inf.Informer())
+	}
+	return ctx, infs
+}
+
+// Get extracts the typed informer from the context.
+func Get(ctx context.Context, selector string) v1beta1.KafkaBindingInformer {
+	untyped := ctx.Value(Key{Selector: selector})
+	if untyped == nil {
+		logging.FromContext(ctx).Panicf(
+			"Unable to fetch knative.dev/eventing-kafka/pkg/client/informers/externalversions/bindings/v1beta1.KafkaBindingInformer with selector %s from context.", selector)
+	}
+	return untyped.(v1beta1.KafkaBindingInformer)
+}
diff --git a/pkg/client/injection/informers/factory/filtered/fake/fake_filtered_factory.go b/pkg/client/injection/informers/factory/filtered/fake/fake_filtered_factory.go
new file mode 100644
index 000000000..5dff766aa
--- /dev/null
+++ b/pkg/client/injection/informers/factory/filtered/fake/fake_filtered_factory.go
@@ -0,0 +1,59 @@
+/*
+Copyright 2020 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.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package fakeFilteredFactory
+
+import (
+	context "context"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	externalversions "knative.dev/eventing-kafka/pkg/client/informers/externalversions"
+	fake "knative.dev/eventing-kafka/pkg/client/injection/client/fake"
+	filtered "knative.dev/eventing-kafka/pkg/client/injection/informers/factory/filtered"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+var Get = filtered.Get
+
+func init() {
+	injection.Fake.RegisterInformerFactory(withInformerFactory)
+}
+
+func withInformerFactory(ctx context.Context) context.Context {
+	c := fake.Get(ctx)
+	opts := []externalversions.SharedInformerOption{}
+	if injection.HasNamespaceScope(ctx) {
+		opts = append(opts, externalversions.WithNamespace(injection.GetNamespaceScope(ctx)))
+	}
+	untyped := ctx.Value(filtered.LabelKey{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch labelkey from context.")
+	}
+	labelSelectors := untyped.([]string)
+	for _, selector := range labelSelectors {
+		thisOpts := append(opts, externalversions.WithTweakListOptions(func(l *v1.ListOptions) {
+			l.LabelSelector = selector
+		}))
+		ctx = context.WithValue(ctx, filtered.Key{Selector: selector},
+			externalversions.NewSharedInformerFactoryWithOptions(c, controller.GetResyncPeriod(ctx), thisOpts...))
+	}
+	return ctx
+}
diff --git a/pkg/client/injection/informers/factory/filtered/filtered_factory.go b/pkg/client/injection/informers/factory/filtered/filtered_factory.go
new file mode 100644
index 000000000..9d3c92063
--- /dev/null
+++ b/pkg/client/injection/informers/factory/filtered/filtered_factory.go
@@ -0,0 +1,77 @@
+/*
+Copyright 2020 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.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package filteredFactory
+
+import (
+	context "context"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	externalversions "knative.dev/eventing-kafka/pkg/client/informers/externalversions"
+	client "knative.dev/eventing-kafka/pkg/client/injection/client"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+func init() {
+	injection.Default.RegisterInformerFactory(withInformerFactory)
+}
+
+// Key is used as the key for associating information with a context.Context.
+type Key struct {
+	Selector string
+}
+
+type LabelKey struct{}
+
+func WithSelectors(ctx context.Context, selector ...string) context.Context {
+	return context.WithValue(ctx, LabelKey{}, selector)
+}
+
+func withInformerFactory(ctx context.Context) context.Context {
+	c := client.Get(ctx)
+	opts := []externalversions.SharedInformerOption{}
+	if injection.HasNamespaceScope(ctx) {
+		opts = append(opts, externalversions.WithNamespace(injection.GetNamespaceScope(ctx)))
+	}
+	untyped := ctx.Value(LabelKey{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch labelkey from context.")
+	}
+	labelSelectors := untyped.([]string)
+	for _, selector := range labelSelectors {
+		thisOpts := append(opts, externalversions.WithTweakListOptions(func(l *v1.ListOptions) {
+			l.LabelSelector = selector
+		}))
+		ctx = context.WithValue(ctx, Key{Selector: selector},
+			externalversions.NewSharedInformerFactoryWithOptions(c, controller.GetResyncPeriod(ctx), thisOpts...))
+	}
+	return ctx
+}
+
+// Get extracts the InformerFactory from the context.
+func Get(ctx context.Context, selector string) externalversions.SharedInformerFactory {
+	untyped := ctx.Value(Key{Selector: selector})
+	if untyped == nil {
+		logging.FromContext(ctx).Panicf(
+			"Unable to fetch knative.dev/eventing-kafka/pkg/client/informers/externalversions.SharedInformerFactory with selector %s from context.", selector)
+	}
+	return untyped.(externalversions.SharedInformerFactory)
+}
diff --git a/pkg/client/injection/informers/messaging/v1alpha1/kafkachannel/filtered/fake/fake.go b/pkg/client/injection/informers/messaging/v1alpha1/kafkachannel/filtered/fake/fake.go
new file mode 100644
index 000000000..7d5fe721b
--- /dev/null
+++ b/pkg/client/injection/informers/messaging/v1alpha1/kafkachannel/filtered/fake/fake.go
@@ -0,0 +1,52 @@
+/*
+Copyright 2020 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.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	context "context"
+
+	factoryfiltered "knative.dev/eventing-kafka/pkg/client/injection/informers/factory/filtered"
+	filtered "knative.dev/eventing-kafka/pkg/client/injection/informers/messaging/v1alpha1/kafkachannel/filtered"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+var Get = filtered.Get
+
+func init() {
+	injection.Fake.RegisterFilteredInformers(withInformer)
+}
+
+func withInformer(ctx context.Context) (context.Context, []controller.Informer) {
+	untyped := ctx.Value(factoryfiltered.LabelKey{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch labelkey from context.")
+	}
+	labelSelectors := untyped.([]string)
+	infs := []controller.Informer{}
+	for _, selector := range labelSelectors {
+		f := factoryfiltered.Get(ctx, selector)
+		inf := f.Messaging().V1alpha1().KafkaChannels()
+		ctx = context.WithValue(ctx, filtered.Key{Selector: selector}, inf)
+		infs = append(infs, inf.Informer())
+	}
+	return ctx, infs
+}
diff --git a/pkg/client/injection/informers/messaging/v1alpha1/kafkachannel/filtered/kafkachannel.go b/pkg/client/injection/informers/messaging/v1alpha1/kafkachannel/filtered/kafkachannel.go
new file mode 100644
index 000000000..9cc546d7d
--- /dev/null
+++ b/pkg/client/injection/informers/messaging/v1alpha1/kafkachannel/filtered/kafkachannel.go
@@ -0,0 +1,65 @@
+/*
+Copyright 2020 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.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package filtered
+
+import (
+	context "context"
+
+	v1alpha1 "knative.dev/eventing-kafka/pkg/client/informers/externalversions/messaging/v1alpha1"
+	filtered "knative.dev/eventing-kafka/pkg/client/injection/informers/factory/filtered"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+func init() {
+	injection.Default.RegisterFilteredInformers(withInformer)
+}
+
+// Key is used for associating the Informer inside the context.Context.
+type Key struct {
+	Selector string
+}
+
+func withInformer(ctx context.Context) (context.Context, []controller.Informer) {
+	untyped := ctx.Value(filtered.LabelKey{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch labelkey from context.")
+	}
+	labelSelectors := untyped.([]string)
+	infs := []controller.Informer{}
+	for _, selector := range labelSelectors {
+		f := filtered.Get(ctx, selector)
+		inf := f.Messaging().V1alpha1().KafkaChannels()
+		ctx = context.WithValue(ctx, Key{Selector: selector}, inf)
+		infs = append(infs, inf.Informer())
+	}
+	return ctx, infs
+}
+
+// Get extracts the typed informer from the context.
+func Get(ctx context.Context, selector string) v1alpha1.KafkaChannelInformer {
+	untyped := ctx.Value(Key{Selector: selector})
+	if untyped == nil {
+		logging.FromContext(ctx).Panicf(
+			"Unable to fetch knative.dev/eventing-kafka/pkg/client/informers/externalversions/messaging/v1alpha1.KafkaChannelInformer with selector %s from context.", selector)
+	}
+	return untyped.(v1alpha1.KafkaChannelInformer)
+}
diff --git a/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/filtered/fake/fake.go b/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/filtered/fake/fake.go
new file mode 100644
index 000000000..64164b00a
--- /dev/null
+++ b/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/filtered/fake/fake.go
@@ -0,0 +1,52 @@
+/*
+Copyright 2020 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.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	context "context"
+
+	factoryfiltered "knative.dev/eventing-kafka/pkg/client/injection/informers/factory/filtered"
+	filtered "knative.dev/eventing-kafka/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/filtered"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+var Get = filtered.Get
+
+func init() {
+	injection.Fake.RegisterFilteredInformers(withInformer)
+}
+
+func withInformer(ctx context.Context) (context.Context, []controller.Informer) {
+	untyped := ctx.Value(factoryfiltered.LabelKey{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch labelkey from context.")
+	}
+	labelSelectors := untyped.([]string)
+	infs := []controller.Informer{}
+	for _, selector := range labelSelectors {
+		f := factoryfiltered.Get(ctx, selector)
+		inf := f.Messaging().V1beta1().KafkaChannels()
+		ctx = context.WithValue(ctx, filtered.Key{Selector: selector}, inf)
+		infs = append(infs, inf.Informer())
+	}
+	return ctx, infs
+}
diff --git a/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/filtered/kafkachannel.go b/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/filtered/kafkachannel.go
new file mode 100644
index 000000000..7e0893f12
--- /dev/null
+++ b/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/filtered/kafkachannel.go
@@ -0,0 +1,65 @@
+/*
+Copyright 2020 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.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package filtered
+
+import (
+	context "context"
+
+	v1beta1 "knative.dev/eventing-kafka/pkg/client/informers/externalversions/messaging/v1beta1"
+	filtered "knative.dev/eventing-kafka/pkg/client/injection/informers/factory/filtered"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+func init() {
+	injection.Default.RegisterFilteredInformers(withInformer)
+}
+
+// Key is used for associating the Informer inside the context.Context.
+type Key struct {
+	Selector string
+}
+
+func withInformer(ctx context.Context) (context.Context, []controller.Informer) {
+	untyped := ctx.Value(filtered.LabelKey{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch labelkey from context.")
+	}
+	labelSelectors := untyped.([]string)
+	infs := []controller.Informer{}
+	for _, selector := range labelSelectors {
+		f := filtered.Get(ctx, selector)
+		inf := f.Messaging().V1beta1().KafkaChannels()
+		ctx = context.WithValue(ctx, Key{Selector: selector}, inf)
+		infs = append(infs, inf.Informer())
+	}
+	return ctx, infs
+}
+
+// Get extracts the typed informer from the context.
+func Get(ctx context.Context, selector string) v1beta1.KafkaChannelInformer {
+	untyped := ctx.Value(Key{Selector: selector})
+	if untyped == nil {
+		logging.FromContext(ctx).Panicf(
+			"Unable to fetch knative.dev/eventing-kafka/pkg/client/informers/externalversions/messaging/v1beta1.KafkaChannelInformer with selector %s from context.", selector)
+	}
+	return untyped.(v1beta1.KafkaChannelInformer)
+}
diff --git a/pkg/client/injection/informers/sources/v1alpha1/kafkasource/filtered/fake/fake.go b/pkg/client/injection/informers/sources/v1alpha1/kafkasource/filtered/fake/fake.go
new file mode 100644
index 000000000..778c46f01
--- /dev/null
+++ b/pkg/client/injection/informers/sources/v1alpha1/kafkasource/filtered/fake/fake.go
@@ -0,0 +1,52 @@
+/*
+Copyright 2020 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.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	context "context"
+
+	factoryfiltered "knative.dev/eventing-kafka/pkg/client/injection/informers/factory/filtered"
+	filtered "knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1alpha1/kafkasource/filtered"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+var Get = filtered.Get
+
+func init() {
+	injection.Fake.RegisterFilteredInformers(withInformer)
+}
+
+func withInformer(ctx context.Context) (context.Context, []controller.Informer) {
+	untyped := ctx.Value(factoryfiltered.LabelKey{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch labelkey from context.")
+	}
+	labelSelectors := untyped.([]string)
+	infs := []controller.Informer{}
+	for _, selector := range labelSelectors {
+		f := factoryfiltered.Get(ctx, selector)
+		inf := f.Sources().V1alpha1().KafkaSources()
+		ctx = context.WithValue(ctx, filtered.Key{Selector: selector}, inf)
+		infs = append(infs, inf.Informer())
+	}
+	return ctx, infs
+}
diff --git a/pkg/client/injection/informers/sources/v1alpha1/kafkasource/filtered/kafkasource.go b/pkg/client/injection/informers/sources/v1alpha1/kafkasource/filtered/kafkasource.go
new file mode 100644
index 000000000..316eb495a
--- /dev/null
+++ b/pkg/client/injection/informers/sources/v1alpha1/kafkasource/filtered/kafkasource.go
@@ -0,0 +1,65 @@
+/*
+Copyright 2020 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.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package filtered
+
+import (
+	context "context"
+
+	v1alpha1 "knative.dev/eventing-kafka/pkg/client/informers/externalversions/sources/v1alpha1"
+	filtered "knative.dev/eventing-kafka/pkg/client/injection/informers/factory/filtered"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+func init() {
+	injection.Default.RegisterFilteredInformers(withInformer)
+}
+
+// Key is used for associating the Informer inside the context.Context.
+type Key struct {
+	Selector string
+}
+
+func withInformer(ctx context.Context) (context.Context, []controller.Informer) {
+	untyped := ctx.Value(filtered.LabelKey{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch labelkey from context.")
+	}
+	labelSelectors := untyped.([]string)
+	infs := []controller.Informer{}
+	for _, selector := range labelSelectors {
+		f := filtered.Get(ctx, selector)
+		inf := f.Sources().V1alpha1().KafkaSources()
+		ctx = context.WithValue(ctx, Key{Selector: selector}, inf)
+		infs = append(infs, inf.Informer())
+	}
+	return ctx, infs
+}
+
+// Get extracts the typed informer from the context.
+func Get(ctx context.Context, selector string) v1alpha1.KafkaSourceInformer {
+	untyped := ctx.Value(Key{Selector: selector})
+	if untyped == nil {
+		logging.FromContext(ctx).Panicf(
+			"Unable to fetch knative.dev/eventing-kafka/pkg/client/informers/externalversions/sources/v1alpha1.KafkaSourceInformer with selector %s from context.", selector)
+	}
+	return untyped.(v1alpha1.KafkaSourceInformer)
+}
diff --git a/pkg/client/injection/informers/sources/v1beta1/kafkasource/filtered/fake/fake.go b/pkg/client/injection/informers/sources/v1beta1/kafkasource/filtered/fake/fake.go
new file mode 100644
index 000000000..fe4abc3be
--- /dev/null
+++ b/pkg/client/injection/informers/sources/v1beta1/kafkasource/filtered/fake/fake.go
@@ -0,0 +1,52 @@
+/*
+Copyright 2020 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.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	context "context"
+
+	factoryfiltered "knative.dev/eventing-kafka/pkg/client/injection/informers/factory/filtered"
+	filtered "knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource/filtered"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+var Get = filtered.Get
+
+func init() {
+	injection.Fake.RegisterFilteredInformers(withInformer)
+}
+
+func withInformer(ctx context.Context) (context.Context, []controller.Informer) {
+	untyped := ctx.Value(factoryfiltered.LabelKey{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch labelkey from context.")
+	}
+	labelSelectors := untyped.([]string)
+	infs := []controller.Informer{}
+	for _, selector := range labelSelectors {
+		f := factoryfiltered.Get(ctx, selector)
+		inf := f.Sources().V1beta1().KafkaSources()
+		ctx = context.WithValue(ctx, filtered.Key{Selector: selector}, inf)
+		infs = append(infs, inf.Informer())
+	}
+	return ctx, infs
+}
diff --git a/pkg/client/injection/informers/sources/v1beta1/kafkasource/filtered/kafkasource.go b/pkg/client/injection/informers/sources/v1beta1/kafkasource/filtered/kafkasource.go
new file mode 100644
index 000000000..a9d3d90d9
--- /dev/null
+++ b/pkg/client/injection/informers/sources/v1beta1/kafkasource/filtered/kafkasource.go
@@ -0,0 +1,65 @@
+/*
+Copyright 2020 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.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package filtered
+
+import (
+	context "context"
+
+	v1beta1 "knative.dev/eventing-kafka/pkg/client/informers/externalversions/sources/v1beta1"
+	filtered "knative.dev/eventing-kafka/pkg/client/injection/informers/factory/filtered"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+func init() {
+	injection.Default.RegisterFilteredInformers(withInformer)
+}
+
+// Key is used for associating the Informer inside the context.Context.
+type Key struct {
+	Selector string
+}
+
+func withInformer(ctx context.Context) (context.Context, []controller.Informer) {
+	untyped := ctx.Value(filtered.LabelKey{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch labelkey from context.")
+	}
+	labelSelectors := untyped.([]string)
+	infs := []controller.Informer{}
+	for _, selector := range labelSelectors {
+		f := filtered.Get(ctx, selector)
+		inf := f.Sources().V1beta1().KafkaSources()
+		ctx = context.WithValue(ctx, Key{Selector: selector}, inf)
+		infs = append(infs, inf.Informer())
+	}
+	return ctx, infs
+}
+
+// Get extracts the typed informer from the context.
+func Get(ctx context.Context, selector string) v1beta1.KafkaSourceInformer {
+	untyped := ctx.Value(Key{Selector: selector})
+	if untyped == nil {
+		logging.FromContext(ctx).Panicf(
+			"Unable to fetch knative.dev/eventing-kafka/pkg/client/informers/externalversions/sources/v1beta1.KafkaSourceInformer with selector %s from context.", selector)
+	}
+	return untyped.(v1beta1.KafkaSourceInformer)
+}
diff --git a/pkg/client/injection/reconciler/bindings/v1alpha1/kafkabinding/controller.go b/pkg/client/injection/reconciler/bindings/v1alpha1/kafkabinding/controller.go
index 4b28b0157..c842d03b5 100644
--- a/pkg/client/injection/reconciler/bindings/v1alpha1/kafkabinding/controller.go
+++ b/pkg/client/injection/reconciler/bindings/v1alpha1/kafkabinding/controller.go
@@ -24,6 +24,7 @@ import (
 	reflect "reflect"
 	strings "strings"
 
+	zap "go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	labels "k8s.io/apimachinery/pkg/labels"
 	types "k8s.io/apimachinery/pkg/types"
@@ -37,6 +38,7 @@ import (
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	controller "knative.dev/pkg/controller"
 	logging "knative.dev/pkg/logging"
+	logkey "knative.dev/pkg/logging/logkey"
 	reconciler "knative.dev/pkg/reconciler"
 )
 
@@ -84,10 +86,16 @@ func NewImpl(ctx context.Context, r Interface, optionsFns ...controller.OptionsF
 		finalizerName: defaultFinalizerName,
 	}
 
-	t := reflect.TypeOf(r).Elem()
-	queueName := fmt.Sprintf("%s.%s", strings.ReplaceAll(t.PkgPath(), "/", "-"), t.Name())
+	ctrType := reflect.TypeOf(r).Elem()
+	ctrTypeName := fmt.Sprintf("%s.%s", ctrType.PkgPath(), ctrType.Name())
+	ctrTypeName = strings.ReplaceAll(ctrTypeName, "/", ".")
 
-	impl := controller.NewImpl(rec, logger, queueName)
+	logger = logger.With(
+		zap.String(logkey.ControllerType, ctrTypeName),
+		zap.String(logkey.Kind, "bindings.knative.dev.KafkaBinding"),
+	)
+
+	impl := controller.NewImpl(rec, logger, ctrTypeName)
 	agentName := defaultControllerAgentName
 
 	// Pass impl to the options. Save any optional results.
diff --git a/pkg/client/injection/reconciler/bindings/v1beta1/kafkabinding/controller.go b/pkg/client/injection/reconciler/bindings/v1beta1/kafkabinding/controller.go
index ec350f6f4..5f8d56992 100644
--- a/pkg/client/injection/reconciler/bindings/v1beta1/kafkabinding/controller.go
+++ b/pkg/client/injection/reconciler/bindings/v1beta1/kafkabinding/controller.go
@@ -24,6 +24,7 @@ import (
 	reflect "reflect"
 	strings "strings"
 
+	zap "go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	labels "k8s.io/apimachinery/pkg/labels"
 	types "k8s.io/apimachinery/pkg/types"
@@ -37,6 +38,7 @@ import (
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	controller "knative.dev/pkg/controller"
 	logging "knative.dev/pkg/logging"
+	logkey "knative.dev/pkg/logging/logkey"
 	reconciler "knative.dev/pkg/reconciler"
 )
 
@@ -84,10 +86,16 @@ func NewImpl(ctx context.Context, r Interface, optionsFns ...controller.OptionsF
 		finalizerName: defaultFinalizerName,
 	}
 
-	t := reflect.TypeOf(r).Elem()
-	queueName := fmt.Sprintf("%s.%s", strings.ReplaceAll(t.PkgPath(), "/", "-"), t.Name())
+	ctrType := reflect.TypeOf(r).Elem()
+	ctrTypeName := fmt.Sprintf("%s.%s", ctrType.PkgPath(), ctrType.Name())
+	ctrTypeName = strings.ReplaceAll(ctrTypeName, "/", ".")
 
-	impl := controller.NewImpl(rec, logger, queueName)
+	logger = logger.With(
+		zap.String(logkey.ControllerType, ctrTypeName),
+		zap.String(logkey.Kind, "bindings.knative.dev.KafkaBinding"),
+	)
+
+	impl := controller.NewImpl(rec, logger, ctrTypeName)
 	agentName := defaultControllerAgentName
 
 	// Pass impl to the options. Save any optional results.
diff --git a/pkg/client/injection/reconciler/messaging/v1alpha1/kafkachannel/controller.go b/pkg/client/injection/reconciler/messaging/v1alpha1/kafkachannel/controller.go
index f60d87799..d63dacc3c 100644
--- a/pkg/client/injection/reconciler/messaging/v1alpha1/kafkachannel/controller.go
+++ b/pkg/client/injection/reconciler/messaging/v1alpha1/kafkachannel/controller.go
@@ -24,6 +24,7 @@ import (
 	reflect "reflect"
 	strings "strings"
 
+	zap "go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	labels "k8s.io/apimachinery/pkg/labels"
 	types "k8s.io/apimachinery/pkg/types"
@@ -37,6 +38,7 @@ import (
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	controller "knative.dev/pkg/controller"
 	logging "knative.dev/pkg/logging"
+	logkey "knative.dev/pkg/logging/logkey"
 	reconciler "knative.dev/pkg/reconciler"
 )
 
@@ -84,10 +86,16 @@ func NewImpl(ctx context.Context, r Interface, optionsFns ...controller.OptionsF
 		finalizerName: defaultFinalizerName,
 	}
 
-	t := reflect.TypeOf(r).Elem()
-	queueName := fmt.Sprintf("%s.%s", strings.ReplaceAll(t.PkgPath(), "/", "-"), t.Name())
+	ctrType := reflect.TypeOf(r).Elem()
+	ctrTypeName := fmt.Sprintf("%s.%s", ctrType.PkgPath(), ctrType.Name())
+	ctrTypeName = strings.ReplaceAll(ctrTypeName, "/", ".")
 
-	impl := controller.NewImpl(rec, logger, queueName)
+	logger = logger.With(
+		zap.String(logkey.ControllerType, ctrTypeName),
+		zap.String(logkey.Kind, "messaging.knative.dev.KafkaChannel"),
+	)
+
+	impl := controller.NewImpl(rec, logger, ctrTypeName)
 	agentName := defaultControllerAgentName
 
 	// Pass impl to the options. Save any optional results.
diff --git a/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel/controller.go b/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel/controller.go
index 8414133f8..9e09a0769 100644
--- a/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel/controller.go
+++ b/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel/controller.go
@@ -24,6 +24,7 @@ import (
 	reflect "reflect"
 	strings "strings"
 
+	zap "go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	labels "k8s.io/apimachinery/pkg/labels"
 	types "k8s.io/apimachinery/pkg/types"
@@ -37,6 +38,7 @@ import (
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	controller "knative.dev/pkg/controller"
 	logging "knative.dev/pkg/logging"
+	logkey "knative.dev/pkg/logging/logkey"
 	reconciler "knative.dev/pkg/reconciler"
 )
 
@@ -84,10 +86,16 @@ func NewImpl(ctx context.Context, r Interface, optionsFns ...controller.OptionsF
 		finalizerName: defaultFinalizerName,
 	}
 
-	t := reflect.TypeOf(r).Elem()
-	queueName := fmt.Sprintf("%s.%s", strings.ReplaceAll(t.PkgPath(), "/", "-"), t.Name())
+	ctrType := reflect.TypeOf(r).Elem()
+	ctrTypeName := fmt.Sprintf("%s.%s", ctrType.PkgPath(), ctrType.Name())
+	ctrTypeName = strings.ReplaceAll(ctrTypeName, "/", ".")
 
-	impl := controller.NewImpl(rec, logger, queueName)
+	logger = logger.With(
+		zap.String(logkey.ControllerType, ctrTypeName),
+		zap.String(logkey.Kind, "messaging.knative.dev.KafkaChannel"),
+	)
+
+	impl := controller.NewImpl(rec, logger, ctrTypeName)
 	agentName := defaultControllerAgentName
 
 	// Pass impl to the options. Save any optional results.
diff --git a/pkg/client/injection/reconciler/sources/v1alpha1/kafkasource/controller.go b/pkg/client/injection/reconciler/sources/v1alpha1/kafkasource/controller.go
index 98a600f22..bf6184d17 100644
--- a/pkg/client/injection/reconciler/sources/v1alpha1/kafkasource/controller.go
+++ b/pkg/client/injection/reconciler/sources/v1alpha1/kafkasource/controller.go
@@ -24,6 +24,7 @@ import (
 	reflect "reflect"
 	strings "strings"
 
+	zap "go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	labels "k8s.io/apimachinery/pkg/labels"
 	types "k8s.io/apimachinery/pkg/types"
@@ -37,6 +38,7 @@ import (
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	controller "knative.dev/pkg/controller"
 	logging "knative.dev/pkg/logging"
+	logkey "knative.dev/pkg/logging/logkey"
 	reconciler "knative.dev/pkg/reconciler"
 )
 
@@ -84,10 +86,16 @@ func NewImpl(ctx context.Context, r Interface, optionsFns ...controller.OptionsF
 		finalizerName: defaultFinalizerName,
 	}
 
-	t := reflect.TypeOf(r).Elem()
-	queueName := fmt.Sprintf("%s.%s", strings.ReplaceAll(t.PkgPath(), "/", "-"), t.Name())
+	ctrType := reflect.TypeOf(r).Elem()
+	ctrTypeName := fmt.Sprintf("%s.%s", ctrType.PkgPath(), ctrType.Name())
+	ctrTypeName = strings.ReplaceAll(ctrTypeName, "/", ".")
 
-	impl := controller.NewImpl(rec, logger, queueName)
+	logger = logger.With(
+		zap.String(logkey.ControllerType, ctrTypeName),
+		zap.String(logkey.Kind, "sources.knative.dev.KafkaSource"),
+	)
+
+	impl := controller.NewImpl(rec, logger, ctrTypeName)
 	agentName := defaultControllerAgentName
 
 	// Pass impl to the options. Save any optional results.
diff --git a/pkg/client/injection/reconciler/sources/v1beta1/kafkasource/controller.go b/pkg/client/injection/reconciler/sources/v1beta1/kafkasource/controller.go
index 3b82ed5e4..d9d60929b 100644
--- a/pkg/client/injection/reconciler/sources/v1beta1/kafkasource/controller.go
+++ b/pkg/client/injection/reconciler/sources/v1beta1/kafkasource/controller.go
@@ -24,6 +24,7 @@ import (
 	reflect "reflect"
 	strings "strings"
 
+	zap "go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	labels "k8s.io/apimachinery/pkg/labels"
 	types "k8s.io/apimachinery/pkg/types"
@@ -37,6 +38,7 @@ import (
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	controller "knative.dev/pkg/controller"
 	logging "knative.dev/pkg/logging"
+	logkey "knative.dev/pkg/logging/logkey"
 	reconciler "knative.dev/pkg/reconciler"
 )
 
@@ -84,10 +86,16 @@ func NewImpl(ctx context.Context, r Interface, optionsFns ...controller.OptionsF
 		finalizerName: defaultFinalizerName,
 	}
 
-	t := reflect.TypeOf(r).Elem()
-	queueName := fmt.Sprintf("%s.%s", strings.ReplaceAll(t.PkgPath(), "/", "-"), t.Name())
+	ctrType := reflect.TypeOf(r).Elem()
+	ctrTypeName := fmt.Sprintf("%s.%s", ctrType.PkgPath(), ctrType.Name())
+	ctrTypeName = strings.ReplaceAll(ctrTypeName, "/", ".")
 
-	impl := controller.NewImpl(rec, logger, queueName)
+	logger = logger.With(
+		zap.String(logkey.ControllerType, ctrTypeName),
+		zap.String(logkey.Kind, "sources.knative.dev.KafkaSource"),
+	)
+
+	impl := controller.NewImpl(rec, logger, ctrTypeName)
 	agentName := defaultControllerAgentName
 
 	// Pass impl to the options. Save any optional results.
diff --git a/pkg/common/client/config.go b/pkg/common/client/config.go
index 4c0429b96..9e63d178e 100644
--- a/pkg/common/client/config.go
+++ b/pkg/common/client/config.go
@@ -17,6 +17,7 @@ limitations under the License.
 package client
 
 import (
+	"context"
 	"crypto/tls"
 	"crypto/x509"
 	"fmt"
@@ -27,6 +28,7 @@ import (
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/constants"
+	"knative.dev/pkg/logging"
 )
 
 type KafkaAuthConfig struct {
@@ -80,8 +82,9 @@ type ConfigBuilder interface {
 	// (if provided) or in the YAML-string
 	WithClientId(clientId string) ConfigBuilder
 
-	// Build builds the Sarama config with the given builder config
-	Build() (*sarama.Config, error)
+	// Build builds the Sarama config with the given context.
+	// Context is used for getting the config at the moment.
+	Build(ctx context.Context) (*sarama.Config, error)
 }
 
 func NewConfigBuilder() ConfigBuilder {
@@ -127,7 +130,8 @@ func (b *configBuilder) WithAuth(kafkaAuthCfg *KafkaAuthConfig) ConfigBuilder {
 	return b
 }
 
-func (b *configBuilder) Build() (*sarama.Config, error) {
+// Build builds the Sarama config.
+func (b *configBuilder) Build(ctx context.Context) (*sarama.Config, error) {
 	var config *sarama.Config
 
 	// check if there's existing first
@@ -221,6 +225,11 @@ func (b *configBuilder) Build() (*sarama.Config, error) {
 		config.ClientID = b.clientId
 	}
 
+	if ctx != nil {
+		logger := logging.FromContext(ctx)
+		logger.Debugf("Built Sarama config: %+v", config)
+	}
+
 	return config, nil
 }
 
diff --git a/pkg/common/constants/constants.go b/pkg/common/constants/constants.go
index 2534edc89..1b4272980 100644
--- a/pkg/common/constants/constants.go
+++ b/pkg/common/constants/constants.go
@@ -21,4 +21,10 @@ const (
 	// KafkaChannel Spec Defaults
 	DefaultNumPartitions     = 1
 	DefaultReplicationFactor = 1
+
+	// The name of the configmap used to hold eventing-kafka settings
+	SettingsConfigMapName          = "config-kafka"
+	EventingKafkaSettingsConfigKey = "eventing-kafka"
+	// The name of the keys in the Data section of the eventing-kafka configmap that holds Sarama and Eventing-Kafka configuration YAML
+	SaramaSettingsConfigKey = "sarama"
 )
diff --git a/pkg/source/client.go b/pkg/source/client.go
index d7e64436c..1a52adf1b 100644
--- a/pkg/source/client.go
+++ b/pkg/source/client.go
@@ -20,10 +20,10 @@ import (
 	"context"
 	"fmt"
 
-	"knative.dev/eventing-kafka/pkg/common/client"
-
 	"github.com/Shopify/sarama"
 	"github.com/kelseyhightower/envconfig"
+	"knative.dev/eventing-kafka/pkg/channel/consolidated/utils"
+	"knative.dev/eventing-kafka/pkg/common/client"
 )
 
 type AdapterSASL struct {
@@ -79,7 +79,7 @@ func NewConfig(ctx context.Context) ([]string, *sarama.Config, error) {
 		WithDefaults().
 		WithAuth(kafkaAuthConfig).
 		WithVersion(&sarama.V2_0_0_0).
-		Build()
+		Build(ctx)
 	if err != nil {
 		return nil, nil, fmt.Errorf("error creating Sarama config: %w", err)
 	}
@@ -100,16 +100,15 @@ func NewProducer(ctx context.Context) (sarama.Client, error) {
 	return sarama.NewClient(bs, cfg)
 }
 
-func MakeAdminClient(clientID string, kafkaAuthCfg *client.KafkaAuthConfig, bootstrapServers []string) (sarama.ClusterAdmin, error) {
+func MakeAdminClient(ctx context.Context, clientID string, kafkaAuthCfg *client.KafkaAuthConfig, kafkaConfig *utils.KafkaConfig) (sarama.ClusterAdmin, error) {
 	saramaConf, err := client.NewConfigBuilder().
 		WithDefaults().
 		WithAuth(kafkaAuthCfg).
-		WithVersion(&sarama.V2_0_0_0).
 		WithClientId(clientID).
-		Build()
+		Build(ctx)
 	if err != nil {
 		return nil, fmt.Errorf("error creating admin client Sarama config: %w", err)
 	}
 
-	return sarama.NewClusterAdmin(bootstrapServers, saramaConf)
+	return sarama.NewClusterAdmin(kafkaConfig.Brokers, saramaConf)
 }
diff --git a/vendor/knative.dev/eventing/pkg/apis/messaging/v1/channel_template_types.go b/vendor/knative.dev/eventing/pkg/apis/messaging/v1/channel_template_types.go
index 6ef23d809..ffa0f4195 100644
--- a/vendor/knative.dev/eventing/pkg/apis/messaging/v1/channel_template_types.go
+++ b/vendor/knative.dev/eventing/pkg/apis/messaging/v1/channel_template_types.go
@@ -17,13 +17,8 @@ limitations under the License.
 package v1
 
 import (
-	"encoding/json"
-
-	"k8s.io/apimachinery/pkg/api/equality"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
-
-	v1 "knative.dev/eventing/pkg/apis/duck/v1"
 )
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
@@ -35,77 +30,3 @@ type ChannelTemplateSpec struct {
 	// +optional
 	Spec *runtime.RawExtension `json:"spec,omitempty"`
 }
-
-// ChannelTemplateSpecInternal is an internal only version that includes ObjectMeta so that
-// we can easily create new Channels off of it.
-// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
-type ChannelTemplateSpecInternal struct {
-	metav1.TypeMeta `json:",inline"`
-
-	// +optional
-	metav1.ObjectMeta `json:"metadata,omitempty"`
-
-	// Spec includes the Channel CR ChannelableSpec and the physical channel spec.
-	// In order to create a new ChannelTemplateSpecInternalSpec, you must use NewChannelTemplateSpecInternalSpec
-	Spec *ChannelTemplateSpecInternalSpec `json:"spec,omitempty"`
-}
-
-// ChannelTemplateSpecInternalSpec merges the "general" spec from Channel CR and the template of the physical channel spec.
-// Note that this struct properly implements only Marshalling, unmarshalling doesn't work!
-type ChannelTemplateSpecInternalSpec struct {
-	// ChannelableSpec includes the fields from the Channel Spec section
-	v1.ChannelableSpec
-
-	// PhysicalChannelSpec includes the fields from the physical channel Spec. Passed
-	// in verbatim to the Channel CRD as Spec section.
-	// +optional
-	PhysicalChannelSpec *runtime.RawExtension
-}
-
-// NewChannelTemplateSpecInternalSpec creates a new ChannelTemplateSpecInternalSpec, returning nil if channelableSpec is empty and physicalChannelSpec is nil.
-func NewChannelTemplateSpecInternalSpec(channelableSpec v1.ChannelableSpec, physicalChannelSpec *runtime.RawExtension) *ChannelTemplateSpecInternalSpec {
-	if physicalChannelSpec == nil && equality.Semantic.DeepEqual(channelableSpec, v1.ChannelableSpec{}) {
-		return nil
-	}
-	return &ChannelTemplateSpecInternalSpec{
-		ChannelableSpec:     channelableSpec,
-		PhysicalChannelSpec: physicalChannelSpec,
-	}
-}
-
-func (s ChannelTemplateSpecInternalSpec) MarshalJSON() ([]byte, error) {
-	// Check if empty
-	if s.PhysicalChannelSpec == nil && equality.Semantic.DeepEqual(s.ChannelableSpec, v1.ChannelableSpec{}) {
-		return []byte{}, nil
-	}
-
-	// Let's merge the channel template spec and the channelable spec from channel
-	channelableSpec := make(map[string]interface{})
-	physicalChannelTemplateSpec := make(map[string]interface{})
-
-	rawChannelSpec, err := json.Marshal(s.ChannelableSpec)
-	if err != nil {
-		return nil, err
-	}
-	if err := json.Unmarshal(rawChannelSpec, &channelableSpec); err != nil {
-		return nil, err
-	}
-
-	if s.PhysicalChannelSpec != nil {
-		rawPhysicalChannelTemplateSpec, err := json.Marshal(s.PhysicalChannelSpec)
-		if err != nil {
-			return nil, err
-		}
-		if err := json.Unmarshal(rawPhysicalChannelTemplateSpec, &physicalChannelTemplateSpec); err != nil {
-			return nil, err
-		}
-	}
-
-	// Merge the two maps into channelableSpec
-	for k, v := range physicalChannelTemplateSpec {
-		channelableSpec[k] = v
-	}
-
-	// Just return the merged map marshalled
-	return json.Marshal(channelableSpec)
-}
diff --git a/vendor/knative.dev/eventing/pkg/apis/messaging/v1/zz_generated.deepcopy.go b/vendor/knative.dev/eventing/pkg/apis/messaging/v1/zz_generated.deepcopy.go
index 90d160a32..0bdfc4e75 100644
--- a/vendor/knative.dev/eventing/pkg/apis/messaging/v1/zz_generated.deepcopy.go
+++ b/vendor/knative.dev/eventing/pkg/apis/messaging/v1/zz_generated.deepcopy.go
@@ -162,59 +162,6 @@ func (in *ChannelTemplateSpec) DeepCopyObject() runtime.Object {
 	return nil
 }
 
-// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
-func (in *ChannelTemplateSpecInternal) DeepCopyInto(out *ChannelTemplateSpecInternal) {
-	*out = *in
-	out.TypeMeta = in.TypeMeta
-	in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
-	if in.Spec != nil {
-		in, out := &in.Spec, &out.Spec
-		*out = new(ChannelTemplateSpecInternalSpec)
-		(*in).DeepCopyInto(*out)
-	}
-	return
-}
-
-// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ChannelTemplateSpecInternal.
-func (in *ChannelTemplateSpecInternal) DeepCopy() *ChannelTemplateSpecInternal {
-	if in == nil {
-		return nil
-	}
-	out := new(ChannelTemplateSpecInternal)
-	in.DeepCopyInto(out)
-	return out
-}
-
-// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
-func (in *ChannelTemplateSpecInternal) DeepCopyObject() runtime.Object {
-	if c := in.DeepCopy(); c != nil {
-		return c
-	}
-	return nil
-}
-
-// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
-func (in *ChannelTemplateSpecInternalSpec) DeepCopyInto(out *ChannelTemplateSpecInternalSpec) {
-	*out = *in
-	in.ChannelableSpec.DeepCopyInto(&out.ChannelableSpec)
-	if in.PhysicalChannelSpec != nil {
-		in, out := &in.PhysicalChannelSpec, &out.PhysicalChannelSpec
-		*out = new(runtime.RawExtension)
-		(*in).DeepCopyInto(*out)
-	}
-	return
-}
-
-// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ChannelTemplateSpecInternalSpec.
-func (in *ChannelTemplateSpecInternalSpec) DeepCopy() *ChannelTemplateSpecInternalSpec {
-	if in == nil {
-		return nil
-	}
-	out := new(ChannelTemplateSpecInternalSpec)
-	in.DeepCopyInto(out)
-	return out
-}
-
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *InMemoryChannel) DeepCopyInto(out *InMemoryChannel) {
 	*out = *in
diff --git a/vendor/knative.dev/eventing/pkg/apis/messaging/v1beta1/channel_template_types.go b/vendor/knative.dev/eventing/pkg/apis/messaging/v1beta1/channel_template_types.go
index 02a5581b1..699fc40c2 100644
--- a/vendor/knative.dev/eventing/pkg/apis/messaging/v1beta1/channel_template_types.go
+++ b/vendor/knative.dev/eventing/pkg/apis/messaging/v1beta1/channel_template_types.go
@@ -30,18 +30,3 @@ type ChannelTemplateSpec struct {
 	// +optional
 	Spec *runtime.RawExtension `json:"spec,omitempty"`
 }
-
-// ChannelTemplateSpecInternal is an internal only version that includes ObjectMeta so that
-// we can easily create new Channels off of it.
-// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
-type ChannelTemplateSpecInternal struct {
-	metav1.TypeMeta `json:",inline"`
-
-	// +optional
-	metav1.ObjectMeta `json:"metadata,omitempty"`
-
-	// Spec defines the Spec to use for each channel created. Passed
-	// in verbatim to the Channel CRD as Spec section.
-	// +optional
-	Spec *runtime.RawExtension `json:"spec,omitempty"`
-}
diff --git a/vendor/knative.dev/eventing/pkg/apis/messaging/v1beta1/zz_generated.deepcopy.go b/vendor/knative.dev/eventing/pkg/apis/messaging/v1beta1/zz_generated.deepcopy.go
index 8d4631b7f..df2dcad7f 100644
--- a/vendor/knative.dev/eventing/pkg/apis/messaging/v1beta1/zz_generated.deepcopy.go
+++ b/vendor/knative.dev/eventing/pkg/apis/messaging/v1beta1/zz_generated.deepcopy.go
@@ -162,37 +162,6 @@ func (in *ChannelTemplateSpec) DeepCopyObject() runtime.Object {
 	return nil
 }
 
-// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
-func (in *ChannelTemplateSpecInternal) DeepCopyInto(out *ChannelTemplateSpecInternal) {
-	*out = *in
-	out.TypeMeta = in.TypeMeta
-	in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
-	if in.Spec != nil {
-		in, out := &in.Spec, &out.Spec
-		*out = new(runtime.RawExtension)
-		(*in).DeepCopyInto(*out)
-	}
-	return
-}
-
-// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ChannelTemplateSpecInternal.
-func (in *ChannelTemplateSpecInternal) DeepCopy() *ChannelTemplateSpecInternal {
-	if in == nil {
-		return nil
-	}
-	out := new(ChannelTemplateSpecInternal)
-	in.DeepCopyInto(out)
-	return out
-}
-
-// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
-func (in *ChannelTemplateSpecInternal) DeepCopyObject() runtime.Object {
-	if c := in.DeepCopy(); c != nil {
-		return c
-	}
-	return nil
-}
-
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *InMemoryChannel) DeepCopyInto(out *InMemoryChannel) {
 	*out = *in
diff --git a/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker/controller.go b/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker/controller.go
index eafcd40a4..74dea84cd 100644
--- a/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker/controller.go
+++ b/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker/controller.go
@@ -24,6 +24,7 @@ import (
 	reflect "reflect"
 	strings "strings"
 
+	zap "go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	labels "k8s.io/apimachinery/pkg/labels"
 	types "k8s.io/apimachinery/pkg/types"
@@ -37,6 +38,7 @@ import (
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	controller "knative.dev/pkg/controller"
 	logging "knative.dev/pkg/logging"
+	logkey "knative.dev/pkg/logging/logkey"
 	reconciler "knative.dev/pkg/reconciler"
 )
 
@@ -88,10 +90,16 @@ func NewImpl(ctx context.Context, r Interface, classValue string, optionsFns ...
 		classValue:    classValue,
 	}
 
-	t := reflect.TypeOf(r).Elem()
-	queueName := fmt.Sprintf("%s.%s", strings.ReplaceAll(t.PkgPath(), "/", "-"), t.Name())
+	ctrType := reflect.TypeOf(r).Elem()
+	ctrTypeName := fmt.Sprintf("%s.%s", ctrType.PkgPath(), ctrType.Name())
+	ctrTypeName = strings.ReplaceAll(ctrTypeName, "/", ".")
 
-	impl := controller.NewImpl(rec, logger, queueName)
+	logger = logger.With(
+		zap.String(logkey.ControllerType, ctrTypeName),
+		zap.String(logkey.Kind, "eventing.knative.dev.Broker"),
+	)
+
+	impl := controller.NewImpl(rec, logger, ctrTypeName)
 	agentName := defaultControllerAgentName
 
 	// Pass impl to the options. Save any optional results.
diff --git a/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker/reconciler.go b/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker/reconciler.go
index 0dd556f70..4b777b492 100644
--- a/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker/reconciler.go
+++ b/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker/reconciler.go
@@ -195,7 +195,7 @@ func (r *reconcilerImpl) Reconcile(ctx context.Context, key string) error {
 	// If we are not the leader, and we don't implement either ReadOnly
 	// observer interfaces, then take a fast-path out.
 	if s.isNotLeaderNorObserver() {
-		return nil
+		return controller.NewSkipKey(key)
 	}
 
 	// If configStore is set, attach the frozen configuration to the context.
diff --git a/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1beta1/broker/controller.go b/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1beta1/broker/controller.go
index 1a521f093..19a4a4454 100644
--- a/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1beta1/broker/controller.go
+++ b/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1beta1/broker/controller.go
@@ -24,6 +24,7 @@ import (
 	reflect "reflect"
 	strings "strings"
 
+	zap "go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	labels "k8s.io/apimachinery/pkg/labels"
 	types "k8s.io/apimachinery/pkg/types"
@@ -37,6 +38,7 @@ import (
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	controller "knative.dev/pkg/controller"
 	logging "knative.dev/pkg/logging"
+	logkey "knative.dev/pkg/logging/logkey"
 	reconciler "knative.dev/pkg/reconciler"
 )
 
@@ -88,10 +90,16 @@ func NewImpl(ctx context.Context, r Interface, classValue string, optionsFns ...
 		classValue:    classValue,
 	}
 
-	t := reflect.TypeOf(r).Elem()
-	queueName := fmt.Sprintf("%s.%s", strings.ReplaceAll(t.PkgPath(), "/", "-"), t.Name())
+	ctrType := reflect.TypeOf(r).Elem()
+	ctrTypeName := fmt.Sprintf("%s.%s", ctrType.PkgPath(), ctrType.Name())
+	ctrTypeName = strings.ReplaceAll(ctrTypeName, "/", ".")
 
-	impl := controller.NewImpl(rec, logger, queueName)
+	logger = logger.With(
+		zap.String(logkey.ControllerType, ctrTypeName),
+		zap.String(logkey.Kind, "eventing.knative.dev.Broker"),
+	)
+
+	impl := controller.NewImpl(rec, logger, ctrTypeName)
 	agentName := defaultControllerAgentName
 
 	// Pass impl to the options. Save any optional results.
diff --git a/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1beta1/broker/reconciler.go b/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1beta1/broker/reconciler.go
index e13d2d663..164866f22 100644
--- a/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1beta1/broker/reconciler.go
+++ b/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1beta1/broker/reconciler.go
@@ -195,7 +195,7 @@ func (r *reconcilerImpl) Reconcile(ctx context.Context, key string) error {
 	// If we are not the leader, and we don't implement either ReadOnly
 	// observer interfaces, then take a fast-path out.
 	if s.isNotLeaderNorObserver() {
-		return nil
+		return controller.NewSkipKey(key)
 	}
 
 	// If configStore is set, attach the frozen configuration to the context.
diff --git a/vendor/knative.dev/eventing/pkg/duck/channel.go b/vendor/knative.dev/eventing/pkg/duck/channel.go
new file mode 100644
index 000000000..4c4604cd5
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/duck/channel.go
@@ -0,0 +1,165 @@
+/*
+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 duck
+
+import (
+	"encoding/json"
+
+	"k8s.io/apimachinery/pkg/api/equality"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+
+	duckv1 "knative.dev/eventing/pkg/apis/duck/v1"
+)
+
+// channelTemplateSpecInternal is an internal only version that includes ObjectMeta so that
+// we can easily create new Channels off of it.
+type channelTemplateSpecInternal struct {
+	metav1.TypeMeta `json:",inline"`
+
+	// +optional
+	metav1.ObjectMeta `json:"metadata,omitempty"`
+
+	// Spec includes the Channel CR ChannelableSpec and the physical channel spec.
+	// In order to create a new channelTemplateSpecInternalSpec, you must use NewChannelTemplateSpecInternalSpec
+	// +optional
+	Spec *channelTemplateSpecInternalSpec `json:"spec,omitempty"`
+}
+
+// channelTemplateSpecInternalSpec merges the "general" spec from Channel CR and the template of the physical channel spec.
+// Note that this struct properly implements only Marshalling, unmarshalling doesn't work!
+type channelTemplateSpecInternalSpec struct {
+	// channelableSpec includes the fields from the Channel Spec section
+	channelableSpec *duckv1.ChannelableSpec
+
+	// physicalChannelSpec includes the fields from the physical channel Spec. Passed
+	// in verbatim to the Channel CRD as Spec section.
+	// +optional
+	physicalChannelSpec *runtime.RawExtension
+}
+
+func (s *channelTemplateSpecInternalSpec) MarshalJSON() ([]byte, error) {
+	// Let's merge the channel template spec and the channelable spec from channel
+	channelableSpec := make(map[string]interface{})
+	physicalChannelTemplateSpec := make(map[string]interface{})
+
+	var cs duckv1.ChannelableSpec
+	if s.channelableSpec != nil {
+		cs = *s.channelableSpec
+	} else {
+		cs = duckv1.ChannelableSpec{}
+	}
+
+	rawChannelSpec, err := json.Marshal(cs)
+	if err != nil {
+		return nil, err
+	}
+	if err := json.Unmarshal(rawChannelSpec, &channelableSpec); err != nil {
+		return nil, err
+	}
+
+	if s.physicalChannelSpec != nil {
+		rawPhysicalChannelTemplateSpec, err := json.Marshal(s.physicalChannelSpec)
+		if err != nil {
+			return nil, err
+		}
+		if err := json.Unmarshal(rawPhysicalChannelTemplateSpec, &physicalChannelTemplateSpec); err != nil {
+			return nil, err
+		}
+	}
+
+	// Merge the two maps into channelableSpec
+	for k, v := range physicalChannelTemplateSpec {
+		channelableSpec[k] = v
+	}
+
+	// Just return the merged map marshalled
+	return json.Marshal(channelableSpec)
+}
+
+// PhysicalChannelOption represents an option for NewPhysicalChannel.
+type PhysicalChannelOption func(*channelTemplateSpecInternal)
+
+// WithChannelableSpec sets the ChannelableSpec of the physical channel.
+func WithChannelableSpec(channelableSpec duckv1.ChannelableSpec) PhysicalChannelOption {
+	return func(internal *channelTemplateSpecInternal) {
+		if equality.Semantic.DeepEqual(channelableSpec, duckv1.ChannelableSpec{}) {
+			// No need to set it
+			return
+		}
+		if internal.Spec == nil {
+			internal.Spec = &channelTemplateSpecInternalSpec{}
+		}
+		internal.Spec.channelableSpec = &channelableSpec
+	}
+}
+
+// WithPhysicalChannelSpec sets the ChannelableSpec of the physical channel.
+func WithPhysicalChannelSpec(physicalChannelSpec *runtime.RawExtension) PhysicalChannelOption {
+	return func(internal *channelTemplateSpecInternal) {
+		if physicalChannelSpec == nil {
+			// No need to set it
+			return
+		}
+		if internal.Spec == nil {
+			internal.Spec = &channelTemplateSpecInternalSpec{}
+		}
+		internal.Spec.physicalChannelSpec = physicalChannelSpec
+	}
+}
+
+// NewPhysicalChannel returns a new physical channel as unstructured.Unstructured, starting from the provided meta and ducks.
+// When developing components that needs to spawn underlying (physical) channels (e.g. the Channel type, brokers backed by channels, parallel, sequence)
+// you should use this function to create the underlying channels.
+// Any physical channel CRD is composed by TypeMeta, ObjectMeta and a Spec, which includes
+// a ChannelableSpec and optionally additional fields (e.g. like KafkaChannel includes partitionCount).
+// You can set the ChannelableSpec using WithChannelableSpec and you can set all the additional fields using WithPhysicalChannelSpec
+// This function returns the channel with the provided TypeMeta and ObjectMeta and with the additional provided options.
+//
+// For example, providing the TypeMeta of an InMemoryChannel and a ChannelableSpec with delivery configured, the return value will look like:
+//
+// apiVersion: messaging.knative.dev/v1
+// kind: InMemoryChannel
+// metadata:
+//   name: "hello"
+//   namespace: "world"
+// spec:
+//   delivery:
+//     retry: 3
+func NewPhysicalChannel(typeMeta metav1.TypeMeta, objMeta metav1.ObjectMeta, opts ...PhysicalChannelOption) (*unstructured.Unstructured, error) {
+	// Set the name of the resource we're creating as well as the namespace, etc.
+	template := channelTemplateSpecInternal{
+		TypeMeta:   typeMeta,
+		ObjectMeta: objMeta,
+	}
+
+	for _, opt := range opts {
+		opt(&template)
+	}
+
+	raw, err := json.Marshal(template)
+	if err != nil {
+		return nil, err
+	}
+	u := &unstructured.Unstructured{}
+	err = json.Unmarshal(raw, u)
+	if err != nil {
+		return nil, err
+	}
+	return u, nil
+}
diff --git a/vendor/knative.dev/eventing/pkg/duck/listable.go b/vendor/knative.dev/eventing/pkg/duck/listable.go
new file mode 100644
index 000000000..95bea2d68
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/duck/listable.go
@@ -0,0 +1,203 @@
+/*
+Copyright 2019 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 duck
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/pkg/errors"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/equality"
+	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/tools/cache"
+	"knative.dev/pkg/apis/duck"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/tracker"
+)
+
+// ListableTracker is a tracker capable of tracking any object that implements the apis.Listable interface.
+type ListableTracker interface {
+	// TrackInNamespace returns a function that can be used to watch arbitrary apis.Listable resources in the same
+	// namespace as obj. Any change will cause a callback for obj.
+	TrackInNamespace(ctx context.Context, obj metav1.Object) Track
+	// TrackInNamespaceKReference returns a function that can be used to watch arbitrary apis.Listable resources
+	// in the same namespace as obj. Any change will cause a callback for obj.
+	TrackInNamespaceKReference(ctx context.Context, obj metav1.Object) TrackKReference
+	// ListerFor returns the lister for the object reference. It returns an error if the lister does not exist.
+	ListerFor(ref corev1.ObjectReference) (cache.GenericLister, error)
+	// InformerFor returns the informer for the object reference. It returns an error if the informer does not exist.
+	InformerFor(ref corev1.ObjectReference) (cache.SharedIndexInformer, error)
+	// ListerFor returns the lister for the KReference. It returns an error if the lister does not exist.
+	ListerForKReference(ref duckv1.KReference) (cache.GenericLister, error)
+	// InformerFor returns the informer for the KReference. It returns an error if the informer does not exist.
+	InformerForKReference(ref duckv1.KReference) (cache.SharedIndexInformer, error)
+}
+
+type Track func(corev1.ObjectReference) error
+type TrackKReference func(duckv1.KReference) error
+
+// NewListableTracker creates a new ListableTracker, backed by a TypedInformerFactory.
+func NewListableTracker(ctx context.Context, getter func(context.Context) duck.InformerFactory, callback func(types.NamespacedName), lease time.Duration) ListableTracker {
+	return &listableTracker{
+		informerFactory: getter(ctx),
+		tracker:         tracker.New(callback, lease),
+		concrete:        map[schema.GroupVersionResource]informerListerPair{},
+	}
+}
+
+type listableTracker struct {
+	informerFactory duck.InformerFactory
+	tracker         tracker.Interface
+
+	concrete     map[schema.GroupVersionResource]informerListerPair
+	concreteLock sync.RWMutex
+}
+
+type informerListerPair struct {
+	informer cache.SharedIndexInformer
+	lister   cache.GenericLister
+}
+
+// ensureTracking ensures that there is an informer watching and sending events to tracker for the
+// concrete GVK. It also ensures that there is the corresponding lister for that informer.
+func (t *listableTracker) ensureTracking(ctx context.Context, ref corev1.ObjectReference) error {
+	if equality.Semantic.DeepEqual(ref, &corev1.ObjectReference{}) {
+		return errors.New("cannot track empty object ref")
+	}
+	gvk := ref.GroupVersionKind()
+	gvr, _ := meta.UnsafeGuessKindToResource(gvk)
+
+	t.concreteLock.RLock()
+	_, present := t.concrete[gvr]
+	t.concreteLock.RUnlock()
+	if present {
+		// There is already an informer running for this GVR, we don't need or want to make
+		// a second one.
+		return nil
+	}
+	// Tracking has not been setup for this GVR.
+	t.concreteLock.Lock()
+	defer t.concreteLock.Unlock()
+	// Now that we have acquired the write lock, make sure no one has added tracking handlers.
+	if _, present = t.concrete[gvr]; present {
+		return nil
+	}
+	informer, lister, err := t.informerFactory.Get(ctx, gvr)
+	if err != nil {
+		return err
+	}
+	informer.AddEventHandler(controller.HandleAll(
+		// Call the tracker's OnChanged method, but we've seen the objects coming through
+		// this path missing TypeMeta, so ensure it is properly populated.
+		controller.EnsureTypeMeta(
+			t.tracker.OnChanged,
+			gvk,
+		),
+	))
+	t.concrete[gvr] = informerListerPair{informer: informer, lister: lister}
+	return nil
+}
+
+// TrackInNamespace satisfies the ListableTracker interface.
+func (t *listableTracker) TrackInNamespace(ctx context.Context, obj metav1.Object) Track {
+	return func(ref corev1.ObjectReference) error {
+		// This is often used by Trigger and Subscription, both of which pass in refs that do not
+		// specify the namespace.
+		ref.Namespace = obj.GetNamespace()
+		if err := t.ensureTracking(ctx, ref); err != nil {
+			return err
+		}
+
+		return t.tracker.TrackReference(tracker.Reference{
+			APIVersion: ref.APIVersion,
+			Kind:       ref.Kind,
+			Namespace:  obj.GetNamespace(),
+			Name:       ref.Name,
+		}, obj)
+	}
+}
+
+// TrackInNamespaceKReference satisfies the ListableTracker interface.
+func (t *listableTracker) TrackInNamespaceKReference(ctx context.Context, obj metav1.Object) TrackKReference {
+	return func(ref duckv1.KReference) error {
+		// This is often used by Trigger and Subscription, both of which pass in refs that do not
+		// specify the namespace.
+		ref.Namespace = obj.GetNamespace()
+		coreRef := corev1.ObjectReference{APIVersion: ref.APIVersion, Kind: ref.Kind, Name: ref.Name, Namespace: ref.Namespace}
+		if err := t.ensureTracking(ctx, coreRef); err != nil {
+			return err
+		}
+
+		return t.tracker.TrackReference(tracker.Reference{
+			APIVersion: ref.APIVersion,
+			Kind:       ref.Kind,
+			Namespace:  obj.GetNamespace(),
+			Name:       ref.Name,
+		}, obj)
+	}
+}
+
+// ListerForKReference satisfies the ListableTracker interface.
+func (t *listableTracker) ListerForKReference(ref duckv1.KReference) (cache.GenericLister, error) {
+	return t.ListerFor(corev1.ObjectReference{APIVersion: ref.APIVersion, Kind: ref.Kind, Name: ref.Name, Namespace: ref.Namespace})
+}
+
+// ListerFor satisfies the ListableTracker interface.
+func (t *listableTracker) ListerFor(ref corev1.ObjectReference) (cache.GenericLister, error) {
+	if equality.Semantic.DeepEqual(ref, &corev1.ObjectReference{}) {
+		return nil, errors.New("cannot get lister for empty object ref")
+	}
+	gvk := ref.GroupVersionKind()
+	gvr, _ := meta.UnsafeGuessKindToResource(gvk)
+
+	t.concreteLock.RLock()
+	defer t.concreteLock.RUnlock()
+	informerListerPair, present := t.concrete[gvr]
+	if !present {
+		return nil, fmt.Errorf("no lister available for GVR %s", gvr.String())
+	}
+	return informerListerPair.lister, nil
+}
+
+// InformerForKReference satisfies the ListableTracker interface.
+func (t *listableTracker) InformerForKReference(ref duckv1.KReference) (cache.SharedIndexInformer, error) {
+	return t.InformerFor(corev1.ObjectReference{APIVersion: ref.APIVersion, Kind: ref.Kind, Name: ref.Name, Namespace: ref.Namespace})
+}
+
+// InformerFor satisfies the ListableTracker interface.
+func (t *listableTracker) InformerFor(ref corev1.ObjectReference) (cache.SharedIndexInformer, error) {
+	if equality.Semantic.DeepEqual(ref, &corev1.ObjectReference{}) {
+		return nil, errors.New("cannot get informer for empty object ref")
+	}
+	gvk := ref.GroupVersionKind()
+	gvr, _ := meta.UnsafeGuessKindToResource(gvk)
+
+	t.concreteLock.RLock()
+	defer t.concreteLock.RUnlock()
+	informerListerPair, present := t.concrete[gvr]
+	if !present {
+		return nil, fmt.Errorf("no informer available for GVR %s", gvr.String())
+	}
+	return informerListerPair.informer, nil
+}
diff --git a/vendor/knative.dev/eventing/pkg/duck/subscriber.go b/vendor/knative.dev/eventing/pkg/duck/subscriber.go
new file mode 100644
index 000000000..8b7c2a91f
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/duck/subscriber.go
@@ -0,0 +1,64 @@
+/*
+Copyright 2019 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 duck
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/url"
+
+	"go.uber.org/zap"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/dynamic"
+
+	duckapis "knative.dev/pkg/apis"
+	"knative.dev/pkg/logging"
+)
+
+// DomainToURL converts a domain into an HTTP URL.
+func DomainToURL(domain string) string {
+	u := url.URL{
+		Scheme: "http",
+		Host:   domain,
+		Path:   "/",
+	}
+	return u.String()
+}
+
+// ResourceInterface creates a resource interface for the given ObjectReference.
+func ResourceInterface(dynamicClient dynamic.Interface, namespace string, gvk schema.GroupVersionKind) (dynamic.ResourceInterface, error) {
+	rc := dynamicClient.Resource(duckapis.KindToResource(gvk))
+
+	if rc == nil {
+		return nil, fmt.Errorf("failed to create dynamic client resource")
+	}
+	return rc.Namespace(namespace), nil
+}
+
+// ObjectReference resolves an object based on an ObjectReference.
+func ObjectReference(ctx context.Context, dynamicClient dynamic.Interface, namespace string, ref *corev1.ObjectReference) (json.Marshaler, error) {
+	resourceClient, err := ResourceInterface(dynamicClient, namespace, ref.GroupVersionKind())
+	if err != nil {
+		logging.FromContext(ctx).Warnw("Failed to create dynamic resource client", zap.Error(err))
+		return nil, err
+	}
+
+	return resourceClient.Get(ctx, ref.Name, metav1.GetOptions{})
+}
diff --git a/vendor/knative.dev/pkg/apis/duck/ABOUT.md b/vendor/knative.dev/pkg/apis/duck/ABOUT.md
index 0c6edabde..98ffc9424 100644
--- a/vendor/knative.dev/pkg/apis/duck/ABOUT.md
+++ b/vendor/knative.dev/pkg/apis/duck/ABOUT.md
@@ -301,7 +301,8 @@ is used with duck types:
                 Base:             reconciler.NewBase(opt, controllerAgentName),
                 ...
         }
-        impl := controller.NewImpl(c, c.Logger, "Revisions")
+        logger := c.Logger.Named("Revisions")
+        impl := controller.NewImpl(c, logger, "Revisions")
 
         // Calls to Track create a 30 minute lease before they must be renewed.
         // Coordinate this value with controller resync periods.
diff --git a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/fakeclient.go b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/fake_client.go
similarity index 100%
rename from vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/fakeclient.go
rename to vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/fake_client.go
diff --git a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/fakeduck.go b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/fake_duck.go
similarity index 100%
rename from vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/fakeduck.go
rename to vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/fake_duck.go
diff --git a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/fakefactory.go b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/fake_factory.go
similarity index 100%
rename from vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/fakefactory.go
rename to vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/fake_factory.go
diff --git a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/fake_filtered_factory.go b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/fake_filtered_factory.go
new file mode 100644
index 000000000..153261dc4
--- /dev/null
+++ b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/fake_filtered_factory.go
@@ -0,0 +1,131 @@
+/*
+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 generators
+
+import (
+	"io"
+
+	"k8s.io/gengo/generator"
+	"k8s.io/gengo/namer"
+	"k8s.io/gengo/types"
+	"k8s.io/klog"
+)
+
+// fakeFilteredFactoryGenerator produces a file of listers for a given GroupVersion and
+// type.
+type fakeFilteredFactoryGenerator struct {
+	generator.DefaultGen
+	outputPackage string
+	imports       namer.ImportTracker
+	filtered      bool
+
+	factoryInjectionPkg          string
+	fakeClientInjectionPkg       string
+	sharedInformerFactoryPackage string
+}
+
+var _ generator.Generator = (*fakeFilteredFactoryGenerator)(nil)
+
+func (g *fakeFilteredFactoryGenerator) Filter(c *generator.Context, t *types.Type) bool {
+	// We generate a single factory, so return true once.
+	if !g.filtered {
+		g.filtered = true
+		return true
+	}
+	return false
+}
+
+func (g *fakeFilteredFactoryGenerator) Namers(c *generator.Context) namer.NameSystems {
+	return namer.NameSystems{
+		"raw": namer.NewRawNamer(g.outputPackage, g.imports),
+	}
+}
+
+func (g *fakeFilteredFactoryGenerator) Imports(c *generator.Context) (imports []string) {
+	imports = append(imports, g.imports.ImportLines()...)
+	return
+}
+
+func (g *fakeFilteredFactoryGenerator) GenerateType(c *generator.Context, t *types.Type, w io.Writer) error {
+	sw := generator.NewSnippetWriter(w, c, "{{", "}}")
+
+	klog.V(5).Info("processing type ", t)
+
+	m := map[string]interface{}{
+		"factoryKey":      c.Universe.Type(types.Name{Package: g.factoryInjectionPkg, Name: "Key"}),
+		"factoryLabelKey": c.Universe.Type(types.Name{Package: g.factoryInjectionPkg, Name: "LabelKey"}),
+		"factoryGet":      c.Universe.Function(types.Name{Package: g.factoryInjectionPkg, Name: "Get"}),
+		"clientGet":       c.Universe.Function(types.Name{Package: g.fakeClientInjectionPkg, Name: "Get"}),
+		"informersNewSharedInformerFactoryWithOptions": c.Universe.Function(types.Name{Package: g.sharedInformerFactoryPackage, Name: "NewSharedInformerFactoryWithOptions"}),
+		"informersSharedInformerOption":                c.Universe.Function(types.Name{Package: g.sharedInformerFactoryPackage, Name: "SharedInformerOption"}),
+		"informersWithNamespace":                       c.Universe.Function(types.Name{Package: g.sharedInformerFactoryPackage, Name: "WithNamespace"}),
+		"informersWithTweakListOptions":                c.Universe.Function(types.Name{Package: g.sharedInformerFactoryPackage, Name: "WithTweakListOptions"}),
+		"injectionRegisterInformerFactory": c.Universe.Function(types.Name{
+			Package: "knative.dev/pkg/injection",
+			Name:    "Fake.RegisterInformerFactory",
+		}),
+		"injectionHasNamespace":     c.Universe.Type(types.Name{Package: "knative.dev/pkg/injection", Name: "HasNamespaceScope"}),
+		"injectionGetNamespace":     c.Universe.Type(types.Name{Package: "knative.dev/pkg/injection", Name: "GetNamespaceScope"}),
+		"controllerGetResyncPeriod": c.Universe.Type(types.Name{Package: "knative.dev/pkg/controller", Name: "GetResyncPeriod"}),
+		"contextContext": c.Universe.Type(types.Name{
+			Package: "context",
+			Name:    "Context",
+		}),
+		"loggingFromContext": c.Universe.Function(types.Name{
+			Package: "knative.dev/pkg/logging",
+			Name:    "FromContext",
+		}),
+		"metav1ListOptions": c.Universe.Type(types.Name{
+			Package: "k8s.io/apimachinery/pkg/apis/meta/v1",
+			Name:    "ListOptions",
+		}),
+	}
+
+	sw.Do(injectionFakeFilteredInformerFactory, m)
+
+	return sw.Error()
+}
+
+var injectionFakeFilteredInformerFactory = `
+var Get = {{.factoryGet|raw}}
+
+func init() {
+	{{.injectionRegisterInformerFactory|raw}}(withInformerFactory)
+}
+
+func withInformerFactory(ctx {{.contextContext|raw}}) {{.contextContext|raw}} {
+	c := {{.clientGet|raw}}(ctx)
+	opts := []{{.informersSharedInformerOption|raw}}{}
+	if {{.injectionHasNamespace|raw}}(ctx) {
+		opts = append(opts, {{.informersWithNamespace|raw}}({{.injectionGetNamespace|raw}}(ctx)))
+	}
+	untyped := ctx.Value({{.factoryLabelKey|raw}}{})
+	if untyped == nil {
+		{{.loggingFromContext|raw}}(ctx).Panic(
+			"Unable to fetch labelkey from context.")
+	}
+	labelSelectors := untyped.([]string)
+	for _, selector := range labelSelectors {
+		thisOpts := append(opts, {{.informersWithTweakListOptions|raw}}(func(l *{{.metav1ListOptions|raw}}) {
+			l.LabelSelector = selector
+		}))
+		ctx = context.WithValue(ctx, {{.factoryKey|raw}}{Selector: selector},
+			{{.informersNewSharedInformerFactoryWithOptions|raw}}(c, {{.controllerGetResyncPeriod|raw}}(ctx), thisOpts...))
+	}
+	return ctx
+}
+`
diff --git a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/fake_filtered_informer.go b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/fake_filtered_informer.go
new file mode 100644
index 000000000..47b700884
--- /dev/null
+++ b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/fake_filtered_informer.go
@@ -0,0 +1,133 @@
+/*
+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 generators
+
+import (
+	"io"
+
+	clientgentypes "k8s.io/code-generator/cmd/client-gen/types"
+	"k8s.io/gengo/generator"
+	"k8s.io/gengo/namer"
+	"k8s.io/gengo/types"
+	"k8s.io/klog"
+)
+
+// fakeFilteredInformerGenerator produces a file of listers for a given GroupVersion and
+// type.
+type fakeFilteredInformerGenerator struct {
+	generator.DefaultGen
+	outputPackage string
+	imports       namer.ImportTracker
+
+	typeToGenerate          *types.Type
+	groupVersion            clientgentypes.GroupVersion
+	groupGoName             string
+	informerInjectionPkg    string
+	fakeFactoryInjectionPkg string
+}
+
+var _ generator.Generator = (*fakeFilteredInformerGenerator)(nil)
+
+func (g *fakeFilteredInformerGenerator) Filter(c *generator.Context, t *types.Type) bool {
+	// Only process the type for this informer generator.
+	return t == g.typeToGenerate
+}
+
+func (g *fakeFilteredInformerGenerator) Namers(c *generator.Context) namer.NameSystems {
+	publicPluralNamer := &ExceptionNamer{
+		Exceptions: map[string]string{
+			// these exceptions are used to deconflict the generated code
+			// you can put your fully qualified package like
+			// to generate a name that doesn't conflict with your group.
+			// "k8s.io/apis/events/v1beta1.Event": "EventResource"
+		},
+		KeyFunc: func(t *types.Type) string {
+			return t.Name.Package + "." + t.Name.Name
+		},
+		Delegate: namer.NewPublicPluralNamer(map[string]string{
+			"Endpoints": "Endpoints",
+		}),
+	}
+
+	return namer.NameSystems{
+		"raw":          namer.NewRawNamer(g.outputPackage, g.imports),
+		"publicPlural": publicPluralNamer,
+	}
+}
+
+func (g *fakeFilteredInformerGenerator) Imports(c *generator.Context) (imports []string) {
+	imports = append(imports, g.imports.ImportLines()...)
+	return
+}
+
+func (g *fakeFilteredInformerGenerator) GenerateType(c *generator.Context, t *types.Type, w io.Writer) error {
+	sw := generator.NewSnippetWriter(w, c, "{{", "}}")
+
+	klog.V(5).Info("processing type ", t)
+
+	m := map[string]interface{}{
+		"informerKey":        c.Universe.Type(types.Name{Package: g.informerInjectionPkg, Name: "Key"}),
+		"informerGet":        c.Universe.Function(types.Name{Package: g.informerInjectionPkg, Name: "Get"}),
+		"factoryGet":         c.Universe.Function(types.Name{Package: g.fakeFactoryInjectionPkg, Name: "Get"}),
+		"factoryLabelKey":    c.Universe.Type(types.Name{Package: g.fakeFactoryInjectionPkg, Name: "LabelKey"}),
+		"group":              namer.IC(g.groupGoName),
+		"type":               t,
+		"version":            namer.IC(g.groupVersion.Version.String()),
+		"controllerInformer": c.Universe.Type(types.Name{Package: "knative.dev/pkg/controller", Name: "Informer"}),
+		"injectionRegisterFilteredInformers": c.Universe.Function(types.Name{
+			Package: "knative.dev/pkg/injection",
+			Name:    "Fake.RegisterFilteredInformers",
+		}),
+		"loggingFromContext": c.Universe.Function(types.Name{
+			Package: "knative.dev/pkg/logging",
+			Name:    "FromContext",
+		}),
+		"contextContext": c.Universe.Type(types.Name{
+			Package: "context",
+			Name:    "Context",
+		}),
+	}
+
+	sw.Do(injectionFakeFilteredInformer, m)
+
+	return sw.Error()
+}
+
+var injectionFakeFilteredInformer = `
+var Get = {{.informerGet|raw}}
+
+func init() {
+	{{.injectionRegisterFilteredInformers|raw}}(withInformer)
+}
+
+func withInformer(ctx {{.contextContext|raw}}) ({{.contextContext|raw}}, []{{.controllerInformer|raw}}) {
+	untyped := ctx.Value({{.factoryLabelKey|raw}}{})
+	if untyped == nil {
+		{{.loggingFromContext|raw}}(ctx).Panic(
+			"Unable to fetch labelkey from context.")
+	}
+	labelSelectors := untyped.([]string)
+	infs := []{{.controllerInformer|raw}}{}
+	for _, selector := range labelSelectors {
+		f := {{.factoryGet|raw}}(ctx, selector)
+		inf := f.{{.group}}().{{.version}}().{{.type|publicPlural}}()
+		ctx = context.WithValue(ctx, {{.informerKey|raw}}{Selector: selector}, inf)
+		infs = append(infs, inf.Informer())
+	}
+	return ctx, infs
+}
+`
diff --git a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/fakeinformer.go b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/fake_informer.go
similarity index 100%
rename from vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/fakeinformer.go
rename to vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/fake_informer.go
diff --git a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/filtered_factory.go b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/filtered_factory.go
new file mode 100644
index 000000000..b036cf1cf
--- /dev/null
+++ b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/filtered_factory.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 generators
+
+import (
+	"io"
+
+	"k8s.io/gengo/generator"
+	"k8s.io/gengo/namer"
+	"k8s.io/gengo/types"
+
+	"k8s.io/klog"
+)
+
+// factoryTestGenerator produces a file of factory injection of a given type.
+type filteredFactoryGenerator struct {
+	generator.DefaultGen
+	outputPackage                string
+	imports                      namer.ImportTracker
+	cachingClientSetPackage      string
+	sharedInformerFactoryPackage string
+	filtered                     bool
+}
+
+var _ generator.Generator = (*filteredFactoryGenerator)(nil)
+
+func (g *filteredFactoryGenerator) Filter(c *generator.Context, t *types.Type) bool {
+	// We generate a single factory, so return true once.
+	if !g.filtered {
+		g.filtered = true
+		return true
+	}
+	return false
+}
+
+func (g *filteredFactoryGenerator) Namers(c *generator.Context) namer.NameSystems {
+	return namer.NameSystems{
+		"raw": namer.NewRawNamer(g.outputPackage, g.imports),
+	}
+}
+
+func (g *filteredFactoryGenerator) Imports(c *generator.Context) (imports []string) {
+	imports = append(imports, g.imports.ImportLines()...)
+	return
+}
+
+func (g *filteredFactoryGenerator) GenerateType(c *generator.Context, t *types.Type, w io.Writer) error {
+	sw := generator.NewSnippetWriter(w, c, "{{", "}}")
+
+	klog.V(5).Info("processing type ", t)
+
+	m := map[string]interface{}{
+		"cachingClientGet": c.Universe.Type(types.Name{Package: g.cachingClientSetPackage, Name: "Get"}),
+		"informersNewSharedInformerFactoryWithOptions": c.Universe.Function(types.Name{Package: g.sharedInformerFactoryPackage, Name: "NewSharedInformerFactoryWithOptions"}),
+		"informersSharedInformerOption":                c.Universe.Function(types.Name{Package: g.sharedInformerFactoryPackage, Name: "SharedInformerOption"}),
+		"informersWithNamespace":                       c.Universe.Function(types.Name{Package: g.sharedInformerFactoryPackage, Name: "WithNamespace"}),
+		"informersWithTweakListOptions":                c.Universe.Function(types.Name{Package: g.sharedInformerFactoryPackage, Name: "WithTweakListOptions"}),
+		"informersSharedInformerFactory":               c.Universe.Function(types.Name{Package: g.sharedInformerFactoryPackage, Name: "SharedInformerFactory"}),
+		"injectionRegisterInformerFactory":             c.Universe.Type(types.Name{Package: "knative.dev/pkg/injection", Name: "Default.RegisterInformerFactory"}),
+		"injectionHasNamespace":                        c.Universe.Type(types.Name{Package: "knative.dev/pkg/injection", Name: "HasNamespaceScope"}),
+		"injectionGetNamespace":                        c.Universe.Type(types.Name{Package: "knative.dev/pkg/injection", Name: "GetNamespaceScope"}),
+		"controllerGetResyncPeriod":                    c.Universe.Type(types.Name{Package: "knative.dev/pkg/controller", Name: "GetResyncPeriod"}),
+		"loggingFromContext": c.Universe.Function(types.Name{
+			Package: "knative.dev/pkg/logging",
+			Name:    "FromContext",
+		}),
+		"contextContext": c.Universe.Type(types.Name{
+			Package: "context",
+			Name:    "Context",
+		}),
+		"metav1ListOptions": c.Universe.Type(types.Name{
+			Package: "k8s.io/apimachinery/pkg/apis/meta/v1",
+			Name:    "ListOptions",
+		}),
+	}
+
+	sw.Do(injectionFilteredFactory, m)
+
+	return sw.Error()
+}
+
+var injectionFilteredFactory = `
+func init() {
+	{{.injectionRegisterInformerFactory|raw}}(withInformerFactory)
+}
+
+// Key is used as the key for associating information with a context.Context.
+type Key struct{
+	Selector string
+}
+
+type LabelKey struct{}
+
+func WithSelectors(ctx {{.contextContext|raw}}, selector ...string) context.Context {
+	return context.WithValue(ctx, LabelKey{}, selector)
+}
+
+func withInformerFactory(ctx {{.contextContext|raw}}) {{.contextContext|raw}} {
+	c := {{.cachingClientGet|raw}}(ctx)
+	opts := []{{.informersSharedInformerOption|raw}}{}
+	if {{.injectionHasNamespace|raw}}(ctx) {
+		opts = append(opts, {{.informersWithNamespace|raw}}({{.injectionGetNamespace|raw}}(ctx)))
+	}
+	untyped := ctx.Value(LabelKey{})
+	if untyped == nil {
+		{{.loggingFromContext|raw}}(ctx).Panic(
+			"Unable to fetch labelkey from context.")
+	}
+	labelSelectors := untyped.([]string)
+	for _, selector := range labelSelectors {
+		thisOpts := append(opts, {{.informersWithTweakListOptions|raw}}(func(l *{{.metav1ListOptions|raw}}) {
+			l.LabelSelector = selector
+		}))
+		ctx = context.WithValue(ctx, Key{Selector: selector},
+			{{.informersNewSharedInformerFactoryWithOptions|raw}}(c, {{.controllerGetResyncPeriod|raw}}(ctx), thisOpts...))
+	}
+	return ctx
+}
+
+// Get extracts the InformerFactory from the context.
+func Get(ctx {{.contextContext|raw}}, selector string) {{.informersSharedInformerFactory|raw}} {
+	untyped := ctx.Value(Key{Selector: selector})
+	if untyped == nil {
+		{{.loggingFromContext|raw}}(ctx).Panicf(
+			"Unable to fetch {{.informersSharedInformerFactory}} with selector %s from context.", selector)
+	}
+	return untyped.({{.informersSharedInformerFactory|raw}})
+}
+`
diff --git a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/filtered_informer.go b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/filtered_informer.go
new file mode 100644
index 000000000..c82a2e59f
--- /dev/null
+++ b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/filtered_informer.go
@@ -0,0 +1,141 @@
+/*
+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 generators
+
+import (
+	"io"
+
+	clientgentypes "k8s.io/code-generator/cmd/client-gen/types"
+	"k8s.io/gengo/generator"
+	"k8s.io/gengo/namer"
+	"k8s.io/gengo/types"
+	"k8s.io/klog"
+)
+
+// injectionTestGenerator produces a file of listers for a given GroupVersion and
+// type.
+type filteredInjectionGenerator struct {
+	generator.DefaultGen
+	outputPackage               string
+	groupVersion                clientgentypes.GroupVersion
+	groupGoName                 string
+	typeToGenerate              *types.Type
+	imports                     namer.ImportTracker
+	typedInformerPackage        string
+	groupInformerFactoryPackage string
+}
+
+var _ generator.Generator = (*filteredInjectionGenerator)(nil)
+
+func (g *filteredInjectionGenerator) Filter(c *generator.Context, t *types.Type) bool {
+	// Only process the type for this informer generator.
+	return t == g.typeToGenerate
+}
+
+func (g *filteredInjectionGenerator) Namers(c *generator.Context) namer.NameSystems {
+	publicPluralNamer := &ExceptionNamer{
+		Exceptions: map[string]string{
+			// these exceptions are used to deconflict the generated code
+			// you can put your fully qualified package like
+			// to generate a name that doesn't conflict with your group.
+			// "k8s.io/apis/events/v1beta1.Event": "EventResource"
+		},
+		KeyFunc: func(t *types.Type) string {
+			return t.Name.Package + "." + t.Name.Name
+		},
+		Delegate: namer.NewPublicPluralNamer(map[string]string{
+			"Endpoints": "Endpoints",
+		}),
+	}
+
+	return namer.NameSystems{
+		"raw":          namer.NewRawNamer(g.outputPackage, g.imports),
+		"publicPlural": publicPluralNamer,
+	}
+}
+
+func (g *filteredInjectionGenerator) Imports(c *generator.Context) (imports []string) {
+	imports = append(imports, g.imports.ImportLines()...)
+	return
+}
+
+func (g *filteredInjectionGenerator) GenerateType(c *generator.Context, t *types.Type, w io.Writer) error {
+	sw := generator.NewSnippetWriter(w, c, "{{", "}}")
+
+	klog.V(5).Info("processing type ", t)
+
+	m := map[string]interface{}{
+		"group":                              namer.IC(g.groupGoName),
+		"type":                               t,
+		"version":                            namer.IC(g.groupVersion.Version.String()),
+		"injectionRegisterFilteredInformers": c.Universe.Type(types.Name{Package: "knative.dev/pkg/injection", Name: "Default.RegisterFilteredInformers"}),
+		"controllerInformer":                 c.Universe.Type(types.Name{Package: "knative.dev/pkg/controller", Name: "Informer"}),
+		"informersTypedInformer":             c.Universe.Type(types.Name{Package: g.typedInformerPackage, Name: t.Name.Name + "Informer"}),
+		"factoryLabelKey":                    c.Universe.Type(types.Name{Package: g.groupInformerFactoryPackage, Name: "LabelKey"}),
+		"factoryGet":                         c.Universe.Function(types.Name{Package: g.groupInformerFactoryPackage, Name: "Get"}),
+		"loggingFromContext": c.Universe.Function(types.Name{
+			Package: "knative.dev/pkg/logging",
+			Name:    "FromContext",
+		}),
+		"contextContext": c.Universe.Type(types.Name{
+			Package: "context",
+			Name:    "Context",
+		}),
+	}
+
+	sw.Do(filteredInjectionInformer, m)
+
+	return sw.Error()
+}
+
+var filteredInjectionInformer = `
+func init() {
+	{{.injectionRegisterFilteredInformers|raw}}(withInformer)
+}
+
+// Key is used for associating the Informer inside the context.Context.
+type Key struct{
+	Selector string
+}
+
+func withInformer(ctx {{.contextContext|raw}}) ({{.contextContext|raw}}, []{{.controllerInformer|raw}}) {
+	untyped := ctx.Value({{.factoryLabelKey|raw}}{})
+	if untyped == nil {
+		{{.loggingFromContext|raw}}(ctx).Panic(
+			"Unable to fetch labelkey from context.")
+	}
+	labelSelectors := untyped.([]string)
+	infs := []{{.controllerInformer|raw}}{}
+	for _, selector := range labelSelectors {
+		f := {{.factoryGet|raw}}(ctx, selector)
+		inf := f.{{.group}}().{{.version}}().{{.type|publicPlural}}()
+		ctx = context.WithValue(ctx, Key{Selector: selector}, inf)
+		infs = append(infs, inf.Informer())
+	}
+	return ctx, infs
+}
+
+// Get extracts the typed informer from the context.
+func Get(ctx {{.contextContext|raw}}, selector string) {{.informersTypedInformer|raw}} {
+	untyped := ctx.Value(Key{Selector: selector})
+	if untyped == nil {
+		{{.loggingFromContext|raw}}(ctx).Panicf(
+			"Unable to fetch {{.informersTypedInformer}} with selector %s from context.", selector)
+	}
+	return untyped.({{.informersTypedInformer|raw}})
+}
+`
diff --git a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/packages.go b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/packages.go
index d1fa91e09..75dafbf1b 100644
--- a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/packages.go
+++ b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/packages.go
@@ -336,11 +336,61 @@ func versionFactoryPackages(basePackage string, boilerplate []byte, customArgs *
 				return tags.NeedsInformerInjection()
 			},
 		},
+
+		// FilterFactoryImpl
+		&generator.DefaultPackage{
+			PackageName: "filteredFactory",
+			PackagePath: filepath.Join(packagePath, "filtered"),
+			HeaderText:  boilerplate,
+			GeneratorFunc: func(c *generator.Context) (generators []generator.Generator) {
+				// Impl
+				generators = append(generators, &filteredFactoryGenerator{
+					DefaultGen: generator.DefaultGen{
+						OptionalName: "filtered_factory",
+					},
+					outputPackage:                filepath.Join(packagePath, "filtered"),
+					cachingClientSetPackage:      filepath.Join(basePackage, "client"),
+					sharedInformerFactoryPackage: customArgs.ExternalVersionsInformersPackage,
+					imports:                      generator.NewImportTracker(),
+				})
+				return generators
+			},
+			FilterFunc: func(c *generator.Context, t *types.Type) bool {
+				tags := MustParseClientGenTags(append(t.SecondClosestCommentLines, t.CommentLines...))
+				return tags.NeedsInformerInjection()
+			},
+		},
+		// FakeFilterFactory
+		&generator.DefaultPackage{
+			PackageName: "fakeFilteredFactory",
+			PackagePath: filepath.Join(packagePath, "filtered", "fake"),
+			HeaderText:  boilerplate,
+			GeneratorFunc: func(c *generator.Context) (generators []generator.Generator) {
+				// Impl
+				generators = append(generators, &fakeFilteredFactoryGenerator{
+					DefaultGen: generator.DefaultGen{
+						OptionalName: "fake_filtered_factory",
+					},
+					outputPackage:                filepath.Join(packagePath, "filtered", "fake"),
+					factoryInjectionPkg:          filepath.Join(packagePath, "filtered"),
+					fakeClientInjectionPkg:       filepath.Join(basePackage, "client", "fake"),
+					sharedInformerFactoryPackage: customArgs.ExternalVersionsInformersPackage,
+					imports:                      generator.NewImportTracker(),
+				})
+				return generators
+			},
+			FilterFunc: func(c *generator.Context, t *types.Type) bool {
+				tags := MustParseClientGenTags(append(t.SecondClosestCommentLines, t.CommentLines...))
+				return tags.NeedsInformerInjection()
+			},
+		},
 	}
 }
 
 func versionInformerPackages(basePackage string, groupPkgName string, gv clientgentypes.GroupVersion, groupGoName string, boilerplate []byte, typesToGenerate []*types.Type, customArgs *informergenargs.CustomArgs) []generator.Package {
 	factoryPackagePath := filepath.Join(basePackage, "informers", "factory")
+	filteredFactoryPackagePath := filepath.Join(basePackage, "informers", "factory", "filtered")
+
 	packagePath := filepath.Join(basePackage, "informers", groupPkgName, strings.ToLower(gv.Version.NonEmpty()))
 
 	vers := make([]generator.Package, 0, 2*len(typesToGenerate))
@@ -404,6 +454,60 @@ func versionInformerPackages(basePackage string, groupPkgName string, gv clientg
 				return tags.NeedsInformerInjection()
 			},
 		})
+		// FilteredInformer
+		vers = append(vers, &generator.DefaultPackage{
+			PackageName: "filtered",
+			PackagePath: filepath.Join(packagePath, "filtered"),
+			HeaderText:  boilerplate,
+			GeneratorFunc: func(c *generator.Context) (generators []generator.Generator) {
+				// Impl
+				generators = append(generators, &filteredInjectionGenerator{
+					DefaultGen: generator.DefaultGen{
+						OptionalName: strings.ToLower(t.Name.Name),
+					},
+					outputPackage:               filepath.Join(packagePath, "filtered"),
+					groupVersion:                gv,
+					groupGoName:                 groupGoName,
+					typeToGenerate:              t,
+					imports:                     generator.NewImportTracker(),
+					typedInformerPackage:        typedInformerPackage,
+					groupInformerFactoryPackage: filteredFactoryPackagePath,
+				})
+				return generators
+			},
+			FilterFunc: func(c *generator.Context, t *types.Type) bool {
+				tags := MustParseClientGenTags(append(t.SecondClosestCommentLines, t.CommentLines...))
+				return tags.NeedsInformerInjection()
+			},
+		})
+
+		// FakeFilteredInformer
+		vers = append(vers, &generator.DefaultPackage{
+			PackageName: "fake",
+			PackagePath: filepath.Join(packagePath, "filtered", "fake"),
+			HeaderText:  boilerplate,
+			GeneratorFunc: func(c *generator.Context) (generators []generator.Generator) {
+				// Impl
+				generators = append(generators, &fakeFilteredInformerGenerator{
+					DefaultGen: generator.DefaultGen{
+						OptionalName: "fake",
+					},
+					outputPackage:           filepath.Join(packagePath, "filtered", "fake"),
+					imports:                 generator.NewImportTracker(),
+					typeToGenerate:          t,
+					groupVersion:            gv,
+					groupGoName:             groupGoName,
+					informerInjectionPkg:    filepath.Join(packagePath, "filtered"),
+					fakeFactoryInjectionPkg: filteredFactoryPackagePath,
+				})
+				return generators
+			},
+			FilterFunc: func(c *generator.Context, t *types.Type) bool {
+				tags := MustParseClientGenTags(append(t.SecondClosestCommentLines, t.CommentLines...))
+				return tags.NeedsInformerInjection()
+			},
+		})
+
 	}
 	return vers
 }
diff --git a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_controller.go b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_controller.go
index 9260fe87f..0f0f417ec 100644
--- a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_controller.go
+++ b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_controller.go
@@ -170,6 +170,18 @@ func (g *reconcilerControllerGenerator) GenerateType(c *generator.Context, t *ty
 			Package: "fmt",
 			Name:    "Sprintf",
 		}),
+		"logkeyControllerType": c.Universe.Constant(types.Name{
+			Package: "knative.dev/pkg/logging/logkey",
+			Name:    "ControllerType",
+		}),
+		"logkeyControllerKind": c.Universe.Constant(types.Name{
+			Package: "knative.dev/pkg/logging/logkey",
+			Name:    "Kind",
+		}),
+		"zapString": c.Universe.Function(types.Name{
+			Package: "go.uber.org/zap",
+			Name:    "String",
+		}),
 	}
 
 	sw.Do(reconcilerControllerNewImpl, m)
@@ -227,10 +239,17 @@ func NewImpl(ctx {{.contextContext|raw}}, r Interface{{if .hasClass}}, classValu
 		{{if .hasClass}}classValue: classValue,{{end}}
 	}
 
-	t := {{.reflectTypeOf|raw}}(r).Elem()
-	queueName := {{.fmtSprintf|raw}}("%s.%s", {{.stringsReplaceAll|raw}}(t.PkgPath(), "/", "-"), t.Name())
+	ctrType := {{.reflectTypeOf|raw}}(r).Elem()
+	ctrTypeName := {{.fmtSprintf|raw}}("%s.%s", ctrType.PkgPath(), ctrType.Name())
+	ctrTypeName = {{.stringsReplaceAll|raw}}(ctrTypeName, "/", ".")
+
+	logger = logger.With(
+			{{.zapString|raw}}({{.logkeyControllerType|raw}}, ctrTypeName),
+			{{.zapString|raw}}({{.logkeyControllerKind|raw}}, "{{ printf "%s.%s" .group .type.Name.Name }}"),
+	)
+
 
-	impl := {{.controllerNewImpl|raw}}(rec, logger, queueName)
+	impl := {{.controllerNewImpl|raw}}(rec, logger, ctrTypeName)
 	agentName := defaultControllerAgentName
 
 	// Pass impl to the options. Save any optional results.
diff --git a/vendor/knative.dev/pkg/controller/controller.go b/vendor/knative.dev/pkg/controller/controller.go
index 953e5ea0c..26030626e 100644
--- a/vendor/knative.dev/pkg/controller/controller.go
+++ b/vendor/knative.dev/pkg/controller/controller.go
@@ -230,7 +230,6 @@ func NewImplWithStats(r Reconciler, logger *zap.SugaredLogger, workQueueName str
 
 // NewImplFull accepts the full set of options available to all controllers.
 func NewImplFull(r Reconciler, options ControllerOptions) *Impl {
-	logger := options.Logger.Named(options.WorkQueueName)
 	if options.RateLimiter == nil {
 		options.RateLimiter = workqueue.DefaultControllerRateLimiter()
 	}
@@ -241,7 +240,7 @@ func NewImplFull(r Reconciler, options ControllerOptions) *Impl {
 		Name:          options.WorkQueueName,
 		Reconciler:    r,
 		workQueue:     newTwoLaneWorkQueue(options.WorkQueueName, options.RateLimiter),
-		logger:        logger,
+		logger:        options.Logger,
 		statsReporter: options.Reporter,
 	}
 }
@@ -528,7 +527,7 @@ func (c *Impl) processNextWorkItem() bool {
 	// Finally, if no error occurs we Forget this item so it does not
 	// have any delay when another change happens.
 	c.workQueue.Forget(key)
-	logger.Info("Reconcile succeeded. Time taken: ", zap.Duration("duration", time.Since(startTime)))
+	logger.Infow("Reconcile succeeded", zap.Duration("duration", time.Since(startTime)))
 
 	return true
 }
diff --git a/vendor/knative.dev/pkg/injection/README.md b/vendor/knative.dev/pkg/injection/README.md
index 131484ab0..761677e88 100644
--- a/vendor/knative.dev/pkg/injection/README.md
+++ b/vendor/knative.dev/pkg/injection/README.md
@@ -138,6 +138,7 @@ func NewController(ctx context.Context, cmw configmap.Watcher) *controller.Impl
 		Client:        kubeclient.Get(ctx),
 		ServiceLister: svcInformer.Lister(),
 	}
+	logger = logger.Named("NameOfController")
 	impl := controller.NewImpl(c, logger, "NameOfController")
 
 	// Set up event handlers.
diff --git a/vendor/knative.dev/pkg/injection/informers.go b/vendor/knative.dev/pkg/injection/informers.go
index 1140174d1..9356f8d7f 100644
--- a/vendor/knative.dev/pkg/injection/informers.go
+++ b/vendor/knative.dev/pkg/injection/informers.go
@@ -28,6 +28,10 @@ import (
 // informer type to a context.
 type InformerInjector func(context.Context) (context.Context, controller.Informer)
 
+// FilteredInformersInjector holds the type of a callback that attaches a set of particular
+// filtered informers type to a context.
+type FilteredInformersInjector func(context.Context) (context.Context, []controller.Informer)
+
 func (i *impl) RegisterInformer(ii InformerInjector) {
 	i.m.Lock()
 	defer i.m.Unlock()
@@ -35,6 +39,13 @@ func (i *impl) RegisterInformer(ii InformerInjector) {
 	i.informers = append(i.informers, ii)
 }
 
+func (i *impl) RegisterFilteredInformers(fii FilteredInformersInjector) {
+	i.m.Lock()
+	defer i.m.Unlock()
+
+	i.filteredInformers = append(i.filteredInformers, fii)
+}
+
 func (i *impl) GetInformers() []InformerInjector {
 	i.m.RLock()
 	defer i.m.RUnlock()
@@ -43,6 +54,14 @@ func (i *impl) GetInformers() []InformerInjector {
 	return append(i.informers[:0:0], i.informers...)
 }
 
+func (i *impl) GetFilteredInformers() []FilteredInformersInjector {
+	i.m.RLock()
+	defer i.m.RUnlock()
+
+	// Copy the slice before returning.
+	return append(i.filteredInformers[:0:0], i.filteredInformers...)
+}
+
 func (i *impl) SetupInformers(ctx context.Context, cfg *rest.Config) (context.Context, []controller.Informer) {
 	// Based on the reconcilers we have linked, build up a set of clients and inject
 	// them onto the context.
@@ -65,10 +84,16 @@ func (i *impl) SetupInformers(ctx context.Context, cfg *rest.Config) (context.Co
 	// Based on the reconcilers we have linked, build up a set of informers
 	// and inject them onto the context.
 	var inf controller.Informer
+	var filteredinfs []controller.Informer
 	informers := make([]controller.Informer, 0, len(i.GetInformers()))
 	for _, ii := range i.GetInformers() {
 		ctx, inf = ii(ctx)
 		informers = append(informers, inf)
 	}
+	for _, fii := range i.GetFilteredInformers() {
+		ctx, filteredinfs = fii(ctx)
+		informers = append(informers, filteredinfs...)
+
+	}
 	return ctx, informers
 }
diff --git a/vendor/knative.dev/pkg/injection/interface.go b/vendor/knative.dev/pkg/injection/interface.go
index 85955db58..7d2764cdc 100644
--- a/vendor/knative.dev/pkg/injection/interface.go
+++ b/vendor/knative.dev/pkg/injection/interface.go
@@ -53,9 +53,16 @@ type Interface interface {
 	// a new informer with a context.
 	RegisterInformer(InformerInjector)
 
+	// RegisterFilteredInformers registers a new filtered informer injector callback for associating
+	// a new set of informers with a context.
+	RegisterFilteredInformers(FilteredInformersInjector)
+
 	// GetInformers fetches all of the registered informer injectors.
 	GetInformers() []InformerInjector
 
+	// GetFilteredInformers fetches all of the registered filtered informer injectors.
+	GetFilteredInformers() []FilteredInformersInjector
+
 	// SetupInformers runs all of the injectors against a context, starting with
 	// the clients and the given rest.Config.  The resulting context is returned
 	// along with a list of the .Informer() for each of the injected informers,
@@ -84,8 +91,9 @@ var (
 type impl struct {
 	m sync.RWMutex
 
-	clients   []ClientInjector
-	factories []InformerFactoryInjector
-	informers []InformerInjector
-	ducks     []DuckFactoryInjector
+	clients           []ClientInjector
+	factories         []InformerFactoryInjector
+	informers         []InformerInjector
+	filteredInformers []FilteredInformersInjector
+	ducks             []DuckFactoryInjector
 }
diff --git a/vendor/knative.dev/pkg/webhook/certificates/controller.go b/vendor/knative.dev/pkg/webhook/certificates/controller.go
index a50acb39a..f6dedb722 100644
--- a/vendor/knative.dev/pkg/webhook/certificates/controller.go
+++ b/vendor/knative.dev/pkg/webhook/certificates/controller.go
@@ -66,7 +66,8 @@ func NewController(
 		secretlister: secretInformer.Lister(),
 	}
 
-	c := controller.NewImpl(wh, logging.FromContext(ctx), "WebhookCertificates")
+	const queueName = "WebhookCertificates"
+	c := controller.NewImpl(wh, logging.FromContext(ctx).Named(queueName), queueName)
 
 	// Reconcile when the cert bundle changes.
 	secretInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
diff --git a/vendor/knative.dev/pkg/webhook/psbinding/README.md b/vendor/knative.dev/pkg/webhook/psbinding/README.md
index 4083093b3..e73ccf858 100644
--- a/vendor/knative.dev/pkg/webhook/psbinding/README.md
+++ b/vendor/knative.dev/pkg/webhook/psbinding/README.md
@@ -142,6 +142,7 @@ func NewController(
 		Recorder: record.NewBroadcaster().NewRecorder(
 			scheme.Scheme, corev1.EventSource{Component: controllerAgentName}),
 	}
+	logger = logger.Named("GithubBindings")
 	impl := controller.NewImpl(c, logger, "GithubBindings")
 
 	logger.Info("Setting up event handlers")
diff --git a/vendor/knative.dev/pkg/webhook/psbinding/controller.go b/vendor/knative.dev/pkg/webhook/psbinding/controller.go
index 3967c053b..1d671d072 100644
--- a/vendor/knative.dev/pkg/webhook/psbinding/controller.go
+++ b/vendor/knative.dev/pkg/webhook/psbinding/controller.go
@@ -87,7 +87,7 @@ func NewAdmissionController(
 
 	// Construct the reconciler for the mutating webhook configuration.
 	wh := NewReconciler(name, path, options.SecretName, client, mwhInformer.Lister(), secretInformer.Lister(), withContext, reconcilerOptions...)
-	c := controller.NewImpl(wh, logging.FromContext(ctx), name)
+	c := controller.NewImpl(wh, logging.FromContext(ctx).Named(name), name)
 
 	// Enqueue a sentinel when we become leader.
 	wh.PromoteFunc = func(bkt pkgreconciler.Bucket, enq func(pkgreconciler.Bucket, types.NamespacedName)) error {
diff --git a/vendor/knative.dev/pkg/webhook/resourcesemantics/conversion/controller.go b/vendor/knative.dev/pkg/webhook/resourcesemantics/conversion/controller.go
index 3b404ba97..7ab4b158b 100644
--- a/vendor/knative.dev/pkg/webhook/resourcesemantics/conversion/controller.go
+++ b/vendor/knative.dev/pkg/webhook/resourcesemantics/conversion/controller.go
@@ -117,7 +117,9 @@ func NewConversionController(
 		crdLister:    crdInformer.Lister(),
 	}
 
-	c := controller.NewImpl(r, logging.FromContext(ctx), "ConversionWebhook")
+	const queueName = "ConversionWebhook"
+	logger := logging.FromContext(ctx)
+	c := controller.NewImpl(r, logger.Named(queueName), queueName)
 
 	// Reconciler when the named CRDs change.
 	for _, gkc := range kinds {
diff --git a/vendor/knative.dev/pkg/webhook/resourcesemantics/defaulting/controller.go b/vendor/knative.dev/pkg/webhook/resourcesemantics/defaulting/controller.go
index b33140588..7a3f22e34 100644
--- a/vendor/knative.dev/pkg/webhook/resourcesemantics/defaulting/controller.go
+++ b/vendor/knative.dev/pkg/webhook/resourcesemantics/defaulting/controller.go
@@ -74,7 +74,8 @@ func NewAdmissionController(
 	}
 
 	logger := logging.FromContext(ctx)
-	c := controller.NewImpl(wh, logger, "DefaultingWebhook")
+	const queueName = "DefaultingWebhook"
+	c := controller.NewImpl(wh, logger.Named(queueName), queueName)
 
 	// Reconcile when the named MutatingWebhookConfiguration changes.
 	mwhInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
diff --git a/vendor/knative.dev/pkg/webhook/resourcesemantics/validation/controller.go b/vendor/knative.dev/pkg/webhook/resourcesemantics/validation/controller.go
index 567515843..942da7079 100644
--- a/vendor/knative.dev/pkg/webhook/resourcesemantics/validation/controller.go
+++ b/vendor/knative.dev/pkg/webhook/resourcesemantics/validation/controller.go
@@ -89,7 +89,8 @@ func NewAdmissionController(
 	}
 
 	logger := logging.FromContext(ctx)
-	c := controller.NewImpl(wh, logger, "ValidationWebhook")
+	const queueName = "ValidationWebhook"
+	c := controller.NewImpl(wh, logger.Named(queueName), queueName)
 
 	// Reconcile when the named ValidatingWebhookConfiguration changes.
 	vwhInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
diff --git a/vendor/modules.txt b/vendor/modules.txt
index faba8ba3b..a634b4a74 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1078,7 +1078,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.20.1-0.20210115015220-1c94ef83b6b5
+# knative.dev/eventing v0.20.1-0.20210120133253-5dc363dcb5be
 ## explicit
 knative.dev/eventing/pkg/adapter/v2
 knative.dev/eventing/pkg/adapter/v2/util/crstatusevent
@@ -1172,6 +1172,7 @@ knative.dev/eventing/pkg/client/listers/sources/v1alpha1
 knative.dev/eventing/pkg/client/listers/sources/v1alpha2
 knative.dev/eventing/pkg/client/listers/sources/v1beta1
 knative.dev/eventing/pkg/client/listers/sources/v1beta2
+knative.dev/eventing/pkg/duck
 knative.dev/eventing/pkg/kncloudevents
 knative.dev/eventing/pkg/reconciler/source
 knative.dev/eventing/pkg/reconciler/sugar
@@ -1213,7 +1214,7 @@ knative.dev/eventing/test/test_images/recordevents
 # knative.dev/hack v0.0.0-20210114150620-4422dcadb3c8
 ## explicit
 knative.dev/hack
-# knative.dev/pkg v0.0.0-20210114223020-f0ea5e6b9c4e
+# knative.dev/pkg v0.0.0-20210119162123-1bbf0a6436c3
 ## explicit
 knative.dev/pkg/apis
 knative.dev/pkg/apis/duck


Test patch:
diff --git a/pkg/channel/distributed/common/config/configwatcher_test.go b/pkg/channel/distributed/common/config/configwatcher_test.go
index e5758017d..eff763eeb 100644
--- a/pkg/channel/distributed/common/config/configwatcher_test.go
+++ b/pkg/channel/distributed/common/config/configwatcher_test.go
@@ -23,12 +23,11 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
-	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes/fake"
-	"knative.dev/eventing-kafka/pkg/channel/distributed/common/config/constants"
 	commontesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/testing"
+	"knative.dev/eventing-kafka/pkg/common/constants"
 	injectionclient "knative.dev/pkg/client/injection/kube/client"
 	logtesting "knative.dev/pkg/logging/testing"
 	"knative.dev/pkg/system"
@@ -106,7 +105,7 @@ func setWatchedMap(configMap *corev1.ConfigMap) {
 }
 
 // Handler function for the ConfigMap watcher
-func configWatcherHandler(_ *zap.SugaredLogger, configMap *corev1.ConfigMap) {
+func configWatcherHandler(ctx context.Context, configMap *corev1.ConfigMap) {
 	// Set the package variable to indicate that the test watcher was called
 	setWatchedMap(configMap)
 }
diff --git a/pkg/channel/distributed/common/config/testing/util.go b/pkg/channel/distributed/common/config/testing/util.go
index c85cc38b5..0698b100b 100644
--- a/pkg/channel/distributed/common/config/testing/util.go
+++ b/pkg/channel/distributed/common/config/testing/util.go
@@ -21,7 +21,7 @@ import (
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"knative.dev/eventing-kafka/pkg/channel/distributed/common/config/constants"
+	"knative.dev/eventing-kafka/pkg/common/constants"
 	"knative.dev/pkg/system"
 )
 
diff --git a/pkg/channel/distributed/common/kafka/sarama/sarama_test.go b/pkg/channel/distributed/common/kafka/sarama/sarama_test.go
index a6b4cfd6e..9541e9fd5 100644
--- a/pkg/channel/distributed/common/kafka/sarama/sarama_test.go
+++ b/pkg/channel/distributed/common/kafka/sarama/sarama_test.go
@@ -29,8 +29,8 @@ import (
 	"k8s.io/apimachinery/pkg/api/resource"
 	"k8s.io/client-go/kubernetes/fake"
 	commonconfig "knative.dev/eventing-kafka/pkg/channel/distributed/common/config"
-	"knative.dev/eventing-kafka/pkg/channel/distributed/common/config/constants"
 	commontesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/testing"
+	"knative.dev/eventing-kafka/pkg/common/constants"
 	injectionclient "knative.dev/pkg/client/injection/kube/client"
 )
 
diff --git a/pkg/channel/distributed/common/testing/util.go b/pkg/channel/distributed/common/testing/util.go
index 726dc45e4..6a5fb0b69 100644
--- a/pkg/channel/distributed/common/testing/util.go
+++ b/pkg/channel/distributed/common/testing/util.go
@@ -23,8 +23,8 @@ import (
 	"github.com/stretchr/testify/assert"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"knative.dev/eventing-kafka/pkg/channel/distributed/common/config/constants"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/env"
+	"knative.dev/eventing-kafka/pkg/common/constants"
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/system"
 )
diff --git a/pkg/channel/distributed/controller/kafkachannel/controller_test.go b/pkg/channel/distributed/controller/kafkachannel/controller_test.go
index b13043191..1fad24642 100644
--- a/pkg/channel/distributed/controller/kafkachannel/controller_test.go
+++ b/pkg/channel/distributed/controller/kafkachannel/controller_test.go
@@ -75,7 +75,7 @@ func TestNewController(t *testing.T) {
 
 	// Verify The Results
 	assert.NotNil(t, controller)
-	assert.Equal(t, "knative.dev-eventing-kafka-pkg-channel-distributed-controller-kafkachannel.Reconciler", controller.Name)
+	assert.True(t, len(controller.Name) > 0)
 	assert.NotNil(t, controller.Reconciler)
 }
 
diff --git a/pkg/channel/distributed/controller/kafkasecret/controller_test.go b/pkg/channel/distributed/controller/kafkasecret/controller_test.go
index a50cf0be9..a0536a045 100644
--- a/pkg/channel/distributed/controller/kafkasecret/controller_test.go
+++ b/pkg/channel/distributed/controller/kafkasecret/controller_test.go
@@ -71,7 +71,7 @@ func TestNewController(t *testing.T) {
 
 	// Verify The Results
 	assert.NotNil(t, controller)
-	assert.Equal(t, "knative.dev-eventing-kafka-pkg-channel-distributed-controller-kafkasecret.Reconciler", controller.Name)
+	assert.True(t, len(controller.Name) > 0)
 	assert.NotNil(t, controller.Reconciler)
 }
 
diff --git a/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go b/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go
index 7391845c5..3fb6d5b4f 100644
--- a/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go
+++ b/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go
@@ -17,6 +17,7 @@ limitations under the License.
 package controller
 
 import (
+	"context"
 	"os"
 	"testing"
 	"time"
@@ -216,6 +217,6 @@ func (m MockDispatcher) UpdateSubscriptions(_ []eventingduck.SubscriberSpec) map
 	return nil
 }
 
-func (m MockDispatcher) ConfigChanged(*corev1.ConfigMap) dispatcher.Dispatcher {
+func (m MockDispatcher) ConfigChanged(context.Context, *corev1.ConfigMap) dispatcher.Dispatcher {
 	return nil
 }
diff --git a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
index ec5c09e3e..a1b260773 100644
--- a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
+++ b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
@@ -17,6 +17,7 @@ limitations under the License.
 package dispatcher
 
 import (
+	"context"
 	"fmt"
 	"testing"
 	"time"
@@ -35,6 +36,7 @@ import (
 	receivertesting "knative.dev/eventing-kafka/pkg/channel/distributed/receiver/testing"
 	commonclient "knative.dev/eventing-kafka/pkg/common/client"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
+	"knative.dev/pkg/logging"
 	logtesting "knative.dev/pkg/logging/testing"
 )
 
@@ -115,17 +117,19 @@ func TestShutdown(t *testing.T) {
 // Test The UpdateSubscriptions() Functionality
 func TestUpdateSubscriptions(t *testing.T) {
 
+	logger := logtesting.TestLogger(t)
+	ctx := logging.WithLogger(context.TODO(), logger)
+
 	// Restore Any Stubbing Of NewConsumerGroupWrapper After Test Is Finished
 	defer consumertesting.RestoreNewConsumerGroupFn()
 
 	// Test Data
 	brokers := []string{dispatchertesting.KafkaBroker}
-	config, err := commonclient.NewConfigBuilder().WithDefaults().FromYaml(commonconfigtesting.DefaultSaramaConfigYaml).Build()
+	config, err := commonclient.NewConfigBuilder().WithDefaults().FromYaml(commonconfigtesting.DefaultSaramaConfigYaml).Build(ctx)
 	assert.Nil(t, err)
 
-	logger := logtesting.TestLogger(t).Desugar()
 	dispatcherConfig := DispatcherConfig{
-		Logger:       logger,
+		Logger:       logger.Desugar(),
 		Brokers:      brokers,
 		SaramaConfig: config,
 	}
@@ -281,6 +285,9 @@ func TestUpdateSubscriptions(t *testing.T) {
 // Test The Dispatcher's ConfigChanged Functionality
 func TestConfigChanged(t *testing.T) {
 
+	logger := logtesting.TestLogger(t)
+	ctx := logging.WithLogger(context.TODO(), logger)
+
 	// Setup Test Environment Namespaces
 	commontesting.SetTestEnvironment(t)
 
@@ -290,7 +297,7 @@ func TestConfigChanged(t *testing.T) {
 		WithDefaults().
 		FromYaml(commonconfigtesting.DefaultSaramaConfigYaml).
 		WithVersion(&sarama.V2_0_0_0).
-		Build()
+		Build(ctx)
 	assert.Nil(t, err)
 
 	// Define The TestCase Struct
@@ -366,7 +373,7 @@ func TestConfigChanged(t *testing.T) {
 			dispatcher := createTestDispatcher(t, brokers, baseSaramaConfig)
 
 			// Perform The Test
-			newDispatcher := dispatcher.ConfigChanged(testCase.newConfigMap)
+			newDispatcher := dispatcher.ConfigChanged(ctx, testCase.newConfigMap)
 
 			// Verify Expected State (Not Much To Verify Due To Interface)
 			assert.Equal(t, testCase.expectNewDispatcher, newDispatcher != nil)
diff --git a/pkg/channel/distributed/receiver/producer/producer_test.go b/pkg/channel/distributed/receiver/producer/producer_test.go
index 0717b02cb..23c979593 100644
--- a/pkg/channel/distributed/receiver/producer/producer_test.go
+++ b/pkg/channel/distributed/receiver/producer/producer_test.go
@@ -33,6 +33,7 @@ import (
 	channelhealth "knative.dev/eventing-kafka/pkg/channel/distributed/receiver/health"
 	receivertesting "knative.dev/eventing-kafka/pkg/channel/distributed/receiver/testing"
 	commonclient "knative.dev/eventing-kafka/pkg/common/client"
+	"knative.dev/pkg/logging"
 	logtesting "knative.dev/pkg/logging/testing"
 )
 
@@ -103,12 +104,15 @@ func TestProduceKafkaMessage(t *testing.T) {
 // Test The Producer's ConfigChanged Functionality
 func TestConfigChanged(t *testing.T) {
 
+	logger := logtesting.TestLogger(t)
+	ctx := logging.WithLogger(context.TODO(), logger)
+
 	// Setup Test Environment Namespaces
 	commontesting.SetTestEnvironment(t)
 
 	// Test Data
 	brokers := []string{receivertesting.KafkaBroker}
-	baseSaramaConfig, err := commonclient.NewConfigBuilder().WithDefaults().FromYaml(commonconfigtesting.DefaultSaramaConfigYaml).Build()
+	baseSaramaConfig, err := commonclient.NewConfigBuilder().WithDefaults().FromYaml(commonconfigtesting.DefaultSaramaConfigYaml).Build(ctx)
 	assert.Nil(t, err)
 
 	// Define The TestCase Struct
@@ -184,7 +188,7 @@ func TestConfigChanged(t *testing.T) {
 			producer := createTestProducer(t, brokers, baseSaramaConfig, mockSyncProducer)
 
 			// Perform The Test
-			newProducer := producer.ConfigChanged(testCase.newConfigMap)
+			newProducer := producer.ConfigChanged(ctx, testCase.newConfigMap)
 
 			// Verify Expected State
 			assert.Equal(t, testCase.expectNewProducer, newProducer != nil)
diff --git a/pkg/common/client/config_test.go b/pkg/common/client/config_test.go
index 6b11a4237..ff5d81746 100644
--- a/pkg/common/client/config_test.go
+++ b/pkg/common/client/config_test.go
@@ -18,6 +18,7 @@ package client
 
 import (
 	"bytes"
+	"context"
 	"crypto/rand"
 	"crypto/rsa"
 	"crypto/tls"
@@ -33,6 +34,8 @@ import (
 	"github.com/Shopify/sarama"
 	"github.com/stretchr/testify/assert"
 	commontesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/testing"
+	"knative.dev/pkg/logging"
+	logtesting "knative.dev/pkg/logging/testing"
 )
 
 const (
@@ -115,6 +118,9 @@ Consumer:
 
 // Verify that the JSON fragment can be loaded into a sarama.Config struct
 func TestBuildSaramaConfig(t *testing.T) {
+	logger := logtesting.TestLogger(t)
+	ctx := logging.WithLogger(context.TODO(), logger)
+
 	// Setup Environment
 	commontesting.SetTestEnvironment(t)
 
@@ -126,7 +132,7 @@ func TestBuildSaramaConfig(t *testing.T) {
 	config, err := NewConfigBuilder().
 		WithDefaults().
 		FromYaml(commontesting.OldSaramaConfig).
-		Build()
+		Build(ctx)
 	assert.Nil(t, err)
 	assert.NotNil(t, config)
 	assert.True(t, config.Net.SASL.Enable)
@@ -137,7 +143,7 @@ func TestBuildSaramaConfig(t *testing.T) {
 	config, err = NewConfigBuilder().
 		WithDefaults().
 		FromYaml(commontesting.NewSaramaConfig).
-		Build()
+		Build(ctx)
 	assert.Nil(t, err)
 	assert.NotNil(t, config)
 	assert.Equal(t, sarama.V2_3_0_0, config.Version)
@@ -151,21 +157,21 @@ func TestBuildSaramaConfig(t *testing.T) {
 	_, err = NewConfigBuilder().
 		WithDefaults().
 		FromYaml(regexVersion.ReplaceAllString(commontesting.NewSaramaConfig, "Version: INVALID")).
-		Build()
+		Build(ctx)
 	assert.NotNil(t, err)
 
 	// Verify error when an invalid RootPEMs is provided
 	_, err = NewConfigBuilder().
 		WithDefaults().
 		FromYaml(strings.Replace(EKDefaultSaramaConfigWithRootCert, "-----BEGIN CERTIFICATE-----", "INVALID CERT DATA", -1)).
-		Build()
+		Build(ctx)
 	assert.NotNil(t, err)
 
 	// Verify that the RootPEMs section is merged properly
 	config, err = NewConfigBuilder().
 		WithDefaults().
 		FromYaml(EKDefaultSaramaConfigWithRootCert).
-		Build()
+		Build(ctx)
 	assert.Nil(t, err)
 	assert.NotNil(t, config.Net.TLS.Config.RootCAs)
 
@@ -173,7 +179,7 @@ func TestBuildSaramaConfig(t *testing.T) {
 	config, err = NewConfigBuilder().
 		WithDefaults().
 		FromYaml(EKDefaultSaramaConfigWithInsecureSkipVerify).
-		Build()
+		Build(ctx)
 	assert.Nil(t, err)
 	assert.True(t, config.Net.TLS.Config.InsecureSkipVerify)
 
@@ -192,7 +198,7 @@ func TestBuildSaramaConfig(t *testing.T) {
 		}).
 		WithVersion(&sarama.V2_0_0_0).
 		WithClientId("newClientId").
-		Build()
+		Build(ctx)
 	assert.Nil(t, err)
 	assert.Equal(t, "foo", config.Net.SASL.User)
 	assert.Equal(t, "newClientId", config.ClientID)
@@ -200,6 +206,9 @@ func TestBuildSaramaConfig(t *testing.T) {
 }
 
 func TestBuildSaramaConfigWithTLSAuth(t *testing.T) {
+	logger := logtesting.TestLogger(t)
+	ctx := logging.WithLogger(context.TODO(), logger)
+
 	// Setup Environment
 	commontesting.SetTestEnvironment(t)
 
@@ -273,7 +282,7 @@ sny569QyyWHk2+FZoWDfjxFZ7CvIdgLJBHc3qUXLsg==
 		WithDefaults().
 		FromYaml(noAuthSaramaYaml).
 		WithAuth(kafkaAuthCfg).
-		Build()
+		Build(ctx)
 	assert.Nil(t, err)
 
 	// Make sure TLS settings are applied from the KafkaAuthConfig
@@ -289,6 +298,9 @@ sny569QyyWHk2+FZoWDfjxFZ7CvIdgLJBHc3qUXLsg==
 }
 
 func TestBuildSaramaConfigWithSASLAuth(t *testing.T) {
+	logger := logtesting.TestLogger(t)
+	ctx := logging.WithLogger(context.TODO(), logger)
+
 	// Setup Environment
 	commontesting.SetTestEnvironment(t)
 
@@ -316,7 +328,7 @@ Metadata:
 		WithDefaults().
 		FromYaml(noAuthSaramaYaml).
 		WithAuth(kafkaAuthCfg).
-		Build()
+		Build(ctx)
 	assert.Nil(t, err)
 
 	// Make sure SASL settings are applied from the KafkaAuthConfig
@@ -332,6 +344,9 @@ Metadata:
 
 // Verify that comparisons of sarama config structs function as expected
 func TestSaramaConfigEqual(t *testing.T) {
+	logger := logtesting.TestLogger(t)
+	ctx := logging.WithLogger(context.TODO(), logger)
+
 	config1 := sarama.NewConfig()
 	config2 := sarama.NewConfig()
 
@@ -387,12 +402,12 @@ func TestSaramaConfigEqual(t *testing.T) {
 	config1, err := NewConfigBuilder().
 		WithDefaults().
 		FromYaml(EKDefaultSaramaConfigWithRootCert).
-		Build()
+		Build(ctx)
 	assert.Nil(t, err)
 	config2, err = NewConfigBuilder().
 		WithDefaults().
 		FromYaml(EKDefaultSaramaConfigWithRootCert).
-		Build()
+		Build(ctx)
 	assert.Nil(t, err)
 	assert.True(t, ConfigEqual(config1, config2))
 }
@@ -490,11 +505,13 @@ func TestUpdateSaramaConfigWithKafkaAuthConfig(t *testing.T) {
 
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
+			logger := logtesting.TestLogger(t)
+			ctx := logging.WithLogger(context.TODO(), logger)
 
 			// Perform The Test
 			config, err := NewConfigBuilder().
 				WithAuth(tc.kafkaAuthCfg).
-				Build()
+				Build(ctx)
 
 			if err != nil {
 				t.Errorf("error configuring Sarama config with auth :%e", err)
diff --git a/pkg/source/client_test.go b/pkg/source/client_test.go
index 4a87eb72a..cee49af9f 100644
--- a/pkg/source/client_test.go
+++ b/pkg/source/client_test.go
@@ -22,6 +22,9 @@ import (
 	"testing"
 
 	"github.com/Shopify/sarama"
+	"knative.dev/eventing-kafka/pkg/channel/consolidated/utils"
+	"knative.dev/pkg/logging"
+	logtesting "knative.dev/pkg/logging/testing"
 
 	"github.com/stretchr/testify/require"
 )
@@ -184,6 +187,8 @@ func TestNewConfig(t *testing.T) {
 }
 
 func TestAdminClient(t *testing.T) {
+	logger := logtesting.TestLogger(t)
+	ctx := logging.WithLogger(context.TODO(), logger)
 
 	seedBroker := sarama.NewMockBroker(t, 1)
 	defer seedBroker.Close()
@@ -195,7 +200,7 @@ func TestAdminClient(t *testing.T) {
 	})
 
 	// mock broker does not support TLS ...
-	admin, err := MakeAdminClient("test-client", nil, []string{seedBroker.Addr()})
+	admin, err := MakeAdminClient(ctx, "test-client", nil, &utils.KafkaConfig{Brokers: []string{seedBroker.Addr()}})
 	if err != nil {
 		t.Fatal(err)
 	}
diff --git a/vendor/knative.dev/eventing/test/conformance/helpers/tracing/traces.go b/vendor/knative.dev/eventing/test/conformance/helpers/tracing/traces.go
index 079039aeb..d9d85e638 100644
--- a/vendor/knative.dev/eventing/test/conformance/helpers/tracing/traces.go
+++ b/vendor/knative.dev/eventing/test/conformance/helpers/tracing/traces.go
@@ -160,18 +160,22 @@ func GetTraceTree(trace []model.SpanModel) (*SpanTree, error) {
 		}
 	}
 
+	if len(roots) == 0 {
+		return nil, fmt.Errorf("no root spans found in the trace: Original: %v", PrettyPrintTrace(trace))
+	}
+
 	children, err := getChildren(parents, roots)
 	if err != nil {
 		return nil, fmt.Errorf("could not create span tree for %v: %v", PrettyPrintTrace(trace), err)
 	}
 
+	if len(parents) != 0 {
+		return nil, fmt.Errorf("left over spans after generating the SpanTree: %v. Original: %v", parents, PrettyPrintTrace(trace))
+	}
 	tree := SpanTree{
 		Root:     true,
 		Children: children,
 	}
-	if len(parents) != 0 {
-		return nil, fmt.Errorf("left over spans after generating the SpanTree: %v. Original: %v", parents, PrettyPrintTrace(trace))
-	}
 	return &tree, nil
 }
 
diff --git a/vendor/knative.dev/eventing/test/lib/duck/resource_creators.go b/vendor/knative.dev/eventing/test/lib/duck/resource_creators.go
index 0869cb8c0..997e1e9eb 100644
--- a/vendor/knative.dev/eventing/test/lib/duck/resource_creators.go
+++ b/vendor/knative.dev/eventing/test/lib/duck/resource_creators.go
@@ -20,18 +20,14 @@ package duck
 
 import (
 	"context"
-	"encoding/json"
 
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/dynamic"
 
+	"knative.dev/eventing/pkg/duck"
 	"knative.dev/eventing/test/lib/resources"
-
-	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
-
-	messagingv1beta1 "knative.dev/eventing/pkg/apis/messaging/v1beta1"
 )
 
 // CreateGenericChannelObject create a generic channel object with the dynamic client and channel's meta data.
@@ -41,7 +37,7 @@ func CreateGenericChannelObject(
 ) ( /*plural*/ schema.GroupVersionResource, error) {
 	// get the resource's gvr
 	gvr, _ := meta.UnsafeGuessKindToResource(obj.GroupVersionKind())
-	newChannel, err := newChannel(obj)
+	newChannel, err := duck.NewPhysicalChannel(obj.TypeMeta, obj.ObjectMeta)
 	if err != nil {
 		return gvr, err
 	}
@@ -50,31 +46,3 @@ func CreateGenericChannelObject(
 	_, err = channelResourceInterface.Create(context.Background(), newChannel, metav1.CreateOptions{})
 	return gvr, err
 }
-
-// newChannel returns an unstructured.Unstructured based on the ChannelTemplateSpec for a given meta resource.
-func newChannel(obj *resources.MetaResource) (*unstructured.Unstructured, error) {
-	// Set the name of the resource we're creating as well as the namespace, etc.
-	template := messagingv1beta1.ChannelTemplateSpecInternal{
-		TypeMeta: metav1.TypeMeta{
-			Kind:       obj.Kind,
-			APIVersion: obj.APIVersion,
-		},
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      obj.Name,
-			Namespace: obj.Namespace,
-		},
-		Spec: messagingv1beta1.ChannelTemplateSpec{
-			TypeMeta: obj.TypeMeta,
-		}.Spec,
-	}
-	raw, err := json.Marshal(template)
-	if err != nil {
-		return nil, err
-	}
-	u := &unstructured.Unstructured{}
-	err = json.Unmarshal(raw, u)
-	if err != nil {
-		return nil, err
-	}
-	return u, nil
-}