Repository: knative-sandbox/eventing-kafka-broker
Pull request #2373 (2022-07-04 14:23:54)

Improve delete lifecycle with safeguards and exception catching
<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Bump Kafka clients to 3.2.0 to bring https://issues.apache.org/jira/browse/KAFKA-12257
- Add additional try-catch blocks for safety in case a future won't complete
  due to exceptions
- Add timeout to `reconcile` to avoid a stuck data plane reconciler

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- :gift: Add new feature
- :bug: Fix bug
- :broom: Update or clean up current behavior
- :wastebasket: Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
:page_facing_up: If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note

```

**Docs**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


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

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

```diff
@@             Coverage Diff              @@
##               main    #2373      +/-   ##
============================================
- Coverage     65.40%   59.71%   -5.69%     
============================================
  Files           140       68      -72     
  Lines          9310     6896    -2414     
  Branches        211        0     -211     
============================================
- Hits           6089     4118    -1971     
+ Misses         2808     2484     -324     
+ Partials        413      294     -119     
```

| 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/2373?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...ive/eventing/kafka/broker/core/AsyncCloseable.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2373/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvQXN5bmNDbG9zZWFibGUuamF2YQ==) | | |
| [.../core/reconciler/impl/ResourcesReconcilerImpl.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2373/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvcmVjb25jaWxlci9pbXBsL1Jlc291cmNlc1JlY29uY2lsZXJJbXBsLmphdmE=) | | |
| [...nciler/impl/ResourcesReconcilerMessageHandler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2373/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvcmVjb25jaWxlci9pbXBsL1Jlc291cmNlc1JlY29uY2lsZXJNZXNzYWdlSGFuZGxlci5qYXZh) | | |
| [.../dispatcher/impl/RecordDispatcherMutatorChain.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2373/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9SZWNvcmREaXNwYXRjaGVyTXV0YXRvckNoYWluLmphdmE=) | | |
| [...atcher/impl/consumer/KafkaConsumerRecordUtils.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2373/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9LYWZrYUNvbnN1bWVyUmVjb3JkVXRpbHMuamF2YQ==) | | |
| [...oker/dispatcher/main/ConsumerDeployerVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2373/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvbWFpbi9Db25zdW1lckRlcGxveWVyVmVydGljbGUuamF2YQ==) | | |
| [...r/dispatcher/main/ConsumerVerticleFactoryImpl.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2373/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvbWFpbi9Db25zdW1lclZlcnRpY2xlRmFjdG9yeUltcGwuamF2YQ==) | | |
| [...tive/eventing/kafka/broker/core/utils/BaseEnv.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2373/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdXRpbHMvQmFzZUVudi5qYXZh) | | |
| [...ting/kafka/broker/dispatcher/CloudEventSender.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2373/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvQ2xvdWRFdmVudFNlbmRlci5qYXZh) | | |
| [...atcher/impl/filter/subscriptionsapi/AnyFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2373/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvc3Vic2NyaXB0aW9uc2FwaS9BbnlGaWx0ZXIuamF2YQ==) | | |
| ... and [62 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2373/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/2373?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/2373?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [decd959...14562cf](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2373?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[bot]: @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
--- | --- | --- | --- | ---
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | 14562cf6a0397efdf9941c13a842e3bdd0e745b8 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2373/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1543963826754424832) | true | `/test channel-integration-tests-sasl-plain_eventing-kafka-broker_main`

[Your PR dashboard](https://prow.knative.dev/pr).

<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 -->
pierDipi: ```
Error from server (InternalError): error when creating "STDIN": Internal error occurred: failed calling webhook "webhook.eventing.knative.dev": failed to call webhook: Post "https://eventing-webhook.knative-eventing.svc:443/defaulting?timeout=10s": EOF
```
/retest-required
knative-prow-robot: @pierDipi: PR needs rebase.

<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.
</details>
knative-prow[bot]: New changes are detected. LGTM label has been removed.
knative-prow[bot]: @pierDipi: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
build-tests_eventing-kafka-broker_main | 3a5c3878383a0b49342d88500e7e6346bca68591 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2373/build-tests_eventing-kafka-broker_main/1544708166992793600) | true | `/test build-tests_eventing-kafka-broker_main`
unit-tests_eventing-kafka-broker_main | 958250fc2d10222a7c9983dbf5ff2f5aefe90563 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2373/unit-tests_eventing-kafka-broker_main/1544711615021060096) | true | `/test unit-tests_eventing-kafka-broker_main`

[Your PR dashboard](https://prow.knative.dev/pr).

<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[bot]: @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
--- | --- | --- | --- | ---
reconciler-tests_eventing-kafka-broker_main | 4f3b6c131e6389f4f679d052bbbc9a5edcaf30f5 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2373/reconciler-tests_eventing-kafka-broker_main/1544721522201989120) | true | `/test reconciler-tests_eventing-kafka-broker_main`

[Your PR dashboard](https://prow.knative.dev/pr).

<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/kafka-controller/main.go b/control-plane/cmd/kafka-controller/main.go
index 23ca7d6dc..c9f0d7fa6 100644
--- a/control-plane/cmd/kafka-controller/main.go
+++ b/control-plane/cmd/kafka-controller/main.go
@@ -112,7 +112,7 @@ func main() {
 		injection.NamedControllerConstructor{
 			Name: "consumer-controller",
 			ControllerConstructor: func(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
-				return consumer.NewController(ctx)
+				return consumer.NewController(ctx, watcher)
 			},
 		},
 	)
diff --git a/control-plane/cmd/kafka-source-controller/main.go b/control-plane/cmd/kafka-source-controller/main.go
index 8912d538f..52073cba9 100644
--- a/control-plane/cmd/kafka-source-controller/main.go
+++ b/control-plane/cmd/kafka-source-controller/main.go
@@ -59,7 +59,7 @@ func main() {
 		injection.NamedControllerConstructor{
 			Name: "consumer-controller",
 			ControllerConstructor: func(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
-				return consumer.NewController(ctx)
+				return consumer.NewController(ctx, watcher)
 			},
 		},
 	)
diff --git a/control-plane/pkg/reconciler/consumer/controller.go b/control-plane/pkg/reconciler/consumer/controller.go
index 97ac54f20..bd4fe5759 100644
--- a/control-plane/pkg/reconciler/consumer/controller.go
+++ b/control-plane/pkg/reconciler/consumer/controller.go
@@ -24,6 +24,7 @@ import (
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	podinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/pod"
 	secretinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/secret"
+	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/resolver"
 
@@ -39,7 +40,7 @@ type ControllerConfig struct {
 	DataPlaneConfigFormat string `required:"true" split_words:"true"`
 }
 
-func NewController(ctx context.Context) *controller.Impl {
+func NewController(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
 
 	controllerConfig := &ControllerConfig{}
 	if err := envconfig.Process("CONSUMER", controllerConfig); err != nil {
@@ -57,14 +58,13 @@ func NewController(ctx context.Context) *controller.Impl {
 		KafkaFeatureFlags:   config.DefaultFeaturesConfig(),
 	}
 
-	impl := creconciler.NewImpl(ctx, r, func(impl *controller.Impl) controller.Options {
-		return controller.Options{
-			ConfigStore: config.NewStore(ctx, func(name string, value *config.KafkaFeatureFlags) {
-				r.KafkaFeatureFlags.Reset(value)
-				impl.GlobalResync(consumerInformer.Informer())
-			}),
-		}
+	impl := creconciler.NewImpl(ctx, r)
+
+	configStore := config.NewStore(ctx, func(name string, value *config.KafkaFeatureFlags) {
+		r.KafkaFeatureFlags.Reset(value)
+		impl.GlobalResync(consumerInformer.Informer())
 	})
+	configStore.WatchConfigs(watcher)
 
 	r.Resolver = resolver.NewURIResolverFromTracker(ctx, impl.Tracker)
 
diff --git a/control-plane/pkg/reconciler/consumergroup/consumergroup.go b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
index a10bc8d14..b83600dcf 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
@@ -112,6 +112,16 @@ func (r Reconciler) ReconcileKind(ctx context.Context, cg *kafkainternals.Consum
 
 func (r Reconciler) FinalizeKind(ctx context.Context, cg *kafkainternals.ConsumerGroup) reconciler.Event {
 
+	cg.Spec.Replicas = pointer.Int32Ptr(0)
+	err := r.schedule(cg) //de-schedule placements
+
+	if err != nil {
+		cg.Status.Placements = nil
+
+		// return an error to 1. update the status. 2. not clear the finalizer
+		return errors.New("placement list was not empty")
+	}
+
 	// Get consumers associated with the ConsumerGroup.
 	existingConsumers, err := r.ConsumerLister.Consumers(cg.GetNamespace()).List(labels.SelectorFromSet(cg.Spec.Selector))
 	if err != nil {
diff --git a/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index 6385309cd..b0c786f9f 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -1,5 +1,5 @@
 
-Lists of 157 third-party dependencies.
+Lists of 162 third-party dependencies.
      (Eclipse Public License - v 1.0) (GNU Lesser General Public License) Logback Classic Module (ch.qos.logback:logback-classic:1.2.11 - http://logback.qos.ch/logback-classic)
      (Eclipse Public License - v 1.0) (GNU Lesser General Public License) Logback Core Module (ch.qos.logback:logback-core:1.2.11 - http://logback.qos.ch/logback-core)
      (The Apache Software License, Version 2.0) Jackson-annotations (com.fasterxml.jackson.core:jackson-annotations:2.13.2 - http://github.com/FasterXML/jackson)
@@ -10,7 +10,7 @@ Lists of 157 third-party dependencies.
      (The Apache Software License, Version 2.0) Jackson datatype: jdk8 (com.fasterxml.jackson.datatype:jackson-datatype-jdk8:2.13.2 - https://github.com/FasterXML/jackson-modules-java8/jackson-datatype-jdk8)
      (The Apache Software License, Version 2.0) Jackson datatype: JSR310 (com.fasterxml.jackson.datatype:jackson-datatype-jsr310:2.13.2 - https://github.com/FasterXML/jackson-modules-java8/jackson-datatype-jsr310)
      (The Apache Software License, Version 2.0) jackson-module-scala (com.fasterxml.jackson.module:jackson-module-scala_2.12:2.13.2 - http://wiki.fasterxml.com/JacksonModuleScala)
-     (BSD 2-Clause License) zstd-jni (com.github.luben:zstd-jni:1.4.9-1 - https://github.com/luben/zstd-jni)
+     (BSD 2-Clause License) zstd-jni (com.github.luben:zstd-jni:1.5.2-1 - https://github.com/luben/zstd-jni)
      (The Apache Software License, Version 2.0) Generex (com.github.mifmif:generex:1.0.2 - https://github.com/mifmif/Generex/tree/master)
      (The Apache Software License, Version 2.0) bucket4j-core (com.github.vladimir-bukhtoyarov:bucket4j-core:7.4.0 - http://github.com/vladimir-bukhtoyarov/bucket4j/bucket4j-core)
      (The Apache Software License, Version 2.0) FindBugs-jsr305 (com.google.code.findbugs:jsr305:3.0.2 - http://findbugs.sourceforge.net/)
@@ -28,7 +28,7 @@ Lists of 157 third-party dependencies.
      (Apache 2.0) OkHttp (com.squareup.okhttp3:okhttp:3.14.9 - https://github.com/square/okhttp/okhttp)
      (Apache 2.0) Okio (com.squareup.okio:okio:1.17.2 - https://github.com/square/okio/okio)
      (BSD) ParaNamer Core (com.thoughtworks.paranamer:paranamer:2.8 - https://github.com/paul-hammant/paranamer/paranamer)
-     (Apache 2.0 License) scala-logging (com.typesafe.scala-logging:scala-logging_2.12:3.9.2 - https://github.com/lightbend/scala-logging)
+     (Apache 2.0 License) scala-logging (com.typesafe.scala-logging:scala-logging_2.12:3.9.4 - https://github.com/lightbend/scala-logging)
      (Apache License 2.0) Metrics Core Library (com.yammer.metrics:metrics-core:2.2.0 - http://metrics.codahale.com/metrics-core/)
      (Apache License, Version 2.0) Apache Commons CLI (commons-cli:commons-cli:1.4 - http://commons.apache.org/proper/commons-cli/)
      (Unknown license) contract (dev.knative.eventing.kafka.broker:contract:1.0-SNAPSHOT - no url defined)
@@ -44,6 +44,7 @@ Lists of 157 third-party dependencies.
      (The Apache Software License, Version 2.0) cloudevents-sql (io.cloudevents:cloudevents-sql:2.3.0 - https://cloudevents.github.io/sdk-java/cloudevents-sql/)
      (Apache Software License 2.0) Debezium API (io.debezium:debezium-api:1.9.4.Final - https://debezium.io/debezium-parent/debezium-api)
      (Apache Software License 2.0) Debezium Core (io.debezium:debezium-core:1.9.4.Final - https://debezium.io/debezium-parent/debezium-core)
+     (Apache License 2.0) Metrics Core (io.dropwizard.metrics:metrics-core:4.1.12.1 - https://metrics.dropwizard.io/metrics-core)
      (Apache License, Version 2.0) Fabric8 :: Kubernetes :: Java Client (io.fabric8:kubernetes-client:5.12.2 - http://fabric8.io/kubernetes-client/)
      (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Admission Registration, Authentication and Authorization (io.fabric8:kubernetes-model-admissionregistration:5.12.2 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-admissionregistration/)
      (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: API Extensions (io.fabric8:kubernetes-model-apiextensions:5.12.2 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-apiextensions/)
@@ -83,8 +84,8 @@ Lists of 157 third-party dependencies.
      (Apache License, Version 2.0) Netty/Resolver/DNS (io.netty:netty-resolver-dns:4.1.74.Final - https://netty.io/netty-resolver-dns/)
      (The Apache Software License, Version 2.0) Netty/TomcatNative [OpenSSL - Classes] (io.netty:netty-tcnative-classes:2.0.48.Final - https://github.com/netty/netty-tcnative/netty-tcnative-classes/)
      (Apache License, Version 2.0) Netty/Transport (io.netty:netty-transport:4.1.74.Final - https://netty.io/netty-transport/)
-     (Apache License, Version 2.0) Netty/Transport/Native/Epoll (io.netty:netty-transport-native-epoll:4.1.50.Final - https://netty.io/netty-transport-native-epoll/)
-     (Apache License, Version 2.0) Netty/Transport/Native/Unix/Common (io.netty:netty-transport-native-unix-common:4.1.50.Final - https://netty.io/netty-transport-native-unix-common/)
+     (Apache License, Version 2.0) Netty/Transport/Native/Epoll (io.netty:netty-transport-native-epoll:4.1.63.Final - https://netty.io/netty-transport-native-epoll/)
+     (Apache License, Version 2.0) Netty/Transport/Native/Unix/Common (io.netty:netty-transport-native-unix-common:4.1.63.Final - https://netty.io/netty-transport-native-unix-common/)
      (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-api:1.13.0 - https://github.com/open-telemetry/opentelemetry-java)
      (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-context:1.13.0 - https://github.com/open-telemetry/opentelemetry-java)
      (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-exporter-logging:1.13.0 - https://github.com/open-telemetry/opentelemetry-java)
@@ -105,7 +106,7 @@ Lists of 157 third-party dependencies.
      (Eclipse Public License - v 1.0) (The Apache Software License, Version 2.0) vertx-auth-common (io.vertx:vertx-auth-common:4.2.5 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-auth/vertx-auth-common)
      (Eclipse Public License - v 2.0) (The Apache Software License, Version 2.0) Vert.x Core (io.vertx:vertx-core:4.2.5 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-core)
      (Eclipse Public License - v 1.0) (The Apache Software License, Version 2.0) Vert.x JUnit 5 support :: Core (io.vertx:vertx-junit5:4.2.5 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-junit5)
-     (Eclipse Public License - v 1.0) (The Apache Software License, Version 2.0) Vert.x Kafka Client (io.vertx:vertx-kafka-client:4.2.5 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-kafka-client)
+     (Eclipse Public License - v 1.0) (The Apache Software License, Version 2.0) Vert.x Kafka Client (io.vertx:vertx-kafka-client:4.3.0 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-kafka-client)
      (Eclipse Public License - v 1.0) (The Apache Software License, Version 2.0) Vert.x metrics implementation for Micrometer.io (io.vertx:vertx-micrometer-metrics:4.2.5 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-micrometer-metrics)
      (Eclipse Public License - v 1.0) (The Apache Software License, Version 2.0) Vert.x OpenTelemetry (io.vertx:vertx-opentelemetry:4.2.5 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-tracing-parent/vertx-opentelemetry)
      (Eclipse Public License - v 1.0) (The Apache Software License, Version 2.0) vertx-web-client (io.vertx:vertx-web-client:4.2.5 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-web-parent/vertx-web-client)
@@ -121,16 +122,20 @@ Lists of 157 third-party dependencies.
      (MIT) argparse4j (net.sourceforge.argparse4j:argparse4j:0.7.0 - http://argparse4j.github.io)
      (The BSD License) ANTLR 4 Runtime (org.antlr:antlr4-runtime:4.9.2 - http://www.antlr.org/antlr4-runtime)
      (The Apache Software License, Version 2.0) Commons Math (org.apache.commons:commons-math3:3.2 - http://commons.apache.org/proper/commons-math/)
-     (The Apache Software License, Version 2.0) Apache Kafka (org.apache.kafka:kafka-clients:2.8.1 - https://kafka.apache.org)
-     (The Apache Software License, Version 2.0) Apache Kafka (org.apache.kafka:kafka-metadata:2.8.1 - https://kafka.apache.org)
-     (The Apache Software License, Version 2.0) Apache Kafka (org.apache.kafka:kafka-raft:2.8.1 - https://kafka.apache.org)
-     (The Apache Software License, Version 2.0) Apache Kafka (org.apache.kafka:kafka_2.12:2.8.1 - https://kafka.apache.org)
+     (The Apache License, Version 2.0) Apache Kafka (org.apache.kafka:kafka-clients:3.2.0 - https://kafka.apache.org)
+     (The Apache License, Version 2.0) Apache Kafka (org.apache.kafka:kafka-metadata:3.2.0 - https://kafka.apache.org)
+     (The Apache License, Version 2.0) Apache Kafka (org.apache.kafka:kafka-raft:3.2.0 - https://kafka.apache.org)
+     (The Apache License, Version 2.0) Apache Kafka (org.apache.kafka:kafka-server-common:3.2.0 - https://kafka.apache.org)
+     (The Apache License, Version 2.0) Apache Kafka (org.apache.kafka:kafka-storage:3.2.0 - https://kafka.apache.org)
+     (The Apache License, Version 2.0) Apache Kafka (org.apache.kafka:kafka-storage-api:3.2.0 - https://kafka.apache.org)
+     (The Apache License, Version 2.0) Apache Kafka (org.apache.kafka:kafka_2.12:3.2.0 - https://kafka.apache.org)
      (Apache License, Version 2.0) Apache Yetus - Audience Annotations (org.apache.yetus:audience-annotations:0.5.0 - https://yetus.apache.org/audience-annotations)
-     (Apache License, Version 2.0) Apache ZooKeeper - Server (org.apache.zookeeper:zookeeper:3.5.9 - http://zookeeper.apache.org/zookeeper)
-     (Apache License, Version 2.0) Apache ZooKeeper - Jute (org.apache.zookeeper:zookeeper-jute:3.5.9 - http://zookeeper.apache.org/zookeeper-jute)
+     (Apache License, Version 2.0) Apache ZooKeeper - Server (org.apache.zookeeper:zookeeper:3.6.3 - http://zookeeper.apache.org/zookeeper)
+     (Apache License, Version 2.0) Apache ZooKeeper - Jute (org.apache.zookeeper:zookeeper-jute:3.6.3 - http://zookeeper.apache.org/zookeeper-jute)
      (The Apache License, Version 2.0) org.apiguardian:apiguardian-api (org.apiguardian:apiguardian-api:1.1.2 - https://github.com/apiguardian-team/apiguardian)
      (Apache License, Version 2.0) AssertJ fluent assertions (org.assertj:assertj-core:3.22.0 - https://assertj.github.io/doc/assertj-core/)
      (Apache 2.0) Awaitility (org.awaitility:awaitility:4.2.0 - http://awaitility.org)
+     (The Apache Software License, Version 2.0) jose4j (org.bitbucket.b_c:jose4j:0.7.9 - https://bitbucket.org/b_c/jose4j/)
      (GNU General Public License, version 2 (GPL2), with the classpath exception) (The MIT License) Checker Qual (org.checkerframework:checker-compat-qual:2.5.5 - https://checkerframework.org)
      (BSD Licence 3) Hamcrest (org.hamcrest:hamcrest:2.1 - http://hamcrest.org/JavaHamcrest/)
      (New BSD License) Hamcrest Core (org.hamcrest:hamcrest-core:1.3 - https://github.com/hamcrest/JavaHamcrest/hamcrest-core)
@@ -142,18 +147,18 @@ Lists of 157 third-party dependencies.
      (Eclipse Public License v2.0) JUnit Platform Commons (org.junit.platform:junit-platform-commons:1.8.2 - https://junit.org/junit5/)
      (Eclipse Public License v2.0) JUnit Platform Engine API (org.junit.platform:junit-platform-engine:1.8.2 - https://junit.org/junit5/)
      (Public Domain, per Creative Commons CC0) LatencyUtils (org.latencyutils:LatencyUtils:2.0.3 - http://latencyutils.github.io/LatencyUtils/)
-     (The Apache Software License, Version 2.0) LZ4 and xxHash (org.lz4:lz4-java:1.7.1 - https://github.com/lz4/lz4-java)
+     (The Apache Software License, Version 2.0) LZ4 and xxHash (org.lz4:lz4-java:1.8.0 - https://github.com/lz4/lz4-java)
      (The MIT License) mockito-core (org.mockito:mockito-core:4.6.0 - https://github.com/mockito/mockito)
      (The MIT License) mockito-junit-jupiter (org.mockito:mockito-junit-jupiter:4.6.0 - https://github.com/mockito/mockito)
      (Apache License, Version 2.0) Objenesis (org.objenesis:objenesis:3.2 - http://objenesis.org/objenesis)
      (GNU General Public License (GPL), version 2, with the Classpath exception) JMH Core (org.openjdk.jmh:jmh-core:1.35 - http://openjdk.java.net/projects/code-tools/jmh/jmh-core/)
      (GNU General Public License (GPL), version 2, with the Classpath exception) JMH Generators: Annotation Processors (org.openjdk.jmh:jmh-generator-annprocess:1.35 - http://openjdk.java.net/projects/code-tools/jmh/jmh-generator-annprocess/)
      (The Apache License, Version 2.0) org.opentest4j:opentest4j (org.opentest4j:opentest4j:1.2.0 - https://github.com/ota4j-team/opentest4j)
-     (Apache-2.0) Scala Library (org.scala-lang:scala-library:2.12.13 - https://www.scala-lang.org/)
-     (Apache-2.0) Scala Compiler (org.scala-lang:scala-reflect:2.12.13 - https://www.scala-lang.org/)
-     (Apache-2.0) scala-collection-compat (org.scala-lang.modules:scala-collection-compat_2.12:2.3.0 - http://www.scala-lang.org/)
-     (Apache-2.0) scala-java8-compat (org.scala-lang.modules:scala-java8-compat_2.12:0.9.1 - http://www.scala-lang.org/)
+     (Apache-2.0) Scala Library (org.scala-lang:scala-library:2.12.15 - https://www.scala-lang.org/)
+     (Apache-2.0) Scala Compiler (org.scala-lang:scala-reflect:2.12.15 - https://www.scala-lang.org/)
+     (Apache-2.0) scala-collection-compat (org.scala-lang.modules:scala-collection-compat_2.12:2.6.0 - http://www.scala-lang.org/)
+     (Apache-2.0) scala-java8-compat (org.scala-lang.modules:scala-java8-compat_2.12:1.0.2 - http://www.scala-lang.org/)
      (MIT License) SLF4J API Module (org.slf4j:slf4j-api:1.7.36 - http://www.slf4j.org)
      (MIT License) SLF4J NOP Binding (org.slf4j:slf4j-nop:1.7.36 - http://www.slf4j.org)
-     (Apache-2.0) snappy-java (org.xerial.snappy:snappy-java:1.1.8.1 - https://github.com/xerial/snappy-java)
+     (Apache-2.0) snappy-java (org.xerial.snappy:snappy-java:1.1.8.4 - https://github.com/xerial/snappy-java)
      (Apache License, Version 2.0) SnakeYAML (org.yaml:snakeyaml:1.30 - https://bitbucket.org/snakeyaml/snakeyaml)
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/AsyncCloseable.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/AsyncCloseable.java
index 80f5062a0..ffe1ab865 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/AsyncCloseable.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/AsyncCloseable.java
@@ -73,15 +73,19 @@ public interface AsyncCloseable extends Closeable {
       return Future.succeededFuture();
     }
 
-    return closeableIterator.
-      next().
-      close().
-      compose(
-        v -> compose(closeableIterator),
-        cause -> {
-          LoggerFactory.getLogger(AsyncCloseable.class).warn("Failed to close closeable", cause);
-          return compose(closeableIterator);
-        }
-      );
+    try {
+      return closeableIterator.
+        next().
+        close().
+        compose(
+          v -> compose(closeableIterator),
+          cause -> {
+            LoggerFactory.getLogger(AsyncCloseable.class).warn("Failed to close closeable", cause);
+            return compose(closeableIterator);
+          }
+        );
+    } catch (final Exception ex) {
+      return compose(closeableIterator);
+    }
   }
 }
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java
index 6201d67b5..3479c6197 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java
@@ -60,8 +60,8 @@ public class ResourcesReconcilerImpl implements ResourcesReconciler {
     this.ingressReconcilerListener = ingressReconcilerListener;
     this.egressReconcilerListener = egressReconcilerListener;
 
-    this.cachedResources = new ConcurrentHashMap<>(CACHED_RESOURCES_INITIAL_CAPACITY);
-    this.cachedEgresses = new ConcurrentHashMap<>(egressReconcilerListener != null ? CACHED_RESOURCES_INITIAL_CAPACITY : 0);
+    this.cachedResources = ingressReconcilerListener == null ? null : new ConcurrentHashMap<>(CACHED_RESOURCES_INITIAL_CAPACITY);
+    this.cachedEgresses = egressReconcilerListener == null ? null : new ConcurrentHashMap<>(CACHED_RESOURCES_INITIAL_CAPACITY);
   }
 
   @Override
@@ -97,7 +97,6 @@ public class ResourcesReconcilerImpl implements ResourcesReconciler {
       futures.add(
         this.egressReconcilerListener.onDeleteEgress(resource, egress)
           // If we succeed to delete the egress we can remove it from the cache.
-          .onSuccess(r -> egresses.remove(uid))
           .onSuccess(r -> this.cachedEgresses.remove(uid))
           .onFailure(cause -> logFailure("Failed to reconcile [onDeleteEgress] egress", egress, cause, generation))
       );
@@ -111,7 +110,6 @@ public class ResourcesReconcilerImpl implements ResourcesReconciler {
       futures.add(
         this.egressReconcilerListener.onNewEgress(resource, egress)
           // If we fail to create the egress we can't put it in the cache.
-          .onFailure(r -> egresses.remove(uid))
           .onSuccess(r -> this.cachedEgresses.put(egress.getUid(), new SimpleImmutableEntry<>(egress, resource)))
           .onFailure(cause -> logFailure("Failed to reconcile [onNewEgress] egress ", egress, cause, generation))
       );
@@ -122,9 +120,11 @@ public class ResourcesReconcilerImpl implements ResourcesReconciler {
       final var newEgress = entry.getKey();
       final var newResource = entry.getValue();
 
-      final var oldResource = this.cachedResources.get(newResource.getUid());
+      final var cachedEgress = this.cachedEgresses.get(uid);
+      final var oldResource = cachedEgress.getValue();
+      final var oldEgress = cachedEgress.getKey();
 
-      if (resourceEquals(newResource, oldResource) && egressEquals(newEgress, this.cachedEgresses.get(uid).getKey())) {
+      if (resourceEquals(newResource, oldResource) && egressEquals(newEgress, oldEgress)) {
         logger.info("Nothing changed for egress {} {} {} {}",
           keyValue("id", newEgress.getUid()),
           keyValue("consumerGroup", newEgress.getConsumerGroup()),
@@ -137,23 +137,13 @@ public class ResourcesReconcilerImpl implements ResourcesReconciler {
       futures.add(
         this.egressReconcilerListener.onUpdateEgress(newResource, newEgress)
           // If we fail to update the egress we can't put it in the cache.
-          .onFailure(r -> egresses.remove(uid))
           .onSuccess(r -> this.cachedEgresses.put(newEgress.getUid(), new SimpleImmutableEntry<>(newEgress, newResource)))
           .onFailure(cause -> logFailure("Failed to reconcile [onUpdateEgress] egress ", newEgress, cause, generation))
       );
     });
 
     // We want to complete the future, once all futures are complete, so use join.
-    return CompositeFuture.join(futures)
-      .onComplete(r -> {
-        this.cachedResources.clear();
-        egresses.values()
-          .forEach(entry -> {
-            final var resource = entry.getValue();
-            this.cachedResources.put(resource.getUid(), resource);
-          });
-      })
-      .mapEmpty();
+    return CompositeFuture.join(futures).mapEmpty();
   }
 
   private Future<Void> reconcileIngress(DataPlaneContract.Contract contract) {
@@ -161,7 +151,7 @@ public class ResourcesReconcilerImpl implements ResourcesReconciler {
     final var generation = contract.getGeneration();
 
     final Map<String, DataPlaneContract.Resource> newResourcesMap = new HashMap<>(
-     contract.getResourcesList()
+      contract.getResourcesList()
         .stream()
         .collect(Collectors.toMap(DataPlaneContract.Resource::getUid, Function.identity()))
     );
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerMessageHandler.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerMessageHandler.java
index cb45d7039..d12176bb5 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerMessageHandler.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerMessageHandler.java
@@ -18,6 +18,7 @@ package dev.knative.eventing.kafka.broker.core.reconciler.impl;
 import dev.knative.eventing.kafka.broker.contract.DataPlaneContract;
 import dev.knative.eventing.kafka.broker.core.reconciler.ResourcesReconciler;
 import io.vertx.core.Handler;
+import io.vertx.core.Promise;
 import io.vertx.core.Vertx;
 import io.vertx.core.eventbus.Message;
 import io.vertx.core.eventbus.MessageConsumer;
@@ -35,12 +36,15 @@ public class ResourcesReconcilerMessageHandler implements Handler<Message<Object
   private static final Logger logger = LoggerFactory.getLogger(ResourcesReconcilerMessageHandler.class);
 
   public final static String ADDRESS = "resourcesreconciler.core";
+  public static final int RECONCILE_TIMEOUT = 10000;
 
+  private final Vertx vertx;
   private final ResourcesReconciler resourcesReconciler;
   private final AtomicBoolean reconciling;
   private final AtomicReference<DataPlaneContract.Contract> last;
 
-  public ResourcesReconcilerMessageHandler(final ResourcesReconciler resourcesReconciler) {
+  public ResourcesReconcilerMessageHandler(final Vertx vertx, final ResourcesReconciler resourcesReconciler) {
+    this.vertx = vertx;
     this.resourcesReconciler = resourcesReconciler;
     reconciling = new AtomicBoolean();
     last = new AtomicReference<>();
@@ -68,7 +72,22 @@ public class ResourcesReconcilerMessageHandler implements Handler<Message<Object
 
       logger.info("Reconciling contract {}", keyValue("contractGeneration", contract.getGeneration()));
 
-      resourcesReconciler.reconcile(contract)
+      final Promise<Void> p = Promise.promise();
+
+      // This is a safety timeout to the `reconcile` phase, there have been multiple times when libraries or our
+      // components will cause the `Future` returned by `reconcile` to never complete (fail or succeed), in those
+      // cases we stop reconciling resources completely.
+      vertx.setTimer(RECONCILE_TIMEOUT, v -> p.tryFail(v + "ms timeout reached"));
+
+      try {
+        resourcesReconciler.reconcile(contract)
+          .onSuccess(p::tryComplete)
+          .onFailure(p::tryFail);
+      } catch (final Exception ex) {
+        p.tryFail(ex);
+      }
+
+      p.future()
         .onComplete(r -> {
           if (r.succeeded()) {
             logger.info(
@@ -96,6 +115,6 @@ public class ResourcesReconcilerMessageHandler implements Handler<Message<Object
   }
 
   public static MessageConsumer<Object> start(final Vertx vertx, final ResourcesReconciler reconciler) {
-    return vertx.eventBus().localConsumer(ADDRESS, new ResourcesReconcilerMessageHandler(reconciler));
+    return vertx.eventBus().localConsumer(ADDRESS, new ResourcesReconcilerMessageHandler(vertx, reconciler));
   }
 }
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherMutatorChain.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherMutatorChain.java
index cb97983ba..31687c53f 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherMutatorChain.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherMutatorChain.java
@@ -47,7 +47,6 @@ public class RecordDispatcherMutatorChain implements RecordDispatcher {
       record.offset(),
       record.timestamp(),
       record.timestampType(),
-      record.checksum(),
       record.serializedKeySize(),
       record.serializedValueSize(),
       record.key(),
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/KafkaConsumerRecordUtils.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/KafkaConsumerRecordUtils.java
index e0adfc092..b6639be82 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/KafkaConsumerRecordUtils.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/KafkaConsumerRecordUtils.java
@@ -31,7 +31,6 @@ public final class KafkaConsumerRecordUtils {
       record.offset(),
       record.timestamp(),
       record.timestampType(),
-      record.checksum(),
       record.serializedKeySize(),
       record.serializedValueSize(),
       record.key(),
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerDeployerVerticle.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerDeployerVerticle.java
index 9b1bfe500..c79ef2257 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerDeployerVerticle.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerDeployerVerticle.java
@@ -130,31 +130,35 @@ public final class ConsumerDeployerVerticle extends AbstractVerticle implements
       return Future.succeededFuture();
     }
 
-    return vertx.undeploy(this.deployedDispatchers.get(egress.getUid()))
-      .compose(
-        v -> {
-          this.deployedDispatchers.remove(egress.getUid());
-          logger.info(
-            "Removed egress {} {}",
-            keyValue("egress.uid", egress.getUid()),
-            keyValue("resource.uid", resource.getUid())
-          );
-          return Future.succeededFuture();
-        },
-        cause -> {
-          // IllegalStateException is thrown when a verticle is already un-deployed.
-          if (cause instanceof IllegalStateException) {
+    try {
+      return vertx.undeploy(this.deployedDispatchers.get(egress.getUid()))
+        .compose(
+          v -> {
             this.deployedDispatchers.remove(egress.getUid());
+            logger.info(
+              "Removed egress {} {}",
+              keyValue("egress.uid", egress.getUid()),
+              keyValue("resource.uid", resource.getUid())
+            );
             return Future.succeededFuture();
+          },
+          cause -> {
+            // IllegalStateException is thrown when a verticle is already un-deployed.
+            if (cause instanceof IllegalStateException) {
+              this.deployedDispatchers.remove(egress.getUid());
+              return Future.succeededFuture();
+            }
+            logger.error(
+              "Failed to un-deploy verticle {} {}",
+              keyValue("egress.uid", egress.getUid()),
+              keyValue("resource.uid", resource.getUid()),
+              cause
+            );
+            return Future.failedFuture(cause);
           }
-          logger.error(
-            "Failed to un-deploy verticle {} {}",
-            keyValue("egress.uid", egress.getUid()),
-            keyValue("resource.uid", resource.getUid()),
-            cause
-          );
-          return Future.failedFuture(cause);
-        }
-      );
+        );
+    } catch (final Exception ex) {
+      return Future.failedFuture(ex);
+    }
   }
 }
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index 2eaab343b..5ae45efc8 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -42,6 +42,7 @@
 
     <!-- dependencies version -->
     <vertx.version>4.2.5</vertx.version>
+    <vertx.kafka.clients.version>4.3.0</vertx.kafka.clients.version>
     <cloudevents.sdk.version>2.3.0</cloudevents.sdk.version>
     <micrometer.version>1.8.5</micrometer.version>
     <opentelemetry.version>1.13.0</opentelemetry.version>
@@ -56,7 +57,7 @@
     <junit.jupiter.version>5.8.2</junit.jupiter.version>
     <mokito.junit.jupiter.version>4.6.0</mokito.junit.jupiter.version>
     <fabric8.kubernetes.version>5.12.2</fabric8.kubernetes.version>
-    <kafka.version>2.8.1</kafka.version>
+    <kafka.version>3.2.0</kafka.version>
     <debezium.version>1.9.4.Final</debezium.version>
     <jib.version>3.2.0</jib.version>
   </properties>
@@ -122,12 +123,16 @@
       <dependency>
         <groupId>io.vertx</groupId>
         <artifactId>vertx-kafka-client</artifactId>
-        <version>${vertx.version}</version>
+        <version>${vertx.kafka.clients.version}</version>
         <exclusions>
           <exclusion>
             <groupId>org.apache.kafka</groupId>
             <artifactId>kafka-clients</artifactId>
           </exclusion>
+          <exclusion>
+            <groupId>io.vertx</groupId>
+            <artifactId>vertx-core</artifactId>
+          </exclusion>
         </exclusions>
       </dependency>
       <dependency>
diff --git a/go.mod b/go.mod
index 20fc0ff69..5d399c1da 100644
--- a/go.mod
+++ b/go.mod
@@ -31,8 +31,8 @@ require (
 	k8s.io/utils v0.0.0-20220210201930-3a6ce19ff2f9
 	knative.dev/eventing v0.32.1-0.20220622112123-0866e62ec48e
 	knative.dev/eventing-kafka v0.32.1-0.20220610014424-689d5055201c
-	knative.dev/hack v0.0.0-20220610014127-dc6c287516dc
-	knative.dev/pkg v0.0.0-20220621173822-9c5a7317fa9d
+	knative.dev/hack v0.0.0-20220701014203-65c463ac8c98
+	knative.dev/pkg v0.0.0-20220705130606-e60d250dc637
 	knative.dev/reconciler-test v0.0.0-20220622130523-57e43ca43bda
 	sigs.k8s.io/yaml v1.3.0
 )
diff --git a/go.sum b/go.sum
index 8cb7e1b41..66961f560 100644
--- a/go.sum
+++ b/go.sum
@@ -1397,14 +1397,16 @@ knative.dev/eventing v0.32.1-0.20220622112123-0866e62ec48e/go.mod h1:WKC5YSu8dhj
 knative.dev/eventing-kafka v0.32.1-0.20220610014424-689d5055201c h1:vLnymcpFVsRfR6YMZcZEiT3HRkAvZriec7PuxJxn+Qk=
 knative.dev/eventing-kafka v0.32.1-0.20220610014424-689d5055201c/go.mod h1:7Nzoitdd3WgLoJKXSifxylBk9+N85H0AVt0CF4wosFs=
 knative.dev/hack v0.0.0-20220524153203-12d3e2a7addc/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/hack v0.0.0-20220610014127-dc6c287516dc h1:LyqyT+rtgZYfOb3ChGE5jTFApCOcUmAcSV+TzgLxnys=
 knative.dev/hack v0.0.0-20220610014127-dc6c287516dc/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
+knative.dev/hack v0.0.0-20220701014203-65c463ac8c98 h1:kAwcKZOwYU0QCGEwvZrgnLrynFCWnXetwR/+a4W9R70=
+knative.dev/hack v0.0.0-20220701014203-65c463ac8c98/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack/schema v0.0.0-20220524153203-12d3e2a7addc/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
 knative.dev/hack/schema v0.0.0-20220610014127-dc6c287516dc/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
 knative.dev/pkg v0.0.0-20220524202603-19adf798efb8/go.mod h1:pApypeWDkGrsMkUDkV6StWXS4CXhwGWuJEID9GGZY0Y=
 knative.dev/pkg v0.0.0-20220610014025-7d607d643ee2/go.mod h1:pApypeWDkGrsMkUDkV6StWXS4CXhwGWuJEID9GGZY0Y=
-knative.dev/pkg v0.0.0-20220621173822-9c5a7317fa9d h1:H/naKtQODjCIr9wFb1YOG17CgbkgSFMy5dX0WZnrz6o=
 knative.dev/pkg v0.0.0-20220621173822-9c5a7317fa9d/go.mod h1:hmwyuQ6uQlqLVEM4lJGg6LZ12gbnUQv9kH88sA5pLDg=
+knative.dev/pkg v0.0.0-20220705130606-e60d250dc637 h1:wvz8Y/b2Ubox0pbTLAbKags19UC72Ik+b/tIW2QP9/E=
+knative.dev/pkg v0.0.0-20220705130606-e60d250dc637/go.mod h1:MxsV3ZGdrH87zv4O9HtxzbNwmiKcGwEP7UdriLINSn0=
 knative.dev/reconciler-test v0.0.0-20220524205904-f750f80bfc7e/go.mod h1:/ps2aEdmtjId+pUGJuuADQN4IucIp4rI7KnrYEahOgE=
 knative.dev/reconciler-test v0.0.0-20220610141926-64ffd15625ec/go.mod h1:/vZ6/AKGYQQPYY/7BSebqt62QrxiJ97T3r1jso9dAJg=
 knative.dev/reconciler-test v0.0.0-20220622130523-57e43ca43bda h1:wr6oFIvSNQq2dPOUIttwh+s0wJg5VRAP3vXxw0H38/U=
diff --git a/vendor/knative.dev/hack/release.sh b/vendor/knative.dev/hack/release.sh
index 915a13035..0448a7752 100644
--- a/vendor/knative.dev/hack/release.sh
+++ b/vendor/knative.dev/hack/release.sh
@@ -234,13 +234,15 @@ function prepare_dot_release() {
   # Use the original tag (ie. potentially with a knative- prefix) when determining the last version commit sha
   local github_tag="$(hub_tool release | grep "${last_version}")"
   local last_release_commit="$(git rev-list -n 1 "${github_tag}")"
+  local last_release_commit_filtered="$(git rev-list --invert-grep --grep "\[skip-dot-release\]" -n 1 "${github_tag}")"
   local release_branch_commit="$(git rev-list -n 1 upstream/"${RELEASE_BRANCH}")"
+  local release_branch_commit_filtered="$(git rev-list --invert-grep --grep "\[skip-dot-release\]" -n 1 upstream/"${RELEASE_BRANCH}")"
   [[ -n "${last_release_commit}" ]] || abort "cannot get last release commit"
   [[ -n "${release_branch_commit}" ]] || abort "cannot get release branch last commit"
-  echo "Version ${last_version} is at commit ${last_release_commit}"
-  echo "Branch ${RELEASE_BRANCH} is at commit ${release_branch_commit}"
-  if [[ "${last_release_commit}" == "${release_branch_commit}" ]]; then
-    echo "*** Branch ${RELEASE_BRANCH} has no new cherry-picks since release ${last_version}"
+  echo "Version ${last_version} is at commit ${last_release_commit}. Comparing using ${last_release_commit_filtered}. If it is different is because commits with the [skip-dot-release] flag in their commit body are not being considered."
+  echo "Branch ${RELEASE_BRANCH} is at commit ${release_branch_commit}. Comparing using ${release_branch_commit_filtered}. If it is different is because commits with the [skip-dot-release] flag in their commit body are not being considered."
+  if [[ "${last_release_commit_filtered}" == "${release_branch_commit_filtered}" ]]; then
+    echo "*** Branch ${RELEASE_BRANCH} has no new cherry-picks (ignoring commits with [skip-dot-release]) since release ${last_version}."
     echo "*** No dot release will be generated, as no changes exist"
     exit 0
   fi
diff --git a/vendor/knative.dev/pkg/apis/duck/v1/zz_generated.deepcopy.go b/vendor/knative.dev/pkg/apis/duck/v1/zz_generated.deepcopy.go
index dcde44940..96638e579 100644
--- a/vendor/knative.dev/pkg/apis/duck/v1/zz_generated.deepcopy.go
+++ b/vendor/knative.dev/pkg/apis/duck/v1/zz_generated.deepcopy.go
@@ -2,7 +2,7 @@
 // +build !ignore_autogenerated
 
 /*
-Copyright 2021 The Knative Authors
+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.
diff --git a/vendor/knative.dev/pkg/apis/duck/v1alpha1/zz_generated.deepcopy.go b/vendor/knative.dev/pkg/apis/duck/v1alpha1/zz_generated.deepcopy.go
index 838f15a17..c845923b1 100644
--- a/vendor/knative.dev/pkg/apis/duck/v1alpha1/zz_generated.deepcopy.go
+++ b/vendor/knative.dev/pkg/apis/duck/v1alpha1/zz_generated.deepcopy.go
@@ -2,7 +2,7 @@
 // +build !ignore_autogenerated
 
 /*
-Copyright 2021 The Knative Authors
+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.
diff --git a/vendor/knative.dev/pkg/apis/duck/v1beta1/zz_generated.deepcopy.go b/vendor/knative.dev/pkg/apis/duck/v1beta1/zz_generated.deepcopy.go
index 2e7d702c4..743081b01 100644
--- a/vendor/knative.dev/pkg/apis/duck/v1beta1/zz_generated.deepcopy.go
+++ b/vendor/knative.dev/pkg/apis/duck/v1beta1/zz_generated.deepcopy.go
@@ -2,7 +2,7 @@
 // +build !ignore_autogenerated
 
 /*
-Copyright 2021 The Knative Authors
+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.
diff --git a/vendor/knative.dev/pkg/apis/zz_generated.deepcopy.go b/vendor/knative.dev/pkg/apis/zz_generated.deepcopy.go
index 512ebea24..367d7941f 100644
--- a/vendor/knative.dev/pkg/apis/zz_generated.deepcopy.go
+++ b/vendor/knative.dev/pkg/apis/zz_generated.deepcopy.go
@@ -2,7 +2,7 @@
 // +build !ignore_autogenerated
 
 /*
-Copyright 2021 The Knative Authors
+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.
diff --git a/vendor/knative.dev/pkg/client/injection/apiextensions/client/client.go b/vendor/knative.dev/pkg/client/injection/apiextensions/client/client.go
index 9c8d95a7c..c44fdb2e3 100644
--- a/vendor/knative.dev/pkg/client/injection/apiextensions/client/client.go
+++ b/vendor/knative.dev/pkg/client/injection/apiextensions/client/client.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2021 The Knative Authors
+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.
diff --git a/vendor/knative.dev/pkg/client/injection/apiextensions/client/fake/fake.go b/vendor/knative.dev/pkg/client/injection/apiextensions/client/fake/fake.go
index 7de9b9aea..c48df7027 100644
--- a/vendor/knative.dev/pkg/client/injection/apiextensions/client/fake/fake.go
+++ b/vendor/knative.dev/pkg/client/injection/apiextensions/client/fake/fake.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2021 The Knative Authors
+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.
diff --git a/vendor/knative.dev/pkg/client/injection/ducks/duck/v1/addressable/addressable.go b/vendor/knative.dev/pkg/client/injection/ducks/duck/v1/addressable/addressable.go
index d40153e95..3bdfb44f0 100644
--- a/vendor/knative.dev/pkg/client/injection/ducks/duck/v1/addressable/addressable.go
+++ b/vendor/knative.dev/pkg/client/injection/ducks/duck/v1/addressable/addressable.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2021 The Knative Authors
+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.
diff --git a/vendor/knative.dev/pkg/client/injection/ducks/duck/v1/addressable/fake/fake.go b/vendor/knative.dev/pkg/client/injection/ducks/duck/v1/addressable/fake/fake.go
index 85225f977..e1aa3035d 100644
--- a/vendor/knative.dev/pkg/client/injection/ducks/duck/v1/addressable/fake/fake.go
+++ b/vendor/knative.dev/pkg/client/injection/ducks/duck/v1/addressable/fake/fake.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2021 The Knative Authors
+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.
diff --git a/vendor/knative.dev/pkg/client/injection/kube/client/client.go b/vendor/knative.dev/pkg/client/injection/kube/client/client.go
index 1fc2956ff..caf6cb793 100644
--- a/vendor/knative.dev/pkg/client/injection/kube/client/client.go
+++ b/vendor/knative.dev/pkg/client/injection/kube/client/client.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2021 The Knative Authors
+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.
diff --git a/vendor/knative.dev/pkg/client/injection/kube/client/fake/fake.go b/vendor/knative.dev/pkg/client/injection/kube/client/fake/fake.go
index 4360b0ec4..11d2f9fbf 100644
--- a/vendor/knative.dev/pkg/client/injection/kube/client/fake/fake.go
+++ b/vendor/knative.dev/pkg/client/injection/kube/client/fake/fake.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2021 The Knative Authors
+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.
diff --git a/vendor/knative.dev/pkg/client/injection/kube/informers/admissionregistration/v1/mutatingwebhookconfiguration/mutatingwebhookconfiguration.go b/vendor/knative.dev/pkg/client/injection/kube/informers/admissionregistration/v1/mutatingwebhookconfiguration/mutatingwebhookconfiguration.go
index c9ca217ff..9c395d26e 100644
--- a/vendor/knative.dev/pkg/client/injection/kube/informers/admissionregistration/v1/mutatingwebhookconfiguration/mutatingwebhookconfiguration.go
+++ b/vendor/knative.dev/pkg/client/injection/kube/informers/admissionregistration/v1/mutatingwebhookconfiguration/mutatingwebhookconfiguration.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2021 The Knative Authors
+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.
diff --git a/vendor/knative.dev/pkg/client/injection/kube/informers/admissionregistration/v1/validatingwebhookconfiguration/validatingwebhookconfiguration.go b/vendor/knative.dev/pkg/client/injection/kube/informers/admissionregistration/v1/validatingwebhookconfiguration/validatingwebhookconfiguration.go
index d89501a03..f4fee2dfd 100644
--- a/vendor/knative.dev/pkg/client/injection/kube/informers/admissionregistration/v1/validatingwebhookconfiguration/validatingwebhookconfiguration.go
+++ b/vendor/knative.dev/pkg/client/injection/kube/informers/admissionregistration/v1/validatingwebhookconfiguration/validatingwebhookconfiguration.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2021 The Knative Authors
+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.
diff --git a/vendor/knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset/fake/fake.go b/vendor/knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset/fake/fake.go
index 518d654ec..704ca3244 100644
--- a/vendor/knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset/fake/fake.go
+++ b/vendor/knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset/fake/fake.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2021 The Knative Authors
+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.
diff --git a/vendor/knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset/statefulset.go b/vendor/knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset/statefulset.go
index 76806dec7..2dd56a4dc 100644
--- a/vendor/knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset/statefulset.go
+++ b/vendor/knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset/statefulset.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2021 The Knative Authors
+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.
diff --git a/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/configmap.go b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/configmap.go
index e8d1516ca..1dafbeb08 100644
--- a/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/configmap.go
+++ b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/configmap.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2021 The Knative Authors
+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.
diff --git a/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/fake/fake.go b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/fake/fake.go
index 475aa4f58..22f414108 100644
--- a/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/fake/fake.go
+++ b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/fake/fake.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2021 The Knative Authors
+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.
diff --git a/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/node/fake/fake.go b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/node/fake/fake.go
index 627d83d7c..e042efaa2 100644
--- a/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/node/fake/fake.go
+++ b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/node/fake/fake.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2021 The Knative Authors
+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.
diff --git a/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/node/node.go b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/node/node.go
index b4712a000..5fc46d776 100644
--- a/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/node/node.go
+++ b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/node/node.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2021 The Knative Authors
+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.
diff --git a/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake/fake.go b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake/fake.go
index db12d56a0..c1d417b42 100644
--- a/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake/fake.go
+++ b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake/fake.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2021 The Knative Authors
+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.
diff --git a/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/pod/pod.go b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/pod/pod.go
index 79776b898..bfdb12f81 100644
--- a/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/pod/pod.go
+++ b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/pod/pod.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2021 The Knative Authors
+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.
diff --git a/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/secret/fake/fake.go b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/secret/fake/fake.go
index 1282f1a23..b876df855 100644
--- a/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/secret/fake/fake.go
+++ b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/secret/fake/fake.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2021 The Knative Authors
+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.
diff --git a/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/secret/secret.go b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/secret/secret.go
index 715fbac5f..27d22b702 100644
--- a/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/secret/secret.go
+++ b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/secret/secret.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2021 The Knative Authors
+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.
diff --git a/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/service/fake/fake.go b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/service/fake/fake.go
index 27974c092..65ffd9138 100644
--- a/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/service/fake/fake.go
+++ b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/service/fake/fake.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2021 The Knative Authors
+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.
diff --git a/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/service/service.go b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/service/service.go
index 2ae3fab57..cc379dade 100644
--- a/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/service/service.go
+++ b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/service/service.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2021 The Knative Authors
+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.
diff --git a/vendor/knative.dev/pkg/client/injection/kube/informers/factory/factory.go b/vendor/knative.dev/pkg/client/injection/kube/informers/factory/factory.go
index c1c136755..8bcb84dcc 100644
--- a/vendor/knative.dev/pkg/client/injection/kube/informers/factory/factory.go
+++ b/vendor/knative.dev/pkg/client/injection/kube/informers/factory/factory.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2021 The Knative Authors
+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.
diff --git a/vendor/knative.dev/pkg/client/injection/kube/informers/factory/fake/fake.go b/vendor/knative.dev/pkg/client/injection/kube/informers/factory/fake/fake.go
index 2c6161513..8a1c1b712 100644
--- a/vendor/knative.dev/pkg/client/injection/kube/informers/factory/fake/fake.go
+++ b/vendor/knative.dev/pkg/client/injection/kube/informers/factory/fake/fake.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2021 The Knative Authors
+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.
diff --git a/vendor/knative.dev/pkg/logging/zz_generated.deepcopy.go b/vendor/knative.dev/pkg/logging/zz_generated.deepcopy.go
index ac7d5e152..a752abd9b 100644
--- a/vendor/knative.dev/pkg/logging/zz_generated.deepcopy.go
+++ b/vendor/knative.dev/pkg/logging/zz_generated.deepcopy.go
@@ -2,7 +2,7 @@
 // +build !ignore_autogenerated
 
 /*
-Copyright 2021 The Knative Authors
+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.
diff --git a/vendor/knative.dev/pkg/metrics/zz_generated.deepcopy.go b/vendor/knative.dev/pkg/metrics/zz_generated.deepcopy.go
index e0108605b..1d90e3a9d 100644
--- a/vendor/knative.dev/pkg/metrics/zz_generated.deepcopy.go
+++ b/vendor/knative.dev/pkg/metrics/zz_generated.deepcopy.go
@@ -2,7 +2,7 @@
 // +build !ignore_autogenerated
 
 /*
-Copyright 2021 The Knative Authors
+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.
diff --git a/vendor/knative.dev/pkg/tracker/zz_generated.deepcopy.go b/vendor/knative.dev/pkg/tracker/zz_generated.deepcopy.go
index 8e2f56b47..814f39909 100644
--- a/vendor/knative.dev/pkg/tracker/zz_generated.deepcopy.go
+++ b/vendor/knative.dev/pkg/tracker/zz_generated.deepcopy.go
@@ -2,7 +2,7 @@
 // +build !ignore_autogenerated
 
 /*
-Copyright 2021 The Knative Authors
+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.
diff --git a/vendor/knative.dev/pkg/webhook/admission.go b/vendor/knative.dev/pkg/webhook/admission.go
index ff14585f7..ea5518260 100644
--- a/vendor/knative.dev/pkg/webhook/admission.go
+++ b/vendor/knative.dev/pkg/webhook/admission.go
@@ -102,7 +102,8 @@ func admissionHandler(rootLogger *zap.SugaredLogger, stats StatsReporter, c Admi
 			logkey.Operation, string(review.Request.Operation),
 			logkey.Resource, review.Request.Resource.String(),
 			logkey.SubResource, review.Request.SubResource,
-			logkey.UserInfo, fmt.Sprint(review.Request.UserInfo))
+			logkey.UserInfo, review.Request.UserInfo.Username,
+		)
 
 		ctx := logging.WithLogger(r.Context(), logger)
 		ctx = apis.WithHTTPRequest(ctx, r)
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 00bc6e53d..0efe56b8a 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1371,11 +1371,11 @@ knative.dev/eventing-kafka/test/rekt/resources/resetoffset
 knative.dev/eventing-kafka/test/upgrade
 knative.dev/eventing-kafka/test/upgrade/continual
 knative.dev/eventing-kafka/test/upgrade/installation
-# knative.dev/hack v0.0.0-20220610014127-dc6c287516dc
+# knative.dev/hack v0.0.0-20220701014203-65c463ac8c98
 ## explicit
 knative.dev/hack
 knative.dev/hack/shell
-# knative.dev/pkg v0.0.0-20220621173822-9c5a7317fa9d
+# knative.dev/pkg v0.0.0-20220705130606-e60d250dc637
 ## explicit
 knative.dev/pkg/apiextensions/storageversion
 knative.dev/pkg/apiextensions/storageversion/cmd/migrate
@@ -1415,6 +1415,7 @@ knative.dev/pkg/codegen/cmd/injection-gen/generators
 knative.dev/pkg/configmap
 knative.dev/pkg/configmap/hash-gen
 knative.dev/pkg/configmap/informer
+knative.dev/pkg/configmap/testing
 knative.dev/pkg/controller
 knative.dev/pkg/environment
 knative.dev/pkg/hack


Test patch:
diff --git a/control-plane/config/eventing-kafka-source/200-controller/100-config-kafka-feautures.yaml b/control-plane/config/eventing-kafka-source/200-controller/100-config-kafka-features.yaml
similarity index 100%
rename from control-plane/config/eventing-kafka-source/200-controller/100-config-kafka-feautures.yaml
rename to control-plane/config/eventing-kafka-source/200-controller/100-config-kafka-features.yaml
diff --git a/control-plane/pkg/apis/config/features_test.go b/control-plane/pkg/apis/config/features_test.go
new file mode 100644
index 000000000..05211ad6b
--- /dev/null
+++ b/control-plane/pkg/apis/config/features_test.go
@@ -0,0 +1,69 @@
+/*
+ * 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 config
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"knative.dev/eventing/pkg/apis/feature"
+	cm "knative.dev/pkg/configmap/testing"
+	_ "knative.dev/pkg/system/testing"
+)
+
+func TestFlags_IsDispatcherRateLimiterEnabled(t *testing.T) {
+	nc := DefaultFeaturesConfig()
+	require.False(t, nc.features.DispatcherRateLimiter == feature.Enabled)
+}
+
+func TestFlags_IsEnabled_ContainingFlag(t *testing.T) {
+	nc := DefaultFeaturesConfig()
+	nc.Reset(&KafkaFeatureFlags{
+		features: features{
+			DispatcherRateLimiter: feature.Enabled,
+		},
+	})
+	require.True(t, nc.features.DispatcherRateLimiter == feature.Enabled)
+}
+
+func TestGetFlags(t *testing.T) {
+	_, example := cm.ConfigMapsFromTestFile(t, FlagsConfigName)
+	flags, err := newFeaturesConfigFromMap(example)
+	require.NoError(t, err)
+
+	require.True(t, flags.IsDispatcherRateLimiterEnabled())
+
+}
+
+func TestStoreLoadWithConfigMap(t *testing.T) {
+	store := NewStore(context.Background())
+
+	_, exampleConfig := cm.ConfigMapsFromTestFile(t, FlagsConfigName)
+	store.OnConfigChanged(exampleConfig)
+
+	have := FromContext(store.ToContext(context.Background()))
+	expected, _ := newFeaturesConfigFromMap(exampleConfig)
+
+	require.Equal(t, expected.IsDispatcherRateLimiterEnabled(), have.IsDispatcherRateLimiterEnabled())
+}
+
+func TestStoreLoadWithContext(t *testing.T) {
+	have := FromContext(context.Background())
+
+	require.False(t, have.IsDispatcherRateLimiterEnabled())
+}
diff --git a/control-plane/pkg/apis/config/testdata/config-kafka-features.yaml b/control-plane/pkg/apis/config/testdata/config-kafka-features.yaml
new file mode 100644
index 000000000..ffa5d183f
--- /dev/null
+++ b/control-plane/pkg/apis/config/testdata/config-kafka-features.yaml
@@ -0,0 +1,8 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: config-kafka-features
+  namespace: knative-eventing
+data:
+  _example: |
+    dispatcher.rate-limiter: "enabled"
diff --git a/control-plane/pkg/reconciler/consumer/controller_test.go b/control-plane/pkg/reconciler/consumer/controller_test.go
index 5e71da0ca..039f7a920 100644
--- a/control-plane/pkg/reconciler/consumer/controller_test.go
+++ b/control-plane/pkg/reconciler/consumer/controller_test.go
@@ -20,7 +20,10 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"knative.dev/pkg/configmap"
 	reconcilertesting "knative.dev/pkg/reconciler/testing"
 
 	fakekubeclientset "k8s.io/client-go/kubernetes/fake"
@@ -49,7 +52,11 @@ func TestNewController(t *testing.T) {
 
 	t.Setenv("CONSUMER_DATA_PLANE_CONFIG_FORMAT", "json")
 
-	controller := NewController(ctx)
+	controller := NewController(ctx, configmap.NewStaticWatcher(&corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "config-kafka-features",
+		},
+	}))
 	if controller == nil {
 		t.Error("failed to create controller: <nil>")
 	}
diff --git a/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go b/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
index 6b4cbda6c..f0ed21d31 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
@@ -1438,6 +1438,8 @@ func TestFinalizeKind(t *testing.T) {
 			Objects: []runtime.Object{
 				NewSASLSSLSecret(ConsumerGroupNamespace, SecretName),
 				NewDeletedConsumeGroup(
+					ConsumerGroupReplicas(2),
+					ConsumerGroupOwnerRef(SourceAsOwnerReference()),
 					ConsumerGroupConsumerSpec(NewConsumerSpec(
 						ConsumerConfigs(
 							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
@@ -1480,7 +1482,12 @@ func TestFinalizeKind(t *testing.T) {
 					)),
 				),
 			},
-			Key:                     testKey,
+			Key: testKey,
+			OtherTestData: map[string]interface{}{
+				testSchedulerKey: SchedulerFunc(func(vpod scheduler.VPod) ([]eventingduckv1alpha1.Placement, error) {
+					return nil, nil
+				}),
+			},
 			SkipNamespaceValidation: true, // WantCreates compare the source namespace with configmap namespace, so skip it
 		},
 		{
@@ -1500,6 +1507,8 @@ func TestFinalizeKind(t *testing.T) {
 					)),
 				),
 				NewDeletedConsumeGroup(
+					ConsumerGroupReplicas(1),
+					ConsumerGroupOwnerRef(SourceAsOwnerReference()),
 					ConsumerGroupConsumerSpec(NewConsumerSpec(
 						ConsumerConfigs(
 							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
@@ -1543,6 +1552,57 @@ func TestFinalizeKind(t *testing.T) {
 				),
 			},
 			Key: testKey,
+			OtherTestData: map[string]interface{}{
+				testSchedulerKey: SchedulerFunc(func(vpod scheduler.VPod) ([]eventingduckv1alpha1.Placement, error) {
+					return nil, nil
+				}),
+			},
+			WantDeletes: []clientgotesting.DeleteActionImpl{
+				{
+					ActionImpl: clientgotesting.ActionImpl{
+						Namespace: ConsumerGroupNamespace,
+						Resource: schema.GroupVersionResource{
+							Group:    kafkainternals.SchemeGroupVersion.Group,
+							Version:  kafkainternals.SchemeGroupVersion.Version,
+							Resource: "consumers",
+						},
+					},
+					Name: fmt.Sprintf("%s-%d", ConsumerNamePrefix, 1),
+				},
+			},
+			SkipNamespaceValidation: true, // WantCreates compare the source namespace with configmap namespace, so skip it
+		},
+		{
+			Name: "Finalize normal - with consumers and existing placements",
+			Objects: []runtime.Object{
+				NewConsumer(1,
+					ConsumerSpec(NewConsumerSpec(
+						ConsumerTopics("t1", "t2"),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+						ConsumerVReplicas(1),
+						ConsumerPlacement(kafkainternals.PodBind{PodName: "p1", PodNamespace: systemNamespace}),
+						ConsumerSubscriber(NewSourceSinkReference()),
+					)),
+				),
+				NewDeletedConsumeGroup(
+					ConsumerGroupOwnerRef(SourceAsOwnerReference()),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+					)),
+				),
+			},
+			Key: testKey,
+			OtherTestData: map[string]interface{}{
+				testSchedulerKey: SchedulerFunc(func(vpod scheduler.VPod) ([]eventingduckv1alpha1.Placement, error) {
+					return nil, nil
+				}),
+			},
 			WantDeletes: []clientgotesting.DeleteActionImpl{
 				{
 					ActionImpl: clientgotesting.ActionImpl{
@@ -1564,6 +1624,9 @@ func TestFinalizeKind(t *testing.T) {
 	table.Test(t, NewFactory(nil, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
 
 		r := &Reconciler{
+			SchedulerFunc: func(s string) scheduler.Scheduler {
+				return row.OtherTestData[testSchedulerKey].(scheduler.Scheduler)
+			},
 			ConsumerLister:  listers.GetConsumerLister(),
 			InternalsClient: fakekafkainternalsclient.Get(ctx).InternalV1alpha1(),
 			SecretLister:    listers.GetSecretLister(),
diff --git a/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerMessageHandlerTest.java b/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerMessageHandlerTest.java
index 27aef4526..9f1506ab6 100644
--- a/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerMessageHandlerTest.java
+++ b/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerMessageHandlerTest.java
@@ -60,14 +60,14 @@ public class ResourcesReconcilerMessageHandlerTest {
 
   @SuppressWarnings("unchecked")
   @Test
-  public void shouldReconcileOnConcurrentEnqueue() {
+  public void shouldReconcileOnConcurrentEnqueue(final Vertx vertx) {
 
     final var reconcileCallsCounter = new AtomicInteger();
 
     final var duringReconcileLatch = new CountDownLatch(1);
     final var newContractSetLatch = new CountDownLatch(1);
 
-    final var handler = new ResourcesReconcilerMessageHandler(resources -> {
+    final var handler = new ResourcesReconcilerMessageHandler(vertx, resources -> {
       reconcileCallsCounter.incrementAndGet();
       try {
         duringReconcileLatch.countDown();
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/InvalidCloudEventInterceptorTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/InvalidCloudEventInterceptorTest.java
index a77c95a57..dd542ec7e 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/InvalidCloudEventInterceptorTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/InvalidCloudEventInterceptorTest.java
@@ -206,7 +206,6 @@ public class InvalidCloudEventInterceptorTest {
           i.offset(),
           i.timestamp(),
           i.timestampType(),
-          i.checksum(),
           i.serializedKeySize(),
           i.serializedValueSize(),
           i.key(),
@@ -294,7 +293,7 @@ public class InvalidCloudEventInterceptorTest {
       ),
       new TopicPartition("t1", 1),
       List.of(new ConsumerRecord<>(
-        "t1", 1, 0, 1L, TimestampType.CREATE_TIME, 2L, 1, 2, "a",
+        "t1", 1, 0, 1L, TimestampType.CREATE_TIME, null, 1, 2, "a",
         new InvalidCloudEvent(new byte[]{1}), new RecordHeaders().add(new RecordHeader("hello", "world".getBytes()))
       ))
     ));
diff --git a/vendor/knative.dev/pkg/configmap/testing/configmap.go b/vendor/knative.dev/pkg/configmap/testing/configmap.go
new file mode 100644
index 000000000..909036892
--- /dev/null
+++ b/vendor/knative.dev/pkg/configmap/testing/configmap.go
@@ -0,0 +1,109 @@
+/*
+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 testing
+
+import (
+	"fmt"
+	"io/ioutil"
+	"strings"
+	"testing"
+	"unicode"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"knative.dev/pkg/configmap"
+	"sigs.k8s.io/yaml"
+)
+
+// ConfigMapFromTestFile creates a v1.ConfigMap from a YAML file
+// It loads the YAML file from the testdata folder.
+func ConfigMapFromTestFile(t testing.TB, name string, allowed ...string) *corev1.ConfigMap {
+	t.Helper()
+
+	cm, _ := ConfigMapsFromTestFile(t, name, allowed...)
+	return cm
+}
+
+// ConfigMapsFromTestFile creates two corev1.ConfigMap resources from the config
+// file read from the testdata directory:
+// 1. The raw configmap read in.
+// 2. A second version of the configmap augmenting `data:` with what's parsed from the value of `_example:`
+func ConfigMapsFromTestFile(t testing.TB, name string, allowed ...string) (*corev1.ConfigMap, *corev1.ConfigMap) {
+	t.Helper()
+
+	b, err := ioutil.ReadFile(fmt.Sprintf("testdata/%s.yaml", name))
+	if err != nil {
+		t.Fatal("ReadFile() =", err)
+	}
+
+	var orig corev1.ConfigMap
+
+	// Use sigs.k8s.io/yaml since it reads json struct
+	// tags so things unmarshal properly.
+	if err := yaml.Unmarshal(b, &orig); err != nil {
+		t.Fatal("yaml.Unmarshal() =", err)
+	}
+
+	// We expect each of the allowed keys, and a key holding an example
+	// configuration for us to validate.
+	allowed = append(allowed, configmap.ExampleKey)
+
+	if len(orig.Data) != len(allowed) {
+		// See here for why we only check in empty ConfigMaps:
+		// https://github.com/knative/serving/issues/2668
+		t.Errorf("Data = %v, wanted %v", orig.Data, allowed)
+	}
+	allow := sets.NewString(allowed...)
+	for key := range orig.Data {
+		if !allow.Has(key) {
+			t.Errorf("Encountered key %q in %q that wasn't on the allowed list", key, name)
+		}
+	}
+	// With the length and membership checks, we know that the keyspace matches.
+
+	exampleBody, hasExampleBody := orig.Data[configmap.ExampleKey]
+	// Check that exampleBody does not have lines that end in a trailing space.
+	for i, line := range strings.Split(exampleBody, "\n") {
+		if strings.TrimRightFunc(line, unicode.IsSpace) != line {
+			t.Errorf("line %d of %q example contains trailing spaces", i, name)
+		}
+	}
+
+	// Check that the hashed exampleBody matches the assigned annotation, if present.
+	gotChecksum, hasExampleChecksumAnnotation := orig.Annotations[configmap.ExampleChecksumAnnotation]
+	if hasExampleBody && hasExampleChecksumAnnotation {
+		wantChecksum := configmap.Checksum(exampleBody)
+		if gotChecksum != wantChecksum {
+			t.Errorf("example checksum annotation = %s, want %s (you may need to re-run ./hack/update-codegen.sh)", gotChecksum, wantChecksum)
+		}
+	}
+
+	// Parse exampleBody into exemplar.Data.
+	exemplar := orig.DeepCopy()
+	if err := yaml.Unmarshal([]byte(exampleBody), &exemplar.Data); err != nil {
+		t.Fatal("yaml.Unmarshal() =", err)
+	}
+	// Augment the sample with actual configuration.
+	for k, v := range orig.Data {
+		if _, ok := exemplar.Data[k]; ok {
+			continue
+		}
+		exemplar.Data[k] = v
+	}
+
+	return &orig, exemplar
+}