Repository: knative-sandbox/eventing-kafka-broker
Pull request #1844 (2022-02-02 09:58:34)

Fix leader election configuration
During tests, resources aren't reconciled fast enough since
chaos duck kills pods continuously and they need to acquire
leases faster.

Similar to https://github.com/knative-sandbox/eventing-kafka/pull/1068/files

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1844#" title="Author self-approved">pierDipi</a>*

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

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

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

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

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

> :exclamation: Current head 2fd1f6e differs from pull request most recent head f3ae668. Consider uploading reports for the commit f3ae668 to get more accurate results
[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1844/graphs/tree.svg?width=650&height=150&src=pr&token=xyPIgGOiwy&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1844?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@             Coverage Diff              @@
##               main    #1844      +/-   ##
============================================
- Coverage     66.53%   58.00%   -8.54%     
============================================
  Files           134       65      -69     
  Lines          5687     3662    -2025     
  Branches        178        0     -178     
============================================
- Hits           3784     2124    -1660     
+ Misses         1600     1329     -271     
+ Partials        303      209      -94     
```

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

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

| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1844?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...lane/pkg/reconciler/base/receiver\_condition\_set.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1844/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9iYXNlL3JlY2VpdmVyX2NvbmRpdGlvbl9zZXQuZ28=) | `0.00% <0.00%> (ø)` | |
| [...nting/kafka/broker/core/security/AuthProvider.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1844/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvc2VjdXJpdHkvQXV0aFByb3ZpZGVyLmphdmE=) | | |
| [...ve/eventing/kafka/broker/core/metrics/Metrics.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1844/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvbWV0cmljcy9NZXRyaWNzLmphdmE=) | | |
| [.../broker/receiver/main/ReceiverVerticleFactory.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1844/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9tYWluL1JlY2VpdmVyVmVydGljbGVGYWN0b3J5LmphdmE=) | | |
| [...ive/eventing/kafka/broker/core/AsyncCloseable.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1844/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvQXN5bmNDbG9zZWFibGUuamF2YQ==) | | |
| [.../dispatcher/impl/RecordDispatcherMutatorChain.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1844/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9SZWNvcmREaXNwYXRjaGVyTXV0YXRvckNoYWluLmphdmE=) | | |
| [...roker/dispatcher/impl/filter/AttributesFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1844/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvQXR0cmlidXRlc0ZpbHRlci5qYXZh) | | |
| [...eventing/kafka/broker/receiver/RequestContext.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1844/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9SZXF1ZXN0Q29udGV4dC5qYXZh) | | |
| [...tive/eventing/kafka/broker/core/utils/BaseEnv.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1844/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdXRpbHMvQmFzZUVudi5qYXZh) | | |
| [...nciler/impl/ResourcesReconcilerMessageHandler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1844/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvcmVjb25jaWxlci9pbXBsL1Jlc291cmNlc1JlY29uY2lsZXJNZXNzYWdlSGFuZGxlci5qYXZh) | | |
| ... and [60 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1844/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

------

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

knative-prow-robot: @pierDipi: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 61b21d93b5be3bcb0f10c85883f0052b31c13065 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1844/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1488815155335663616) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->

Patch:
diff --git a/control-plane/cmd/webhook-kafka/main.go b/control-plane/cmd/webhook-kafka/main.go
index 27b2a7aa3..817f85797 100644
--- a/control-plane/cmd/webhook-kafka/main.go
+++ b/control-plane/cmd/webhook-kafka/main.go
@@ -19,6 +19,7 @@ package main
 import (
 	"context"
 
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
@@ -33,9 +34,11 @@ import (
 	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
 	eventingcorev1beta1 "knative.dev/eventing/pkg/apis/eventing/v1"
 
+	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/core"
 	eventingv1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1"
 	eventingv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
-	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 )
 
 const (
@@ -48,10 +51,14 @@ var types = map[schema.GroupVersionKind]resourcesemantics.GenericCRD{
 	messagingv1beta1.SchemeGroupVersion.WithKind("KafkaChannel"): &messagingv1beta1.KafkaChannel{},
 }
 
-var callbacks = map[schema.GroupVersionKind]validation.Callback{
+var validationCallbacks = map[schema.GroupVersionKind]validation.Callback{
 	eventingcorev1beta1.SchemeGroupVersion.WithKind("Broker"): eventingv1.BrokerValidationCallback(),
 }
 
+var defaultingCallbacks = map[schema.GroupVersionKind]defaulting.Callback{
+	corev1.SchemeGroupVersion.WithKind("Pod"): core.DispatcherPodsDefaulting(),
+}
+
 func NewDefaultingAdmissionController(ctx context.Context, _ configmap.Watcher) *controller.Impl {
 
 	// A function that infuses the context passed to Validate/SetDefaults with custom metadata.
@@ -77,6 +84,35 @@ func NewDefaultingAdmissionController(ctx context.Context, _ configmap.Watcher)
 	)
 }
 
+func NewPodDefaultingAdmissionController(ctx context.Context, _ configmap.Watcher) *controller.Impl {
+
+	// A function that infuses the context passed to Validate/SetDefaults with custom metadata.
+	ctxFunc := func(ctx context.Context) context.Context {
+		return ctx
+	}
+
+	return defaulting.NewAdmissionController(ctx,
+		// Name of the resource webhook.
+		"pods.defaulting.webhook.kafka.eventing.knative.dev",
+
+		// The path on which to serve the webhook.
+		"/pods-defaulting",
+
+		// The resources to default.
+		// We use only defaulting callbacks for pods.
+		map[schema.GroupVersionKind]resourcesemantics.GenericCRD{},
+
+		// A function that infuses the context passed to Validate/SetDefaults with custom metadata.
+		ctxFunc,
+
+		// Whether to disallow unknown fields.
+		false,
+
+		// Extra defaulting callbacks to be applied to resources.
+		defaultingCallbacks,
+	)
+}
+
 func NewValidationAdmissionController(ctx context.Context, _ configmap.Watcher) *controller.Impl {
 	return validation.NewAdmissionController(ctx,
 		// Name of the resource webhook.
@@ -97,7 +133,7 @@ func NewValidationAdmissionController(ctx context.Context, _ configmap.Watcher)
 		true,
 
 		// Extra validating callbacks to be applied to resources.
-		callbacks,
+		validationCallbacks,
 	)
 }
 
@@ -114,6 +150,7 @@ func main() {
 	sharedmain.MainWithContext(ctx, webhook.NameFromEnv(),
 		certificates.NewController,
 		NewDefaultingAdmissionController,
+		NewPodDefaultingAdmissionController,
 		NewValidationAdmissionController,
 	)
 }
diff --git a/control-plane/config/200-controller/100-config-kafka-descheduler.yaml b/control-plane/config/200-controller/100-config-kafka-descheduler.yaml
new file mode 100644
index 000000000..00b90633a
--- /dev/null
+++ b/control-plane/config/200-controller/100-config-kafka-descheduler.yaml
@@ -0,0 +1,61 @@
+# Copyright 2021 The Knative Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: config-kafka-descheduler
+  namespace: knative-eventing
+  labels:
+    kafka.eventing.knative.dev/release: devel
+data:
+  _example: |
+    ################################
+    #                              #
+    #    EXAMPLE CONFIGURATION     #
+    #                              #
+    ################################
+
+    # This block is not actually functional configuration,
+    # but serves to illustrate the available configuration
+    # options and document them in a way that is accessible
+    # to users that `kubectl edit` this config map.
+    #
+    # These sample configuration options may be copied out of
+    # this example block and unindented to be in the data block
+    # to actually change the configuration.
+    #
+    # These predicate plugins are used to filter out pods that
+    # a vreplica cannot be removed from due to not satisfying the
+    # predicate condition. For each pod, the descheduler
+    # will call filter plugins in their configured order.
+    # predicates: |+
+                    []
+    #
+    # These priority plugins are used to score each pod that
+    # has passed the filtering phase. The scheduler will then
+    # select the pod with the highest weighted score sum to
+    # remove vreplica from. If two pods have an equal score,
+    # a pod is chosen randomly.
+    # priorities: |+
+                  [
+                    {"Name": "RemoveWithEvenPodSpreadPriority",
+                    "Weight": 10,
+                    "Args": "{\"MaxSkew\": 2}"},
+                    {"Name": "RemoveWithAvailabilityZonePriority",
+                    "Weight": 10,
+                    "Args":  "{\"MaxSkew\": 2}"},
+                    {"Name": "RemoveWithHighestOrdinalPriority",
+                    "Weight": 2}
+                  ]
diff --git a/control-plane/config/200-controller/100-config-kafka-leader-election.yaml b/control-plane/config/200-controller/100-config-kafka-leader-election.yaml
new file mode 100644
index 000000000..8e1163e6a
--- /dev/null
+++ b/control-plane/config/200-controller/100-config-kafka-leader-election.yaml
@@ -0,0 +1,47 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  labels:
+    kafka.eventing.knative.dev/release: devel
+  name: config-kafka-leader-election
+  namespace: knative-eventing
+  annotations:
+    knative.dev/example-checksum: "96896b00"
+data:
+  _example: |
+    ################################
+    #                              #
+    #    EXAMPLE CONFIGURATION     #
+    #                              #
+    ################################
+
+    # This block is not actually functional configuration,
+    # but serves to illustrate the available configuration
+    # options and document them in a way that is accessible
+    # to users that `kubectl edit` this config map.
+    #
+    # These sample configuration options may be copied out of
+    # this example block and unindented to be in the data block
+    # to actually change the configuration.
+
+    # leaseDuration is how long non-leaders will wait to try to acquire the
+    # lock; 15 seconds is the value used by core kubernetes controllers.
+    leaseDuration: "15s"
+
+    # renewDeadline is how long a leader will try to renew the lease before
+    # giving up; 10 seconds is the value used by core kubernetes controllers.
+    renewDeadline: "10s"
+
+    # retryPeriod is how long the leader election client waits between tries of
+    # actions; 2 seconds is the value used by core kubernetes controllers.
+    retryPeriod: "2s"
+
+    # buckets is the number of buckets used to partition key space of each
+    # Reconciler. If this number is M and the replica number of the controller
+    # is N, the N replicas will compete for the M buckets. The owner of a
+    # bucket will take care of the reconciling for the keys partitioned into
+    # that bucket.
+    buckets: "1"
+  leaseDuration: "15s"
+  renewDeadline: "10s"
+  retryPeriod: "2s"
diff --git a/control-plane/config/200-controller/100-config-kafka-scheduler.yaml b/control-plane/config/200-controller/100-config-kafka-scheduler.yaml
new file mode 100644
index 000000000..a419b8471
--- /dev/null
+++ b/control-plane/config/200-controller/100-config-kafka-scheduler.yaml
@@ -0,0 +1,63 @@
+# Copyright 2021 The Knative Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: config-kafka-scheduler
+  namespace: knative-eventing
+  labels:
+    kafka.eventing.knative.dev/release: devel
+data:
+  _example: |
+    ################################
+    #                              #
+    #    EXAMPLE CONFIGURATION     #
+    #                              #
+    ################################
+
+    # This block is not actually functional configuration,
+    # but serves to illustrate the available configuration
+    # options and document them in a way that is accessible
+    # to users that `kubectl edit` this config map.
+    #
+    # These sample configuration options may be copied out of
+    # this example block and unindented to be in the data block
+    # to actually change the configuration.
+    #
+    # These predicate plugins are used to filter out pods that
+    # a vreplica cannot be placed on due to not satisfying the
+    # predicate condition. For each pod, the scheduler
+    # will call filter plugins in their configured order.
+    # predicates: |+
+                  [
+                    {"Name": "PodFitsResources"},
+                    {"Name": "NoMaxResourceCount",
+                    "Args": "{\"NumPartitions\": 100}"},
+                    {"Name": "EvenPodSpread",
+                    "Args": "{\"MaxSkew\": 2}"}
+                  ]
+    #
+    # These priority plugins are used to score each pod that
+    # has passed the filtering phase. The scheduler will then
+    # select the pod with the highest weighted score sum. If
+    # two pods have an equal score, a pod is chosen randomly.
+    # priorities: |+
+                  [
+                    {"Name": "AvailabilityZonePriority",
+                    "Weight": 10,
+                    "Args":  "{\"MaxSkew\": 2}"},
+                    {"Name": "LowestOrdinalPriority",
+                    "Weight": 2}
+                  ]
diff --git a/control-plane/config/200-controller/500-controller.yaml b/control-plane/config/200-controller/500-controller.yaml
index bc327ce36..ea50753e8 100644
--- a/control-plane/config/200-controller/500-controller.yaml
+++ b/control-plane/config/200-controller/500-controller.yaml
@@ -129,6 +129,8 @@ spec:
                 fieldRef:
                   fieldPath: metadata.namespace
 
+            - name: CONFIG_LEADERELECTION_NAME
+              value: config-kafka-leader-election
             - name: CONFIG_LOGGING_NAME
               value: config-logging
             - name: CONFIG_OBSERVABILITY_NAME
diff --git a/control-plane/config/200-webhook/400-webhook-defaulting.yaml b/control-plane/config/200-webhook/400-webhook-defaulting.yaml
index f2b6c6634..718b3edef 100644
--- a/control-plane/config/200-webhook/400-webhook-defaulting.yaml
+++ b/control-plane/config/200-webhook/400-webhook-defaulting.yaml
@@ -19,7 +19,7 @@ metadata:
   labels:
     kafka.eventing.knative.dev/release: devel
 webhooks:
-  - admissionReviewVersions: ["v1", "v1beta1"]
+  - admissionReviewVersions: [ "v1", "v1beta1" ]
     clientConfig:
       service:
         name: kafka-webhook-eventing
@@ -28,3 +28,22 @@ webhooks:
     failurePolicy: Fail
     name: defaulting.webhook.kafka.eventing.knative.dev
     timeoutSeconds: 2
+
+  # Dispatcher pods webhook config.
+  - admissionReviewVersions: [ "v1", "v1beta1" ]
+    clientConfig:
+      service:
+        name: kafka-webhook-eventing
+        namespace: knative-eventing
+    sideEffects: None
+    failurePolicy: Fail
+    name: pods.defaulting.webhook.kafka.eventing.knative.dev
+    timeoutSeconds: 2
+    reinvocationPolicy: IfNeeded
+    matchPolicy: Equivalent
+    namespaceSelector:
+      matchLabels:
+        app.kubernetes.io/name: knative-eventing
+    objectSelector:
+      matchLabels:
+        app.kubernetes.io/component: kafka-dispatcher
diff --git a/control-plane/config/200-webhook/500-webhook.yaml b/control-plane/config/200-webhook/500-webhook.yaml
index 1beaf3019..b7cc21b07 100644
--- a/control-plane/config/200-webhook/500-webhook.yaml
+++ b/control-plane/config/200-webhook/500-webhook.yaml
@@ -64,6 +64,8 @@ spec:
               valueFrom:
                 fieldRef:
                   fieldPath: metadata.namespace
+            - name: CONFIG_LEADERELECTION_NAME
+              value: config-kafka-leader-election
             - name: CONFIG_LOGGING_NAME
               value: config-logging
             - name: METRICS_DOMAIN
diff --git a/control-plane/pkg/apis/core/pod_defaulting.go b/control-plane/pkg/apis/core/pod_defaulting.go
new file mode 100644
index 000000000..2ffaa586b
--- /dev/null
+++ b/control-plane/pkg/apis/core/pod_defaulting.go
@@ -0,0 +1,99 @@
+/*
+ * Copyright 2022 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package core
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/google/uuid"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+	"knative.dev/pkg/system"
+	"knative.dev/pkg/webhook"
+	"knative.dev/pkg/webhook/resourcesemantics/defaulting"
+
+	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
+)
+
+// DispatcherPodsDefaulting returns the defaulting Callback for dispatcher pods
+func DispatcherPodsDefaulting() defaulting.Callback {
+	return defaulting.NewCallback(
+		podDefaultingCallback(func(u *unstructured.Unstructured) uuid.UUID { return uuid.New() }),
+		webhook.Create,
+	)
+}
+
+// generatorFunc generates uuid to set as ConfigMap names.
+type generatorFunc func(*unstructured.Unstructured) uuid.UUID
+
+func podDefaultingCallback(generator generatorFunc) defaulting.CallbackFunc {
+	namespace := system.Namespace()
+	return func(ctx context.Context, unstructured *unstructured.Unstructured) error {
+		if unstructured.GetNamespace() != namespace {
+			return nil
+		}
+
+		var cmName string
+		if unstructured.GetName() == "" {
+			cmName = generator(unstructured).String()
+		} else {
+			cmName = unstructured.GetName()
+		}
+
+		volume := &corev1.Volume{
+			Name: kafkainternals.DispatcherVolumeName,
+			VolumeSource: corev1.VolumeSource{
+				ConfigMap: &corev1.ConfigMapVolumeSource{
+					LocalObjectReference: corev1.LocalObjectReference{Name: cmName},
+				},
+			},
+		}
+
+		unstrVolume, err := runtime.DefaultUnstructuredConverter.ToUnstructured(volume)
+		if err != nil {
+			return fmt.Errorf("failed to convert volume struct to unstructured object: %w", err)
+		}
+
+		pod := unstructured.Object
+		spec := pod["spec"].(map[string]interface{})
+
+		volumes, ok := spec["volumes"].([]interface{})
+		if !ok {
+			volumes = []interface{}{}
+			spec["volumes"] = volumes
+		}
+
+		found := false
+		for k, v := range volumes {
+			vt := v.(map[string]interface{})
+			if vt["name"].(string) == "contract-resources" {
+				found = true
+				volumes[k] = unstrVolume
+			}
+		}
+
+		if !found {
+			spec["volumes"] = append(volumes, unstrVolume)
+		}
+
+		unstructured.Object = pod
+
+		return nil
+	}
+}
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
index ec5c7d83c..92d17daca 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
@@ -30,6 +30,10 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 )
 
+const (
+	DispatcherVolumeName = "contract-resources"
+)
+
 // +genclient
 // +genreconciler
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
diff --git a/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index e732754de..775825b80 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -20,8 +20,8 @@ Lists of 158 third-party dependencies.
      (Apache License, Version 2.0) Guava: Google Core Libraries for Java (com.google.guava:guava:30.1.1-android - https://github.com/google/guava/guava)
      (The Apache Software License, Version 2.0) Guava ListenableFuture only (com.google.guava:listenablefuture:9999.0-empty-to-avoid-conflict-with-guava - https://github.com/google/guava/listenablefuture)
      (The Apache Software License, Version 2.0) J2ObjC Annotations (com.google.j2objc:j2objc-annotations:1.3 - https://github.com/google/j2objc/)
-     (3-Clause BSD License) Protocol Buffers [Core] (com.google.protobuf:protobuf-java:3.19.3 - https://developers.google.com/protocol-buffers/protobuf-java/)
-     (3-Clause BSD License) Protocol Buffers [Util] (com.google.protobuf:protobuf-java-util:3.19.3 - https://developers.google.com/protocol-buffers/protobuf-java-util/)
+     (3-Clause BSD License) Protocol Buffers [Core] (com.google.protobuf:protobuf-java:3.19.4 - https://developers.google.com/protocol-buffers/protobuf-java/)
+     (3-Clause BSD License) Protocol Buffers [Util] (com.google.protobuf:protobuf-java-util:3.19.4 - https://developers.google.com/protocol-buffers/protobuf-java-util/)
      (Apache 2.0) OkHttp Logging Interceptor (com.squareup.okhttp3:logging-interceptor:3.12.12 - https://github.com/square/okhttp/logging-interceptor)
      (Apache 2.0) MockWebServer (com.squareup.okhttp3:mockwebserver:3.12.12 - https://github.com/square/okhttp/mockwebserver)
      (Apache 2.0) OkHttp (com.squareup.okhttp3:okhttp:3.14.9 - https://github.com/square/okhttp/okhttp)
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index 910aa857a..ee60953fc 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -46,7 +46,7 @@
     <micrometer.version>1.8.2</micrometer.version>
     <opentelemetry.version>1.10.1</opentelemetry.version>
     <jackson.version>2.13.1</jackson.version>
-    <protobuf.version>3.19.3</protobuf.version>
+    <protobuf.version>3.19.4</protobuf.version>
     <slf4j.version>1.7.35</slf4j.version>
     <ch.qos.logback.version>1.2.10</ch.qos.logback.version>
     <net.logstash.logback.encoder.version>7.0.1</net.logstash.logback.encoder.version>
diff --git a/go.mod b/go.mod
index 378cb5b92..0319af2e3 100644
--- a/go.mod
+++ b/go.mod
@@ -23,9 +23,9 @@ require (
 	k8s.io/apiserver v0.22.5
 	k8s.io/client-go v0.22.5
 	k8s.io/utils v0.0.0-20211208161948-7d6a63dca704
-	knative.dev/eventing v0.29.0
-	knative.dev/eventing-kafka v0.28.1-0.20220126011444-734561d578b9
-	knative.dev/hack v0.0.0-20220118141833-9b2ed8471e30
-	knative.dev/pkg v0.0.0-20220118160532-77555ea48cd4
-	knative.dev/reconciler-test v0.0.0-20220118183433-c8bfbe66bada
+	knative.dev/eventing v0.29.1-0.20220201083031-25bf6fe36dfe
+	knative.dev/eventing-kafka v0.28.1-0.20220201082931-c084e4c42ad3
+	knative.dev/hack v0.0.0-20220201013531-82bfca153560
+	knative.dev/pkg v0.0.0-20220201132031-8681fe2035b9
+	knative.dev/reconciler-test v0.0.0-20220126171745-740e77ebaace
 )
diff --git a/go.sum b/go.sum
index 2860ff725..b8d5bb772 100644
--- a/go.sum
+++ b/go.sum
@@ -1345,17 +1345,26 @@ k8s.io/utils v0.0.0-20210819203725-bdf08cb9a70a/go.mod h1:jPW/WVKK9YHAvNhRxK0md/
 k8s.io/utils v0.0.0-20211208161948-7d6a63dca704 h1:ZKMMxTvduyf5WUtREOqg5LiXaN1KO/+0oOQPRFrClpo=
 k8s.io/utils v0.0.0-20211208161948-7d6a63dca704/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 knative.dev/control-protocol v0.0.0-20220118185434-ad4ce9e9d455/go.mod h1:xiTC5Qt9GdPuhRknr0Oq744SsUqbkDx0oe04L9a+qYY=
-knative.dev/eventing v0.29.0 h1:+AhcYEuuz37x1op71bknZ/lFgsvuZJl0AKFOOFiH03s=
-knative.dev/eventing v0.29.0/go.mod h1:u5T5NZTDUsLR7yJwp5MDnBnDX5MhywD3yK3Rq+7gTtI=
-knative.dev/eventing-kafka v0.28.1-0.20220126011444-734561d578b9 h1:UCtzXoRZvZf4dXDZ7oKjXny2FLCDqIIZnYQdtu2A3/o=
-knative.dev/eventing-kafka v0.28.1-0.20220126011444-734561d578b9/go.mod h1:Y9+hHdaYCmD46PPOU2pAfWCUXJyIfgJOnkWZwsBAEoU=
-knative.dev/hack v0.0.0-20220118141833-9b2ed8471e30 h1:UkNpCWCMM5C4AeQ8aTrPTuR/6OeARiqk+LEQ6tuMP7c=
+knative.dev/eventing v0.29.1-0.20220128133147-7a44b07f299d/go.mod h1:82rJhzGDSMl1J2HQSdC3Z7foKJO5s5koL3e3S6mSvug=
+knative.dev/eventing v0.29.1-0.20220201083031-25bf6fe36dfe h1:VLVkZaToXNixhY//FZfAr9fAF7jQQVGo64qT/3pJpfk=
+knative.dev/eventing v0.29.1-0.20220201083031-25bf6fe36dfe/go.mod h1:jsc5u3aC+BYmfUR2cRz9jb77krfULStxXkR1EgMUWp0=
+knative.dev/eventing-kafka v0.28.1-0.20220201082931-c084e4c42ad3 h1:BeGfGTTnZ6A1QILh3kPQcoR4CkGOacBAct50BW7iip0=
+knative.dev/eventing-kafka v0.28.1-0.20220201082931-c084e4c42ad3/go.mod h1:bnEFmqZhjVpZbzrChBki4c9HWAqyUnQVK+VPe0wcAqo=
 knative.dev/hack v0.0.0-20220118141833-9b2ed8471e30/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
+knative.dev/hack v0.0.0-20220128200847-51a42b2eb63e/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
+knative.dev/hack v0.0.0-20220201013531-82bfca153560 h1:JTQwe928hZokY181gjYKeVdxr+s3v95LDkMs7R0fxfE=
+knative.dev/hack v0.0.0-20220201013531-82bfca153560/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack/schema v0.0.0-20220118141833-9b2ed8471e30/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
-knative.dev/pkg v0.0.0-20220118160532-77555ea48cd4 h1:b9aXVrcfM/ajjHE/lGvlJOHZNAR5FF2TOTLWG7eMhzQ=
+knative.dev/hack/schema v0.0.0-20220128200847-51a42b2eb63e/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
 knative.dev/pkg v0.0.0-20220118160532-77555ea48cd4/go.mod h1:etVT7Tm8pSDf4RKhGk4r7j/hj3dNBpvT7bO6a6wpahs=
-knative.dev/reconciler-test v0.0.0-20220118183433-c8bfbe66bada h1:YSxoVY3Z2GF8QsNrWlJ8qQlJ6pUXKVsX7xkhY6odOXU=
+knative.dev/pkg v0.0.0-20220121092305-3ba5d72e310a/go.mod h1:etVT7Tm8pSDf4RKhGk4r7j/hj3dNBpvT7bO6a6wpahs=
+knative.dev/pkg v0.0.0-20220129160148-ca64d3b3ea8c/go.mod h1:BML2uvSEBhY/cpEsAyWv1+vLy2fxNk8VDJ9fT+D0tj4=
+knative.dev/pkg v0.0.0-20220131144930-f4b57aef0006/go.mod h1:bZMFTPDPHV3wXuiQ09UJuEGYYQnfpe81MCxNvsMAiJk=
+knative.dev/pkg v0.0.0-20220201132031-8681fe2035b9 h1:oxeI9FbCIDxLrG2BAIgguv1tWHOpIiXv56VuKWD2S5Y=
+knative.dev/pkg v0.0.0-20220201132031-8681fe2035b9/go.mod h1:bZMFTPDPHV3wXuiQ09UJuEGYYQnfpe81MCxNvsMAiJk=
 knative.dev/reconciler-test v0.0.0-20220118183433-c8bfbe66bada/go.mod h1:XV4cghzCtdASkfUsfMYSnGfGRyd/naDjy9h7Tnae22g=
+knative.dev/reconciler-test v0.0.0-20220126171745-740e77ebaace h1:ECS0b2WOBhCTm7PHmYeDDXvD5oXVeXGkd93QW33Y3Us=
+knative.dev/reconciler-test v0.0.0-20220126171745-740e77ebaace/go.mod h1:XV4cghzCtdASkfUsfMYSnGfGRyd/naDjy9h7Tnae22g=
 modernc.org/cc v1.0.0/go.mod h1:1Sk4//wdnYJiUIxnW8ddKpaOJCF37yAdqYnkxUpaYxw=
 modernc.org/golex v1.0.0/go.mod h1:b/QX9oBD/LhixY6NDh+IdGv17hgB+51fET1i2kPSmvk=
 modernc.org/mathutil v1.0.0/go.mod h1:wU0vUrJsVWBZ4P6e7xtFJEhFSNsfRLJ8H458uRjg03k=
diff --git a/hack/update-checksums.sh b/hack/update-checksums.sh
index cefc4ec77..54f0d5d0f 100755
--- a/hack/update-checksums.sh
+++ b/hack/update-checksums.sh
@@ -27,3 +27,5 @@ fi
 source $(dirname $0)/../vendor/knative.dev/hack/library.sh
 
 go run "${REPO_ROOT_DIR}/vendor/knative.dev/pkg/configmap/hash-gen" "${REPO_ROOT_DIR}"/control-plane/config/200-controller/100-config-tracing.yaml
+go run "${REPO_ROOT_DIR}/vendor/knative.dev/pkg/configmap/hash-gen" "${REPO_ROOT_DIR}"/control-plane/config/200-controller/100-config-logging.yaml
+go run "${REPO_ROOT_DIR}/vendor/knative.dev/pkg/configmap/hash-gen" "${REPO_ROOT_DIR}"/control-plane/config/200-controller/100-config-kafka-leader-election.yaml
diff --git a/vendor/knative.dev/hack/README.md b/vendor/knative.dev/hack/README.md
index a6a98c784..570f37097 100644
--- a/vendor/knative.dev/hack/README.md
+++ b/vendor/knative.dev/hack/README.md
@@ -167,6 +167,11 @@ This is a helper script for Knative E2E test scripts. To use it:
 1. By default Istio is installed on the cluster via Addon, use
    `--skip-istio-addon` if you choose not to have it preinstalled.
 
+1. By default Google Kubernetes Engine telemetry to Cloud Logging and Monitoring is disabled.
+   This can be enabled by setting `ENABLE_GKE_TELEMETRY` to `true`.
+   
+1. By default Spot Worker nodes are disabled. This can be enabled by setting `ENABLE_PREEMPTIBLE_NODES`
+   to `true`.
 ### Sample end-to-end test script
 
 This script will test that the latest Knative Serving nightly release works. It
diff --git a/vendor/knative.dev/hack/infra-library.sh b/vendor/knative.dev/hack/infra-library.sh
index f87c259d1..a96210e2e 100644
--- a/vendor/knative.dev/hack/infra-library.sh
+++ b/vendor/knative.dev/hack/infra-library.sh
@@ -77,20 +77,6 @@ function dump_cluster_state() {
   echo "***************************************"
 }
 
-# Sets the current user as cluster admin for the given cluster.
-# Parameters: $1 - cluster context name
-function acquire_cluster_admin_role() {
-  if [[ -z "$(kubectl get clusterrolebinding cluster-admin-binding 2> /dev/null)" ]]; then
-    if [[ "$1" =~ ^gke_.* ]]; then
-      kubectl create clusterrolebinding cluster-admin-binding \
-        --clusterrole=cluster-admin --user="$(gcloud config get-value core/account)"
-    else
-      kubectl create clusterrolebinding cluster-admin-binding \
-        --clusterrole=cluster-admin --user="prow"
-    fi
-  fi
-}
-
 # Create a test cluster and run the tests if provided.
 # Parameters: $1 - cluster provider name, e.g. gke
 #             $2 - custom flags supported by kntest
@@ -137,5 +123,15 @@ function create_gke_test_cluster() {
   local -n _custom_flags=$1
   local -n _test_command=$2
 
-  run_kntest kubetest2 gke "${_custom_flags[@]}" --test-command="${_test_command[*]}"
+  # We are disabling logs and metrics on Boskos Clusters by default as they are not used. Manually set ENABLE_GKE_TELEMETRY to true to enable telemetry
+  # and ENABLE_PREEMPTIBLE_NODES to true to create preemptible/spot VMs. VM Preemption is a rare event and shouldn't be distruptive given the fault tolerant nature of our tests.
+  local extra_gcloud_flags=""
+  if [[ "$ENABLE_GKE_TELEMETRY" != "true" ]]; then
+    extra_gcloud_flags="${extra_gcloud_flags} --logging=NONE --monitoring=NONE"
+  fi
+
+  if [[ "$ENABLE_PREEMPTIBLE_NODES" == "true" ]]; then
+    extra_gcloud_flags="${extra_gcloud_flags} --preemptible"
+  fi
+  run_kntest kubetest2 gke "${_custom_flags[@]}" --test-command="${_test_command[*]}" --extra-gcloud-flags="${extra_gcloud_flags}"
 }
diff --git a/vendor/knative.dev/hack/shell/project.go b/vendor/knative.dev/hack/shell/project.go
index f4cd1681a..a6df6c4a6 100644
--- a/vendor/knative.dev/hack/shell/project.go
+++ b/vendor/knative.dev/hack/shell/project.go
@@ -37,11 +37,12 @@ var (
 // NewProjectLocation creates a ProjectLocation that is used to calculate
 // relative paths within the project.
 func NewProjectLocation(pathToRoot string) (ProjectLocation, error) {
-	_, filename, _, ok := runtime.Caller(1)
+	pc, filename, _, ok := runtime.Caller(1)
 	if !ok {
 		return nil, ErrCantGetCaller
 	}
-	err := ensureIsValid(filename)
+	funcName := runtime.FuncForPC(pc).Name()
+	err := isCallsiteAllowed(funcName)
 	if err != nil {
 		return nil, err
 	}
@@ -64,17 +65,17 @@ type callerLocation struct {
 	pathToRoot string
 }
 
-func ensureIsValid(filename string) error {
+func isCallsiteAllowed(funcName string) error {
 	validPaths := []string{
-		"knative.+/test/upgrade/",
-		"knative(:?\\.dev/|-)hack/shell/",
+		"knative.+/test/upgrade",
+		"knative(:?\\.dev/|-)hack/shell",
 	}
 	for _, validPath := range validPaths {
 		r := regexp.MustCompile(validPath)
-		if loc := r.FindStringIndex(filename); loc != nil {
+		if loc := r.FindStringIndex(funcName); loc != nil {
 			return nil
 		}
 	}
 	return fmt.Errorf("%w, tried using from: %s",
-		ErrCallerNotAllowed, filename)
+		ErrCallerNotAllowed, funcName)
 }
diff --git a/vendor/knative.dev/pkg/resolver/addressable_resolver.go b/vendor/knative.dev/pkg/resolver/addressable_resolver.go
index 0e50a2578..d6995947a 100644
--- a/vendor/knative.dev/pkg/resolver/addressable_resolver.go
+++ b/vendor/knative.dev/pkg/resolver/addressable_resolver.go
@@ -23,12 +23,14 @@ import (
 
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/tools/cache"
+
 	"knative.dev/pkg/client/injection/ducks/duck/v1/addressable"
 	"knative.dev/pkg/controller"
 
 	corev1 "k8s.io/api/core/v1"
 	apierrs "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
+
 	"knative.dev/pkg/apis"
 	pkgapisduck "knative.dev/pkg/apis/duck"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
@@ -169,17 +171,17 @@ func (r *URIResolver) URIFromObjectReference(ctx context.Context, ref *corev1.Ob
 		Namespace:  ref.Namespace,
 		Name:       ref.Name,
 	}, parent); err != nil {
-		return nil, apierrs.NewNotFound(gvr.GroupResource(), ref.Name)
+		return nil, fmt.Errorf("failed to track reference %s %s/%s: %w", gvr.String(), ref.Namespace, ref.Name, err)
 	}
 
 	lister, err := r.listerFactory(gvr)
 	if err != nil {
-		return nil, apierrs.NewNotFound(gvr.GroupResource(), "Lister")
+		return nil, fmt.Errorf("failed to get lister for %s: %w", gvr.String(), err)
 	}
 
 	obj, err := lister.ByNamespace(ref.Namespace).Get(ref.Name)
 	if err != nil {
-		return nil, apierrs.NewNotFound(gvr.GroupResource(), ref.Name)
+		return nil, fmt.Errorf("failed to get object %s/%s: %w", ref.Namespace, ref.Name, err)
 	}
 
 	// K8s Services are special cased. They can be called, even though they do not satisfy the
diff --git a/vendor/knative.dev/pkg/webhook/resourcesemantics/defaulting/user_info.go b/vendor/knative.dev/pkg/webhook/resourcesemantics/defaulting/user_info.go
index fcd97a456..f206588e4 100644
--- a/vendor/knative.dev/pkg/webhook/resourcesemantics/defaulting/user_info.go
+++ b/vendor/knative.dev/pkg/webhook/resourcesemantics/defaulting/user_info.go
@@ -27,6 +27,11 @@ import (
 	"knative.dev/pkg/apis"
 )
 
+var (
+	emptyGroupUpdaterAnnotation = apis.UpdaterAnnotationSuffix[1:]
+	emptyGroupCreatorAnnotation = apis.CreatorAnnotationSuffix[1:]
+)
+
 // setUserInfoAnnotations sets creator and updater annotations on a resource.
 func setUserInfoAnnotations(ctx context.Context, resource apis.HasSpec, groupName string) {
 	if ui := apis.GetUserInfo(ctx); ui != nil {
@@ -41,15 +46,22 @@ func setUserInfoAnnotations(ctx context.Context, resource apis.HasSpec, groupNam
 			objectMetaAccessor.GetObjectMeta().SetAnnotations(annotations)
 		}
 
+		updaterAnnotation := emptyGroupUpdaterAnnotation
+		creatorAnnotation := emptyGroupCreatorAnnotation
+		if groupName != "" {
+			updaterAnnotation = groupName + apis.UpdaterAnnotationSuffix
+			creatorAnnotation = groupName + apis.CreatorAnnotationSuffix
+		}
+
 		if apis.IsInUpdate(ctx) {
 			old := apis.GetBaseline(ctx).(apis.HasSpec)
 			if equality.Semantic.DeepEqual(old.GetUntypedSpec(), resource.GetUntypedSpec()) {
 				return
 			}
-			annotations[groupName+apis.UpdaterAnnotationSuffix] = ui.Username
+			annotations[updaterAnnotation] = ui.Username
 		} else {
-			annotations[groupName+apis.CreatorAnnotationSuffix] = ui.Username
-			annotations[groupName+apis.UpdaterAnnotationSuffix] = ui.Username
+			annotations[creatorAnnotation] = ui.Username
+			annotations[updaterAnnotation] = ui.Username
 		}
 		objectMetaAccessor.GetObjectMeta().SetAnnotations(annotations)
 	}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 694d7d209..3de10ee32 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1124,7 +1124,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.29.0
+# knative.dev/eventing v0.29.1-0.20220201083031-25bf6fe36dfe
 ## explicit
 knative.dev/eventing/pkg/apis/config
 knative.dev/eventing/pkg/apis/duck
@@ -1272,7 +1272,7 @@ knative.dev/eventing/test/upgrade/prober/wathola/fetcher
 knative.dev/eventing/test/upgrade/prober/wathola/forwarder
 knative.dev/eventing/test/upgrade/prober/wathola/receiver
 knative.dev/eventing/test/upgrade/prober/wathola/sender
-# knative.dev/eventing-kafka v0.28.1-0.20220126011444-734561d578b9
+# knative.dev/eventing-kafka v0.28.1-0.20220201082931-c084e4c42ad3
 ## explicit
 knative.dev/eventing-kafka/pkg/apis/bindings
 knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1
@@ -1333,11 +1333,11 @@ knative.dev/eventing-kafka/test/rekt/features/kafkachannel
 knative.dev/eventing-kafka/test/rekt/resources/kafkachannel
 knative.dev/eventing-kafka/test/rekt/resources/resetoffset
 knative.dev/eventing-kafka/test/upgrade/continual
-# knative.dev/hack v0.0.0-20220118141833-9b2ed8471e30
+# knative.dev/hack v0.0.0-20220201013531-82bfca153560
 ## explicit
 knative.dev/hack
 knative.dev/hack/shell
-# knative.dev/pkg v0.0.0-20220118160532-77555ea48cd4
+# knative.dev/pkg v0.0.0-20220201132031-8681fe2035b9
 ## explicit
 knative.dev/pkg/apis
 knative.dev/pkg/apis/duck
@@ -1436,7 +1436,7 @@ knative.dev/pkg/webhook/json
 knative.dev/pkg/webhook/resourcesemantics
 knative.dev/pkg/webhook/resourcesemantics/defaulting
 knative.dev/pkg/webhook/resourcesemantics/validation
-# knative.dev/reconciler-test v0.0.0-20220118183433-c8bfbe66bada
+# knative.dev/reconciler-test v0.0.0-20220126171745-740e77ebaace
 ## explicit
 knative.dev/reconciler-test/cmd/eventshub
 knative.dev/reconciler-test/pkg/environment


Test patch:
diff --git a/control-plane/pkg/apis/core/pod_defaulting_test.go b/control-plane/pkg/apis/core/pod_defaulting_test.go
new file mode 100644
index 000000000..ce051a5b7
--- /dev/null
+++ b/control-plane/pkg/apis/core/pod_defaulting_test.go
@@ -0,0 +1,501 @@
+/*
+ * Copyright 2022 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package core
+
+import (
+	"context"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+	"knative.dev/pkg/system"
+)
+
+func TestPodDefaulting(t *testing.T) {
+	t.Setenv("SYSTEM_NAMESPACE", "knative-eventing")
+
+	id := uuid.New()
+
+	tests := []struct {
+		name    string
+		ctx     context.Context
+		given   corev1.Pod
+		want    corev1.Pod
+		wantErr bool
+	}{
+		{
+			name: "add volume source",
+			ctx:  context.Background(),
+			given: corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "kafka-source-dispatcher",
+					Namespace: system.Namespace(),
+				},
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{
+						{
+							Name: "dispatcher",
+							VolumeMounts: []corev1.VolumeMount{
+								{
+									Name:      "contract-resources",
+									ReadOnly:  false,
+									MountPath: "/etc/contract/contract-resources",
+								},
+							},
+						},
+					},
+				},
+			},
+			want: corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "kafka-source-dispatcher",
+					Namespace: system.Namespace(),
+				},
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{
+						{
+							Name: "dispatcher",
+							VolumeMounts: []corev1.VolumeMount{
+								{
+									Name:      "contract-resources",
+									ReadOnly:  false,
+									MountPath: "/etc/contract/contract-resources",
+								},
+							},
+						},
+					},
+					Volumes: []corev1.Volume{
+						{
+							Name: "contract-resources",
+							VolumeSource: corev1.VolumeSource{
+								ConfigMap: &corev1.ConfigMapVolumeSource{
+									LocalObjectReference: corev1.LocalObjectReference{Name: "kafka-source-dispatcher"},
+								},
+							},
+						},
+					},
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "add volume source, preserving another volume",
+			ctx:  context.Background(),
+			given: corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "kafka-source-dispatcher",
+					Namespace: system.Namespace(),
+				},
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{
+						{
+							Name: "dispatcher",
+							VolumeMounts: []corev1.VolumeMount{
+								{
+									Name:      "contract-resources",
+									ReadOnly:  false,
+									MountPath: "/etc/contract/contract-resources",
+								},
+							},
+						},
+					},
+					Volumes: []corev1.Volume{
+						{
+							Name: "contract-resources-2",
+							VolumeSource: corev1.VolumeSource{
+								ConfigMap: &corev1.ConfigMapVolumeSource{
+									LocalObjectReference: corev1.LocalObjectReference{Name: "kafka-source-dispatcher-2"},
+								},
+							},
+						},
+					},
+				},
+			},
+			want: corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "kafka-source-dispatcher",
+					Namespace: system.Namespace(),
+				},
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{
+						{
+							Name: "dispatcher",
+							VolumeMounts: []corev1.VolumeMount{
+								{
+									Name:      "contract-resources",
+									ReadOnly:  false,
+									MountPath: "/etc/contract/contract-resources",
+								},
+							},
+						},
+					},
+					Volumes: []corev1.Volume{
+						{
+							Name: "contract-resources-2",
+							VolumeSource: corev1.VolumeSource{
+								ConfigMap: &corev1.ConfigMapVolumeSource{
+									LocalObjectReference: corev1.LocalObjectReference{Name: "kafka-source-dispatcher-2"},
+								},
+							},
+						},
+						{
+							Name: "contract-resources",
+							VolumeSource: corev1.VolumeSource{
+								ConfigMap: &corev1.ConfigMapVolumeSource{
+									LocalObjectReference: corev1.LocalObjectReference{Name: "kafka-source-dispatcher"},
+								},
+							},
+						},
+					},
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "replace volume source, preserving another volume",
+			ctx:  context.Background(),
+			given: corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "kafka-source-dispatcher",
+					Namespace: system.Namespace(),
+				},
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{
+						{
+							Name: "dispatcher",
+							VolumeMounts: []corev1.VolumeMount{
+								{
+									Name:      "contract-resources",
+									ReadOnly:  false,
+									MountPath: "/etc/contract/contract-resources",
+								},
+							},
+						},
+					},
+					Volumes: []corev1.Volume{
+						{
+							Name: "contract-resources-2",
+							VolumeSource: corev1.VolumeSource{
+								ConfigMap: &corev1.ConfigMapVolumeSource{
+									LocalObjectReference: corev1.LocalObjectReference{Name: "kafka-source-dispatcher-2"},
+								},
+							},
+						},
+						{
+							Name: "contract-resources",
+							VolumeSource: corev1.VolumeSource{
+								ConfigMap: &corev1.ConfigMapVolumeSource{
+									LocalObjectReference: corev1.LocalObjectReference{Name: id.String() + "-aaa"},
+								},
+							},
+						},
+					},
+				},
+			},
+			want: corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "kafka-source-dispatcher",
+					Namespace: system.Namespace(),
+				},
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{
+						{
+							Name: "dispatcher",
+							VolumeMounts: []corev1.VolumeMount{
+								{
+									Name:      "contract-resources",
+									ReadOnly:  false,
+									MountPath: "/etc/contract/contract-resources",
+								},
+							},
+						},
+					},
+					Volumes: []corev1.Volume{
+						{
+							Name: "contract-resources-2",
+							VolumeSource: corev1.VolumeSource{
+								ConfigMap: &corev1.ConfigMapVolumeSource{
+									LocalObjectReference: corev1.LocalObjectReference{Name: "kafka-source-dispatcher-2"},
+								},
+							},
+						},
+						{
+							Name: "contract-resources",
+							VolumeSource: corev1.VolumeSource{
+								ConfigMap: &corev1.ConfigMapVolumeSource{
+									LocalObjectReference: corev1.LocalObjectReference{Name: "kafka-source-dispatcher"},
+								},
+							},
+						},
+					},
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "no changes to volume source, preserving another volume",
+			ctx:  context.Background(),
+			given: corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "kafka-source-dispatcher",
+					Namespace: system.Namespace(),
+				},
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{
+						{
+							Name: "dispatcher",
+							VolumeMounts: []corev1.VolumeMount{
+								{
+									Name:      "contract-resources",
+									ReadOnly:  false,
+									MountPath: "/etc/contract/contract-resources",
+								},
+							},
+						},
+					},
+					Volumes: []corev1.Volume{
+						{
+							Name: "contract-resources-2",
+							VolumeSource: corev1.VolumeSource{
+								ConfigMap: &corev1.ConfigMapVolumeSource{
+									LocalObjectReference: corev1.LocalObjectReference{Name: "kafka-source-dispatcher-2"},
+								},
+							},
+						},
+						{
+							Name: "contract-resources",
+							VolumeSource: corev1.VolumeSource{
+								ConfigMap: &corev1.ConfigMapVolumeSource{
+									LocalObjectReference: corev1.LocalObjectReference{Name: "kafka-source-dispatcher"},
+								},
+							},
+						},
+					},
+				},
+			},
+			want: corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "kafka-source-dispatcher",
+					Namespace: system.Namespace(),
+				},
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{
+						{
+							Name: "dispatcher",
+							VolumeMounts: []corev1.VolumeMount{
+								{
+									Name:      "contract-resources",
+									ReadOnly:  false,
+									MountPath: "/etc/contract/contract-resources",
+								},
+							},
+						},
+					},
+					Volumes: []corev1.Volume{
+						{
+							Name: "contract-resources-2",
+							VolumeSource: corev1.VolumeSource{
+								ConfigMap: &corev1.ConfigMapVolumeSource{
+									LocalObjectReference: corev1.LocalObjectReference{Name: "kafka-source-dispatcher-2"},
+								},
+							},
+						},
+						{
+							Name: "contract-resources",
+							VolumeSource: corev1.VolumeSource{
+								ConfigMap: &corev1.ConfigMapVolumeSource{
+									LocalObjectReference: corev1.LocalObjectReference{Name: "kafka-source-dispatcher"},
+								},
+							},
+						},
+					},
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "no namespace, no changes",
+			ctx:  context.Background(),
+			given: corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{Name: "kafka-source-dispatcher"},
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{
+						{
+							Name: "dispatcher",
+							VolumeMounts: []corev1.VolumeMount{
+								{
+									Name:      "contract-resources",
+									ReadOnly:  false,
+									MountPath: "/etc/contract/contract-resources",
+								},
+							},
+						},
+					},
+					Volumes: []corev1.Volume{
+						{
+							Name: "contract-resources-2",
+							VolumeSource: corev1.VolumeSource{
+								ConfigMap: &corev1.ConfigMapVolumeSource{
+									LocalObjectReference: corev1.LocalObjectReference{Name: "kafka-source-dispatcher-2"},
+								},
+							},
+						},
+						{
+							Name: "contract-resources",
+							VolumeSource: corev1.VolumeSource{
+								ConfigMap: &corev1.ConfigMapVolumeSource{
+									LocalObjectReference: corev1.LocalObjectReference{Name: "kafka-source-dispatcher"},
+								},
+							},
+						},
+					},
+				},
+			},
+			want: corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{Name: "kafka-source-dispatcher"},
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{
+						{
+							Name: "dispatcher",
+							VolumeMounts: []corev1.VolumeMount{
+								{
+									Name:      "contract-resources",
+									ReadOnly:  false,
+									MountPath: "/etc/contract/contract-resources",
+								},
+							},
+						},
+					},
+					Volumes: []corev1.Volume{
+						{
+							Name: "contract-resources-2",
+							VolumeSource: corev1.VolumeSource{
+								ConfigMap: &corev1.ConfigMapVolumeSource{
+									LocalObjectReference: corev1.LocalObjectReference{Name: "kafka-source-dispatcher-2"},
+								},
+							},
+						},
+						{
+							Name: "contract-resources",
+							VolumeSource: corev1.VolumeSource{
+								ConfigMap: &corev1.ConfigMapVolumeSource{
+									LocalObjectReference: corev1.LocalObjectReference{Name: "kafka-source-dispatcher"},
+								},
+							},
+						},
+					},
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "No name, use generator",
+			ctx:  context.Background(),
+			given: corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{Namespace: system.Namespace()},
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{
+						{
+							Name: "dispatcher",
+							VolumeMounts: []corev1.VolumeMount{
+								{
+									Name:      "contract-resources",
+									ReadOnly:  false,
+									MountPath: "/etc/contract/contract-resources",
+								},
+							},
+						},
+					},
+					Volumes: []corev1.Volume{
+						{
+							Name: "contract-resources-2",
+							VolumeSource: corev1.VolumeSource{
+								ConfigMap: &corev1.ConfigMapVolumeSource{
+									LocalObjectReference: corev1.LocalObjectReference{Name: "kafka-source-dispatcher-2"},
+								},
+							},
+						},
+					},
+				},
+			},
+			want: corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{Namespace: system.Namespace()},
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{
+						{
+							Name: "dispatcher",
+							VolumeMounts: []corev1.VolumeMount{
+								{
+									Name:      "contract-resources",
+									ReadOnly:  false,
+									MountPath: "/etc/contract/contract-resources",
+								},
+							},
+						},
+					},
+					Volumes: []corev1.Volume{
+						{
+							Name: "contract-resources-2",
+							VolumeSource: corev1.VolumeSource{
+								ConfigMap: &corev1.ConfigMapVolumeSource{
+									LocalObjectReference: corev1.LocalObjectReference{Name: "kafka-source-dispatcher-2"},
+								},
+							},
+						},
+						{
+							Name: "contract-resources",
+							VolumeSource: corev1.VolumeSource{
+								ConfigMap: &corev1.ConfigMapVolumeSource{
+									LocalObjectReference: corev1.LocalObjectReference{Name: id.String()},
+								},
+							},
+						},
+					},
+				},
+			},
+			wantErr: false,
+		},
+	}
+
+	var generator generatorFunc = func(*unstructured.Unstructured) uuid.UUID { return id }
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+
+			given, err := runtime.DefaultUnstructuredConverter.ToUnstructured(&tt.given)
+			require.Nil(t, err)
+
+			unstr := &unstructured.Unstructured{Object: given}
+			if err := podDefaultingCallback(generator)(tt.ctx, unstr); (err != nil) != tt.wantErr {
+				t.Errorf("podDefaultingCallback() error = %v, wantErr %v", err, tt.wantErr)
+			}
+
+			got := corev1.Pod{}
+			err = runtime.DefaultUnstructuredConverter.FromUnstructured(unstr.UnstructuredContent(), &got)
+			require.Nil(t, err)
+
+			if diff := cmp.Diff(tt.want, got); diff != "" {
+				t.Error("(-want,+got)", diff)
+			}
+		})
+	}
+}
diff --git a/control-plane/pkg/reconciler/source/source_test.go b/control-plane/pkg/reconciler/source/source_test.go
index b3736ca3a..1220f782a 100644
--- a/control-plane/pkg/reconciler/source/source_test.go
+++ b/control-plane/pkg/reconciler/source/source_test.go
@@ -478,7 +478,7 @@ func TestReconcileKind(t *testing.T) {
 						StatusTopicReadyWithName(strings.Join(SourceTopics, ", ")),
 						StatusDataPlaneAvailable,
 						StatusInitialOffsetsCommitted,
-						StatusSourceSinkNotResolved("failed to resolve destination: services \"test-service\" not found"),
+						StatusSourceSinkNotResolved("failed to resolve destination: failed to get object test-service-namespace/test-service: services \"test-service\" not found"),
 					),
 				},
 			},
@@ -487,7 +487,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			WantEvents: []string{
 				finalizerUpdatedEvent,
-				"Warning InternalError failed to resolve sink: failed to resolve destination: services \"test-service\" not found",
+				"Warning InternalError failed to resolve sink: failed to resolve destination: failed to get object test-service-namespace/test-service: services \"test-service\" not found",
 			},
 			WantErr: true,
 		},
diff --git a/test/e2e_new/features/broker_deleted_recreated.go b/test/e2e_new/features/broker_deleted_recreated.go
index e54458c15..8d250f284 100644
--- a/test/e2e_new/features/broker_deleted_recreated.go
+++ b/test/e2e_new/features/broker_deleted_recreated.go
@@ -36,7 +36,7 @@ func BrokerDeletedRecreated() *feature.Feature {
 	triggerName := feature.MakeRandomK8sName("trigger")
 
 	f.Setup("test broker", featuressteps.BrokerSmokeTest(brokerName, triggerName))
-	f.Requirement("delete resources", featuressteps.DeleteResources(f))
+	f.Requirement("delete broker", featuressteps.DeleteBroker(brokerName))
 	f.Assert("test broker after deletion", featuressteps.BrokerSmokeTest(brokerName, triggerName))
 
 	return f
diff --git a/test/e2e_new/features/featuressteps/broker.go b/test/e2e_new/features/featuressteps/broker.go
index 6be502fc8..4ec8ecd2a 100644
--- a/test/e2e_new/features/featuressteps/broker.go
+++ b/test/e2e_new/features/featuressteps/broker.go
@@ -18,10 +18,14 @@ package featuressteps
 
 import (
 	"context"
+	"encoding/json"
 
 	cetest "github.com/cloudevents/sdk-go/v2/test"
 	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/utils/pointer"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	eventingclient "knative.dev/eventing/pkg/client/injection/client"
@@ -108,5 +112,27 @@ func DeleteBroker(name string) feature.StepFn {
 		if err != nil {
 			t.Fatalf("Failed to delete Broker %s/%s: %w\n", ns, name, err)
 		}
+
+		interval, timeout := environment.PollTimingsFromContext(ctx)
+
+		err = wait.PollImmediate(interval, timeout, func() (bool, error) {
+			br, err := eventingclient.Get(ctx).
+				EventingV1().
+				Brokers(ns).
+				Get(ctx, name, metav1.GetOptions{})
+			if apierrors.IsNotFound(err) {
+				return true, nil
+			}
+			if err != nil {
+				t.Logf("Failed to get broker %s: %w", name, err)
+				return false, err
+			}
+
+			b, _ := json.MarshalIndent(br, "", " ")
+			t.Logf("Broker %s still present\n%s\n", name, b)
+
+			return false, nil
+		})
+		require.Nil(t, err)
 	}
 }
diff --git a/test/upgrade/postupgrade.go b/test/upgrade/postupgrade.go
index f555ce771..04cf80a79 100644
--- a/test/upgrade/postupgrade.go
+++ b/test/upgrade/postupgrade.go
@@ -32,7 +32,7 @@ func BrokerPostUpgradeTest() pkgupgrade.Operation {
 
 // SinkPostUpgradeTest tests sink basic operations post upgrade.
 func SinkPostUpgradeTest() pkgupgrade.Operation {
-	return pkgupgrade.NewOperation("SinkPostDowngradeTest", func(c pkgupgrade.Context) {
+	return pkgupgrade.NewOperation("SinkPostUpgradeTest", func(c pkgupgrade.Context) {
 		e2e.RunTestKafkaSink(c.T, eventing.ModeBinary, nil)
 	})
 }
diff --git a/test/upgrade/preupgrade.go b/test/upgrade/preupgrade.go
index c46b23f41..1df23a95b 100644
--- a/test/upgrade/preupgrade.go
+++ b/test/upgrade/preupgrade.go
@@ -32,7 +32,7 @@ func BrokerPreUpgradeTest() pkgupgrade.Operation {
 
 // SinkPreUpgradeTest tests sink basic operations pre upgrade.
 func SinkPreUpgradeTest() pkgupgrade.Operation {
-	return pkgupgrade.NewOperation("SinkPostDowngradeTest", func(c pkgupgrade.Context) {
+	return pkgupgrade.NewOperation("SinkPreUpgradeTest", func(c pkgupgrade.Context) {
 		e2e.RunTestKafkaSink(c.T, eventing.ModeBinary, nil)
 	})
 }
diff --git a/vendor/knative.dev/eventing-kafka/test/e2e-tests.sh b/vendor/knative.dev/eventing-kafka/test/e2e-tests.sh
index c8daf4693..ffa8a1f9c 100644
--- a/vendor/knative.dev/eventing-kafka/test/e2e-tests.sh
+++ b/vendor/knative.dev/eventing-kafka/test/e2e-tests.sh
@@ -98,6 +98,12 @@ if [[ $TEST_CONSOLIDATED_CHANNEL_SASL == 1 ]]; then
   test_consolidated_channel_sasl || exit 1
 fi
 
+if ! command -v ps &> /dev/null && command -v apt-get &> /dev/null; then
+    # need to install ps
+    echo "Couldn't find ps command. Installing it. "
+    apt-get update && apt-get install -y procps
+fi
+
 # Terminate any zipkin port-forward processes that are still present on the system
 ps -e -o pid,command | grep 'kubectl port-forward zipkin[^*]*9411:9411 -n' | sed 's/^ *\([0-9][0-9]*\) .*/\1/' | xargs kill
 
diff --git a/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/inmemorychannel.go b/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/inmemorychannel.go
index 9f12b169a..81d802a9b 100644
--- a/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/inmemorychannel.go
+++ b/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/inmemorychannel.go
@@ -25,11 +25,12 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"knative.dev/pkg/apis"
 
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+
 	eventingv1 "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/eventing/pkg/apis/eventing"
 	"knative.dev/eventing/pkg/apis/messaging"
 	v1 "knative.dev/eventing/pkg/apis/messaging/v1"
-	duckv1 "knative.dev/pkg/apis/duck/v1"
 )
 
 // InMemoryChannelOption enables further configuration of a v1.InMemoryChannel.
@@ -219,7 +220,7 @@ func WithInMemoryChannelDLSResolvedFailed() InMemoryChannelOption {
 	return func(imc *v1.InMemoryChannel) {
 		imc.Status.MarkDeadLetterSinkResolvedFailed(
 			"Unable to get the DeadLetterSink's URI",
-			fmt.Sprintf(`services "%s" not found`,
+			fmt.Sprintf(`failed to get object test-namespace/test-dls: services "%s" not found`,
 				imc.Spec.Delivery.DeadLetterSink.Ref.Name,
 			),
 		)
diff --git a/vendor/knative.dev/hack/e2e-tests.sh b/vendor/knative.dev/hack/e2e-tests.sh
index 537b3867e..9178eac2c 100644
--- a/vendor/knative.dev/hack/e2e-tests.sh
+++ b/vendor/knative.dev/hack/e2e-tests.sh
@@ -73,9 +73,6 @@ function setup_test_cluster() {
   is_protected_cluster "${k8s_cluster}" && \
     abort "kubeconfig context set to ${k8s_cluster}, which is forbidden"
 
-  # Acquire cluster admin role for the current user.
-  acquire_cluster_admin_role "${k8s_cluster}"
-
   # Setup KO_DOCKER_REPO if it is a GKE cluster. Incorporate an element of
   # randomness to ensure that each run properly publishes images. Don't
   # owerwrite KO_DOCKER_REPO if already set.
diff --git a/vendor/knative.dev/hack/presubmit-tests.sh b/vendor/knative.dev/hack/presubmit-tests.sh
index 70a399c8e..08bd85d4a 100644
--- a/vendor/knative.dev/hack/presubmit-tests.sh
+++ b/vendor/knative.dev/hack/presubmit-tests.sh
@@ -266,8 +266,6 @@ function main() {
     git version
     echo ">> ko version"
     [[ -f /ko_version ]] && cat /ko_version || echo "unknown"
-    echo ">> bazel version"
-    [[ -f /bazel_version ]] && cat /bazel_version || echo "unknown"
     if [[ "${DOCKER_IN_DOCKER_ENABLED}" == "true" ]]; then
       echo ">> docker version"
       docker version
@@ -281,6 +279,8 @@ function main() {
       echo ">> maven version"
       mvn --version
     fi
+    echo ">> prow-tests image version"
+    [[ -f /commit_hash ]] && echo "Prow test image was built from $(cat /commit_hash) commit which is viewable at https://github.com/knative/test-infra/tree/$(cat /commit_hash) " || echo "unknown"
   fi
 
   [[ -z ${1:-} ]] && set -- "--all-tests"
diff --git a/vendor/knative.dev/pkg/test/upgrade/functions.go b/vendor/knative.dev/pkg/test/upgrade/functions.go
index 07cd80e92..98bf28340 100644
--- a/vendor/knative.dev/pkg/test/upgrade/functions.go
+++ b/vendor/knative.dev/pkg/test/upgrade/functions.go
@@ -17,19 +17,26 @@ limitations under the License.
 package upgrade
 
 import (
+	"bytes"
+	"strings"
 	"time"
 
+	"go.uber.org/zap/zapcore"
+
 	"go.uber.org/zap"
 )
 
 // Execute the Suite of upgrade tests with a Configuration given.
 func (s *Suite) Execute(c Configuration) {
-	l := c.logger()
+	l, err := c.logger(c.T)
+	if err != nil {
+		c.T.Fatal("Failed to build logger:", err)
+		return
+	}
 	se := suiteExecution{
 		suite:         enrichSuite(s),
 		configuration: c,
 		failed:        false,
-		logger:        l,
 	}
 	l.Info("🏃 Running upgrade test suite...")
 
@@ -93,10 +100,6 @@ func WaitForStopEvent(bc BackgroundContext, w WaitForStopEventConfiguration) {
 	}
 }
 
-func (c Configuration) logger() *zap.SugaredLogger {
-	return c.Log.Sugar()
-}
-
 // Name returns a friendly human readable text.
 func (s *StopEvent) Name() string {
 	return s.name
@@ -107,8 +110,19 @@ func handleStopEvent(
 	bc BackgroundContext,
 	wc WaitForStopEventConfiguration,
 ) {
-	bc.Log.Infof("%s have received a stop event: %s", wc.Name, se.Name())
-	defer close(se.Finished)
+	bc.Log.Debugf("%s have received a stop event: %s", wc.Name, se.Name())
+
+	logFn := se.logger.Info
+	logFn(wrapLog(bc.logBuffer.Dump()))
+
+	defer func() {
+		defer close(se.Finished)
+		if se.T.Failed() {
+			logFn = se.logger.Error
+		}
+		logFn(wrapLog(bc.logBuffer.Dump()))
+	}()
+
 	wc.OnStop(se)
 }
 
@@ -160,3 +174,58 @@ func (s *simpleBackgroundOperation) Setup() func(c Context) {
 func (s *simpleBackgroundOperation) Handler() func(bc BackgroundContext) {
 	return s.handler
 }
+
+func (b *threadSafeBuffer) Read(p []byte) (n int, err error) {
+	b.Mutex.Lock()
+	defer b.Mutex.Unlock()
+	return b.Buffer.Read(p)
+}
+
+func (b *threadSafeBuffer) Write(p []byte) (n int, err error) {
+	b.Mutex.Lock()
+	defer b.Mutex.Unlock()
+	return b.Buffer.Write(p)
+}
+
+// Dump returns the previous content of the buffer and creates a new
+// empty buffer for future writes.
+func (b *threadSafeBuffer) Dump() string {
+	b.Mutex.Lock()
+	defer b.Mutex.Unlock()
+	buf := b.Buffer.String()
+	b.Buffer = bytes.Buffer{}
+	return buf
+}
+
+// newInMemoryLoggerBuffer creates a logger that writes logs into a byte buffer.
+// This byte buffer is returned and can be used to process the logs at later stage.
+func newInMemoryLoggerBuffer(config Configuration) (*zap.Logger, *threadSafeBuffer) {
+	logConfig := config.logConfig()
+	buf := &threadSafeBuffer{}
+	core := zapcore.NewCore(
+		zapcore.NewConsoleEncoder(logConfig.Config.EncoderConfig),
+		zapcore.AddSync(buf),
+		logConfig.Config.Level)
+
+	var opts []zap.Option
+	if logConfig.Config.Development {
+		opts = append(opts, zap.Development())
+	}
+	if !logConfig.Config.DisableCaller {
+		opts = append(opts, zap.AddCaller())
+	}
+	stackLevel := zap.ErrorLevel
+	if logConfig.Config.Development {
+		stackLevel = zap.WarnLevel
+	}
+	if !logConfig.Config.DisableStacktrace {
+		opts = append(opts, zap.AddStacktrace(stackLevel))
+	}
+	return zap.New(core, opts...), buf
+}
+
+func wrapLog(log string) string {
+	s := strings.Join(strings.Split(log, "\n"), "\n⏳ ")
+	return "Rewind of background logs (prefix ⏳):\n" +
+		"⏳ " + strings.TrimSuffix(s, "⏳ ")
+}
diff --git a/vendor/knative.dev/pkg/test/upgrade/logger.go b/vendor/knative.dev/pkg/test/upgrade/logger.go
new file mode 100644
index 000000000..903544286
--- /dev/null
+++ b/vendor/knative.dev/pkg/test/upgrade/logger.go
@@ -0,0 +1,110 @@
+/*
+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 upgrade
+
+import (
+	"errors"
+	"fmt"
+	"net/url"
+	"strings"
+	"testing"
+
+	"go.uber.org/zap"
+)
+
+const testingTScheme = "testing-t"
+
+var (
+	// ErrInvalidTestingOutputMethod when given invalid testing output method.
+	ErrInvalidTestingOutputMethod = errors.New("invalid testing output method")
+	// ErrNoTestingTRegisteredForTest when no testing.TB registered for test.
+	ErrNoTestingTRegisteredForTest = errors.New("no testing.TB registered for test")
+
+	testingTeeBees = make(map[string]testing.TB)
+)
+
+func (c Configuration) logger(tb testing.TB) (*zap.SugaredLogger, error) {
+	// TODO(mgencur): Remove when dependent repositories use LogConfig instead of Log.
+	// This is for backwards compatibility.
+	if c.Log != nil {
+		return c.Log.Sugar(), nil
+	}
+
+	testName := tb.Name()
+	testingTeeBees[testName] = tb
+	cfg := c.logConfig().Config
+	outs := []string{fmt.Sprintf("%s://Log/%s", testingTScheme, testName)}
+	if isStandardOutputOnly(cfg.OutputPaths) {
+		cfg.OutputPaths = outs
+	} else {
+		cfg.OutputPaths = append(cfg.OutputPaths, outs...)
+	}
+	errOuts := []string{fmt.Sprintf("%s://Error/%s", testingTScheme, testName)}
+	if isStandardOutputOnly(cfg.ErrorOutputPaths) {
+		cfg.ErrorOutputPaths = errOuts
+	} else {
+		cfg.ErrorOutputPaths = append(cfg.ErrorOutputPaths, errOuts...)
+	}
+	err := zap.RegisterSink(testingTScheme, func(url *url.URL) (zap.Sink, error) {
+		var ok bool
+		testName = strings.TrimPrefix(url.Path, "/")
+		tb, ok = testingTeeBees[testName]
+		if !ok {
+			return nil, fmt.Errorf("%w: %s", ErrNoTestingTRegisteredForTest, testName)
+		}
+		switch url.Host {
+		case "Log":
+			return sink{writer: tb.Log}, nil
+		case "Error":
+			return sink{writer: tb.Error}, nil
+		}
+		return nil, fmt.Errorf("%w: %s", ErrInvalidTestingOutputMethod, url.Host)
+	})
+	if err != nil && !strings.HasPrefix(err.Error(), "sink factory already registered") {
+		return nil, err
+	}
+	var logger *zap.Logger
+	logger, err = cfg.Build(c.logConfig().Options...)
+	if err != nil {
+		return nil, err
+	}
+	return logger.Sugar(), nil
+}
+
+type sink struct {
+	writer func(args ...interface{})
+}
+
+func (t sink) Write(bytes []byte) (n int, err error) {
+	msg := string(bytes)
+	t.writer(msg)
+	return len(bytes), nil
+}
+
+func (t sink) Sync() error {
+	return nil
+}
+
+func (t sink) Close() error {
+	return nil
+}
+
+// isStandardOutputOnly checks the output paths and returns true, if they
+// contain only standard outputs.
+func isStandardOutputOnly(paths []string) bool {
+	return len(paths) == 1 && (paths[0] == "stderr" || paths[0] == "stdout")
+}
diff --git a/vendor/knative.dev/pkg/test/upgrade/private_types.go b/vendor/knative.dev/pkg/test/upgrade/private_types.go
index ab930e41b..d301b5d68 100644
--- a/vendor/knative.dev/pkg/test/upgrade/private_types.go
+++ b/vendor/knative.dev/pkg/test/upgrade/private_types.go
@@ -16,13 +16,15 @@ limitations under the License.
 
 package upgrade
 
-import "go.uber.org/zap"
+import (
+	"bytes"
+	"sync"
+)
 
 type suiteExecution struct {
 	suite         *enrichedSuite
 	configuration Configuration
 	failed        bool
-	logger        *zap.SugaredLogger
 }
 
 type enrichedSuite struct {
@@ -61,3 +63,10 @@ type simpleBackgroundOperation struct {
 	setup   func(c Context)
 	handler func(bc BackgroundContext)
 }
+
+// threadSafeBuffer avoids race conditions on bytes.Buffer.
+// See: https://stackoverflow.com/a/36226525/844449
+type threadSafeBuffer struct {
+	bytes.Buffer
+	sync.Mutex
+}
diff --git a/vendor/knative.dev/pkg/test/upgrade/steps.go b/vendor/knative.dev/pkg/test/upgrade/steps.go
index 1c08038ec..e4c270c0a 100644
--- a/vendor/knative.dev/pkg/test/upgrade/steps.go
+++ b/vendor/knative.dev/pkg/test/upgrade/steps.go
@@ -46,13 +46,16 @@ func (se *suiteExecution) preUpgradeTests(num int) {
 }
 
 func (se *suiteExecution) startContinualTests(num int) {
-	l := se.logger
 	operations := se.suite.tests.continual
 	groupTemplate := "%d) 🔄 Starting continual tests. " +
 		"%d tests are registered."
 	elementTemplate := `%d.%d) Starting continual tests of "%s".`
 	numOps := len(operations)
 	se.configuration.T.Run("ContinualTests", func(t *testing.T) {
+		l, err := se.configuration.logger(t)
+		if err != nil {
+			t.Fatal(err)
+		}
 		if numOps > 0 {
 			l.Infof(groupTemplate, num, numOps)
 			for i := range operations {
@@ -63,17 +66,28 @@ func (se *suiteExecution) startContinualTests(num int) {
 					return
 				}
 				setup := operation.Setup()
+
+				logger, buffer := newInMemoryLoggerBuffer(se.configuration)
 				t.Run("Setup"+operation.Name(), func(t *testing.T) {
-					setup(Context{T: t, Log: l})
+					l, err = se.configuration.logger(t)
+					if err != nil {
+						t.Fatal(err)
+					}
+					setup(Context{T: t, Log: logger.Sugar()})
 				})
+
 				handler := operation.Handler()
 				go func() {
-					bc := BackgroundContext{Log: l, Stop: operation.stop}
-					handler(bc)
+					handler(BackgroundContext{
+						Log:       logger.Sugar(),
+						Stop:      operation.stop,
+						logBuffer: buffer,
+					})
 				}()
-
 				se.failed = se.failed || t.Failed()
 				if se.failed {
+					// need to dump logs here, because verify will not be executed.
+					l.Error(wrapLog(buffer.Dump()))
 					return
 				}
 			}
@@ -85,18 +99,26 @@ func (se *suiteExecution) startContinualTests(num int) {
 }
 
 func (se *suiteExecution) verifyContinualTests(num int) {
-	l := se.logger
 	testsCount := len(se.suite.tests.continual)
 	if testsCount > 0 {
 		se.configuration.T.Run("VerifyContinualTests", func(t *testing.T) {
+			l, err := se.configuration.logger(t)
+			if err != nil {
+				t.Fatal(err)
+			}
 			l.Infof("%d) ✋ Verifying %d running continual tests.", num, testsCount)
 			for i, operation := range se.suite.tests.continual {
 				t.Run(operation.Name(), func(t *testing.T) {
+					l, err = se.configuration.logger(t)
+					if err != nil {
+						t.Fatal(err)
+					}
 					l.Infof(`%d.%d) Verifying "%s".`, num, i+1, operation.Name())
 					finished := make(chan struct{})
 					operation.stop <- StopEvent{
 						T:        t,
 						Finished: finished,
+						logger:   l,
 						name:     "Stop of " + operation.Name(),
 					}
 					<-finished
diff --git a/vendor/knative.dev/pkg/test/upgrade/suite_execution.go b/vendor/knative.dev/pkg/test/upgrade/suite_execution.go
index d926d4076..9f1783e68 100644
--- a/vendor/knative.dev/pkg/test/upgrade/suite_execution.go
+++ b/vendor/knative.dev/pkg/test/upgrade/suite_execution.go
@@ -21,8 +21,11 @@ import (
 )
 
 func (se *suiteExecution) processOperationGroup(op operationGroup) {
-	l := se.logger
 	se.configuration.T.Run(op.groupName, func(t *testing.T) {
+		l, err := se.configuration.logger(t)
+		if err != nil {
+			t.Fatal(err)
+		}
 		if len(op.operations) > 0 {
 			l.Infof(op.groupTemplate, op.num, len(op.operations))
 			for i, operation := range op.operations {
@@ -33,6 +36,10 @@ func (se *suiteExecution) processOperationGroup(op operationGroup) {
 				}
 				handler := operation.Handler()
 				t.Run(operation.Name(), func(t *testing.T) {
+					l, err = se.configuration.logger(t)
+					if err != nil {
+						t.Fatal(err)
+					}
 					handler(Context{T: t, Log: l})
 				})
 				se.failed = se.failed || t.Failed()
diff --git a/vendor/knative.dev/pkg/test/upgrade/types.go b/vendor/knative.dev/pkg/test/upgrade/types.go
index e4324f4a0..e46a516a4 100644
--- a/vendor/knative.dev/pkg/test/upgrade/types.go
+++ b/vendor/knative.dev/pkg/test/upgrade/types.go
@@ -84,10 +84,12 @@ type Context struct {
 // down to each handler of BackgroundOperation. It contains a StopEvent channel
 // which end user should use to obtain a testing.T for error reporting. Until
 // StopEvent is sent user may use zap.SugaredLogger to log state of execution if
-// necessary.
+// necessary. The logs are stored in a threadSafeBuffer and flushed to the test
+// output when the test fails.
 type BackgroundContext struct {
-	Log  *zap.SugaredLogger
-	Stop <-chan StopEvent
+	Log       *zap.SugaredLogger
+	Stop      <-chan StopEvent
+	logBuffer *threadSafeBuffer
 }
 
 // StopEvent represents an event that is to be received by background operation
@@ -98,6 +100,7 @@ type StopEvent struct {
 	T        *testing.T
 	Finished chan<- struct{}
 	name     string
+	logger   *zap.SugaredLogger
 }
 
 // WaitForStopEventConfiguration holds a values to be used be WaitForStopEvent
@@ -113,8 +116,28 @@ type WaitForStopEventConfiguration struct {
 
 // Configuration holds required and optional configuration to run upgrade tests.
 type Configuration struct {
-	T   *testing.T
+	T *testing.T
+	// TODO(mgencur): Remove when dependent repositories migrate to LogConfig.
+	// Keep this for backwards compatibility.
 	Log *zap.Logger
+	LogConfig
+}
+
+func (c Configuration) logConfig() LogConfig {
+	if len(c.LogConfig.Config.OutputPaths) == 0 {
+		c.LogConfig.Config = zap.NewDevelopmentConfig()
+	}
+	return c.LogConfig
+}
+
+// LogConfig holds the logger configuration. It allows for passing just the
+// logger configuration and also a custom function for building the resulting
+// logger.
+type LogConfig struct {
+	// Config from which the zap.Logger be created.
+	Config zap.Config
+	// Options holds options for the zap.Logger.
+	Options []zap.Option
 }
 
 // SuiteExecutor is to execute upgrade test suite.