Repository: knative-sandbox/eventing-kafka-broker
Pull request #1795 (2022-01-24 20:03:04)

[data-plane]: Bump mockito-junit-jupiter from 4.2.0 to 4.3.0 in /data-plane
Bumps [mockito-junit-jupiter](https://github.com/mockito/mockito) from 4.2.0 to 4.3.0.
<details>
<summary>Release notes</summary>
<p><em>Sourced from <a href="https://github.com/mockito/mockito/releases">mockito-junit-jupiter's releases</a>.</em></p>
<blockquote>
<h2>v4.3.0</h2>
<p><!-- raw HTML omitted --><!-- raw HTML omitted --><em>Changelog generated by <a href="https://github.com/shipkit/shipkit-changelog">Shipkit Changelog Gradle Plugin</a></em><!-- raw HTML omitted --><!-- raw HTML omitted --></p>
<h4>4.3.0</h4>
<ul>
<li>2022-01-24 - <a href="https://github.com/mockito/mockito/compare/v4.2.0...v4.3.0">20 commit(s)</a> by Andrew Kozel, John Pyeatt, Liam Miller-Cushon, Thomas Keller, Tim van der Lippe, dependabot[bot], temp-droid</li>
<li>Fixes <a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2489">#2489</a> : Fixed issue related to exceptions thrown from the nested spies [(<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2546">#2546</a>)](<a href="https://github-redirect.dependabot.com/mockito/mockito/pull/2546">mockito/mockito#2546</a>)</li>
<li>Issue 2544 [(<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2545">#2545</a>)](<a href="https://github-redirect.dependabot.com/mockito/mockito/pull/2545">mockito/mockito#2545</a>)</li>
<li>Bump versions.bytebuddy from 1.12.6 to 1.12.7 [(<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2543">#2543</a>)](<a href="https://github-redirect.dependabot.com/mockito/mockito/pull/2543">mockito/mockito#2543</a>)</li>
<li>Bump com.diffplug.spotless from 6.1.2 to 6.2.0 [(<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2542">#2542</a>)](<a href="https://github-redirect.dependabot.com/mockito/mockito/pull/2542">mockito/mockito#2542</a>)</li>
<li>Bump material from 1.4.0 to 1.5.0 [(<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2541">#2541</a>)](<a href="https://github-redirect.dependabot.com/mockito/mockito/pull/2541">mockito/mockito#2541</a>)</li>
<li>Bump appcompat from 1.4.0 to 1.4.1 [(<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2539">#2539</a>)](<a href="https://github-redirect.dependabot.com/mockito/mockito/pull/2539">mockito/mockito#2539</a>)</li>
<li>Bump com.diffplug.spotless from 6.1.1 to 6.1.2 [(<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2536">#2536</a>)](<a href="https://github-redirect.dependabot.com/mockito/mockito/pull/2536">mockito/mockito#2536</a>)</li>
<li>Remove an <code>@link</code> [(<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2535">#2535</a>)](<a href="https://github-redirect.dependabot.com/mockito/mockito/pull/2535">mockito/mockito#2535</a>)</li>
<li>Bump com.diffplug.spotless from 6.1.0 to 6.1.1 [(<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2534">#2534</a>)](<a href="https://github-redirect.dependabot.com/mockito/mockito/pull/2534">mockito/mockito#2534</a>)</li>
<li>Bump com.github.ben-manes.versions from 0.40.0 to 0.41.0 [(<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2533">#2533</a>)](<a href="https://github-redirect.dependabot.com/mockito/mockito/pull/2533">mockito/mockito#2533</a>)</li>
<li>Bump assertj-core from 3.21.0 to 3.22.0 [(<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2531">#2531</a>)](<a href="https://github-redirect.dependabot.com/mockito/mockito/pull/2531">mockito/mockito#2531</a>)</li>
<li>Bump com.github.ben-manes.versions from 0.39.0 to 0.40.0 [(<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2529">#2529</a>)](<a href="https://github-redirect.dependabot.com/mockito/mockito/pull/2529">mockito/mockito#2529</a>)</li>
<li>Bump com.diffplug.spotless from 6.0.5 to 6.1.0 [(<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2527">#2527</a>)](<a href="https://github-redirect.dependabot.com/mockito/mockito/pull/2527">mockito/mockito#2527</a>)</li>
<li>Bump kotlinx-coroutines-core from 1.5.2-native-mt to 1.6.0-native-mt [(<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2526">#2526</a>)](<a href="https://github-redirect.dependabot.com/mockito/mockito/pull/2526">mockito/mockito#2526</a>)</li>
<li>Bump versions.bytebuddy from 1.12.5 to 1.12.6 [(<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2524">#2524</a>)](<a href="https://github-redirect.dependabot.com/mockito/mockito/pull/2524">mockito/mockito#2524</a>)</li>
<li>Bump com.diffplug.spotless from 6.0.4 to 6.0.5 [(<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2520">#2520</a>)](<a href="https://github-redirect.dependabot.com/mockito/mockito/pull/2520">mockito/mockito#2520</a>)</li>
<li>Bump versions.bytebuddy from 1.12.4 to 1.12.5 [(<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2519">#2519</a>)](<a href="https://github-redirect.dependabot.com/mockito/mockito/pull/2519">mockito/mockito#2519</a>)</li>
<li>Fixes <a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2510">#2510</a>: Remove ExpectedException from internal test suite [(<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2518">#2518</a>)](<a href="https://github-redirect.dependabot.com/mockito/mockito/pull/2518">mockito/mockito#2518</a>)</li>
<li>Fix JavaDoc warnings and enforce zero errors in Gradle [(<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2513">#2513</a>)](<a href="https://github-redirect.dependabot.com/mockito/mockito/pull/2513">mockito/mockito#2513</a>)</li>
<li>Remove <code>ExpectedException</code> from internal test suite [(<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2510">#2510</a>)](<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2510">mockito/mockito#2510</a>)</li>
<li>Incomplete stack trace returned from spy inside another spy [(<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2489">#2489</a>)](<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2489">mockito/mockito#2489</a>)</li>
<li>Introduce a BOM for Mockito's artifacts (closes <a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2321">#2321</a>) [(<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2323">#2323</a>)](<a href="https://github-redirect.dependabot.com/mockito/mockito/pull/2323">mockito/mockito#2323</a>)</li>
<li>Provide a bill of materials (BOM) [(<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2321">#2321</a>)](<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2321">mockito/mockito#2321</a>)</li>
</ul>
</blockquote>
</details>
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/mockito/mockito/commit/8cdf0cce796a3497460de93abbe395108feae74c"><code>8cdf0cc</code></a> Fixes <a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2489">#2489</a> : Fixed issue related to exceptions thrown from the nested spies ...</li>
<li><a href="https://github.com/mockito/mockito/commit/faa6e92aec0c5b854c11422224e0628fa787a1cb"><code>faa6e92</code></a> Bump appcompat from 1.4.0 to 1.4.1 (<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2539">#2539</a>)</li>
<li><a href="https://github.com/mockito/mockito/commit/55059417fbba63c3c63cd7ab65aeb556424c967e"><code>5505941</code></a> Fix Mockito.mockConstructionWithAnswer with more than 2 invocations (<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2545">#2545</a>)</li>
<li><a href="https://github.com/mockito/mockito/commit/9e95f1a3bafc66b4d8ce6388a184cdbef3bfa4ea"><code>9e95f1a</code></a> Bump versions.bytebuddy from 1.12.6 to 1.12.7 (<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2543">#2543</a>)</li>
<li><a href="https://github.com/mockito/mockito/commit/8e94aeb5c206d41a1145617ac2aea14937b56ed2"><code>8e94aeb</code></a> Bump material from 1.4.0 to 1.5.0 (<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2541">#2541</a>)</li>
<li><a href="https://github.com/mockito/mockito/commit/88b958d1b7e8b197670615442c1ceaa8f4c8d77a"><code>88b958d</code></a> Bump com.diffplug.spotless from 6.1.2 to 6.2.0 (<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2542">#2542</a>)</li>
<li><a href="https://github.com/mockito/mockito/commit/cb948d231041a7639e46e39d3456a98bf5a3daf3"><code>cb948d2</code></a> Bump com.diffplug.spotless from 6.1.1 to 6.1.2 (<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2536">#2536</a>)</li>
<li><a href="https://github.com/mockito/mockito/commit/d5013001c5289be1a15125fc51d61493c9d1603f"><code>d501300</code></a> Remove <code>@link</code> to ProxyMockMaker from main Mockito javadoc (<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2535">#2535</a>)</li>
<li><a href="https://github.com/mockito/mockito/commit/3dca21f6f7c644654ac0d68ce3347e9b7be3426e"><code>3dca21f</code></a> Bump com.diffplug.spotless from 6.1.0 to 6.1.1 (<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2534">#2534</a>)</li>
<li><a href="https://github.com/mockito/mockito/commit/d5cad0648a454b237ee30f1a2b9919cf73fcee68"><code>d5cad06</code></a> Bump com.github.ben-manes.versions from 0.40.0 to 0.41.0 (<a href="https://github-redirect.dependabot.com/mockito/mockito/issues/2533">#2533</a>)</li>
<li>Additional commits viewable in <a href="https://github.com/mockito/mockito/compare/v4.2.0...v4.3.0">compare view</a></li>
</ul>
</details>
<br />


[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=org.mockito:mockito-junit-jupiter&package-manager=maven&previous-version=4.2.0&new-version=4.3.0)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores)

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)


</details>

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

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

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["aliok"]} -->
knative-prow-robot: Hi @dependabot[bot]. Thanks for your PR.

I'm waiting for a [knative-sandbox](https://github.com/orgs/knative-sandbox/people) member to verify that this patch is reasonable to test. If it is, they should reply with `/ok-to-test` on its own line. Until that is done, I will not automatically test new commits in this PR, but the usual testing commands by org members will still work. Regular contributors should [join the org](https://github.com/orgs/knative-sandbox/people) to skip this step.

Once the patch is verified, the new status will be reflected by the `ok-to-test` label.

I understand the commands that are listed [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka-broker).

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1795#" title="Author self-approved">dependabot[bot]</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please ask for approval from **pierdipi** after the PR has been reviewed.

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

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

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

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

> :exclamation: Current head 3f1f7bc differs from pull request most recent head fefb1d4. Consider uploading reports for the commit fefb1d4 to get more accurate results
[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1795/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/1795?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    #1795      +/-   ##
============================================
- Coverage     68.63%   60.39%   -8.25%     
============================================
  Files           129       60      -69     
  Lines          5334     3310    -2024     
  Branches        178        0     -178     
============================================
- Hits           3661     1999    -1662     
+ Misses         1382     1115     -267     
+ Partials        291      196      -95     
```

| 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/1795?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/consumergroup/consumergroup.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1795/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2NvbnN1bWVyZ3JvdXAuZ28=) | `76.82% <0.00%> (-2.44%)` | :arrow_down: |
| [...patcher/impl/consumer/OrderedConsumerVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1795/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9PcmRlcmVkQ29uc3VtZXJWZXJ0aWNsZS5qYXZh) | | |
| [...roker/dispatcher/impl/filter/AttributesFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1795/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvQXR0cmlidXRlc0ZpbHRlci5qYXZh) | | |
| [.../broker/receiver/main/ReceiverVerticleFactory.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1795/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9tYWluL1JlY2VpdmVyVmVydGljbGVGYWN0b3J5LmphdmE=) | | |
| [...e/eventing/kafka/broker/core/file/FileWatcher.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1795/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvZmlsZS9GaWxlV2F0Y2hlci5qYXZh) | | |
| [...ka/broker/core/reconciler/ResourcesReconciler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1795/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvcmVjb25jaWxlci9SZXNvdXJjZXNSZWNvbmNpbGVyLmphdmE=) | | |
| [...nting/kafka/broker/core/security/AuthProvider.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1795/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvc2VjdXJpdHkvQXV0aFByb3ZpZGVyLmphdmE=) | | |
| [...fka/broker/core/security/PlaintextCredentials.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1795/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvc2VjdXJpdHkvUGxhaW50ZXh0Q3JlZGVudGlhbHMuamF2YQ==) | | |
| [...ting/kafka/broker/core/utils/CollectionsUtils.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1795/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdXRpbHMvQ29sbGVjdGlvbnNVdGlscy5qYXZh) | | |
| [...g/kafka/broker/core/security/KafkaClientsAuth.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1795/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvc2VjdXJpdHkvS2Fma2FDbGllbnRzQXV0aC5qYXZh) | | |
| ... and [62 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1795/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/1795?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/1795?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [0058cf3...fefb1d4](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1795?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).


Patch:
diff --git a/control-plane/config/200-controller/200-controller-cluster-role.yaml b/control-plane/config/200-controller/200-controller-cluster-role.yaml
index fd54da4f3..dfb85a5d7 100644
--- a/control-plane/config/200-controller/200-controller-cluster-role.yaml
+++ b/control-plane/config/200-controller/200-controller-cluster-role.yaml
@@ -108,6 +108,15 @@ rules:
       - watch
       - update
       - patch
+  - apiGroups:
+      - messaging.knative.dev
+    resources:
+      - subscriptions
+      - subscriptions/status
+    verbs:
+      - get
+      - list
+      - watch
   - apiGroups:
       - messaging.knative.dev
     resources:
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go
index 0f769d11b..4693af4db 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go
@@ -167,3 +167,11 @@ func (cg *ConsumerGroup) IsReady() bool {
 	return cg.Generation == cg.Status.ObservedGeneration &&
 		cg.GetConditionSet().Manage(cg.GetStatus()).IsHappy()
 }
+
+func (cg *ConsumerGroup) IsNotScheduled() bool {
+	// We want to return true when:
+	// - the condition isn't present, or
+	// - the condition isn't ready (aka status=false)
+	cond := cg.Status.GetCondition(ConditionConsumerGroupConsumersScheduled)
+	return cond.IsFalse() || cond.IsUnknown()
+}
diff --git a/control-plane/pkg/contract/contract.pb.go b/control-plane/pkg/contract/contract.pb.go
index 3ae8af054..49076b003 100644
--- a/control-plane/pkg/contract/contract.pb.go
+++ b/control-plane/pkg/contract/contract.pb.go
@@ -534,6 +534,12 @@ type Egress struct {
 	DeliveryOrder DeliveryOrder `protobuf:"varint,8,opt,name=deliveryOrder,proto3,enum=DeliveryOrder" json:"deliveryOrder,omitempty"`
 	// Kafka record key type.
 	KeyType KeyType `protobuf:"varint,10,opt,name=keyType,proto3,enum=KeyType" json:"keyType,omitempty"`
+	// Resource reference.
+	//
+	// This reference is used to reference the associated resource for data plane
+	// activities such as:
+	// - tagging metrics
+	Reference *Reference `protobuf:"bytes,11,opt,name=reference,proto3" json:"reference,omitempty"`
 }
 
 func (x *Egress) Reset() {
@@ -645,6 +651,13 @@ func (x *Egress) GetKeyType() KeyType {
 	return KeyType_String
 }
 
+func (x *Egress) GetReference() *Reference {
+	if x != nil {
+		return x.Reference
+	}
+	return nil
+}
+
 type isEgress_ReplyStrategy interface {
 	isEgress_ReplyStrategy()
 }
@@ -1348,7 +1361,7 @@ var file_contract_proto_rawDesc = []byte{
 	0x6b, 0x6f, 0x66, 0x66, 0x44, 0x65, 0x6c, 0x61, 0x79, 0x18, 0x04, 0x20, 0x01, 0x28, 0x04, 0x52,
 	0x0c, 0x62, 0x61, 0x63, 0x6b, 0x6f, 0x66, 0x66, 0x44, 0x65, 0x6c, 0x61, 0x79, 0x12, 0x18, 0x0a,
 	0x07, 0x74, 0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74, 0x18, 0x05, 0x20, 0x01, 0x28, 0x04, 0x52, 0x07,
-	0x74, 0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74, 0x22, 0xab, 0x03, 0x0a, 0x06, 0x45, 0x67, 0x72, 0x65,
+	0x74, 0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74, 0x22, 0xd5, 0x03, 0x0a, 0x06, 0x45, 0x67, 0x72, 0x65,
 	0x73, 0x73, 0x12, 0x24, 0x0a, 0x0d, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6d, 0x65, 0x72, 0x47, 0x72,
 	0x6f, 0x75, 0x70, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0d, 0x63, 0x6f, 0x6e, 0x73, 0x75,
 	0x6d, 0x65, 0x72, 0x47, 0x72, 0x6f, 0x75, 0x70, 0x12, 0x20, 0x0a, 0x0b, 0x64, 0x65, 0x73, 0x74,
@@ -1374,118 +1387,121 @@ var file_contract_proto_rawDesc = []byte{
 	0x72, 0x52, 0x0d, 0x64, 0x65, 0x6c, 0x69, 0x76, 0x65, 0x72, 0x79, 0x4f, 0x72, 0x64, 0x65, 0x72,
 	0x12, 0x22, 0x0a, 0x07, 0x6b, 0x65, 0x79, 0x54, 0x79, 0x70, 0x65, 0x18, 0x0a, 0x20, 0x01, 0x28,
 	0x0e, 0x32, 0x08, 0x2e, 0x4b, 0x65, 0x79, 0x54, 0x79, 0x70, 0x65, 0x52, 0x07, 0x6b, 0x65, 0x79,
-	0x54, 0x79, 0x70, 0x65, 0x42, 0x0f, 0x0a, 0x0d, 0x72, 0x65, 0x70, 0x6c, 0x79, 0x53, 0x74, 0x72,
-	0x61, 0x74, 0x65, 0x67, 0x79, 0x22, 0x74, 0x0a, 0x07, 0x49, 0x6e, 0x67, 0x72, 0x65, 0x73, 0x73,
-	0x12, 0x2e, 0x0a, 0x0b, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x4d, 0x6f, 0x64, 0x65, 0x18,
-	0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x0c, 0x2e, 0x43, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x4d,
-	0x6f, 0x64, 0x65, 0x52, 0x0b, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x4d, 0x6f, 0x64, 0x65,
-	0x12, 0x14, 0x0a, 0x04, 0x70, 0x61, 0x74, 0x68, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x48, 0x00,
-	0x52, 0x04, 0x70, 0x61, 0x74, 0x68, 0x12, 0x14, 0x0a, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x18, 0x03,
-	0x20, 0x01, 0x28, 0x09, 0x48, 0x00, 0x52, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x42, 0x0d, 0x0a, 0x0b,
-	0x69, 0x6e, 0x67, 0x72, 0x65, 0x73, 0x73, 0x54, 0x79, 0x70, 0x65, 0x22, 0x6b, 0x0a, 0x09, 0x52,
-	0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x75, 0x75, 0x69, 0x64,
-	0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x75, 0x75, 0x69, 0x64, 0x12, 0x1c, 0x0a, 0x09,
-	0x6e, 0x61, 0x6d, 0x65, 0x73, 0x70, 0x61, 0x63, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52,
-	0x09, 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x70, 0x61, 0x63, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61,
-	0x6d, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x18,
-	0x0a, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52,
-	0x07, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x22, 0x7f, 0x0a, 0x0f, 0x53, 0x65, 0x63, 0x72,
-	0x65, 0x74, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x12, 0x28, 0x0a, 0x09, 0x72,
-	0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0a,
-	0x2e, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x52, 0x09, 0x72, 0x65, 0x66, 0x65,
-	0x72, 0x65, 0x6e, 0x63, 0x65, 0x12, 0x42, 0x0a, 0x12, 0x6b, 0x65, 0x79, 0x46, 0x69, 0x65, 0x6c,
-	0x64, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28,
-	0x0b, 0x32, 0x12, 0x2e, 0x4b, 0x65, 0x79, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x52, 0x65, 0x66, 0x65,
-	0x72, 0x65, 0x6e, 0x63, 0x65, 0x52, 0x12, 0x6b, 0x65, 0x79, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x52,
-	0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x73, 0x22, 0x55, 0x0a, 0x11, 0x4b, 0x65, 0x79,
-	0x46, 0x69, 0x65, 0x6c, 0x64, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x12, 0x1c,
-	0x0a, 0x09, 0x73, 0x65, 0x63, 0x72, 0x65, 0x74, 0x4b, 0x65, 0x79, 0x18, 0x02, 0x20, 0x01, 0x28,
-	0x09, 0x52, 0x09, 0x73, 0x65, 0x63, 0x72, 0x65, 0x74, 0x4b, 0x65, 0x79, 0x12, 0x22, 0x0a, 0x05,
-	0x66, 0x69, 0x65, 0x6c, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x0c, 0x2e, 0x53, 0x65,
-	0x63, 0x72, 0x65, 0x74, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x52, 0x05, 0x66, 0x69, 0x65, 0x6c, 0x64,
-	0x22, 0x6f, 0x0a, 0x14, 0x4d, 0x75, 0x6c, 0x74, 0x69, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x52,
-	0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x12, 0x25, 0x0a, 0x08, 0x70, 0x72, 0x6f, 0x74,
-	0x6f, 0x63, 0x6f, 0x6c, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x09, 0x2e, 0x50, 0x72, 0x6f,
-	0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x52, 0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x12,
-	0x30, 0x0a, 0x0a, 0x72, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x73, 0x18, 0x02, 0x20,
-	0x03, 0x28, 0x0b, 0x32, 0x10, 0x2e, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x52, 0x65, 0x66, 0x65,
-	0x72, 0x65, 0x6e, 0x63, 0x65, 0x52, 0x0a, 0x72, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65,
-	0x73, 0x22, 0x9a, 0x01, 0x0a, 0x13, 0x43, 0x6c, 0x6f, 0x75, 0x64, 0x45, 0x76, 0x65, 0x6e, 0x74,
-	0x4f, 0x76, 0x65, 0x72, 0x72, 0x69, 0x64, 0x65, 0x73, 0x12, 0x44, 0x0a, 0x0a, 0x65, 0x78, 0x74,
-	0x65, 0x6e, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x24, 0x2e,
+	0x54, 0x79, 0x70, 0x65, 0x12, 0x28, 0x0a, 0x09, 0x72, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63,
+	0x65, 0x18, 0x0b, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0a, 0x2e, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65,
+	0x6e, 0x63, 0x65, 0x52, 0x09, 0x72, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x42, 0x0f,
+	0x0a, 0x0d, 0x72, 0x65, 0x70, 0x6c, 0x79, 0x53, 0x74, 0x72, 0x61, 0x74, 0x65, 0x67, 0x79, 0x22,
+	0x74, 0x0a, 0x07, 0x49, 0x6e, 0x67, 0x72, 0x65, 0x73, 0x73, 0x12, 0x2e, 0x0a, 0x0b, 0x63, 0x6f,
+	0x6e, 0x74, 0x65, 0x6e, 0x74, 0x4d, 0x6f, 0x64, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32,
+	0x0c, 0x2e, 0x43, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x4d, 0x6f, 0x64, 0x65, 0x52, 0x0b, 0x63,
+	0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x4d, 0x6f, 0x64, 0x65, 0x12, 0x14, 0x0a, 0x04, 0x70, 0x61,
+	0x74, 0x68, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x48, 0x00, 0x52, 0x04, 0x70, 0x61, 0x74, 0x68,
+	0x12, 0x14, 0x0a, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x48, 0x00,
+	0x52, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x42, 0x0d, 0x0a, 0x0b, 0x69, 0x6e, 0x67, 0x72, 0x65, 0x73,
+	0x73, 0x54, 0x79, 0x70, 0x65, 0x22, 0x6b, 0x0a, 0x09, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e,
+	0x63, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x75, 0x75, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09,
+	0x52, 0x04, 0x75, 0x75, 0x69, 0x64, 0x12, 0x1c, 0x0a, 0x09, 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x70,
+	0x61, 0x63, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x6e, 0x61, 0x6d, 0x65, 0x73,
+	0x70, 0x61, 0x63, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x03, 0x20, 0x01,
+	0x28, 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x18, 0x0a, 0x07, 0x76, 0x65, 0x72, 0x73,
+	0x69, 0x6f, 0x6e, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69,
+	0x6f, 0x6e, 0x22, 0x7f, 0x0a, 0x0f, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x52, 0x65, 0x66, 0x65,
+	0x72, 0x65, 0x6e, 0x63, 0x65, 0x12, 0x28, 0x0a, 0x09, 0x72, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e,
+	0x63, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0a, 0x2e, 0x52, 0x65, 0x66, 0x65, 0x72,
+	0x65, 0x6e, 0x63, 0x65, 0x52, 0x09, 0x72, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x12,
+	0x42, 0x0a, 0x12, 0x6b, 0x65, 0x79, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x52, 0x65, 0x66, 0x65, 0x72,
+	0x65, 0x6e, 0x63, 0x65, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x4b, 0x65,
+	0x79, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x52,
+	0x12, 0x6b, 0x65, 0x79, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e,
+	0x63, 0x65, 0x73, 0x22, 0x55, 0x0a, 0x11, 0x4b, 0x65, 0x79, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x52,
+	0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x12, 0x1c, 0x0a, 0x09, 0x73, 0x65, 0x63, 0x72,
+	0x65, 0x74, 0x4b, 0x65, 0x79, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x73, 0x65, 0x63,
+	0x72, 0x65, 0x74, 0x4b, 0x65, 0x79, 0x12, 0x22, 0x0a, 0x05, 0x66, 0x69, 0x65, 0x6c, 0x64, 0x18,
+	0x03, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x0c, 0x2e, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x46, 0x69,
+	0x65, 0x6c, 0x64, 0x52, 0x05, 0x66, 0x69, 0x65, 0x6c, 0x64, 0x22, 0x6f, 0x0a, 0x14, 0x4d, 0x75,
+	0x6c, 0x74, 0x69, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e,
+	0x63, 0x65, 0x12, 0x25, 0x0a, 0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x18, 0x01,
+	0x20, 0x01, 0x28, 0x0e, 0x32, 0x09, 0x2e, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x52,
+	0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x12, 0x30, 0x0a, 0x0a, 0x72, 0x65, 0x66,
+	0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x10, 0x2e,
+	0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x52,
+	0x0a, 0x72, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x73, 0x22, 0x9a, 0x01, 0x0a, 0x13,
 	0x43, 0x6c, 0x6f, 0x75, 0x64, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x4f, 0x76, 0x65, 0x72, 0x72, 0x69,
-	0x64, 0x65, 0x73, 0x2e, 0x45, 0x78, 0x74, 0x65, 0x6e, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x45, 0x6e,
-	0x74, 0x72, 0x79, 0x52, 0x0a, 0x65, 0x78, 0x74, 0x65, 0x6e, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x1a,
-	0x3d, 0x0a, 0x0f, 0x45, 0x78, 0x74, 0x65, 0x6e, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x45, 0x6e, 0x74,
-	0x72, 0x79, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52,
-	0x03, 0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20,
-	0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x22, 0xf1,
-	0x03, 0x0a, 0x08, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x12, 0x10, 0x0a, 0x03, 0x75,
-	0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x75, 0x69, 0x64, 0x12, 0x16, 0x0a,
-	0x06, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x09, 0x52, 0x06, 0x74,
-	0x6f, 0x70, 0x69, 0x63, 0x73, 0x12, 0x2a, 0x0a, 0x10, 0x62, 0x6f, 0x6f, 0x74, 0x73, 0x74, 0x72,
-	0x61, 0x70, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52,
-	0x10, 0x62, 0x6f, 0x6f, 0x74, 0x73, 0x74, 0x72, 0x61, 0x70, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72,
-	0x73, 0x12, 0x22, 0x0a, 0x07, 0x69, 0x6e, 0x67, 0x72, 0x65, 0x73, 0x73, 0x18, 0x04, 0x20, 0x01,
-	0x28, 0x0b, 0x32, 0x08, 0x2e, 0x49, 0x6e, 0x67, 0x72, 0x65, 0x73, 0x73, 0x52, 0x07, 0x69, 0x6e,
-	0x67, 0x72, 0x65, 0x73, 0x73, 0x12, 0x31, 0x0a, 0x0c, 0x65, 0x67, 0x72, 0x65, 0x73, 0x73, 0x43,
-	0x6f, 0x6e, 0x66, 0x69, 0x67, 0x18, 0x05, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0d, 0x2e, 0x45, 0x67,
-	0x72, 0x65, 0x73, 0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x52, 0x0c, 0x65, 0x67, 0x72, 0x65,
-	0x73, 0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x23, 0x0a, 0x08, 0x65, 0x67, 0x72, 0x65,
-	0x73, 0x73, 0x65, 0x73, 0x18, 0x06, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x07, 0x2e, 0x45, 0x67, 0x72,
-	0x65, 0x73, 0x73, 0x52, 0x08, 0x65, 0x67, 0x72, 0x65, 0x73, 0x73, 0x65, 0x73, 0x12, 0x28, 0x0a,
-	0x0a, 0x61, 0x62, 0x73, 0x65, 0x6e, 0x74, 0x41, 0x75, 0x74, 0x68, 0x18, 0x07, 0x20, 0x01, 0x28,
-	0x0b, 0x32, 0x06, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x48, 0x00, 0x52, 0x0a, 0x61, 0x62, 0x73,
-	0x65, 0x6e, 0x74, 0x41, 0x75, 0x74, 0x68, 0x12, 0x2c, 0x0a, 0x0a, 0x61, 0x75, 0x74, 0x68, 0x53,
-	0x65, 0x63, 0x72, 0x65, 0x74, 0x18, 0x08, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0a, 0x2e, 0x52, 0x65,
-	0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x48, 0x00, 0x52, 0x0a, 0x61, 0x75, 0x74, 0x68, 0x53,
-	0x65, 0x63, 0x72, 0x65, 0x74, 0x12, 0x41, 0x0a, 0x0f, 0x6d, 0x75, 0x6c, 0x74, 0x69, 0x41, 0x75,
-	0x74, 0x68, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x18, 0x09, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x15,
-	0x2e, 0x4d, 0x75, 0x6c, 0x74, 0x69, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x52, 0x65, 0x66, 0x65,
-	0x72, 0x65, 0x6e, 0x63, 0x65, 0x48, 0x00, 0x52, 0x0f, 0x6d, 0x75, 0x6c, 0x74, 0x69, 0x41, 0x75,
-	0x74, 0x68, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x12, 0x46, 0x0a, 0x13, 0x63, 0x6c, 0x6f, 0x75,
-	0x64, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x4f, 0x76, 0x65, 0x72, 0x72, 0x69, 0x64, 0x65, 0x73, 0x18,
-	0x0a, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x14, 0x2e, 0x43, 0x6c, 0x6f, 0x75, 0x64, 0x45, 0x76, 0x65,
-	0x6e, 0x74, 0x4f, 0x76, 0x65, 0x72, 0x72, 0x69, 0x64, 0x65, 0x73, 0x52, 0x13, 0x63, 0x6c, 0x6f,
-	0x75, 0x64, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x4f, 0x76, 0x65, 0x72, 0x72, 0x69, 0x64, 0x65, 0x73,
-	0x12, 0x28, 0x0a, 0x09, 0x72, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x18, 0x0b, 0x20,
-	0x01, 0x28, 0x0b, 0x32, 0x0a, 0x2e, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x52,
-	0x09, 0x72, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x42, 0x06, 0x0a, 0x04, 0x41, 0x75,
-	0x74, 0x68, 0x22, 0x53, 0x0a, 0x08, 0x43, 0x6f, 0x6e, 0x74, 0x72, 0x61, 0x63, 0x74, 0x12, 0x1e,
-	0x0a, 0x0a, 0x67, 0x65, 0x6e, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x01, 0x20, 0x01,
-	0x28, 0x04, 0x52, 0x0a, 0x67, 0x65, 0x6e, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x27,
-	0x0a, 0x09, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28,
-	0x0b, 0x32, 0x09, 0x2e, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x52, 0x09, 0x72, 0x65,
-	0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x73, 0x2a, 0x2c, 0x0a, 0x0d, 0x42, 0x61, 0x63, 0x6b, 0x6f,
-	0x66, 0x66, 0x50, 0x6f, 0x6c, 0x69, 0x63, 0x79, 0x12, 0x0f, 0x0a, 0x0b, 0x45, 0x78, 0x70, 0x6f,
-	0x6e, 0x65, 0x6e, 0x74, 0x69, 0x61, 0x6c, 0x10, 0x00, 0x12, 0x0a, 0x0a, 0x06, 0x4c, 0x69, 0x6e,
-	0x65, 0x61, 0x72, 0x10, 0x01, 0x2a, 0x2b, 0x0a, 0x0d, 0x44, 0x65, 0x6c, 0x69, 0x76, 0x65, 0x72,
-	0x79, 0x4f, 0x72, 0x64, 0x65, 0x72, 0x12, 0x0d, 0x0a, 0x09, 0x55, 0x4e, 0x4f, 0x52, 0x44, 0x45,
-	0x52, 0x45, 0x44, 0x10, 0x00, 0x12, 0x0b, 0x0a, 0x07, 0x4f, 0x52, 0x44, 0x45, 0x52, 0x45, 0x44,
-	0x10, 0x01, 0x2a, 0x3d, 0x0a, 0x07, 0x4b, 0x65, 0x79, 0x54, 0x79, 0x70, 0x65, 0x12, 0x0a, 0x0a,
-	0x06, 0x53, 0x74, 0x72, 0x69, 0x6e, 0x67, 0x10, 0x00, 0x12, 0x0b, 0x0a, 0x07, 0x49, 0x6e, 0x74,
-	0x65, 0x67, 0x65, 0x72, 0x10, 0x01, 0x12, 0x0a, 0x0a, 0x06, 0x44, 0x6f, 0x75, 0x62, 0x6c, 0x65,
-	0x10, 0x02, 0x12, 0x0d, 0x0a, 0x09, 0x42, 0x79, 0x74, 0x65, 0x41, 0x72, 0x72, 0x61, 0x79, 0x10,
-	0x03, 0x2a, 0x29, 0x0a, 0x0b, 0x43, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x4d, 0x6f, 0x64, 0x65,
-	0x12, 0x0a, 0x0a, 0x06, 0x42, 0x49, 0x4e, 0x41, 0x52, 0x59, 0x10, 0x00, 0x12, 0x0e, 0x0a, 0x0a,
-	0x53, 0x54, 0x52, 0x55, 0x43, 0x54, 0x55, 0x52, 0x45, 0x44, 0x10, 0x01, 0x2a, 0x61, 0x0a, 0x0b,
-	0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x12, 0x12, 0x0a, 0x0e, 0x53,
-	0x41, 0x53, 0x4c, 0x5f, 0x4d, 0x45, 0x43, 0x48, 0x41, 0x4e, 0x49, 0x53, 0x4d, 0x10, 0x00, 0x12,
-	0x0a, 0x0a, 0x06, 0x43, 0x41, 0x5f, 0x43, 0x52, 0x54, 0x10, 0x01, 0x12, 0x0c, 0x0a, 0x08, 0x55,
-	0x53, 0x45, 0x52, 0x5f, 0x43, 0x52, 0x54, 0x10, 0x02, 0x12, 0x0c, 0x0a, 0x08, 0x55, 0x53, 0x45,
-	0x52, 0x5f, 0x4b, 0x45, 0x59, 0x10, 0x03, 0x12, 0x08, 0x0a, 0x04, 0x55, 0x53, 0x45, 0x52, 0x10,
-	0x04, 0x12, 0x0c, 0x0a, 0x08, 0x50, 0x41, 0x53, 0x53, 0x57, 0x4f, 0x52, 0x44, 0x10, 0x05, 0x2a,
-	0x44, 0x0a, 0x08, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x12, 0x0d, 0x0a, 0x09, 0x50,
-	0x4c, 0x41, 0x49, 0x4e, 0x54, 0x45, 0x58, 0x54, 0x10, 0x00, 0x12, 0x12, 0x0a, 0x0e, 0x53, 0x41,
-	0x53, 0x4c, 0x5f, 0x50, 0x4c, 0x41, 0x49, 0x4e, 0x54, 0x45, 0x58, 0x54, 0x10, 0x01, 0x12, 0x07,
-	0x0a, 0x03, 0x53, 0x53, 0x4c, 0x10, 0x02, 0x12, 0x0c, 0x0a, 0x08, 0x53, 0x41, 0x53, 0x4c, 0x5f,
-	0x53, 0x53, 0x4c, 0x10, 0x03, 0x42, 0x5b, 0x0a, 0x2a, 0x64, 0x65, 0x76, 0x2e, 0x6b, 0x6e, 0x61,
-	0x74, 0x69, 0x76, 0x65, 0x2e, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x69, 0x6e, 0x67, 0x2e, 0x6b, 0x61,
-	0x66, 0x6b, 0x61, 0x2e, 0x62, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72,
-	0x61, 0x63, 0x74, 0x42, 0x11, 0x44, 0x61, 0x74, 0x61, 0x50, 0x6c, 0x61, 0x6e, 0x65, 0x43, 0x6f,
-	0x6e, 0x74, 0x72, 0x61, 0x63, 0x74, 0x5a, 0x1a, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x2d,
-	0x70, 0x6c, 0x61, 0x6e, 0x65, 0x2f, 0x70, 0x6b, 0x67, 0x2f, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x61,
-	0x63, 0x74, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x64, 0x65, 0x73, 0x12, 0x44, 0x0a, 0x0a, 0x65, 0x78, 0x74, 0x65, 0x6e, 0x73, 0x69, 0x6f, 0x6e,
+	0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x24, 0x2e, 0x43, 0x6c, 0x6f, 0x75, 0x64, 0x45,
+	0x76, 0x65, 0x6e, 0x74, 0x4f, 0x76, 0x65, 0x72, 0x72, 0x69, 0x64, 0x65, 0x73, 0x2e, 0x45, 0x78,
+	0x74, 0x65, 0x6e, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x0a, 0x65,
+	0x78, 0x74, 0x65, 0x6e, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x1a, 0x3d, 0x0a, 0x0f, 0x45, 0x78, 0x74,
+	0x65, 0x6e, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x12, 0x10, 0x0a, 0x03,
+	0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14,
+	0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76,
+	0x61, 0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x22, 0xf1, 0x03, 0x0a, 0x08, 0x52, 0x65, 0x73,
+	0x6f, 0x75, 0x72, 0x63, 0x65, 0x12, 0x10, 0x0a, 0x03, 0x75, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01,
+	0x28, 0x09, 0x52, 0x03, 0x75, 0x69, 0x64, 0x12, 0x16, 0x0a, 0x06, 0x74, 0x6f, 0x70, 0x69, 0x63,
+	0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x09, 0x52, 0x06, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x73, 0x12,
+	0x2a, 0x0a, 0x10, 0x62, 0x6f, 0x6f, 0x74, 0x73, 0x74, 0x72, 0x61, 0x70, 0x53, 0x65, 0x72, 0x76,
+	0x65, 0x72, 0x73, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x10, 0x62, 0x6f, 0x6f, 0x74, 0x73,
+	0x74, 0x72, 0x61, 0x70, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x12, 0x22, 0x0a, 0x07, 0x69,
+	0x6e, 0x67, 0x72, 0x65, 0x73, 0x73, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x08, 0x2e, 0x49,
+	0x6e, 0x67, 0x72, 0x65, 0x73, 0x73, 0x52, 0x07, 0x69, 0x6e, 0x67, 0x72, 0x65, 0x73, 0x73, 0x12,
+	0x31, 0x0a, 0x0c, 0x65, 0x67, 0x72, 0x65, 0x73, 0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x18,
+	0x05, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0d, 0x2e, 0x45, 0x67, 0x72, 0x65, 0x73, 0x73, 0x43, 0x6f,
+	0x6e, 0x66, 0x69, 0x67, 0x52, 0x0c, 0x65, 0x67, 0x72, 0x65, 0x73, 0x73, 0x43, 0x6f, 0x6e, 0x66,
+	0x69, 0x67, 0x12, 0x23, 0x0a, 0x08, 0x65, 0x67, 0x72, 0x65, 0x73, 0x73, 0x65, 0x73, 0x18, 0x06,
+	0x20, 0x03, 0x28, 0x0b, 0x32, 0x07, 0x2e, 0x45, 0x67, 0x72, 0x65, 0x73, 0x73, 0x52, 0x08, 0x65,
+	0x67, 0x72, 0x65, 0x73, 0x73, 0x65, 0x73, 0x12, 0x28, 0x0a, 0x0a, 0x61, 0x62, 0x73, 0x65, 0x6e,
+	0x74, 0x41, 0x75, 0x74, 0x68, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x06, 0x2e, 0x45, 0x6d,
+	0x70, 0x74, 0x79, 0x48, 0x00, 0x52, 0x0a, 0x61, 0x62, 0x73, 0x65, 0x6e, 0x74, 0x41, 0x75, 0x74,
+	0x68, 0x12, 0x2c, 0x0a, 0x0a, 0x61, 0x75, 0x74, 0x68, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x18,
+	0x08, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0a, 0x2e, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63,
+	0x65, 0x48, 0x00, 0x52, 0x0a, 0x61, 0x75, 0x74, 0x68, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x12,
+	0x41, 0x0a, 0x0f, 0x6d, 0x75, 0x6c, 0x74, 0x69, 0x41, 0x75, 0x74, 0x68, 0x53, 0x65, 0x63, 0x72,
+	0x65, 0x74, 0x18, 0x09, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x15, 0x2e, 0x4d, 0x75, 0x6c, 0x74, 0x69,
+	0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x48,
+	0x00, 0x52, 0x0f, 0x6d, 0x75, 0x6c, 0x74, 0x69, 0x41, 0x75, 0x74, 0x68, 0x53, 0x65, 0x63, 0x72,
+	0x65, 0x74, 0x12, 0x46, 0x0a, 0x13, 0x63, 0x6c, 0x6f, 0x75, 0x64, 0x45, 0x76, 0x65, 0x6e, 0x74,
+	0x4f, 0x76, 0x65, 0x72, 0x72, 0x69, 0x64, 0x65, 0x73, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x0b, 0x32,
+	0x14, 0x2e, 0x43, 0x6c, 0x6f, 0x75, 0x64, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x4f, 0x76, 0x65, 0x72,
+	0x72, 0x69, 0x64, 0x65, 0x73, 0x52, 0x13, 0x63, 0x6c, 0x6f, 0x75, 0x64, 0x45, 0x76, 0x65, 0x6e,
+	0x74, 0x4f, 0x76, 0x65, 0x72, 0x72, 0x69, 0x64, 0x65, 0x73, 0x12, 0x28, 0x0a, 0x09, 0x72, 0x65,
+	0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x18, 0x0b, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0a, 0x2e,
+	0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x52, 0x09, 0x72, 0x65, 0x66, 0x65, 0x72,
+	0x65, 0x6e, 0x63, 0x65, 0x42, 0x06, 0x0a, 0x04, 0x41, 0x75, 0x74, 0x68, 0x22, 0x53, 0x0a, 0x08,
+	0x43, 0x6f, 0x6e, 0x74, 0x72, 0x61, 0x63, 0x74, 0x12, 0x1e, 0x0a, 0x0a, 0x67, 0x65, 0x6e, 0x65,
+	0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0a, 0x67, 0x65,
+	0x6e, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x27, 0x0a, 0x09, 0x72, 0x65, 0x73, 0x6f,
+	0x75, 0x72, 0x63, 0x65, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x09, 0x2e, 0x52, 0x65,
+	0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x52, 0x09, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
+	0x73, 0x2a, 0x2c, 0x0a, 0x0d, 0x42, 0x61, 0x63, 0x6b, 0x6f, 0x66, 0x66, 0x50, 0x6f, 0x6c, 0x69,
+	0x63, 0x79, 0x12, 0x0f, 0x0a, 0x0b, 0x45, 0x78, 0x70, 0x6f, 0x6e, 0x65, 0x6e, 0x74, 0x69, 0x61,
+	0x6c, 0x10, 0x00, 0x12, 0x0a, 0x0a, 0x06, 0x4c, 0x69, 0x6e, 0x65, 0x61, 0x72, 0x10, 0x01, 0x2a,
+	0x2b, 0x0a, 0x0d, 0x44, 0x65, 0x6c, 0x69, 0x76, 0x65, 0x72, 0x79, 0x4f, 0x72, 0x64, 0x65, 0x72,
+	0x12, 0x0d, 0x0a, 0x09, 0x55, 0x4e, 0x4f, 0x52, 0x44, 0x45, 0x52, 0x45, 0x44, 0x10, 0x00, 0x12,
+	0x0b, 0x0a, 0x07, 0x4f, 0x52, 0x44, 0x45, 0x52, 0x45, 0x44, 0x10, 0x01, 0x2a, 0x3d, 0x0a, 0x07,
+	0x4b, 0x65, 0x79, 0x54, 0x79, 0x70, 0x65, 0x12, 0x0a, 0x0a, 0x06, 0x53, 0x74, 0x72, 0x69, 0x6e,
+	0x67, 0x10, 0x00, 0x12, 0x0b, 0x0a, 0x07, 0x49, 0x6e, 0x74, 0x65, 0x67, 0x65, 0x72, 0x10, 0x01,
+	0x12, 0x0a, 0x0a, 0x06, 0x44, 0x6f, 0x75, 0x62, 0x6c, 0x65, 0x10, 0x02, 0x12, 0x0d, 0x0a, 0x09,
+	0x42, 0x79, 0x74, 0x65, 0x41, 0x72, 0x72, 0x61, 0x79, 0x10, 0x03, 0x2a, 0x29, 0x0a, 0x0b, 0x43,
+	0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x4d, 0x6f, 0x64, 0x65, 0x12, 0x0a, 0x0a, 0x06, 0x42, 0x49,
+	0x4e, 0x41, 0x52, 0x59, 0x10, 0x00, 0x12, 0x0e, 0x0a, 0x0a, 0x53, 0x54, 0x52, 0x55, 0x43, 0x54,
+	0x55, 0x52, 0x45, 0x44, 0x10, 0x01, 0x2a, 0x61, 0x0a, 0x0b, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74,
+	0x46, 0x69, 0x65, 0x6c, 0x64, 0x12, 0x12, 0x0a, 0x0e, 0x53, 0x41, 0x53, 0x4c, 0x5f, 0x4d, 0x45,
+	0x43, 0x48, 0x41, 0x4e, 0x49, 0x53, 0x4d, 0x10, 0x00, 0x12, 0x0a, 0x0a, 0x06, 0x43, 0x41, 0x5f,
+	0x43, 0x52, 0x54, 0x10, 0x01, 0x12, 0x0c, 0x0a, 0x08, 0x55, 0x53, 0x45, 0x52, 0x5f, 0x43, 0x52,
+	0x54, 0x10, 0x02, 0x12, 0x0c, 0x0a, 0x08, 0x55, 0x53, 0x45, 0x52, 0x5f, 0x4b, 0x45, 0x59, 0x10,
+	0x03, 0x12, 0x08, 0x0a, 0x04, 0x55, 0x53, 0x45, 0x52, 0x10, 0x04, 0x12, 0x0c, 0x0a, 0x08, 0x50,
+	0x41, 0x53, 0x53, 0x57, 0x4f, 0x52, 0x44, 0x10, 0x05, 0x2a, 0x44, 0x0a, 0x08, 0x50, 0x72, 0x6f,
+	0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x12, 0x0d, 0x0a, 0x09, 0x50, 0x4c, 0x41, 0x49, 0x4e, 0x54, 0x45,
+	0x58, 0x54, 0x10, 0x00, 0x12, 0x12, 0x0a, 0x0e, 0x53, 0x41, 0x53, 0x4c, 0x5f, 0x50, 0x4c, 0x41,
+	0x49, 0x4e, 0x54, 0x45, 0x58, 0x54, 0x10, 0x01, 0x12, 0x07, 0x0a, 0x03, 0x53, 0x53, 0x4c, 0x10,
+	0x02, 0x12, 0x0c, 0x0a, 0x08, 0x53, 0x41, 0x53, 0x4c, 0x5f, 0x53, 0x53, 0x4c, 0x10, 0x03, 0x42,
+	0x5b, 0x0a, 0x2a, 0x64, 0x65, 0x76, 0x2e, 0x6b, 0x6e, 0x61, 0x74, 0x69, 0x76, 0x65, 0x2e, 0x65,
+	0x76, 0x65, 0x6e, 0x74, 0x69, 0x6e, 0x67, 0x2e, 0x6b, 0x61, 0x66, 0x6b, 0x61, 0x2e, 0x62, 0x72,
+	0x6f, 0x6b, 0x65, 0x72, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x61, 0x63, 0x74, 0x42, 0x11, 0x44,
+	0x61, 0x74, 0x61, 0x50, 0x6c, 0x61, 0x6e, 0x65, 0x43, 0x6f, 0x6e, 0x74, 0x72, 0x61, 0x63, 0x74,
+	0x5a, 0x1a, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x2d, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x2f,
+	0x70, 0x6b, 0x67, 0x2f, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x61, 0x63, 0x74, 0x62, 0x06, 0x70, 0x72,
+	0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
@@ -1533,27 +1549,28 @@ var file_contract_proto_depIdxs = []int32{
 	8,  // 5: Egress.egressConfig:type_name -> EgressConfig
 	1,  // 6: Egress.deliveryOrder:type_name -> DeliveryOrder
 	2,  // 7: Egress.keyType:type_name -> KeyType
-	3,  // 8: Ingress.contentMode:type_name -> ContentMode
-	11, // 9: SecretReference.reference:type_name -> Reference
-	13, // 10: SecretReference.keyFieldReferences:type_name -> KeyFieldReference
-	4,  // 11: KeyFieldReference.field:type_name -> SecretField
-	5,  // 12: MultiSecretReference.protocol:type_name -> Protocol
-	12, // 13: MultiSecretReference.references:type_name -> SecretReference
-	19, // 14: CloudEventOverrides.extensions:type_name -> CloudEventOverrides.ExtensionsEntry
-	10, // 15: Resource.ingress:type_name -> Ingress
-	8,  // 16: Resource.egressConfig:type_name -> EgressConfig
-	9,  // 17: Resource.egresses:type_name -> Egress
-	6,  // 18: Resource.absentAuth:type_name -> Empty
-	11, // 19: Resource.authSecret:type_name -> Reference
-	14, // 20: Resource.multiAuthSecret:type_name -> MultiSecretReference
-	15, // 21: Resource.cloudEventOverrides:type_name -> CloudEventOverrides
-	11, // 22: Resource.reference:type_name -> Reference
-	16, // 23: Contract.resources:type_name -> Resource
-	24, // [24:24] is the sub-list for method output_type
-	24, // [24:24] is the sub-list for method input_type
-	24, // [24:24] is the sub-list for extension type_name
-	24, // [24:24] is the sub-list for extension extendee
-	0,  // [0:24] is the sub-list for field type_name
+	11, // 8: Egress.reference:type_name -> Reference
+	3,  // 9: Ingress.contentMode:type_name -> ContentMode
+	11, // 10: SecretReference.reference:type_name -> Reference
+	13, // 11: SecretReference.keyFieldReferences:type_name -> KeyFieldReference
+	4,  // 12: KeyFieldReference.field:type_name -> SecretField
+	5,  // 13: MultiSecretReference.protocol:type_name -> Protocol
+	12, // 14: MultiSecretReference.references:type_name -> SecretReference
+	19, // 15: CloudEventOverrides.extensions:type_name -> CloudEventOverrides.ExtensionsEntry
+	10, // 16: Resource.ingress:type_name -> Ingress
+	8,  // 17: Resource.egressConfig:type_name -> EgressConfig
+	9,  // 18: Resource.egresses:type_name -> Egress
+	6,  // 19: Resource.absentAuth:type_name -> Empty
+	11, // 20: Resource.authSecret:type_name -> Reference
+	14, // 21: Resource.multiAuthSecret:type_name -> MultiSecretReference
+	15, // 22: Resource.cloudEventOverrides:type_name -> CloudEventOverrides
+	11, // 23: Resource.reference:type_name -> Reference
+	16, // 24: Contract.resources:type_name -> Resource
+	25, // [25:25] is the sub-list for method output_type
+	25, // [25:25] is the sub-list for method input_type
+	25, // [25:25] is the sub-list for extension type_name
+	25, // [25:25] is the sub-list for extension extendee
+	0,  // [0:25] is the sub-list for field type_name
 }
 
 func init() { file_contract_proto_init() }
diff --git a/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index 620b29361..c3a6be022 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -18,13 +18,15 @@ package channel
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"strconv"
 
 	"github.com/Shopify/sarama"
-
 	"go.uber.org/multierr"
 	"go.uber.org/zap"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -37,6 +39,7 @@ import (
 	"knative.dev/pkg/system"
 
 	v1 "knative.dev/eventing/pkg/apis/duck/v1"
+	messaginglisters "knative.dev/eventing/pkg/client/listers/messaging/v1"
 
 	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	"knative.dev/eventing-kafka/pkg/common/constants"
@@ -77,7 +80,8 @@ type Reconciler struct {
 	// reconciliation loop.
 	InitOffsetsFunc kafka.InitOffsetsFunc
 
-	ConfigMapLister corelisters.ConfigMapLister
+	ConfigMapLister    corelisters.ConfigMapLister
+	SubscriptionLister messaginglisters.SubscriptionLister
 
 	Prober prober.Prober
 
@@ -455,7 +459,7 @@ func (r *Reconciler) reconcileSubscriber(ctx context.Context, kafkaClient sarama
 	subscriberIndex := coreconfig.FindEgress(channelContractResource.Egresses, subscriberSpec.UID)
 	subscriberConfig, err := r.getSubscriberConfig(ctx, channel, subscriberSpec)
 	if err != nil {
-		return fmt.Errorf("failed to resolve subscriber config: %v", zap.Error(err))
+		return fmt.Errorf("failed to resolve subscriber config: %w", err)
 	}
 
 	coreconfig.AddOrUpdateEgressConfigForResource(channelContractResource, subscriberConfig, subscriberIndex)
@@ -476,11 +480,21 @@ func (r *Reconciler) reconcileInitialOffset(ctx context.Context, channel *messag
 }
 
 func (r *Reconciler) getSubscriberConfig(ctx context.Context, channel *messagingv1beta1.KafkaChannel, subscriber *v1.SubscriberSpec) (*contract.Egress, error) {
+	subscriptionName, err := r.getSubscriptionName(channel, subscriber)
+	if err != nil {
+		return nil, fmt.Errorf("failed to extract subscription name for subscriber %v: %w", subscriber, err)
+	}
+
 	egress := &contract.Egress{
 		Destination:   subscriber.SubscriberURI.String(),
 		ConsumerGroup: consumerGroup(channel, subscriber),
 		DeliveryOrder: DefaultDeliveryOrder,
 		Uid:           string(subscriber.UID),
+		Reference: &contract.Reference{
+			Uuid:      string(subscriber.UID),
+			Namespace: channel.GetNamespace(),
+			Name:      subscriptionName,
+		},
 		ReplyStrategy: &contract.Egress_DiscardReply{},
 	}
 
@@ -581,6 +595,21 @@ func (r *Reconciler) getChannelContractResource(ctx context.Context, topic strin
 	return resource, nil
 }
 
+func (r *Reconciler) getSubscriptionName(channel *messagingv1beta1.KafkaChannel, subscriber *v1.SubscriberSpec) (string, error) {
+	subscriptions, err := r.SubscriptionLister.Subscriptions(channel.GetNamespace()).List(labels.Everything())
+	if err != nil {
+		return "", fmt.Errorf("failed to list subscriptions in namespace %s: %w", channel.GetNamespace(), err)
+	}
+
+	for _, s := range subscriptions {
+		if s.UID == subscriber.UID {
+			return s.Name, nil
+		}
+	}
+
+	return "", errors.New(string(metav1.StatusReasonNotFound))
+}
+
 // consumerGroup returns a consumerGroup name for the given channel and subscription
 func consumerGroup(channel *messagingv1beta1.KafkaChannel, sub *v1.SubscriberSpec) string {
 	return fmt.Sprintf("kafka.%s.%s.%s", channel.Namespace, channel.Name, string(sub.UID))
diff --git a/control-plane/pkg/reconciler/channel/controller.go b/control-plane/pkg/reconciler/channel/controller.go
index 8ccfa800b..db8296757 100644
--- a/control-plane/pkg/reconciler/channel/controller.go
+++ b/control-plane/pkg/reconciler/channel/controller.go
@@ -25,6 +25,8 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/tools/cache"
 
+	subscriptioninformer "knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription"
+
 	messagingv1beta "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	kafkachannelinformer "knative.dev/eventing-kafka/pkg/client/injection/informers/messaging/v1beta1/kafkachannel"
 	kafkachannelreconciler "knative.dev/eventing-kafka/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel"
@@ -65,6 +67,7 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *conf
 			DispatcherLabel:             base.ChannelDispatcherLabel,
 			ReceiverLabel:               base.ChannelReceiverLabel,
 		},
+		SubscriptionLister:         subscriptioninformer.Get(ctx).Lister(),
 		NewKafkaClient:             sarama.NewClient,
 		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
 		InitOffsetsFunc:            offset.InitOffsets,
diff --git a/control-plane/pkg/reconciler/consumergroup/evictor.go b/control-plane/pkg/reconciler/consumergroup/evictor.go
new file mode 100644
index 000000000..28b2149d6
--- /dev/null
+++ b/control-plane/pkg/reconciler/consumergroup/evictor.go
@@ -0,0 +1,129 @@
+/*
+ * 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 consumergroup
+
+import (
+	"context"
+	"fmt"
+
+	"go.uber.org/zap"
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	eventingduckv1alpha1 "knative.dev/eventing/pkg/apis/duck/v1alpha1"
+	"knative.dev/eventing/pkg/scheduler"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	"knative.dev/pkg/logging"
+
+	internalv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/clientset/versioned/typed/eventing/v1alpha1"
+	kafkainternalsclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client"
+)
+
+type evictor struct {
+	ctx context.Context
+
+	kubeClient      kubernetes.Interface
+	InternalsClient internalv1alpha1.InternalV1alpha1Interface
+
+	logger *zap.Logger
+}
+
+// newEvictor creates a new evictor.
+//
+// fields are additional logger fields to be attached to the evictor logger.
+func newEvictor(ctx context.Context, fields ...zap.Field) evictor {
+	return evictor{
+		ctx:             ctx,
+		kubeClient:      kubeclient.Get(ctx),
+		InternalsClient: kafkainternalsclient.Get(ctx).InternalV1alpha1(),
+		logger: logging.FromContext(ctx).
+			Desugar().
+			With(zap.String("component", "evictor")).
+			With(fields...),
+	}
+}
+
+func (e *evictor) evict(pod *corev1.Pod, vpod scheduler.VPod, from *eventingduckv1alpha1.Placement) error {
+	key := vpod.GetKey()
+
+	logger := e.logger.
+		With(zap.String("consumergroup", key.String())).
+		With(zap.String("pod", fmt.Sprintf("%s/%s", pod.GetNamespace(), pod.GetName())))
+
+	if err := e.disablePodScheduling(logger, pod.DeepCopy() /* Do not modify informer copy. */); err != nil {
+		return fmt.Errorf("failed to mark pod unschedulable: %w", err)
+	}
+
+	cg, err := e.InternalsClient.
+		ConsumerGroups(key.Namespace).
+		Get(e.ctx, key.Name, metav1.GetOptions{})
+	if apierrors.IsNotFound(err) {
+		return nil
+	}
+	if err != nil {
+		return fmt.Errorf("failed to get consumer group %s/%s: %w", key.Namespace, key.Name, err)
+	}
+
+	// Do not evict when the consumer group isn't scheduled yet.
+	if cg.IsNotScheduled() {
+		return nil
+	}
+
+	cg.Status.Placements = removePlacement(cg.GetPlacements(), from)
+
+	_, err = e.InternalsClient.
+		ConsumerGroups(cg.GetNamespace()).
+		Update(e.ctx, cg, metav1.UpdateOptions{})
+	if apierrors.IsNotFound(err) {
+		return nil
+	}
+	if err != nil {
+		return fmt.Errorf("failed to update consumer group %s/%s: %w", cg.GetNamespace(), cg.GetName(), err)
+	}
+
+	return nil
+}
+
+func (e *evictor) disablePodScheduling(logger *zap.Logger, pod *corev1.Pod) error {
+	if pod.Annotations == nil {
+		pod.Annotations = make(map[string]string, 1)
+	}
+	// scheduling disabled.
+	pod.Annotations[scheduler.PodAnnotationKey] = "true"
+
+	_, err := e.kubeClient.CoreV1().
+		Pods(pod.GetNamespace()).
+		Update(e.ctx, pod, metav1.UpdateOptions{})
+	if err != nil {
+		return fmt.Errorf("failed to update pod %s/%s: %w", pod.GetNamespace(), pod.GetName(), err)
+	}
+
+	logger.Info("Marked pod as unschedulable")
+
+	return nil
+}
+
+func removePlacement(before []eventingduckv1alpha1.Placement, toRemove *eventingduckv1alpha1.Placement) []eventingduckv1alpha1.Placement {
+	after := make([]eventingduckv1alpha1.Placement, 0, len(before))
+	for _, p := range before {
+		if p.PodName != toRemove.PodName {
+			after = append(after, p)
+		}
+	}
+	return after
+}
diff --git a/control-plane/pkg/reconciler/source/source.go b/control-plane/pkg/reconciler/source/source.go
index e3980959c..935fb5f16 100644
--- a/control-plane/pkg/reconciler/source/source.go
+++ b/control-plane/pkg/reconciler/source/source.go
@@ -275,6 +275,11 @@ func (r *Reconciler) reconcileKafkaSourceResource(ctx context.Context, ks *sourc
 		Uid:           string(ks.GetUID()),
 		EgressConfig:  egressConfig,
 		DeliveryOrder: DefaultDeliveryOrder,
+		Reference: &contract.Reference{
+			Uuid:      string(ks.GetUID()),
+			Namespace: ks.GetNamespace(),
+			Name:      ks.GetName(),
+		},
 	}
 	// Set key type hint (if any).
 	if keyType, ok := ks.Labels[sources.KafkaKeyTypeLabel]; ok {
@@ -286,10 +291,7 @@ func (r *Reconciler) reconcileKafkaSourceResource(ctx context.Context, ks *sourc
 		BootstrapServers: strings.Join(ks.Spec.BootstrapServers, ","),
 		Egresses:         []*contract.Egress{egress},
 		Auth:             &contract.Resource_AbsentAuth{},
-		Reference: &contract.Reference{
-			Namespace: ks.GetNamespace(),
-			Name:      ks.GetName(),
-		},
+		Reference:        egress.Reference,
 	}
 	if ks.Spec.CloudEventOverrides != nil {
 		resource.CloudEventOverrides = &contract.CloudEventOverrides{
diff --git a/control-plane/pkg/reconciler/trigger/trigger.go b/control-plane/pkg/reconciler/trigger/trigger.go
index e1614b381..a4a8b8b55 100644
--- a/control-plane/pkg/reconciler/trigger/trigger.go
+++ b/control-plane/pkg/reconciler/trigger/trigger.go
@@ -276,6 +276,11 @@ func (r *Reconciler) reconcileTriggerEgress(ctx context.Context, broker *eventin
 		Destination:   destination.String(),
 		ConsumerGroup: string(trigger.UID),
 		Uid:           string(trigger.UID),
+		Reference: &contract.Reference{
+			Uuid:      string(trigger.GetUID()),
+			Namespace: trigger.GetNamespace(),
+			Name:      trigger.GetName(),
+		},
 	}
 
 	if trigger.Spec.Filter != nil && trigger.Spec.Filter.Attributes != nil {
diff --git a/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index 8ea9aebed..7efab1a74 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -115,8 +115,8 @@ Lists of 158 third-party dependencies.
      (The Apache Software License, Version 2.0) Zipkin Sender: OkHttp 3 (io.zipkin.reporter2:zipkin-sender-okhttp3:2.16.3 - https://github.com/openzipkin/zipkin-reporter-java/zipkin-sender-okhttp3)
      (The Apache Software License, Version 2.0) Zipkin Core Library (io.zipkin.zipkin2:zipkin:2.23.2 - https://github.com/openzipkin/zipkin/zipkin)
      (Eclipse Public License 1.0) JUnit (junit:junit:4.12 - http://junit.org)
-     (Apache License, Version 2.0) Byte Buddy (without dependencies) (net.bytebuddy:byte-buddy:1.12.4 - https://bytebuddy.net/byte-buddy)
-     (Apache License, Version 2.0) Byte Buddy agent (net.bytebuddy:byte-buddy-agent:1.12.4 - https://bytebuddy.net/byte-buddy-agent)
+     (Apache License, Version 2.0) Byte Buddy (without dependencies) (net.bytebuddy:byte-buddy:1.12.7 - https://bytebuddy.net/byte-buddy)
+     (Apache License, Version 2.0) Byte Buddy agent (net.bytebuddy:byte-buddy-agent:1.12.7 - https://bytebuddy.net/byte-buddy-agent)
      (Apache License, Version 2.0) (MIT License) Logstash Logback Encoder (net.logstash.logback:logstash-logback-encoder:7.0.1 - https://github.com/logfellow/logstash-logback-encoder)
      (The MIT License) JOpt Simple (net.sf.jopt-simple:jopt-simple:5.0.4 - http://jopt-simple.github.io/jopt-simple)
      (MIT) argparse4j (net.sourceforge.argparse4j:argparse4j:0.7.0 - http://argparse4j.github.io)
@@ -144,8 +144,8 @@ Lists of 158 third-party dependencies.
      (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 MIT License) mockito-core (org.mockito:mockito-core:4.2.0 - https://github.com/mockito/mockito)
-     (The MIT License) mockito-junit-jupiter (org.mockito:mockito-junit-jupiter:4.2.0 - https://github.com/mockito/mockito)
+     (The MIT License) mockito-core (org.mockito:mockito-core:4.3.0 - https://github.com/mockito/mockito)
+     (The MIT License) mockito-junit-jupiter (org.mockito:mockito-junit-jupiter:4.3.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.34 - 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.34 - http://openjdk.java.net/projects/code-tools/jmh/jmh-generator-annprocess/)
diff --git a/data-plane/contract/src/main/java/dev/knative/eventing/kafka/broker/contract/DataPlaneContract.java b/data-plane/contract/src/main/java/dev/knative/eventing/kafka/broker/contract/DataPlaneContract.java
index 1d81cacea..fc4b899fe 100644
--- a/data-plane/contract/src/main/java/dev/knative/eventing/kafka/broker/contract/DataPlaneContract.java
+++ b/data-plane/contract/src/main/java/dev/knative/eventing/kafka/broker/contract/DataPlaneContract.java
@@ -3325,6 +3325,42 @@ public final class DataPlaneContract {
      */
     dev.knative.eventing.kafka.broker.contract.DataPlaneContract.KeyType getKeyType();
 
+    /**
+     * <pre>
+     * Resource reference.
+     * This reference is used to reference the associated resource for data plane
+     * activities such as:
+     * - tagging metrics
+     * </pre>
+     *
+     * <code>.Reference reference = 11;</code>
+     * @return Whether the reference field is set.
+     */
+    boolean hasReference();
+    /**
+     * <pre>
+     * Resource reference.
+     * This reference is used to reference the associated resource for data plane
+     * activities such as:
+     * - tagging metrics
+     * </pre>
+     *
+     * <code>.Reference reference = 11;</code>
+     * @return The reference.
+     */
+    dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference getReference();
+    /**
+     * <pre>
+     * Resource reference.
+     * This reference is used to reference the associated resource for data plane
+     * activities such as:
+     * - tagging metrics
+     * </pre>
+     *
+     * <code>.Reference reference = 11;</code>
+     */
+    dev.knative.eventing.kafka.broker.contract.DataPlaneContract.ReferenceOrBuilder getReferenceOrBuilder();
+
     public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Egress.ReplyStrategyCase getReplyStrategyCase();
   }
   /**
@@ -3467,6 +3503,19 @@ public final class DataPlaneContract {
               keyType_ = rawValue;
               break;
             }
+            case 90: {
+              dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference.Builder subBuilder = null;
+              if (reference_ != null) {
+                subBuilder = reference_.toBuilder();
+              }
+              reference_ = input.readMessage(dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference.parser(), extensionRegistry);
+              if (subBuilder != null) {
+                subBuilder.mergeFrom(reference_);
+                reference_ = subBuilder.buildPartial();
+              }
+
+              break;
+            }
             default: {
               if (!parseUnknownField(
                   input, unknownFields, extensionRegistry, tag)) {
@@ -3955,6 +4004,53 @@ public final class DataPlaneContract {
       return result == null ? dev.knative.eventing.kafka.broker.contract.DataPlaneContract.KeyType.UNRECOGNIZED : result;
     }
 
+    public static final int REFERENCE_FIELD_NUMBER = 11;
+    private dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference reference_;
+    /**
+     * <pre>
+     * Resource reference.
+     * This reference is used to reference the associated resource for data plane
+     * activities such as:
+     * - tagging metrics
+     * </pre>
+     *
+     * <code>.Reference reference = 11;</code>
+     * @return Whether the reference field is set.
+     */
+    @java.lang.Override
+    public boolean hasReference() {
+      return reference_ != null;
+    }
+    /**
+     * <pre>
+     * Resource reference.
+     * This reference is used to reference the associated resource for data plane
+     * activities such as:
+     * - tagging metrics
+     * </pre>
+     *
+     * <code>.Reference reference = 11;</code>
+     * @return The reference.
+     */
+    @java.lang.Override
+    public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference getReference() {
+      return reference_ == null ? dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference.getDefaultInstance() : reference_;
+    }
+    /**
+     * <pre>
+     * Resource reference.
+     * This reference is used to reference the associated resource for data plane
+     * activities such as:
+     * - tagging metrics
+     * </pre>
+     *
+     * <code>.Reference reference = 11;</code>
+     */
+    @java.lang.Override
+    public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.ReferenceOrBuilder getReferenceOrBuilder() {
+      return getReference();
+    }
+
     private byte memoizedIsInitialized = -1;
     @java.lang.Override
     public final boolean isInitialized() {
@@ -3999,6 +4095,9 @@ public final class DataPlaneContract {
       if (keyType_ != dev.knative.eventing.kafka.broker.contract.DataPlaneContract.KeyType.String.getNumber()) {
         output.writeEnum(10, keyType_);
       }
+      if (reference_ != null) {
+        output.writeMessage(11, getReference());
+      }
       unknownFields.writeTo(output);
     }
 
@@ -4044,6 +4143,10 @@ public final class DataPlaneContract {
         size += com.google.protobuf.CodedOutputStream
           .computeEnumSize(10, keyType_);
       }
+      if (reference_ != null) {
+        size += com.google.protobuf.CodedOutputStream
+          .computeMessageSize(11, getReference());
+      }
       size += unknownFields.getSerializedSize();
       memoizedSize = size;
       return size;
@@ -4077,6 +4180,11 @@ public final class DataPlaneContract {
       }
       if (deliveryOrder_ != other.deliveryOrder_) return false;
       if (keyType_ != other.keyType_) return false;
+      if (hasReference() != other.hasReference()) return false;
+      if (hasReference()) {
+        if (!getReference()
+            .equals(other.getReference())) return false;
+      }
       if (!getReplyStrategyCase().equals(other.getReplyStrategyCase())) return false;
       switch (replyStrategyCase_) {
         case 3:
@@ -4123,6 +4231,10 @@ public final class DataPlaneContract {
       hash = (53 * hash) + deliveryOrder_;
       hash = (37 * hash) + KEYTYPE_FIELD_NUMBER;
       hash = (53 * hash) + keyType_;
+      if (hasReference()) {
+        hash = (37 * hash) + REFERENCE_FIELD_NUMBER;
+        hash = (53 * hash) + getReference().hashCode();
+      }
       switch (replyStrategyCase_) {
         case 3:
           hash = (37 * hash) + REPLYURL_FIELD_NUMBER;
@@ -4294,6 +4406,12 @@ public final class DataPlaneContract {
 
         keyType_ = 0;
 
+        if (referenceBuilder_ == null) {
+          reference_ = null;
+        } else {
+          reference_ = null;
+          referenceBuilder_ = null;
+        }
         replyStrategyCase_ = 0;
         replyStrategy_ = null;
         return this;
@@ -4354,6 +4472,11 @@ public final class DataPlaneContract {
         }
         result.deliveryOrder_ = deliveryOrder_;
         result.keyType_ = keyType_;
+        if (referenceBuilder_ == null) {
+          result.reference_ = reference_;
+        } else {
+          result.reference_ = referenceBuilder_.build();
+        }
         result.replyStrategyCase_ = replyStrategyCase_;
         onBuilt();
         return result;
@@ -4427,6 +4550,9 @@ public final class DataPlaneContract {
         if (other.keyType_ != 0) {
           setKeyTypeValue(other.getKeyTypeValue());
         }
+        if (other.hasReference()) {
+          mergeReference(other.getReference());
+        }
         switch (other.getReplyStrategyCase()) {
           case REPLYURL: {
             replyStrategyCase_ = 3;
@@ -5693,6 +5819,188 @@ public final class DataPlaneContract {
         onChanged();
         return this;
       }
+
+      private dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference reference_;
+      private com.google.protobuf.SingleFieldBuilderV3<
+          dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference.Builder, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.ReferenceOrBuilder> referenceBuilder_;
+      /**
+       * <pre>
+       * Resource reference.
+       * This reference is used to reference the associated resource for data plane
+       * activities such as:
+       * - tagging metrics
+       * </pre>
+       *
+       * <code>.Reference reference = 11;</code>
+       * @return Whether the reference field is set.
+       */
+      public boolean hasReference() {
+        return referenceBuilder_ != null || reference_ != null;
+      }
+      /**
+       * <pre>
+       * Resource reference.
+       * This reference is used to reference the associated resource for data plane
+       * activities such as:
+       * - tagging metrics
+       * </pre>
+       *
+       * <code>.Reference reference = 11;</code>
+       * @return The reference.
+       */
+      public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference getReference() {
+        if (referenceBuilder_ == null) {
+          return reference_ == null ? dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference.getDefaultInstance() : reference_;
+        } else {
+          return referenceBuilder_.getMessage();
+        }
+      }
+      /**
+       * <pre>
+       * Resource reference.
+       * This reference is used to reference the associated resource for data plane
+       * activities such as:
+       * - tagging metrics
+       * </pre>
+       *
+       * <code>.Reference reference = 11;</code>
+       */
+      public Builder setReference(dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference value) {
+        if (referenceBuilder_ == null) {
+          if (value == null) {
+            throw new NullPointerException();
+          }
+          reference_ = value;
+          onChanged();
+        } else {
+          referenceBuilder_.setMessage(value);
+        }
+
+        return this;
+      }
+      /**
+       * <pre>
+       * Resource reference.
+       * This reference is used to reference the associated resource for data plane
+       * activities such as:
+       * - tagging metrics
+       * </pre>
+       *
+       * <code>.Reference reference = 11;</code>
+       */
+      public Builder setReference(
+          dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference.Builder builderForValue) {
+        if (referenceBuilder_ == null) {
+          reference_ = builderForValue.build();
+          onChanged();
+        } else {
+          referenceBuilder_.setMessage(builderForValue.build());
+        }
+
+        return this;
+      }
+      /**
+       * <pre>
+       * Resource reference.
+       * This reference is used to reference the associated resource for data plane
+       * activities such as:
+       * - tagging metrics
+       * </pre>
+       *
+       * <code>.Reference reference = 11;</code>
+       */
+      public Builder mergeReference(dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference value) {
+        if (referenceBuilder_ == null) {
+          if (reference_ != null) {
+            reference_ =
+              dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference.newBuilder(reference_).mergeFrom(value).buildPartial();
+          } else {
+            reference_ = value;
+          }
+          onChanged();
+        } else {
+          referenceBuilder_.mergeFrom(value);
+        }
+
+        return this;
+      }
+      /**
+       * <pre>
+       * Resource reference.
+       * This reference is used to reference the associated resource for data plane
+       * activities such as:
+       * - tagging metrics
+       * </pre>
+       *
+       * <code>.Reference reference = 11;</code>
+       */
+      public Builder clearReference() {
+        if (referenceBuilder_ == null) {
+          reference_ = null;
+          onChanged();
+        } else {
+          reference_ = null;
+          referenceBuilder_ = null;
+        }
+
+        return this;
+      }
+      /**
+       * <pre>
+       * Resource reference.
+       * This reference is used to reference the associated resource for data plane
+       * activities such as:
+       * - tagging metrics
+       * </pre>
+       *
+       * <code>.Reference reference = 11;</code>
+       */
+      public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference.Builder getReferenceBuilder() {
+        
+        onChanged();
+        return getReferenceFieldBuilder().getBuilder();
+      }
+      /**
+       * <pre>
+       * Resource reference.
+       * This reference is used to reference the associated resource for data plane
+       * activities such as:
+       * - tagging metrics
+       * </pre>
+       *
+       * <code>.Reference reference = 11;</code>
+       */
+      public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.ReferenceOrBuilder getReferenceOrBuilder() {
+        if (referenceBuilder_ != null) {
+          return referenceBuilder_.getMessageOrBuilder();
+        } else {
+          return reference_ == null ?
+              dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference.getDefaultInstance() : reference_;
+        }
+      }
+      /**
+       * <pre>
+       * Resource reference.
+       * This reference is used to reference the associated resource for data plane
+       * activities such as:
+       * - tagging metrics
+       * </pre>
+       *
+       * <code>.Reference reference = 11;</code>
+       */
+      private com.google.protobuf.SingleFieldBuilderV3<
+          dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference.Builder, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.ReferenceOrBuilder> 
+          getReferenceFieldBuilder() {
+        if (referenceBuilder_ == null) {
+          referenceBuilder_ = new com.google.protobuf.SingleFieldBuilderV3<
+              dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference.Builder, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.ReferenceOrBuilder>(
+                  getReference(),
+                  getParentForChildren(),
+                  isClean());
+          reference_ = null;
+        }
+        return referenceBuilder_;
+      }
       @java.lang.Override
       public final Builder setUnknownFields(
           final com.google.protobuf.UnknownFieldSet unknownFields) {
@@ -16745,51 +17053,52 @@ public final class DataPlaneContract {
       "e\030\002 \001(\t:\0028\001\"\177\n\014EgressConfig\022\022\n\ndeadLette" +
       "r\030\001 \001(\t\022\r\n\005retry\030\002 \001(\r\022%\n\rbackoffPolicy\030" +
       "\003 \001(\0162\016.BackoffPolicy\022\024\n\014backoffDelay\030\004 " +
-      "\001(\004\022\017\n\007timeout\030\005 \001(\004\"\256\002\n\006Egress\022\025\n\rconsu" +
+      "\001(\004\022\017\n\007timeout\030\005 \001(\004\"\315\002\n\006Egress\022\025\n\rconsu" +
       "merGroup\030\001 \001(\t\022\023\n\013destination\030\002 \001(\t\022\022\n\010r" +
       "eplyUrl\030\003 \001(\tH\000\022&\n\024replyToOriginalTopic\030" +
       "\004 \001(\0132\006.EmptyH\000\022\036\n\014discardReply\030\t \001(\0132\006." +
       "EmptyH\000\022\027\n\006filter\030\005 \001(\0132\007.Filter\022\013\n\003uid\030" +
       "\006 \001(\t\022#\n\014egressConfig\030\007 \001(\0132\r.EgressConf" +
       "ig\022%\n\rdeliveryOrder\030\010 \001(\0162\016.DeliveryOrde" +
-      "r\022\031\n\007keyType\030\n \001(\0162\010.KeyTypeB\017\n\rreplyStr" +
-      "ategy\"[\n\007Ingress\022!\n\013contentMode\030\001 \001(\0162\014." +
-      "ContentMode\022\016\n\004path\030\002 \001(\tH\000\022\016\n\004host\030\003 \001(" +
-      "\tH\000B\r\n\013ingressType\"K\n\tReference\022\014\n\004uuid\030" +
-      "\001 \001(\t\022\021\n\tnamespace\030\002 \001(\t\022\014\n\004name\030\003 \001(\t\022\017" +
-      "\n\007version\030\004 \001(\t\"`\n\017SecretReference\022\035\n\tre" +
-      "ference\030\001 \001(\0132\n.Reference\022.\n\022keyFieldRef" +
-      "erences\030\002 \003(\0132\022.KeyFieldReference\"C\n\021Key" +
-      "FieldReference\022\021\n\tsecretKey\030\002 \001(\t\022\033\n\005fie" +
-      "ld\030\003 \001(\0162\014.SecretField\"Y\n\024MultiSecretRef" +
-      "erence\022\033\n\010protocol\030\001 \001(\0162\t.Protocol\022$\n\nr" +
-      "eferences\030\002 \003(\0132\020.SecretReference\"\202\001\n\023Cl" +
-      "oudEventOverrides\0228\n\nextensions\030\001 \003(\0132$." +
-      "CloudEventOverrides.ExtensionsEntry\0321\n\017E" +
-      "xtensionsEntry\022\013\n\003key\030\001 \001(\t\022\r\n\005value\030\002 \001" +
-      "(\t:\0028\001\"\350\002\n\010Resource\022\013\n\003uid\030\001 \001(\t\022\016\n\006topi" +
-      "cs\030\002 \003(\t\022\030\n\020bootstrapServers\030\003 \001(\t\022\031\n\007in" +
-      "gress\030\004 \001(\0132\010.Ingress\022#\n\014egressConfig\030\005 " +
-      "\001(\0132\r.EgressConfig\022\031\n\010egresses\030\006 \003(\0132\007.E" +
-      "gress\022\034\n\nabsentAuth\030\007 \001(\0132\006.EmptyH\000\022 \n\na" +
-      "uthSecret\030\010 \001(\0132\n.ReferenceH\000\0220\n\017multiAu" +
-      "thSecret\030\t \001(\0132\025.MultiSecretReferenceH\000\022" +
-      "1\n\023cloudEventOverrides\030\n \001(\0132\024.CloudEven" +
-      "tOverrides\022\035\n\treference\030\013 \001(\0132\n.Referenc" +
-      "eB\006\n\004Auth\"<\n\010Contract\022\022\n\ngeneration\030\001 \001(" +
-      "\004\022\034\n\tresources\030\002 \003(\0132\t.Resource*,\n\rBacko" +
-      "ffPolicy\022\017\n\013Exponential\020\000\022\n\n\006Linear\020\001*+\n" +
-      "\rDeliveryOrder\022\r\n\tUNORDERED\020\000\022\013\n\007ORDERED" +
-      "\020\001*=\n\007KeyType\022\n\n\006String\020\000\022\013\n\007Integer\020\001\022\n" +
-      "\n\006Double\020\002\022\r\n\tByteArray\020\003*)\n\013ContentMode" +
-      "\022\n\n\006BINARY\020\000\022\016\n\nSTRUCTURED\020\001*a\n\013SecretFi" +
-      "eld\022\022\n\016SASL_MECHANISM\020\000\022\n\n\006CA_CRT\020\001\022\014\n\010U" +
-      "SER_CRT\020\002\022\014\n\010USER_KEY\020\003\022\010\n\004USER\020\004\022\014\n\010PAS" +
-      "SWORD\020\005*D\n\010Protocol\022\r\n\tPLAINTEXT\020\000\022\022\n\016SA" +
-      "SL_PLAINTEXT\020\001\022\007\n\003SSL\020\002\022\014\n\010SASL_SSL\020\003B[\n" +
-      "*dev.knative.eventing.kafka.broker.contr" +
-      "actB\021DataPlaneContractZ\032control-plane/pk" +
-      "g/contractb\006proto3"
+      "r\022\031\n\007keyType\030\n \001(\0162\010.KeyType\022\035\n\treferenc" +
+      "e\030\013 \001(\0132\n.ReferenceB\017\n\rreplyStrategy\"[\n\007" +
+      "Ingress\022!\n\013contentMode\030\001 \001(\0162\014.ContentMo" +
+      "de\022\016\n\004path\030\002 \001(\tH\000\022\016\n\004host\030\003 \001(\tH\000B\r\n\013in" +
+      "gressType\"K\n\tReference\022\014\n\004uuid\030\001 \001(\t\022\021\n\t" +
+      "namespace\030\002 \001(\t\022\014\n\004name\030\003 \001(\t\022\017\n\007version" +
+      "\030\004 \001(\t\"`\n\017SecretReference\022\035\n\treference\030\001" +
+      " \001(\0132\n.Reference\022.\n\022keyFieldReferences\030\002" +
+      " \003(\0132\022.KeyFieldReference\"C\n\021KeyFieldRefe" +
+      "rence\022\021\n\tsecretKey\030\002 \001(\t\022\033\n\005field\030\003 \001(\0162" +
+      "\014.SecretField\"Y\n\024MultiSecretReference\022\033\n" +
+      "\010protocol\030\001 \001(\0162\t.Protocol\022$\n\nreferences" +
+      "\030\002 \003(\0132\020.SecretReference\"\202\001\n\023CloudEventO" +
+      "verrides\0228\n\nextensions\030\001 \003(\0132$.CloudEven" +
+      "tOverrides.ExtensionsEntry\0321\n\017Extensions" +
+      "Entry\022\013\n\003key\030\001 \001(\t\022\r\n\005value\030\002 \001(\t:\0028\001\"\350\002" +
+      "\n\010Resource\022\013\n\003uid\030\001 \001(\t\022\016\n\006topics\030\002 \003(\t\022" +
+      "\030\n\020bootstrapServers\030\003 \001(\t\022\031\n\007ingress\030\004 \001" +
+      "(\0132\010.Ingress\022#\n\014egressConfig\030\005 \001(\0132\r.Egr" +
+      "essConfig\022\031\n\010egresses\030\006 \003(\0132\007.Egress\022\034\n\n" +
+      "absentAuth\030\007 \001(\0132\006.EmptyH\000\022 \n\nauthSecret" +
+      "\030\010 \001(\0132\n.ReferenceH\000\0220\n\017multiAuthSecret\030" +
+      "\t \001(\0132\025.MultiSecretReferenceH\000\0221\n\023cloudE" +
+      "ventOverrides\030\n \001(\0132\024.CloudEventOverride" +
+      "s\022\035\n\treference\030\013 \001(\0132\n.ReferenceB\006\n\004Auth" +
+      "\"<\n\010Contract\022\022\n\ngeneration\030\001 \001(\004\022\034\n\treso" +
+      "urces\030\002 \003(\0132\t.Resource*,\n\rBackoffPolicy\022" +
+      "\017\n\013Exponential\020\000\022\n\n\006Linear\020\001*+\n\rDelivery" +
+      "Order\022\r\n\tUNORDERED\020\000\022\013\n\007ORDERED\020\001*=\n\007Key" +
+      "Type\022\n\n\006String\020\000\022\013\n\007Integer\020\001\022\n\n\006Double\020" +
+      "\002\022\r\n\tByteArray\020\003*)\n\013ContentMode\022\n\n\006BINAR" +
+      "Y\020\000\022\016\n\nSTRUCTURED\020\001*a\n\013SecretField\022\022\n\016SA" +
+      "SL_MECHANISM\020\000\022\n\n\006CA_CRT\020\001\022\014\n\010USER_CRT\020\002" +
+      "\022\014\n\010USER_KEY\020\003\022\010\n\004USER\020\004\022\014\n\010PASSWORD\020\005*D" +
+      "\n\010Protocol\022\r\n\tPLAINTEXT\020\000\022\022\n\016SASL_PLAINT" +
+      "EXT\020\001\022\007\n\003SSL\020\002\022\014\n\010SASL_SSL\020\003B[\n*dev.knat" +
+      "ive.eventing.kafka.broker.contractB\021Data" +
+      "PlaneContractZ\032control-plane/pkg/contrac" +
+      "tb\006proto3"
     };
     descriptor = com.google.protobuf.Descriptors.FileDescriptor
       .internalBuildGeneratedFileFrom(descriptorData,
@@ -16824,7 +17133,7 @@ public final class DataPlaneContract {
     internal_static_Egress_fieldAccessorTable = new
       com.google.protobuf.GeneratedMessageV3.FieldAccessorTable(
         internal_static_Egress_descriptor,
-        new java.lang.String[] { "ConsumerGroup", "Destination", "ReplyUrl", "ReplyToOriginalTopic", "DiscardReply", "Filter", "Uid", "EgressConfig", "DeliveryOrder", "KeyType", "ReplyStrategy", });
+        new java.lang.String[] { "ConsumerGroup", "Destination", "ReplyUrl", "ReplyToOriginalTopic", "DiscardReply", "Filter", "Uid", "EgressConfig", "DeliveryOrder", "KeyType", "Reference", "ReplyStrategy", });
     internal_static_Ingress_descriptor =
       getDescriptor().getMessageTypes().get(4);
     internal_static_Ingress_fieldAccessorTable = new
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 2a3c2a896..f7dab59a0 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
@@ -66,7 +66,7 @@ public class ResourcesReconcilerImpl implements ResourcesReconciler {
   }
 
   @Override
-  public Future<Void> reconcile(final Collection<DataPlaneContract.Resource> newResources) {
+  public synchronized Future<Void> reconcile(final Collection<DataPlaneContract.Resource> newResources) {
     if (isReconcilingIngress()) {
       return reconcileIngress(newResources);
     }
@@ -86,7 +86,7 @@ public class ResourcesReconcilerImpl implements ResourcesReconciler {
     final List<Future> futures = new ArrayList<>(egresses.size() + this.cachedEgresses.size());
 
     final var diff = CollectionsUtils.diff(this.cachedEgresses.keySet(), egresses.keySet());
-    logger.debug("Reconcile egress diff {}", diff);
+    logger.info("Reconcile egress diff {}", diff);
 
     diff.getRemoved().forEach(uid -> {
       final var entry = this.cachedEgresses.get(uid);
@@ -124,7 +124,7 @@ public class ResourcesReconcilerImpl implements ResourcesReconciler {
       final var oldResource = this.cachedResources.get(newResource.getUid());
 
       if (resourceEquals(newResource, oldResource) && egressEquals(newEgress, this.cachedEgresses.get(uid).getKey())) {
-        logger.debug("Nothing changed for egress {} {} {}",
+        logger.info("Nothing changed for egress {} {} {}",
           keyValue("id", newEgress.getUid()),
           keyValue("consumerGroup", newEgress.getConsumerGroup()),
           keyValue("destination", newEgress.getDestination())
@@ -165,7 +165,7 @@ public class ResourcesReconcilerImpl implements ResourcesReconciler {
     final List<Future> futures = new ArrayList<>(newResourcesMap.size() + this.cachedResources.size());
 
     final var diff = CollectionsUtils.diff(this.cachedResources.keySet(), newResourcesMap.keySet());
-    logger.debug("Reconcile ingress diff {}", diff);
+    logger.info("Reconcile ingress diff {}", diff);
 
     diff.getRemoved().stream()
       .map(this.cachedResources::get)
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index 9eb0b1664..0dbfd92ea 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -53,7 +53,7 @@
     <assertj.version>3.22.0</assertj.version>
     <awaitility.version>4.1.1</awaitility.version>
     <junit.jupiter.version>5.8.2</junit.jupiter.version>
-    <mokito.junit.jupiter.version>4.2.0</mokito.junit.jupiter.version>
+    <mokito.junit.jupiter.version>4.3.0</mokito.junit.jupiter.version>
     <fabric8.kubernetes.version>5.11.1</fabric8.kubernetes.version>
     <kafka.version>2.8.1</kafka.version>
     <debezium.version>1.8.0.Final</debezium.version>
diff --git a/proto/contract.proto b/proto/contract.proto
index 543f19938..45e9ed909 100644
--- a/proto/contract.proto
+++ b/proto/contract.proto
@@ -98,6 +98,13 @@ message Egress {
 
   // Kafka record key type.
   KeyType keyType = 10;
+
+  // Resource reference.
+  //
+  // This reference is used to reference the associated resource for data plane
+  // activities such as:
+  // - tagging metrics
+  Reference reference = 11;
 }
 
 // CloudEvent content mode
diff --git a/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake/fake.go b/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake/fake.go
new file mode 100644
index 000000000..eb18a1fac
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake/fake.go
@@ -0,0 +1,40 @@
+/*
+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.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	context "context"
+
+	fake "knative.dev/eventing/pkg/client/injection/informers/factory/fake"
+	subscription "knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+)
+
+var Get = subscription.Get
+
+func init() {
+	injection.Fake.RegisterInformer(withInformer)
+}
+
+func withInformer(ctx context.Context) (context.Context, controller.Informer) {
+	f := fake.Get(ctx)
+	inf := f.Messaging().V1().Subscriptions()
+	return context.WithValue(ctx, subscription.Key{}, inf), inf.Informer()
+}
diff --git a/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/subscription.go b/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/subscription.go
new file mode 100644
index 000000000..37862d8a5
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/subscription.go
@@ -0,0 +1,116 @@
+/*
+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.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package subscription
+
+import (
+	context "context"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	cache "k8s.io/client-go/tools/cache"
+	apismessagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
+	versioned "knative.dev/eventing/pkg/client/clientset/versioned"
+	v1 "knative.dev/eventing/pkg/client/informers/externalversions/messaging/v1"
+	client "knative.dev/eventing/pkg/client/injection/client"
+	factory "knative.dev/eventing/pkg/client/injection/informers/factory"
+	messagingv1 "knative.dev/eventing/pkg/client/listers/messaging/v1"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+func init() {
+	injection.Default.RegisterInformer(withInformer)
+	injection.Dynamic.RegisterDynamicInformer(withDynamicInformer)
+}
+
+// Key is used for associating the Informer inside the context.Context.
+type Key struct{}
+
+func withInformer(ctx context.Context) (context.Context, controller.Informer) {
+	f := factory.Get(ctx)
+	inf := f.Messaging().V1().Subscriptions()
+	return context.WithValue(ctx, Key{}, inf), inf.Informer()
+}
+
+func withDynamicInformer(ctx context.Context) context.Context {
+	inf := &wrapper{client: client.Get(ctx), resourceVersion: injection.GetResourceVersion(ctx)}
+	return context.WithValue(ctx, Key{}, inf)
+}
+
+// Get extracts the typed informer from the context.
+func Get(ctx context.Context) v1.SubscriptionInformer {
+	untyped := ctx.Value(Key{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch knative.dev/eventing/pkg/client/informers/externalversions/messaging/v1.SubscriptionInformer from context.")
+	}
+	return untyped.(v1.SubscriptionInformer)
+}
+
+type wrapper struct {
+	client versioned.Interface
+
+	namespace string
+
+	resourceVersion string
+}
+
+var _ v1.SubscriptionInformer = (*wrapper)(nil)
+var _ messagingv1.SubscriptionLister = (*wrapper)(nil)
+
+func (w *wrapper) Informer() cache.SharedIndexInformer {
+	return cache.NewSharedIndexInformer(nil, &apismessagingv1.Subscription{}, 0, nil)
+}
+
+func (w *wrapper) Lister() messagingv1.SubscriptionLister {
+	return w
+}
+
+func (w *wrapper) Subscriptions(namespace string) messagingv1.SubscriptionNamespaceLister {
+	return &wrapper{client: w.client, namespace: namespace, resourceVersion: w.resourceVersion}
+}
+
+// SetResourceVersion allows consumers to adjust the minimum resourceVersion
+// used by the underlying client.  It is not accessible via the standard
+// lister interface, but can be accessed through a user-defined interface and
+// an implementation check e.g. rvs, ok := foo.(ResourceVersionSetter)
+func (w *wrapper) SetResourceVersion(resourceVersion string) {
+	w.resourceVersion = resourceVersion
+}
+
+func (w *wrapper) List(selector labels.Selector) (ret []*apismessagingv1.Subscription, err error) {
+	lo, err := w.client.MessagingV1().Subscriptions(w.namespace).List(context.TODO(), metav1.ListOptions{
+		LabelSelector:   selector.String(),
+		ResourceVersion: w.resourceVersion,
+	})
+	if err != nil {
+		return nil, err
+	}
+	for idx := range lo.Items {
+		ret = append(ret, &lo.Items[idx])
+	}
+	return ret, nil
+}
+
+func (w *wrapper) Get(name string) (*apismessagingv1.Subscription, error) {
+	return w.client.MessagingV1().Subscriptions(w.namespace).Get(context.TODO(), name, metav1.GetOptions{
+		ResourceVersion: w.resourceVersion,
+	})
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 343796a6e..1b0c9f168 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1179,6 +1179,8 @@ knative.dev/eventing/pkg/client/injection/informers/eventing/v1/trigger
 knative.dev/eventing/pkg/client/injection/informers/eventing/v1/trigger/fake
 knative.dev/eventing/pkg/client/injection/informers/factory
 knative.dev/eventing/pkg/client/injection/informers/factory/fake
+knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription
+knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake
 knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker
 knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/trigger
 knative.dev/eventing/pkg/client/listers/eventing/v1


Test patch:
diff --git a/control-plane/pkg/reconciler/channel/channel_test.go b/control-plane/pkg/reconciler/channel/channel_test.go
index 64f9e2831..ffd577dbc 100644
--- a/control-plane/pkg/reconciler/channel/channel_test.go
+++ b/control-plane/pkg/reconciler/channel/channel_test.go
@@ -22,6 +22,7 @@ import (
 	"testing"
 
 	"github.com/Shopify/sarama"
+	eventingrekttesting "knative.dev/eventing/pkg/reconciler/testing/v1"
 	"knative.dev/pkg/network"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
@@ -350,6 +351,9 @@ func TestReconcileKind(t *testing.T) {
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
+				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
+					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
+				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -372,6 +376,11 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
+								Reference: &contract.Reference{
+									Name:      Subscription1Name,
+									Namespace: ChannelNamespace,
+									Uuid:      Subscription1UUID,
+								},
 							}},
 						},
 					},
@@ -425,6 +434,9 @@ func TestReconcileKind(t *testing.T) {
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
+				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
+					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
+				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -447,6 +459,11 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
+								Reference: &contract.Reference{
+									Name:      Subscription1Name,
+									Namespace: ChannelNamespace,
+									Uuid:      Subscription1UUID,
+								},
 							}},
 						},
 					},
@@ -500,6 +517,9 @@ func TestReconcileKind(t *testing.T) {
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
+				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
+					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
+				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -522,6 +542,11 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
+								Reference: &contract.Reference{
+									Name:      Subscription1Name,
+									Namespace: ChannelNamespace,
+									Uuid:      Subscription1UUID,
+								},
 							}},
 						},
 					},
@@ -578,6 +603,12 @@ func TestReconcileKind(t *testing.T) {
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
+				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
+					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
+				),
+				eventingrekttesting.NewSubscription(Subscription2Name, ChannelNamespace,
+					eventingrekttesting.WithSubscriptionUID(Subscription2UUID),
+				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -600,12 +631,22 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
+								Reference: &contract.Reference{
+									Name:      Subscription1Name,
+									Namespace: ChannelNamespace,
+									Uuid:      Subscription1UUID,
+								},
 							}, {
 								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription2UUID,
 								Destination:   "http://" + Subscription2URI,
 								Uid:           Subscription2UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_DiscardReply{},
+								Reference: &contract.Reference{
+									Name:      Subscription2Name,
+									Namespace: ChannelNamespace,
+									Uuid:      Subscription2UUID,
+								},
 							}},
 						},
 					},
@@ -933,6 +974,9 @@ func TestReconcileKind(t *testing.T) {
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
 				NewSSLSecret(system.Namespace(), "secret-1"),
+				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
+					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
+				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -963,6 +1007,11 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
+								Reference: &contract.Reference{
+									Name:      Subscription1Name,
+									Namespace: ChannelNamespace,
+									Uuid:      Subscription1UUID,
+								},
 							}},
 						},
 					},
@@ -1217,8 +1266,9 @@ func useTable(t *testing.T, table TableTest, env config.Env) {
 				DispatcherLabel:             base.ChannelDispatcherLabel,
 				ReceiverLabel:               base.ChannelReceiverLabel,
 			},
-			Env:             env,
-			ConfigMapLister: listers.GetConfigMapLister(),
+			Env:                env,
+			ConfigMapLister:    listers.GetConfigMapLister(),
+			SubscriptionLister: listers.GetSubscriptionLister(),
 			InitOffsetsFunc: func(ctx context.Context, kafkaClient sarama.Client, kafkaAdminClient sarama.ClusterAdmin, topics []string, consumerGroup string) (int32, error) {
 				return 1, nil
 			},
diff --git a/control-plane/pkg/reconciler/channel/controller_test.go b/control-plane/pkg/reconciler/channel/controller_test.go
index cc3ca16ce..f3a49c227 100644
--- a/control-plane/pkg/reconciler/channel/controller_test.go
+++ b/control-plane/pkg/reconciler/channel/controller_test.go
@@ -24,6 +24,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	fakekubeclientset "k8s.io/client-go/kubernetes/fake"
 	_ "knative.dev/eventing-kafka/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/fake"
+	_ "knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake"
 	_ "knative.dev/pkg/client/injection/ducks/duck/v1/addressable/fake"
 	fakekubeclient "knative.dev/pkg/client/injection/kube/client/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/fake"
diff --git a/control-plane/pkg/reconciler/consumergroup/evictor_test.go b/control-plane/pkg/reconciler/consumergroup/evictor_test.go
new file mode 100644
index 000000000..e334ede0d
--- /dev/null
+++ b/control-plane/pkg/reconciler/consumergroup/evictor_test.go
@@ -0,0 +1,155 @@
+/*
+ * 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 consumergroup
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/pointer"
+	eventingduckv1alpha1 "knative.dev/eventing/pkg/apis/duck/v1alpha1"
+	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
+	reconcilertesting "knative.dev/pkg/reconciler/testing"
+
+	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
+	_ "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client/fake"
+	kafkainternalsclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client/fake"
+)
+
+func TestNewEvictor(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	require.NotPanics(t, func() { newEvictor(ctx, zap.String("k", "n")) })
+}
+
+func TestEvictorEvictSuccess(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "name"},
+	}
+	cg := &kafkainternals.ConsumerGroup{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "ns-1", Name: "cg-name"},
+		Status: kafkainternals.ConsumerGroupStatus{
+			PlaceableStatus: eventingduckv1alpha1.PlaceableStatus{Placeable: eventingduckv1alpha1.Placeable{
+				MaxAllowedVReplicas: pointer.Int32(1),
+				Placements: []eventingduckv1alpha1.Placement{
+					{PodName: pod.GetName(), VReplicas: 1},
+				},
+			}},
+		},
+	}
+	cg.GetConditionSet().Manage(cg.GetStatus()).InitializeConditions()
+	cg.MarkScheduleSucceeded()
+
+	placement := &eventingduckv1alpha1.Placement{PodName: pod.GetName(), VReplicas: 1}
+
+	ctx, _ = kubeclient.With(ctx, pod)
+	ctx, _ = kafkainternalsclient.With(ctx, cg)
+
+	e := newEvictor(ctx)
+	err := e.evict(pod, cg, placement)
+
+	require.Nil(t, err)
+}
+
+func TestEvictorEvictSuccessConsumerGroupSchedulingInProgress(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "name"},
+	}
+	cg := &kafkainternals.ConsumerGroup{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "ns-1", Name: "cg-name"},
+		Status: kafkainternals.ConsumerGroupStatus{
+			PlaceableStatus: eventingduckv1alpha1.PlaceableStatus{Placeable: eventingduckv1alpha1.Placeable{
+				MaxAllowedVReplicas: pointer.Int32(1),
+				Placements: []eventingduckv1alpha1.Placement{
+					{PodName: pod.GetName(), VReplicas: 1},
+				},
+			}},
+		},
+	}
+	cg.GetConditionSet().Manage(cg.GetStatus()).InitializeConditions()
+
+	placement := &eventingduckv1alpha1.Placement{PodName: pod.GetName(), VReplicas: 1}
+
+	ctx, _ = kubeclient.With(ctx, pod)
+	ctx, _ = kafkainternalsclient.With(ctx, cg)
+
+	e := newEvictor(ctx)
+	err := e.evict(pod, cg, placement)
+
+	require.True(t, cg.IsNotScheduled())
+	require.Nil(t, err)
+}
+
+func TestEvictorEvictPodNotFound(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "name"},
+	}
+	cg := &kafkainternals.ConsumerGroup{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "ns-1", Name: "cg-name"},
+		Status: kafkainternals.ConsumerGroupStatus{
+			PlaceableStatus: eventingduckv1alpha1.PlaceableStatus{Placeable: eventingduckv1alpha1.Placeable{
+				MaxAllowedVReplicas: pointer.Int32(1),
+				Placements: []eventingduckv1alpha1.Placement{
+					{PodName: pod.GetName(), VReplicas: 1},
+				},
+			}},
+		},
+	}
+	placement := &eventingduckv1alpha1.Placement{PodName: pod.GetName(), VReplicas: 1}
+
+	ctx, _ = kafkainternalsclient.With(ctx, cg)
+
+	e := newEvictor(ctx)
+	err := e.evict(pod, cg, placement)
+
+	require.NotNil(t, err)
+}
+func TestEvictorEvictConsumerGroupNotFound(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "name"},
+	}
+	cg := &kafkainternals.ConsumerGroup{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "ns-1", Name: "cg-name"},
+		Status: kafkainternals.ConsumerGroupStatus{
+			PlaceableStatus: eventingduckv1alpha1.PlaceableStatus{Placeable: eventingduckv1alpha1.Placeable{
+				MaxAllowedVReplicas: pointer.Int32(1),
+				Placements: []eventingduckv1alpha1.Placement{
+					{PodName: pod.GetName(), VReplicas: 1},
+				},
+			}},
+		},
+	}
+	placement := &eventingduckv1alpha1.Placement{PodName: pod.GetName(), VReplicas: 1}
+
+	ctx, _ = kubeclient.With(ctx, pod)
+
+	e := newEvictor(ctx)
+	err := e.evict(pod, cg, placement)
+
+	require.Nil(t, err)
+}
diff --git a/control-plane/pkg/reconciler/source/source_test.go b/control-plane/pkg/reconciler/source/source_test.go
index d9b86eca8..b3736ca3a 100644
--- a/control-plane/pkg/reconciler/source/source_test.go
+++ b/control-plane/pkg/reconciler/source/source_test.go
@@ -105,10 +105,11 @@ func TestReconcileKind(t *testing.T) {
 									EgressConfig:  &DefaultEgressConfig,
 									DeliveryOrder: DefaultDeliveryOrder,
 									ReplyStrategy: &contract.Egress_DiscardReply{},
+									Reference:     SourceReference(),
 								},
 							},
 							Auth:      &contract.Resource_AbsentAuth{},
-							Reference: &contract.Reference{Namespace: SourceNamespace, Name: SourceName},
+							Reference: SourceReference(),
 						},
 					},
 				}),
@@ -170,10 +171,11 @@ func TestReconcileKind(t *testing.T) {
 									EgressConfig:  &DefaultEgressConfig,
 									DeliveryOrder: DefaultDeliveryOrder,
 									ReplyStrategy: &contract.Egress_DiscardReply{},
+									Reference:     SourceReference(),
 								},
 							},
 							Auth:      &contract.Resource_AbsentAuth{},
-							Reference: &contract.Reference{Namespace: SourceNamespace, Name: SourceName},
+							Reference: SourceReference(),
 						},
 					},
 				}),
@@ -234,10 +236,11 @@ func TestReconcileKind(t *testing.T) {
 									DeliveryOrder: DefaultDeliveryOrder,
 									KeyType:       contract.KeyType_String,
 									ReplyStrategy: &contract.Egress_DiscardReply{},
+									Reference:     SourceReference(),
 								},
 							},
 							Auth:      &contract.Resource_AbsentAuth{},
-							Reference: &contract.Reference{Namespace: SourceNamespace, Name: SourceName},
+							Reference: SourceReference(),
 						},
 					},
 				}),
@@ -296,10 +299,11 @@ func TestReconcileKind(t *testing.T) {
 									DeliveryOrder: DefaultDeliveryOrder,
 									KeyType:       contract.KeyType_Integer,
 									ReplyStrategy: &contract.Egress_DiscardReply{},
+									Reference:     SourceReference(),
 								},
 							},
 							Auth:      &contract.Resource_AbsentAuth{},
-							Reference: &contract.Reference{Namespace: SourceNamespace, Name: SourceName},
+							Reference: SourceReference(),
 						},
 					},
 				}),
@@ -358,10 +362,11 @@ func TestReconcileKind(t *testing.T) {
 									DeliveryOrder: DefaultDeliveryOrder,
 									KeyType:       contract.KeyType_ByteArray,
 									ReplyStrategy: &contract.Egress_DiscardReply{},
+									Reference:     SourceReference(),
 								},
 							},
 							Auth:      &contract.Resource_AbsentAuth{},
-							Reference: &contract.Reference{Namespace: SourceNamespace, Name: SourceName},
+							Reference: SourceReference(),
 						},
 					},
 				}),
@@ -420,10 +425,11 @@ func TestReconcileKind(t *testing.T) {
 									DeliveryOrder: DefaultDeliveryOrder,
 									KeyType:       contract.KeyType_Double,
 									ReplyStrategy: &contract.Egress_DiscardReply{},
+									Reference:     SourceReference(),
 								},
 							},
 							Auth:      &contract.Resource_AbsentAuth{},
-							Reference: &contract.Reference{Namespace: SourceNamespace, Name: SourceName},
+							Reference: SourceReference(),
 						},
 					},
 				}),
diff --git a/control-plane/pkg/reconciler/testing/listers.go b/control-plane/pkg/reconciler/testing/listers.go
index 779496b29..3fd1175ef 100644
--- a/control-plane/pkg/reconciler/testing/listers.go
+++ b/control-plane/pkg/reconciler/testing/listers.go
@@ -24,8 +24,10 @@ import (
 	corelisters "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/tools/cache"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
+	messaging "knative.dev/eventing/pkg/apis/messaging/v1"
 	fakeeventingclientset "knative.dev/eventing/pkg/client/clientset/versioned/fake"
 	eventinglisters "knative.dev/eventing/pkg/client/listers/eventing/v1"
+	messaginglisters "knative.dev/eventing/pkg/client/listers/messaging/v1"
 	"knative.dev/pkg/reconciler/testing"
 
 	eventingkafkachannels "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
@@ -120,6 +122,10 @@ func (l *Listers) GetTriggerLister() eventinglisters.TriggerLister {
 	return eventinglisters.NewTriggerLister(l.indexerFor(&eventing.Trigger{}))
 }
 
+func (l *Listers) GetSubscriptionLister() messaginglisters.SubscriptionLister {
+	return messaginglisters.NewSubscriptionLister(l.indexerFor(&messaging.Subscription{}))
+}
+
 func (l *Listers) GetConfigMapLister() corelisters.ConfigMapLister {
 	return corelisters.NewConfigMapLister(l.indexerFor(&corev1.ConfigMap{}))
 }
diff --git a/control-plane/pkg/reconciler/testing/objects_broker.go b/control-plane/pkg/reconciler/testing/objects_broker.go
index 5e1e368bf..7ab0e6936 100644
--- a/control-plane/pkg/reconciler/testing/objects_broker.go
+++ b/control-plane/pkg/reconciler/testing/objects_broker.go
@@ -47,6 +47,9 @@ const (
 	BrokerNamespace = "test-namespace"
 	BrokerName      = "test-broker"
 
+	TriggerName      = "test-trigger"
+	TriggerNamespace = "test-namespace"
+
 	ConfigMapFinalizerName = "kafka.brokers.eventing.knative.dev/" + BrokerNamespace + "-" + BrokerName
 )
 
@@ -240,6 +243,14 @@ func BrokerReference() *contract.Reference {
 	}
 }
 
+func TriggerReference() *contract.Reference {
+	return &contract.Reference{
+		Uuid:      TriggerUUID,
+		Namespace: TriggerNamespace,
+		Name:      TriggerName,
+	}
+}
+
 func BrokerDLSResolved(uri string) func(broker *eventing.Broker) {
 	return func(broker *eventing.Broker) {
 		broker.Status.DeadLetterSinkURI, _ = apis.ParseURL(uri)
diff --git a/control-plane/pkg/reconciler/testing/objects_source.go b/control-plane/pkg/reconciler/testing/objects_source.go
index 04ce2639f..bb79545db 100644
--- a/control-plane/pkg/reconciler/testing/objects_source.go
+++ b/control-plane/pkg/reconciler/testing/objects_source.go
@@ -25,6 +25,7 @@ import (
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 )
 
@@ -156,3 +157,11 @@ func StatusSourceSinkNotResolved(err string) KRShapedOption {
 		)
 	}
 }
+
+func SourceReference() *contract.Reference {
+	return &contract.Reference{
+		Namespace: SourceNamespace,
+		Name:      SourceName,
+		Uuid:      SourceUUID,
+	}
+}
diff --git a/control-plane/pkg/reconciler/trigger/trigger_test.go b/control-plane/pkg/reconciler/trigger/trigger_test.go
index 437bdb85a..114f6a7db 100644
--- a/control-plane/pkg/reconciler/trigger/trigger_test.go
+++ b/control-plane/pkg/reconciler/trigger/trigger_test.go
@@ -24,12 +24,13 @@ import (
 
 	"google.golang.org/protobuf/testing/protocmp"
 	"k8s.io/utils/pointer"
-	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/tracker"
 
+	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 
 	"github.com/google/go-cmp/cmp"
@@ -53,13 +54,6 @@ import (
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/testing"
 )
 
-const (
-	// name of the trigger under test
-	triggerName = "test-trigger"
-	// namespace of the trigger under test
-	triggerNamespace = "test-namespace"
-)
-
 var DefaultEnv = &config.Env{
 	DataPlaneConfigMapNamespace: "knative-eventing",
 	DataPlaneConfigMapName:      "kafka-broker-brokers-triggers",
@@ -74,7 +68,7 @@ var (
 	finalizerUpdatedEvent = Eventf(
 		corev1.EventTypeNormal,
 		"FinalizerUpdate",
-		fmt.Sprintf(`Updated %q finalizers`, triggerName),
+		fmt.Sprintf(`Updated %q finalizers`, TriggerName),
 	)
 
 	url = &apis.URL{
@@ -98,7 +92,7 @@ func TestTriggerReconciler(t *testing.T) {
 
 func triggerReconciliation(t *testing.T, format string, env config.Env) {
 
-	testKey := fmt.Sprintf("%s/%s", triggerNamespace, triggerName)
+	testKey := fmt.Sprintf("%s/%s", TriggerNamespace, TriggerName)
 
 	env.DataPlaneConfigFormat = format
 
@@ -141,6 +135,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Reference:     TriggerReference(),
 								},
 							},
 						},
@@ -205,6 +200,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
 									EgressConfig:  &contract.EgressConfig{DeadLetter: ServiceURL},
+									Reference:     TriggerReference(),
 								},
 							},
 						},
@@ -267,6 +263,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Reference:     TriggerReference(),
 									EgressConfig: &contract.EgressConfig{
 										DeadLetter:    url.String(),
 										Retry:         3,
@@ -337,6 +334,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Reference:     TriggerReference(),
 									DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								},
 							},
@@ -401,6 +399,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Reference:     TriggerReference(),
 									DeliveryOrder: contract.DeliveryOrder_UNORDERED,
 								},
 							},
@@ -465,6 +464,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Reference:     TriggerReference(),
 									EgressConfig: &contract.EgressConfig{
 										DeadLetter:    url.String(),
 										Retry:         3,
@@ -517,6 +517,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Reference:     TriggerReference(),
 								},
 							},
 						},
@@ -540,6 +541,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Reference:     TriggerReference(),
 								},
 							},
 						},
@@ -567,6 +569,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Reference:     TriggerReference(),
 								},
 							},
 						},
@@ -587,6 +590,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Reference:     TriggerReference(),
 								},
 							},
 						},
@@ -807,6 +811,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Reference:     TriggerReference(),
 								},
 							},
 						},
@@ -874,6 +879,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Reference:     TriggerReference(),
 									Filter: &contract.Filter{Attributes: map[string]string{
 										"type": "type1",
 									}},
@@ -932,6 +938,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Reference:     TriggerReference(),
 								},
 								{
 									Destination:   "http://example.com/1",
@@ -1012,6 +1019,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Reference:     TriggerReference(),
 									Filter: &contract.Filter{Attributes: map[string]string{
 										"type": "type1",
 									}},
@@ -1248,6 +1256,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Reference:     TriggerReference(),
 									Filter: &contract.Filter{Attributes: map[string]string{
 										"ext": "extval",
 									}},
@@ -1459,6 +1468,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Reference:     TriggerReference(),
 								},
 								{
 									Destination:   "http://example.com/3",
@@ -1580,7 +1590,7 @@ func TestTriggerFinalizer(t *testing.T) {
 
 func triggerFinalizer(t *testing.T, format string, env config.Env) {
 
-	testKey := fmt.Sprintf("%s/%s", triggerNamespace, triggerName)
+	testKey := fmt.Sprintf("%s/%s", TriggerNamespace, TriggerName)
 
 	env.DataPlaneConfigFormat = format
 
@@ -2316,8 +2326,8 @@ func useTable(t *testing.T, table TableTest, env *config.Env) {
 
 func newTrigger(options ...reconcilertesting.TriggerOption) runtime.Object {
 	return reconcilertesting.NewTrigger(
-		triggerName,
-		triggerNamespace,
+		TriggerName,
+		TriggerNamespace,
 		BrokerName,
 		append(
 			options,
@@ -2368,8 +2378,8 @@ func withTriggerSubscriberResolvedSucceeded(deliveryOrder contract.DeliveryOrder
 
 func patchFinalizers() clientgotesting.PatchActionImpl {
 	action := clientgotesting.PatchActionImpl{}
-	action.Name = triggerName
-	action.Namespace = triggerNamespace
+	action.Name = TriggerName
+	action.Namespace = TriggerNamespace
 	patch := `{"metadata":{"finalizers":["` + FinalizerName + `"],"resourceVersion":""}}`
 	action.Patch = []byte(patch)
 	return action
diff --git a/test/config/cm-watcher.yaml b/test/config/cm-watcher-broker.yaml
similarity index 100%
rename from test/config/cm-watcher.yaml
rename to test/config/cm-watcher-broker.yaml
diff --git a/test/config/cm-watcher-sink.yaml b/test/config/cm-watcher-sink.yaml
new file mode 100644
index 000000000..e98238b3d
--- /dev/null
+++ b/test/config/cm-watcher-sink.yaml
@@ -0,0 +1,80 @@
+# Copyright 2020 The Knative Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: test-kafka-sink-cm-watcher
+  namespace: knative-eventing
+  labels:
+    app: test-kafka-sink-cm-watcher
+    kafka.eventing.knative.dev/release: devel
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: test-kafka-sink-cm-watcher
+  template:
+    metadata:
+      name: test-kafka-sink-cm-watcher
+      labels:
+        app: test-kafka-sink-cm-watcher
+        kafka.eventing.knative.dev/release: devel
+    spec:
+      securityContext:
+        runAsNonRoot: true
+      serviceAccountName: kafka-controller # use the same service account of our controller
+      containers:
+        - name: test-kafka-sink-cm-watcher
+          image: ko://knative.dev/eventing-kafka-broker/test/cmd/watch-cm
+          imagePullPolicy: IfNotPresent
+          volumeMounts:
+            - name: config-logging
+              mountPath: /etc/config-logging
+              readOnly: true
+          env:
+            - name: DATA_PLANE_CONFIG_MAP_NAMESPACE
+              value: knative-eventing
+            - name: DATA_PLANE_CONFIG_MAP_NAME
+              value: kafka-sink-sinks
+            - name: CONFIG_LOGGING_NAME
+              value: config-logging
+            - name: CONFIG_OBSERVABILITY_NAME
+              value: config-observability
+            - name: METRICS_DOMAIN
+              value: knative.dev/eventing
+            - name: DATA_PLANE_CONFIG_FORMAT
+              value: json
+            - name: INGRESS_NAME
+              value: kafka-sink-ingress
+            - name: GENERAL_CONFIG_MAP_NAME
+              value: kafka-sink-config
+            - name: SYSTEM_NAMESPACE
+              valueFrom:
+                fieldRef:
+                  fieldPath: metadata.namespace
+          ports:
+            - containerPort: 9090
+              name: metrics
+          terminationMessagePolicy: FallbackToLogsOnError
+          terminationMessagePath: /dev/temination-log
+          securityContext:
+            allowPrivilegeEscalation: false
+            privileged: false
+            readOnlyRootFilesystem: true
+      volumes:
+        - name: config-logging
+          configMap:
+            name: config-logging
+      restartPolicy: Always
diff --git a/test/upgrade-tests.sh b/test/upgrade-tests.sh
index 954548670..c2f66792c 100755
--- a/test/upgrade-tests.sh
+++ b/test/upgrade-tests.sh
@@ -35,6 +35,8 @@ if ! ${SKIP_INITIALIZE}; then
   save_release_artifacts || fail_test "Failed to save release artifacts"
 fi
 
+export_logs_continuously
+
 set -Eeuo pipefail
 
 TIMEOUT=${TIMEOUT:-60m}
@@ -44,7 +46,7 @@ EVENTING_KAFKA_BROKER_UPGRADE_TESTS_FINISHEDSLEEP="5m" \
 go_test_e2e -v \
   -tags=upgrade \
   -timeout="${TIMEOUT}" \
-  ./test/upgrade \
-  || fail_test
+  ./test/upgrade ||
+  fail_test
 
 success
diff --git a/third_party/eventing-latest/eventing-core.yaml b/third_party/eventing-latest/eventing-core.yaml
index fc912c3f7..83f275c51 100644
--- a/third_party/eventing-latest/eventing-core.yaml
+++ b/third_party/eventing-latest/eventing-core.yaml
@@ -16,8 +16,8 @@ kind: Namespace
 metadata:
   name: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 
 ---
@@ -41,8 +41,8 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 ---
 apiVersion: rbac.authorization.k8s.io/v1
@@ -50,8 +50,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -67,8 +67,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-resolver
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -84,8 +84,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-source-observer
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -101,8 +101,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-sources-controller
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -118,8 +118,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-manipulator
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -151,8 +151,8 @@ metadata:
   name: pingsource-mt-adapter
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 ---
 apiVersion: rbac.authorization.k8s.io/v1
@@ -160,8 +160,8 @@ kind: ClusterRoleBinding
 metadata:
   name: knative-eventing-pingsource-mt-adapter
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -193,8 +193,8 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 ---
 apiVersion: rbac.authorization.k8s.io/v1
@@ -202,8 +202,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -220,8 +220,8 @@ metadata:
   namespace: knative-eventing
   name: eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -237,8 +237,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-webhook-resolver
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -254,8 +254,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-webhook-podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -287,8 +287,8 @@ metadata:
   name: config-br-default-channel
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 data:
   channel-template-spec: |
@@ -316,8 +316,8 @@ metadata:
   name: config-br-defaults
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 data:
   # Configures the default for any Broker that does not specify a spec.config or Broker class.
@@ -328,6 +328,10 @@ data:
       kind: ConfigMap
       name: config-br-default-channel
       namespace: knative-eventing
+      delivery:
+        retry: 10
+        backoffPolicy: exponential
+        backoffDelay: PT0.2S
 
 ---
 # Copyright 2018 The Knative Authors
@@ -350,8 +354,8 @@ metadata:
   name: default-ch-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 data:
   # Configuration for defaulting channels that do not specify CRD implementations.
@@ -385,10 +389,10 @@ metadata:
   name: config-ping-defaults
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
   annotations:
     knative.dev/example-checksum: "9185c153"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 data:
   _example: |
@@ -432,10 +436,10 @@ metadata:
   name: config-features
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 data:
   # ALPHA feature: The kreference-group allows you to use the Group field in KReferences.
@@ -479,7 +483,7 @@ metadata:
   name: config-kreference-mapping
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
   annotations:
@@ -528,8 +532,8 @@ metadata:
   name: config-leader-election
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
   annotations:
     knative.dev/example-checksum: "f7948630"
@@ -590,10 +594,10 @@ metadata:
   name: config-logging
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 data:
   # Common configuration for all Knative codebase
@@ -644,10 +648,10 @@ metadata:
   name: config-observability
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
   annotations:
     knative.dev/example-checksum: "f46cf09d"
@@ -721,10 +725,10 @@ metadata:
   name: config-tracing
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
   annotations:
     knative.dev/example-checksum: "0492ceb0"
@@ -779,10 +783,10 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     knative.dev/high-availability: "true"
     app.kubernetes.io/component: eventing-controller
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 spec:
   selector:
@@ -792,9 +796,9 @@ spec:
     metadata:
       labels:
         app: eventing-controller
-        eventing.knative.dev/release: "v20220118-f6fec1088"
+        eventing.knative.dev/release: "v20220122-f07487898"
         app.kubernetes.io/component: eventing-autoscaler
-        app.kubernetes.io/version: "20220118-f6fec1088"
+        app.kubernetes.io/version: "20220122-f07487898"
         app.kubernetes.io/name: knative-eventing
     spec:
       # To avoid node becoming SPOF, spread our replicas to different nodes.
@@ -812,7 +816,7 @@ spec:
       containers:
         - name: eventing-controller
           terminationMessagePolicy: FallbackToLogsOnError
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:9328449502724216bd52e31dc13d25818b3d3c93f537717eb161ff85926a7ef4
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:d51f4e8a1e19a657fdccde63a465d4a4fd00240faa73d847e13d8c96f327ce87
           resources:
             requests:
               cpu: 100m
@@ -830,7 +834,7 @@ spec:
               value: knative.dev/eventing
             # APIServerSource
             - name: APISERVER_RA_IMAGE
-              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:033a012326798e22f0fb6f707c5f15b6b1b601eb76b1b7d7b2712cca9b1da058
+              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:c0bde79565c48e297ebf9c7072b205a7127a5090ca2a3e9285f6456555a45bee
             - name: POD_NAME
               valueFrom:
                 fieldRef:
@@ -879,9 +883,9 @@ metadata:
   name: pingsource-mt-adapter
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     app.kubernetes.io/component: pingsource-mt-adapter
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 spec:
   # when set to 0 (and only 0) will be set to 1 when the first PingSource is created.
@@ -894,9 +898,9 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20220118-f6fec1088"
+        eventing.knative.dev/release: "v20220122-f07487898"
         app.kubernetes.io/component: pingsource-mt-adapter
-        app.kubernetes.io/version: "20220118-f6fec1088"
+        app.kubernetes.io/version: "20220122-f07487898"
         app.kubernetes.io/name: knative-eventing
     spec:
       affinity:
@@ -910,7 +914,7 @@ spec:
       enableServiceLinks: false
       containers:
         - name: dispatcher
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:11f8c49f7531debd5348a945522fbe9634890804528a53bf2b3148d4831f166b
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:17f30435a9f39d3121263186c90927a869437d7d7babe51fe1fbecfcececc9fe
           env:
             - name: SYSTEM_NAMESPACE
               value: ''
@@ -975,9 +979,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 spec:
   scaleTargetRef:
@@ -1001,9 +1005,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 spec:
   minAvailable: 80%
@@ -1032,9 +1036,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 spec:
   selector:
@@ -1045,9 +1049,9 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20220118-f6fec1088"
+        eventing.knative.dev/release: "v20220122-f07487898"
         app.kubernetes.io/component: eventing-webhook
-        app.kubernetes.io/version: "20220118-f6fec1088"
+        app.kubernetes.io/version: "20220122-f07487898"
         app.kubernetes.io/name: knative-eventing
     spec:
       # To avoid node becoming SPOF, spread our replicas to different nodes.
@@ -1067,7 +1071,7 @@ spec:
           terminationMessagePolicy: FallbackToLogsOnError
           # This is the Go import path for the binary that is containerized
           # and substituted here.
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/webhook@sha256:0e8afc747d9adfa2c66596713f530dd0faa0eab760c4ac4191f68c51378baf43
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/webhook@sha256:ace72d0992e9526b1ea7825eb7647809a4eeb3f6d1f71ff36fc10c92dfd11e0b
           resources:
             requests:
               # taken from serving.
@@ -1136,10 +1140,10 @@ apiVersion: v1
 kind: Service
 metadata:
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     role: eventing-webhook
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
   name: eventing-webhook
   namespace: knative-eventing
@@ -1171,11 +1175,11 @@ kind: CustomResourceDefinition
 metadata:
   creationTimestamp: null
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -1387,10 +1391,10 @@ kind: CustomResourceDefinition
 metadata:
   name: brokers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -1554,11 +1558,11 @@ kind: CustomResourceDefinition
 metadata:
   name: channels.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -1826,11 +1830,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
   name: containersources.sources.knative.dev
 spec:
@@ -1980,9 +1984,9 @@ kind: CustomResourceDefinition
 metadata:
   name: eventtypes.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -2109,10 +2113,10 @@ kind: CustomResourceDefinition
 metadata:
   name: parallels.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: flows.knative.dev
@@ -2364,11 +2368,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -2555,10 +2559,10 @@ kind: CustomResourceDefinition
 metadata:
   name: sequences.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: flows.knative.dev
@@ -2875,12 +2879,12 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
   name: sinkbindings.sources.knative.dev
 spec:
@@ -3067,9 +3071,9 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -3279,9 +3283,9 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -3465,8 +3469,8 @@ kind: ClusterRole
 metadata:
   name: addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -3479,9 +3483,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: service-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3499,9 +3503,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: serving-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3522,9 +3526,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: channel-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3549,9 +3553,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: broker-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3570,9 +3574,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: flows-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3608,8 +3612,8 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-filter
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3635,8 +3639,8 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-ingress
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3653,8 +3657,8 @@ kind: ClusterRole
 metadata:
   name: eventing-config-reader
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3687,8 +3691,8 @@ kind: ClusterRole
 metadata:
   name: channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -3701,9 +3705,9 @@ kind: ClusterRole
 metadata:
   name: meta-channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     duck.knative.dev/channelable: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "channelable-manipulator" role.
 rules:
@@ -3741,9 +3745,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-eventing-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev"]
@@ -3755,9 +3759,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-messaging-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["messaging.knative.dev"]
@@ -3769,9 +3773,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-flows-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["flows.knative.dev"]
@@ -3783,9 +3787,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-sources-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["sources.knative.dev"]
@@ -3797,9 +3801,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-bindings-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["bindings.knative.dev"]
@@ -3812,8 +3816,8 @@ metadata:
   name: knative-eventing-namespaced-edit
   labels:
     rbac.authorization.k8s.io/aggregate-to-edit: "true"
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
@@ -3826,8 +3830,8 @@ metadata:
   name: knative-eventing-namespaced-view
   labels:
     rbac.authorization.k8s.io/aggregate-to-view: "true"
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
@@ -3854,8 +3858,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-controller
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3990,8 +3994,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-pingsource-mt-adapter
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -4058,8 +4062,8 @@ kind: ClusterRole
 metadata:
   name: podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -4072,9 +4076,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: builtin-podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     duck.knative.dev/podspecable: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "podspecable-binding role.
 rules:
@@ -4120,8 +4124,8 @@ kind: ClusterRole
 metadata:
   name: source-observer
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -4134,9 +4138,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: eventing-sources-source-observer
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     duck.knative.dev/source: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "source-observer" role.
 rules:
@@ -4172,8 +4176,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-sources-controller
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -4259,8 +4263,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 rules:
   # For watching logging configuration and getting certs.
@@ -4359,8 +4363,8 @@ metadata:
   namespace: knative-eventing
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 rules:
   # For manipulating certs into secrets.
@@ -4396,8 +4400,8 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: config.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4434,8 +4438,8 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4468,8 +4472,8 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: validation.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4503,8 +4507,8 @@ metadata:
   name: eventing-webhook-certs
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 # The data is populated at install time.
 
@@ -4528,8 +4532,8 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: sinkbindings.webhook.sources.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
diff --git a/third_party/eventing-latest/eventing-crds.yaml b/third_party/eventing-latest/eventing-crds.yaml
index 26eebdebe..9c9f22eff 100644
--- a/third_party/eventing-latest/eventing-crds.yaml
+++ b/third_party/eventing-latest/eventing-crds.yaml
@@ -17,11 +17,11 @@ kind: CustomResourceDefinition
 metadata:
   creationTimestamp: null
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -233,10 +233,10 @@ kind: CustomResourceDefinition
 metadata:
   name: brokers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -400,11 +400,11 @@ kind: CustomResourceDefinition
 metadata:
   name: channels.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -672,11 +672,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
   name: containersources.sources.knative.dev
 spec:
@@ -826,9 +826,9 @@ kind: CustomResourceDefinition
 metadata:
   name: eventtypes.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -955,10 +955,10 @@ kind: CustomResourceDefinition
 metadata:
   name: parallels.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: flows.knative.dev
@@ -1210,11 +1210,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -1401,10 +1401,10 @@ kind: CustomResourceDefinition
 metadata:
   name: sequences.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: flows.knative.dev
@@ -1721,12 +1721,12 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
   name: sinkbindings.sources.knative.dev
 spec:
@@ -1913,9 +1913,9 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -2125,9 +2125,9 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220118-f6fec1088"
+    eventing.knative.dev/release: "v20220122-f07487898"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220118-f6fec1088"
+    app.kubernetes.io/version: "20220122-f07487898"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev