Repository: knative-sandbox/eventing-kafka-broker
Pull request #1886 (2022-02-09 20:02:05)

[data-plane]: Bump slf4j.version from 1.7.35 to 1.7.36 in /data-plane
Bumps `slf4j.version` from 1.7.35 to 1.7.36.
Updates `slf4j-api` from 1.7.35 to 1.7.36
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/qos-ch/slf4j/commit/e9ee55cca93c2bf26f14482a9bdf961c750d2a56"><code>e9ee55c</code></a> update timestamp before release</li>
<li><a href="https://github.com/qos-ch/slf4j/commit/3b6a92f66815605ce851671dd1d6e0758c29ed4c"><code>3b6a92f</code></a> reproducible builds</li>
<li><a href="https://github.com/qos-ch/slf4j/commit/882502706063b460997a3c6018223018f36cecd3"><code>8825027</code></a> prepare release 1.7.36, fix SLF4J-541</li>
<li><a href="https://github.com/qos-ch/slf4j/commit/561738747da5813b8a9d2e2ebd10f0128a78a960"><code>5617387</code></a> add relase.sh</li>
<li>See full diff in <a href="https://github.com/qos-ch/slf4j/compare/v_1.7.35...v_1.7.36">compare view</a></li>
</ul>
</details>
<br />

Updates `slf4j-nop` from 1.7.35 to 1.7.36
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/qos-ch/slf4j/commit/e9ee55cca93c2bf26f14482a9bdf961c750d2a56"><code>e9ee55c</code></a> update timestamp before release</li>
<li><a href="https://github.com/qos-ch/slf4j/commit/3b6a92f66815605ce851671dd1d6e0758c29ed4c"><code>3b6a92f</code></a> reproducible builds</li>
<li><a href="https://github.com/qos-ch/slf4j/commit/882502706063b460997a3c6018223018f36cecd3"><code>8825027</code></a> prepare release 1.7.36, fix SLF4J-541</li>
<li><a href="https://github.com/qos-ch/slf4j/commit/561738747da5813b8a9d2e2ebd10f0128a78a960"><code>5617387</code></a> add relase.sh</li>
<li>See full diff in <a href="https://github.com/qos-ch/slf4j/compare/v_1.7.35...v_1.7.36">compare view</a></li>
</ul>
</details>
<br />


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/1886#" 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 **matzew** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @matzew` 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":["matzew"]} -->
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>

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1886?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1886](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1886?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (d9766ba) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/90275cbf2d9c665551912bdf68d60cf41f94a959?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (90275cb) will **increase** coverage by `0.08%`.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1886/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/1886?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    #1886      +/-   ##
============================================
+ Coverage     66.87%   66.96%   +0.08%     
  Complexity      614      614              
============================================
  Files           135      135              
  Lines          5748     5748              
  Branches        178      178              
============================================
+ Hits           3844     3849       +5     
+ Misses         1589     1585       -4     
+ Partials        315      314       -1     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `82.01% <ø> (ø)` | |

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/1886?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [control-plane/pkg/prober/cache.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1886/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcHJvYmVyL2NhY2hlLmdv) | `100.00% <0.00%> (+4.76%)` | :arrow_up: |
| [control-plane/pkg/prober/prober.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1886/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcHJvYmVyL3Byb2Jlci5nbw==) | `75.00% <0.00%> (+15.00%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1886?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/1886?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [90275cb...d9766ba](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1886?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1886#" 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"]} -->

Patch:
diff --git a/.github/workflows/knative-release-notes.yaml b/.github/workflows/knative-release-notes.yaml
index 999231ca7..5b011ba2a 100644
--- a/.github/workflows/knative-release-notes.yaml
+++ b/.github/workflows/knative-release-notes.yaml
@@ -21,13 +21,13 @@ on:
   workflow_dispatch:
     inputs:
       branch:
-        description: 'Branch? (main)'
+        description: 'Branch'
         required: true
         default: 'main'
-      start-sha:
-        description: 'Starting SHA? (last tag on branch)'
-      end-sha:
-        description: 'Ending SHA? (latest HEAD)'
+      start-rev:
+        description: 'Start Tag (defaults to merge-base(branch, prev-branch))'
+      end-rev:
+        description: 'End Tag (defaults to HEAD of the target branch)'
 
 jobs:
   release-notes:
@@ -44,6 +44,7 @@ jobs:
           go-version: 1.17.x
 
       - name: Install Dependencies
+        # https://github.com/kubernetes/release/tree/master/cmd/release-notes
         run: go install k8s.io/release/cmd/release-notes@latest
 
       - name: Check out code
@@ -52,38 +53,45 @@ jobs:
           # fetch-depth of 0 indicates all history for all branches and tags.
           fetch-depth: 0
 
-      # Note: Defaults needs to run after we check out the repo.
-      - name: Defaults
+      - name: Generate Notes
         run: |
-          echo ORG=$(echo '${{ github.repository }}' | awk -F '/' '{print $1}') >> $GITHUB_ENV
-          echo REPO=$(echo '${{ github.repository }}' | awk -F '/' '{print $2}') >> $GITHUB_ENV
-
-          echo "BRANCH=${{ github.event.inputs.branch }}" >> $GITHUB_ENV
-
-          if [[ "${{ github.event.inputs.start-sha }}" != "" ]]; then
-            echo "START_SHA=${{ github.event.inputs.start-sha }}" >> $GITHUB_ENV
-          else
-            # Default Starting SHA (thanks @dprotaso)
-            export semver=$(git describe --match "v[0-9]*" --abbrev=0)
-            echo "Using ${semver} tag for starting sha."
-            echo START_SHA=$(git rev-list -n 1 "${semver}") >> $GITHUB_ENV
+          set -x
+          # The release-notes tool access ENV vars as options
+          # https://github.com/kubernetes/release/tree/master/cmd/release-notes#options
+
+          export ORG=$(echo '${{ github.repository }}' | awk -F '/' '{print $1}')
+          export REPO=$(echo '${{ github.repository }}' | awk -F '/' '{print $2}')
+          export BRANCH="${{ github.event.inputs.branch }}"
+
+          export START_REV=${{ github.event.inputs.start-rev }}
+          export END_REV=${{ github.event.inputs.end-rev }}
+
+          if [[ -z "$END_REV" ]]; then
+            END_REV="origin/$BRANCH"
           fi
 
-          if [[ "${{ github.event.inputs.end-sha }}" != "" ]]; then
-            echo "END_SHA=${{ github.event.inputs.end-sha }}" >> $GITHUB_ENV
-          else
-            # Default Ending SHA (thanks @dprotaso)
-            echo "END_SHA=$(git rev-list -n 1 HEAD)" >> $GITHUB_ENV
+          # If start rev isn't set find the merge base of
+          # the target branch and the previous branch
+          if [[ -z "$START_REV" ]]; then
+            BRANCHES=$(mktemp)
+            # List of branches sorted by semver descending
+            git branch -r -l "origin/release-[0-9]*\.[0-9]*" | sed 's/ //g' | sort -Vr > "$BRANCHES"
+
+            if [[ "$BRANCH" == "main" ]]; then
+              LAST_BRANCH="$(head -n1 "$BRANCHES")"
+            else
+              # use grep magic to find the next branch
+              # '-A 1' - prints the line after the match which we can parse
+              LAST_BRANCH="$(grep -A 1 "$BRANCH" "$BRANCHES" | tail -n1)"
+            fi
+
+            export START_SHA=$(git merge-base $LAST_BRANCH origin/$BRANCH)
           fi
 
-      - name: Generate Notes
-        run: |
-          # See  https://github.com/kubernetes/release/tree/master/cmd/release-notes for options.
-          # Note: we are setting env vars in the Defaults step.
           release-notes \
-            --required-author "" \
-            --repo-path       "$(pwd)" \
-            --output          release-notes.md
+           --required-author="" \
+           --output=release-notes.md \
+           --repo-path="$PWD" \
 
       - name: Display Notes
         run: |
diff --git a/.github/workflows/knative-security.yaml b/.github/workflows/knative-security.yaml
index eb427a3e6..9f7bb3eda 100644
--- a/.github/workflows/knative-security.yaml
+++ b/.github/workflows/knative-security.yaml
@@ -50,7 +50,7 @@ jobs:
       uses: github/codeql-action/analyze@v1
 
     - name: Find Unicode Control Characters
-      uses: pierdipi/unicode-control-characters-action@v0.1.0
+      uses: pierdipi/unicode-control-characters-action@v0.1.1
       with:
         args: -d .
 
diff --git a/.gitignore b/.gitignore
index ccbc1fcce..c81faa7b7 100644
--- a/.gitignore
+++ b/.gitignore
@@ -34,6 +34,8 @@ eventing-kafka-broker.yaml
 eventing-kafka-channel.yaml
 eventing-kafka-sink.yaml
 
+eventing-kafka-post-install.yaml
+
 eventing-kafka-controller-prometheus-operator.yaml
 eventing-kafka-source-prometheus-operator.yaml
 eventing-kafka-channel-prometheus-operator.yaml
diff --git a/control-plane/cmd/post-install/kafka_source_migrator.go b/control-plane/cmd/post-install/kafka_source_migrator.go
new file mode 100644
index 000000000..85e5af09c
--- /dev/null
+++ b/control-plane/cmd/post-install/kafka_source_migrator.go
@@ -0,0 +1,78 @@
+/*
+ * 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 main
+
+import (
+	"context"
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/kubernetes"
+	kcs "knative.dev/eventing-kafka/pkg/client/clientset/versioned"
+)
+
+type KafkaSourceMigrator struct {
+	kcs kcs.Interface
+	k8s kubernetes.Interface
+}
+
+func (m *KafkaSourceMigrator) Migrate(ctx context.Context) error {
+
+	// cont takes care of results pagination.
+	cont := ""
+	first := true
+	for first || cont != "" {
+		kss, err := m.kcs.
+			SourcesV1beta1().
+			KafkaSources(corev1.NamespaceAll).
+			List(ctx, metav1.ListOptions{Continue: cont})
+		if err != nil {
+			return fmt.Errorf("failed to list KafkaSources: %w", err)
+		}
+
+		for _, ks := range kss.Items {
+
+			// The KafkaSource deployment name isn't predictable, so we use a label selector to get
+			// the associated deployment.
+			ksLabels := map[string]string{
+				// See https://github.com/knative-sandbox/eventing-kafka/blob/2b185a9b59347cc4a16471ed0fb2ba3bdf6e1672/pkg/source/reconciler/source/resources/labels.go#L24-L39
+				"eventing.knative.dev/source":     "kafka-source-controller",
+				"eventing.knative.dev/sourceName": ks.GetName(),
+			}
+			lsOpts := metav1.ListOptions{LabelSelector: labels.SelectorFromSet(ksLabels).String()}
+
+			// Instead of listing and then deleting a single deployment,
+			// we use delete collection that allows deleting objects based
+			// on a label selector.
+			pp := metav1.DeletePropagationForeground
+			err := m.k8s.AppsV1().
+				Deployments(ks.GetNamespace()).
+				DeleteCollection(ctx, metav1.DeleteOptions{PropagationPolicy: &pp}, lsOpts)
+			if err != nil && !apierrors.IsNotFound(err) {
+				return fmt.Errorf("failed to delete deployments with selector %v: %w", lsOpts.LabelSelector, err)
+			}
+		}
+
+		cont = kss.Continue
+		first = false
+	}
+
+	return nil
+}
diff --git a/control-plane/cmd/post-install/main.go b/control-plane/cmd/post-install/main.go
new file mode 100644
index 000000000..7442b5be6
--- /dev/null
+++ b/control-plane/cmd/post-install/main.go
@@ -0,0 +1,67 @@
+/*
+ * 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 main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"log"
+
+	"k8s.io/client-go/kubernetes"
+	kcs "knative.dev/eventing-kafka/pkg/client/clientset/versioned"
+	"knative.dev/pkg/environment"
+	"knative.dev/pkg/logging"
+	"knative.dev/pkg/signals"
+)
+
+func main() {
+	ctx := signals.NewContext()
+
+	config, err := logging.NewConfigFromMap(nil)
+	if err != nil {
+		log.Fatal("Failed to create logging config: ", err)
+	}
+
+	logger, _ := logging.NewLoggerFromConfig(config, "kafka-broker-post-install")
+	defer logger.Sync()
+
+	logging.WithLogger(ctx, logger)
+
+	if err := run(ctx); err != nil {
+		logger.Fatal(err)
+	}
+}
+
+func run(ctx context.Context) error {
+
+	env := environment.ClientConfig{}
+	env.InitFlags(flag.CommandLine)
+	flag.Parse()
+
+	config, err := env.GetRESTConfig()
+	if err != nil {
+		return fmt.Errorf("failed to get kubeconfig: %w", err)
+	}
+
+	migrator := &KafkaSourceMigrator{
+		kcs: kcs.NewForConfigOrDie(config),
+		k8s: kubernetes.NewForConfigOrDie(config),
+	}
+
+	return migrator.Migrate(ctx)
+}
diff --git a/control-plane/config/100-channel/100-kafka-channel.yaml b/control-plane/config/100-channel/100-kafka-channel.yaml
deleted file mode 120000
index d60cf59f1..000000000
--- a/control-plane/config/100-channel/100-kafka-channel.yaml
+++ /dev/null
@@ -1,1 +0,0 @@
-../../../third_party/eventing-kafka-latest/channel-crds.yaml
\ No newline at end of file
diff --git a/control-plane/config/100-source/100-kafka-source.yaml b/control-plane/config/100-source/100-kafka-source.yaml
deleted file mode 120000
index 1e45756e5..000000000
--- a/control-plane/config/100-source/100-kafka-source.yaml
+++ /dev/null
@@ -1,1 +0,0 @@
-../../../third_party/eventing-kafka-latest/source-crd.yaml
\ No newline at end of file
diff --git a/control-plane/config/100-broker/100-kafka-broker-configmap.yaml b/control-plane/config/eventing-kafka-broker/100-broker/100-kafka-broker-configmap.yaml
similarity index 100%
rename from control-plane/config/100-broker/100-kafka-broker-configmap.yaml
rename to control-plane/config/eventing-kafka-broker/100-broker/100-kafka-broker-configmap.yaml
diff --git a/control-plane/config/100-channel/100-kafka-channel-configmap.yaml b/control-plane/config/eventing-kafka-broker/100-channel/100-kafka-channel-configmap.yaml
similarity index 100%
rename from control-plane/config/100-channel/100-kafka-channel-configmap.yaml
rename to control-plane/config/eventing-kafka-broker/100-channel/100-kafka-channel-configmap.yaml
diff --git a/control-plane/config/eventing-kafka-broker/100-channel/100-kafka-channel.yaml b/control-plane/config/eventing-kafka-broker/100-channel/100-kafka-channel.yaml
new file mode 120000
index 000000000..0f2cef1b2
--- /dev/null
+++ b/control-plane/config/eventing-kafka-broker/100-channel/100-kafka-channel.yaml
@@ -0,0 +1,1 @@
+../../../../third_party/eventing-kafka-latest/channel-crds.yaml
\ No newline at end of file
diff --git a/control-plane/config/100-kafka-internal/100-consumer.yaml b/control-plane/config/eventing-kafka-broker/100-kafka-internal/100-consumer.yaml
similarity index 100%
rename from control-plane/config/100-kafka-internal/100-consumer.yaml
rename to control-plane/config/eventing-kafka-broker/100-kafka-internal/100-consumer.yaml
diff --git a/control-plane/config/100-kafka-internal/100-consumergroup.yaml b/control-plane/config/eventing-kafka-broker/100-kafka-internal/100-consumergroup.yaml
similarity index 100%
rename from control-plane/config/100-kafka-internal/100-consumergroup.yaml
rename to control-plane/config/eventing-kafka-broker/100-kafka-internal/100-consumergroup.yaml
diff --git a/control-plane/config/100-sink/100-kafka-sink.yaml b/control-plane/config/eventing-kafka-broker/100-sink/100-kafka-sink.yaml
similarity index 100%
rename from control-plane/config/100-sink/100-kafka-sink.yaml
rename to control-plane/config/eventing-kafka-broker/100-sink/100-kafka-sink.yaml
diff --git a/control-plane/config/eventing-kafka-broker/100-source/100-kafka-source.yaml b/control-plane/config/eventing-kafka-broker/100-source/100-kafka-source.yaml
new file mode 120000
index 000000000..595cfd702
--- /dev/null
+++ b/control-plane/config/eventing-kafka-broker/100-source/100-kafka-source.yaml
@@ -0,0 +1,1 @@
+../../../../third_party/eventing-kafka-latest/source-crd.yaml
\ No newline at end of file
diff --git a/control-plane/config/100-source/200-sources-observer-aggregated-clusterrole.yaml b/control-plane/config/eventing-kafka-broker/100-source/200-sources-observer-aggregated-clusterrole.yaml
similarity index 100%
rename from control-plane/config/100-source/200-sources-observer-aggregated-clusterrole.yaml
rename to control-plane/config/eventing-kafka-broker/100-source/200-sources-observer-aggregated-clusterrole.yaml
diff --git a/control-plane/config/200-controller/100-config-kafka-descheduler.yaml b/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-descheduler.yaml
similarity index 100%
rename from control-plane/config/200-controller/100-config-kafka-descheduler.yaml
rename to control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-descheduler.yaml
diff --git a/control-plane/config/200-controller/100-config-kafka-leader-election.yaml b/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-leader-election.yaml
similarity index 100%
rename from control-plane/config/200-controller/100-config-kafka-leader-election.yaml
rename to control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-leader-election.yaml
diff --git a/control-plane/config/200-controller/100-config-kafka-scheduler.yaml b/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-scheduler.yaml
similarity index 100%
rename from control-plane/config/200-controller/100-config-kafka-scheduler.yaml
rename to control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-scheduler.yaml
diff --git a/control-plane/config/200-controller/100-config-logging.yaml b/control-plane/config/eventing-kafka-broker/200-controller/100-config-logging.yaml
similarity index 100%
rename from control-plane/config/200-controller/100-config-logging.yaml
rename to control-plane/config/eventing-kafka-broker/200-controller/100-config-logging.yaml
diff --git a/control-plane/config/200-controller/100-config-tracing.yaml b/control-plane/config/eventing-kafka-broker/200-controller/100-config-tracing.yaml
similarity index 100%
rename from control-plane/config/200-controller/100-config-tracing.yaml
rename to control-plane/config/eventing-kafka-broker/200-controller/100-config-tracing.yaml
diff --git a/control-plane/config/200-controller/200-addressable-resolver-cluster-role.yaml b/control-plane/config/eventing-kafka-broker/200-controller/200-addressable-resolver-cluster-role.yaml
similarity index 100%
rename from control-plane/config/200-controller/200-addressable-resolver-cluster-role.yaml
rename to control-plane/config/eventing-kafka-broker/200-controller/200-addressable-resolver-cluster-role.yaml
diff --git a/control-plane/config/200-controller/200-channelable-manipulator-clusterrole.yaml b/control-plane/config/eventing-kafka-broker/200-controller/200-channelable-manipulator-clusterrole.yaml
similarity index 100%
rename from control-plane/config/200-controller/200-channelable-manipulator-clusterrole.yaml
rename to control-plane/config/eventing-kafka-broker/200-controller/200-channelable-manipulator-clusterrole.yaml
diff --git a/control-plane/config/200-controller/200-controller-cluster-role.yaml b/control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml
similarity index 100%
rename from control-plane/config/200-controller/200-controller-cluster-role.yaml
rename to control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml
diff --git a/control-plane/config/200-controller/200-controller-service-account.yaml b/control-plane/config/eventing-kafka-broker/200-controller/200-controller-service-account.yaml
similarity index 100%
rename from control-plane/config/200-controller/200-controller-service-account.yaml
rename to control-plane/config/eventing-kafka-broker/200-controller/200-controller-service-account.yaml
diff --git a/control-plane/config/200-controller/201-controller-cluster-role-binding.yaml b/control-plane/config/eventing-kafka-broker/200-controller/201-controller-cluster-role-binding.yaml
similarity index 100%
rename from control-plane/config/200-controller/201-controller-cluster-role-binding.yaml
rename to control-plane/config/eventing-kafka-broker/200-controller/201-controller-cluster-role-binding.yaml
diff --git a/control-plane/config/200-controller/500-controller.yaml b/control-plane/config/eventing-kafka-broker/200-controller/500-controller.yaml
similarity index 100%
rename from control-plane/config/200-controller/500-controller.yaml
rename to control-plane/config/eventing-kafka-broker/200-controller/500-controller.yaml
diff --git a/control-plane/config/200-webhook/100-webhook-cluster-role.yaml b/control-plane/config/eventing-kafka-broker/200-webhook/100-webhook-cluster-role.yaml
similarity index 100%
rename from control-plane/config/200-webhook/100-webhook-cluster-role.yaml
rename to control-plane/config/eventing-kafka-broker/200-webhook/100-webhook-cluster-role.yaml
diff --git a/control-plane/config/200-webhook/100-webhook-service-account.yaml b/control-plane/config/eventing-kafka-broker/200-webhook/100-webhook-service-account.yaml
similarity index 100%
rename from control-plane/config/200-webhook/100-webhook-service-account.yaml
rename to control-plane/config/eventing-kafka-broker/200-webhook/100-webhook-service-account.yaml
diff --git a/control-plane/config/200-webhook/200-webhook-cluster-role-binding.yaml b/control-plane/config/eventing-kafka-broker/200-webhook/200-webhook-cluster-role-binding.yaml
similarity index 100%
rename from control-plane/config/200-webhook/200-webhook-cluster-role-binding.yaml
rename to control-plane/config/eventing-kafka-broker/200-webhook/200-webhook-cluster-role-binding.yaml
diff --git a/control-plane/config/200-webhook/400-webhook-defaulting.yaml b/control-plane/config/eventing-kafka-broker/200-webhook/400-webhook-defaulting.yaml
similarity index 100%
rename from control-plane/config/200-webhook/400-webhook-defaulting.yaml
rename to control-plane/config/eventing-kafka-broker/200-webhook/400-webhook-defaulting.yaml
diff --git a/control-plane/config/200-webhook/400-webhook-secret.yaml b/control-plane/config/eventing-kafka-broker/200-webhook/400-webhook-secret.yaml
similarity index 100%
rename from control-plane/config/200-webhook/400-webhook-secret.yaml
rename to control-plane/config/eventing-kafka-broker/200-webhook/400-webhook-secret.yaml
diff --git a/control-plane/config/200-webhook/400-webhook-validation.yaml b/control-plane/config/eventing-kafka-broker/200-webhook/400-webhook-validation.yaml
similarity index 100%
rename from control-plane/config/200-webhook/400-webhook-validation.yaml
rename to control-plane/config/eventing-kafka-broker/200-webhook/400-webhook-validation.yaml
diff --git a/control-plane/config/200-webhook/500-webhook.yaml b/control-plane/config/eventing-kafka-broker/200-webhook/500-webhook.yaml
similarity index 100%
rename from control-plane/config/200-webhook/500-webhook.yaml
rename to control-plane/config/eventing-kafka-broker/200-webhook/500-webhook.yaml
diff --git a/control-plane/config/post-install/200-controller-cluster-role.yaml b/control-plane/config/post-install/200-controller-cluster-role.yaml
new file mode 100644
index 000000000..09c9707e2
--- /dev/null
+++ b/control-plane/config/post-install/200-controller-cluster-role.yaml
@@ -0,0 +1,40 @@
+---
+
+# 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.
+
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: knative-kafka-controller-post-install
+  labels:
+    kafka.eventing.knative.dev/release: devel
+rules:
+  - apiGroups:
+      - apps
+    resources:
+      - deployments
+    verbs:
+      - delete
+      - deletecollection
+
+  # sources.knative.dev resources and finalizers we care about.
+  - apiGroups:
+      - sources.knative.dev
+    resources:
+      - kafkasources
+      - kafkasources/status
+    verbs:
+      - get
+      - list
diff --git a/control-plane/config/post-install/200-controller-service-account.yaml b/control-plane/config/post-install/200-controller-service-account.yaml
new file mode 100644
index 000000000..91f65a3d9
--- /dev/null
+++ b/control-plane/config/post-install/200-controller-service-account.yaml
@@ -0,0 +1,23 @@
+---
+
+# 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.
+
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: knative-kafka-controller-post-install
+  namespace: knative-eventing
+  labels:
+    kafka.eventing.knative.dev/release: devel
diff --git a/control-plane/config/post-install/201-controller-cluster-role-binding.yaml b/control-plane/config/post-install/201-controller-cluster-role-binding.yaml
new file mode 100644
index 000000000..9f36b0c8c
--- /dev/null
+++ b/control-plane/config/post-install/201-controller-cluster-role-binding.yaml
@@ -0,0 +1,30 @@
+---
+
+# 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.
+
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: knative-kafka-controller-post-install
+  labels:
+    kafka.eventing.knative.dev/release: devel
+subjects:
+  - kind: ServiceAccount
+    name: knative-kafka-controller-post-install
+    namespace: knative-eventing
+roleRef:
+  kind: ClusterRole
+  name: knative-kafka-controller-post-install
+  apiGroup: rbac.authorization.k8s.io
diff --git a/control-plane/config/post-install/500-post-install-job.yaml b/control-plane/config/post-install/500-post-install-job.yaml
new file mode 100644
index 000000000..89798e8c8
--- /dev/null
+++ b/control-plane/config/post-install/500-post-install-job.yaml
@@ -0,0 +1,38 @@
+# 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
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: kafka-controller-post-install
+  namespace: knative-eventing
+  labels:
+    app: kafka-controller-post-install
+    kafka.eventing.knative.dev/release: devel
+spec:
+  ttlSecondsAfterFinished: 600
+  backoffLimit: 10
+  template:
+    metadata:
+      labels:
+        app: kafka-controller-post-install
+        kafka.eventing.knative.dev/release: devel
+      annotations:
+        sidecar.istio.io/inject: "false"
+    spec:
+      serviceAccountName: knative-kafka-controller-post-install
+      restartPolicy: OnFailure
+      containers:
+        - name: post-install
+          image: ko://knative.dev/eventing-kafka-broker/control-plane/cmd/post-install
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
index 612d55d57..7beb3bd2d 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
@@ -98,22 +98,28 @@ type ConsumerSpec struct {
 	CloudEventOverrides *duckv1.CloudEventOverrides `json:"ceOverrides,omitempty"`
 
 	// VReplicas is the number of virtual replicas for a consumer.
-	VReplicas *int32
+	VReplicas *int32 `json:"vReplicas"`
 
 	// PodBind represents a reference to the pod in which the consumer should be placed.
 	PodBind *PodBind `json:"podBind"`
 }
 
 type ReplyStrategy struct {
-	TopicReply *TopicReply
-	URLReply   *DestinationReply
-	NoReply    *struct{}
+	TopicReply *TopicReply       `json:"topicReply,omitempty"`
+	URLReply   *DestinationReply `json:"URLReply,omitempty"`
+	NoReply    *NoReply          `json:"noReply,omitempty"`
+}
+
+type NoReply struct {
+	Enabled bool `json:"enabled"`
 }
 
 type TopicReply struct {
+	Enabled bool `json:"enabled"`
 }
 
 type DestinationReply struct {
+	Enabled     bool `json:"enabled"`
 	Destination duckv1.Destination
 }
 
@@ -197,21 +203,6 @@ func (c *Consumer) GetStatus() *duckv1.Status {
 	return &c.Status.Status
 }
 
-// IsLessThan returns true if c is less than other.
-//
-// if c is less than other, other might be deleted before c.
-func (c *Consumer) IsLessThan(other *Consumer) bool {
-	// Prefer ready instances.
-	if c.IsReady() {
-		return true
-	}
-	if other.IsReady() {
-		return false
-	}
-	// Prefer older instances.
-	return c.CreationTimestamp.Time.Before(other.CreationTimestamp.Time)
-}
-
 func (c *Consumer) IsReady() bool {
 	return c.Generation == c.Status.ObservedGeneration &&
 		c.GetConditionSet().Manage(c.GetStatus()).IsHappy()
@@ -234,3 +225,29 @@ func (c *Consumer) GetConsumerGroup() *metav1.OwnerReference {
 func (c Consumer) HasDeadLetterSink() bool {
 	return hasDeadLetterSink(c.Spec.Delivery)
 }
+
+type ByReadinessAndCreationTime []*Consumer
+
+func (consumers ByReadinessAndCreationTime) Len() int {
+	return len(consumers)
+}
+
+func (consumers ByReadinessAndCreationTime) Less(i, j int) bool {
+	c, other := consumers[i], consumers[j]
+
+	// Prefer ready instances.
+	if c.IsReady() {
+		return true
+	}
+	if other.IsReady() {
+		return false
+	}
+	// Prefer older instances.
+	return c.CreationTimestamp.Time.Before(other.CreationTimestamp.Time)
+}
+
+func (consumers ByReadinessAndCreationTime) Swap(i, j int) {
+	tmp := consumers[i]
+	consumers[i] = consumers[j]
+	consumers[j] = tmp
+}
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_validation.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_validation.go
index f8dd77591..db02d7e4b 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_validation.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_validation.go
@@ -45,6 +45,7 @@ func (cs *ConsumerSpec) Validate(ctx context.Context) *apis.FieldError {
 		cs.Subscriber.Validate(ctx).ViaField("subscriber"),
 		cs.PodBind.Validate(ctx).ViaField("podBind"),
 		cs.CloudEventOverrides.Validate(ctx).ViaField("ceOverrides"),
+		cs.Reply.Validate(ctx).ViaField("reply"),
 	)
 	return err
 }
@@ -73,6 +74,27 @@ func (f *Filters) Validate(ctx context.Context) *apis.FieldError {
 	return nil
 }
 
+func (in *ReplyStrategy) Validate(ctx context.Context) *apis.FieldError {
+	if in == nil {
+		return nil
+	}
+	setCounter := 0
+	if in.TopicReply != nil && in.TopicReply.Enabled {
+		setCounter++
+	}
+	if in.URLReply != nil && in.URLReply.Enabled {
+		setCounter++
+	}
+	if in.NoReply != nil && in.NoReply.Enabled {
+		setCounter++
+	}
+
+	if setCounter > 1 {
+		return apis.ErrMultipleOneOf("topicReply", "URLReply", "NoReply")
+	}
+	return nil
+}
+
 func (p *PodBind) Validate(ctx context.Context) *apis.FieldError {
 	if p == nil {
 		return apis.ErrMissingField("")
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go
index b9e527115..2498b1e2d 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go
@@ -57,6 +57,32 @@ func (in *Auth) DeepCopy() *Auth {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in ByReadinessAndCreationTime) DeepCopyInto(out *ByReadinessAndCreationTime) {
+	{
+		in := &in
+		*out = make(ByReadinessAndCreationTime, len(*in))
+		for i := range *in {
+			if (*in)[i] != nil {
+				in, out := &(*in)[i], &(*out)[i]
+				*out = new(Consumer)
+				(*in).DeepCopyInto(*out)
+			}
+		}
+		return
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ByReadinessAndCreationTime.
+func (in ByReadinessAndCreationTime) DeepCopy() ByReadinessAndCreationTime {
+	if in == nil {
+		return nil
+	}
+	out := new(ByReadinessAndCreationTime)
+	in.DeepCopyInto(out)
+	return *out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *Consumer) DeepCopyInto(out *Consumer) {
 	*out = *in
@@ -425,6 +451,22 @@ func (in *Filters) DeepCopy() *Filters {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *NoReply) DeepCopyInto(out *NoReply) {
+	*out = *in
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new NoReply.
+func (in *NoReply) DeepCopy() *NoReply {
+	if in == nil {
+		return nil
+	}
+	out := new(NoReply)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *PodBind) DeepCopyInto(out *PodBind) {
 	*out = *in
@@ -456,7 +498,7 @@ func (in *ReplyStrategy) DeepCopyInto(out *ReplyStrategy) {
 	}
 	if in.NoReply != nil {
 		in, out := &in.NoReply, &out.NoReply
-		*out = new(struct{})
+		*out = new(NoReply)
 		**out = **in
 	}
 	return
diff --git a/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index 57fb269c2..d63fee73b 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -430,7 +430,7 @@ func (r *Reconciler) reconcileSubscribers(ctx context.Context, kafkaClient saram
 				UID:                s.UID,
 				ObservedGeneration: s.Generation,
 				Ready:              corev1.ConditionFalse,
-				Message:            fmt.Sprint("Subscription not ready", err),
+				Message:            fmt.Sprintf("Subscription not ready: %v", err),
 			})
 			globalErr = multierr.Append(globalErr, err)
 		} else {
@@ -479,8 +479,13 @@ 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) {
+	subscriberURI := subscriber.SubscriberURI.String()
+	if subscriberURI == "" {
+		return nil, fmt.Errorf("failed to resolve Subscription.Spec.Subscriber: empty subscriber URI")
+	}
+
 	egress := &contract.Egress{
-		Destination:   subscriber.SubscriberURI.String(),
+		Destination:   subscriberURI,
 		ConsumerGroup: consumerGroup(channel, subscriber),
 		DeliveryOrder: DefaultDeliveryOrder,
 		Uid:           string(subscriber.UID),
diff --git a/control-plane/pkg/reconciler/consumer/consumer.go b/control-plane/pkg/reconciler/consumer/consumer.go
index 65b7a6078..b4b58c264 100644
--- a/control-plane/pkg/reconciler/consumer/consumer.go
+++ b/control-plane/pkg/reconciler/consumer/consumer.go
@@ -252,11 +252,11 @@ func (r Reconciler) reconcileReplyStrategy(ctx context.Context, c *kafkainternal
 	if c.Spec.Reply == nil {
 		return nil
 	}
-	if c.Spec.Reply.NoReply != nil {
+	if c.Spec.Reply.NoReply != nil && c.Spec.Reply.NoReply.Enabled {
 		egress.ReplyStrategy = &contract.Egress_DiscardReply{}
 		return nil
 	}
-	if c.Spec.Reply.URLReply != nil {
+	if c.Spec.Reply.URLReply != nil && c.Spec.Reply.URLReply.Enabled {
 		destination, err := r.Resolver.URIFromDestinationV1(ctx, c.Spec.Reply.URLReply.Destination, c)
 		if err != nil {
 			return fmt.Errorf("failed to resolve reply destination: %w", err)
@@ -266,7 +266,7 @@ func (r Reconciler) reconcileReplyStrategy(ctx context.Context, c *kafkainternal
 		}
 		return nil
 	}
-	if c.Spec.Reply.TopicReply != nil {
+	if c.Spec.Reply.TopicReply != nil && c.Spec.Reply.TopicReply.Enabled {
 		egress.ReplyStrategy = &contract.Egress_ReplyToOriginalTopic{}
 		return nil
 	}
@@ -316,9 +316,6 @@ func (r Reconciler) schedule(ctx context.Context, logger *zap.Logger, c *kafkain
 		// will get re-queued when the pod is added.
 		return false, nil
 	}
-	if err != nil {
-		return false, err
-	}
 
 	// Get contract associated with the pod.
 	cmName, err := cmNameFromPod(p, c)
diff --git a/control-plane/pkg/reconciler/consumergroup/consumergroup.go b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
index 401de505b..8ee716437 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
@@ -128,10 +128,10 @@ func (r Reconciler) reconcileConsumersInPlacement(
 	}
 
 	// Stable sort consumers so that we give consumers different deletion
-	// priorities based on their state.
+	// priorities based on their state (readiness, etc).
 	//
 	// Consumers at the tail of the list are deleted.
-	sort.SliceStable(consumers, func(i, j int) bool { return consumers[i].IsLessThan(consumers[j]) })
+	sort.Stable(kafkainternals.ByReadinessAndCreationTime(consumers))
 
 	for _, c := range consumers[1:] {
 		if err := r.finalizeConsumer(ctx, c); err != nil {
@@ -139,9 +139,7 @@ func (r Reconciler) reconcileConsumersInPlacement(
 		}
 	}
 
-	// Do not modify informer copy.
-	c := &kafkainternals.Consumer{}
-	consumers[0].DeepCopyInto(c)
+	c := consumers[0]
 
 	expectedSpec := kafkainternals.ConsumerSpec{}
 	cg.Spec.Template.Spec.DeepCopyInto(&expectedSpec)
@@ -153,11 +151,21 @@ func (r Reconciler) reconcileConsumersInPlacement(
 		return nil
 	}
 
-	c.Spec.PodBind = &kafkainternals.PodBind{PodName: placement.PodName, PodNamespace: r.SystemNamespace}
-	c.Spec.VReplicas = pointer.Int32Ptr(*expectedSpec.VReplicas)
+	expectedSpec.PodBind = &kafkainternals.PodBind{
+		PodName:      placement.PodName,
+		PodNamespace: r.SystemNamespace,
+	}
+
+	// Do not modify informer copy.
+	expectedC := &kafkainternals.Consumer{
+		TypeMeta:   c.TypeMeta,
+		ObjectMeta: c.ObjectMeta,
+		Spec:       expectedSpec,
+		Status:     c.Status,
+	}
 
 	// Update existing Consumer.
-	if _, err := r.InternalsClient.Consumers(cg.GetNamespace()).Update(ctx, c, metav1.UpdateOptions{}); err != nil {
+	if _, err := r.InternalsClient.Consumers(cg.GetNamespace()).Update(ctx, expectedC, metav1.UpdateOptions{}); err != nil {
 		return fmt.Errorf("failed to update consumer %s/%s: %w", c.GetNamespace(), c.GetName(), err)
 	}
 
diff --git a/control-plane/pkg/reconciler/source/v2/sourcev2.go b/control-plane/pkg/reconciler/source/v2/sourcev2.go
index e68eb8950..6558ccf0a 100644
--- a/control-plane/pkg/reconciler/source/v2/sourcev2.go
+++ b/control-plane/pkg/reconciler/source/v2/sourcev2.go
@@ -66,7 +66,11 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, ks *sources.KafkaSource)
 	} else {
 		topLevelCondition := cg.GetConditionSet().Manage(cg.GetStatus()).GetTopLevelCondition()
 		if topLevelCondition == nil {
-			ks.GetConditionSet().Manage(&ks.Status).MarkUnknown(KafkaConditionConsumerGroup, "failed to reconcile consumer group", "consumer group not ready")
+			ks.GetConditionSet().Manage(&ks.Status).MarkUnknown(
+				KafkaConditionConsumerGroup,
+				"failed to reconcile consumer group",
+				"consumer group not ready",
+			)
 		} else {
 			ks.GetConditionSet().Manage(&ks.Status).MarkFalse(
 				KafkaConditionConsumerGroup,
@@ -87,7 +91,7 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, ks *sources.KafkaSource)
 
 func (r Reconciler) reconcileConsumerGroup(ctx context.Context, ks *sources.KafkaSource) (*internalscg.ConsumerGroup, error) {
 
-	newcg := &internalscg.ConsumerGroup{
+	expectedCg := &internalscg.ConsumerGroup{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      string(ks.UID),
 			Namespace: ks.Namespace,
@@ -110,6 +114,7 @@ func (r Reconciler) reconcileConsumerGroup(ctx context.Context, ks *sources.Kafk
 						Ordering: DefaultDeliveryOrder,
 					},
 					Subscriber: ks.Spec.Sink,
+					Reply:      &internalscg.ReplyStrategy{NoReply: &internalscg.NoReply{Enabled: true}},
 				},
 			},
 			Replicas: ks.Spec.Consumers,
@@ -117,7 +122,7 @@ func (r Reconciler) reconcileConsumerGroup(ctx context.Context, ks *sources.Kafk
 	}
 
 	if ks.Spec.CloudEventOverrides != nil {
-		newcg.Spec.Template.Spec.CloudEventOverrides = &duckv1.CloudEventOverrides{
+		expectedCg.Spec.Template.Spec.CloudEventOverrides = &duckv1.CloudEventOverrides{
 			Extensions: ks.Spec.CloudEventOverrides.Extensions,
 		}
 	}
@@ -126,26 +131,22 @@ func (r Reconciler) reconcileConsumerGroup(ctx context.Context, ks *sources.Kafk
 	if err != nil && !apierrors.IsNotFound(err) {
 		return nil, err
 	}
-
 	if apierrors.IsNotFound(err) {
-		cg, err := r.InternalsClient.InternalV1alpha1().ConsumerGroups(newcg.GetNamespace()).Create(ctx, newcg, metav1.CreateOptions{})
+		cg, err := r.InternalsClient.InternalV1alpha1().ConsumerGroups(expectedCg.GetNamespace()).Create(ctx, expectedCg, metav1.CreateOptions{})
 		if err != nil && !apierrors.IsAlreadyExists(err) {
-			return nil, fmt.Errorf("failed to create consumer group %s/%s: %w", newcg.GetNamespace(), newcg.GetName(), err)
-		}
-		if apierrors.IsAlreadyExists(err) {
-			return newcg, nil
+			return nil, fmt.Errorf("failed to create consumer group %s/%s: %w", expectedCg.GetNamespace(), expectedCg.GetName(), err)
 		}
 		return cg, nil
 	}
 
-	if equality.Semantic.DeepDerivative(newcg.Spec, cg.Spec) {
+	if equality.Semantic.DeepDerivative(expectedCg.Spec, cg.Spec) {
 		return cg, nil
 	}
 
 	newCg := &internalscg.ConsumerGroup{
 		TypeMeta:   cg.TypeMeta,
 		ObjectMeta: cg.ObjectMeta,
-		Spec:       newcg.Spec,
+		Spec:       expectedCg.Spec,
 		Status:     cg.Status,
 	}
 	if cg, err = r.InternalsClient.InternalV1alpha1().ConsumerGroups(cg.GetNamespace()).Update(ctx, newCg, metav1.UpdateOptions{}); err != nil {
diff --git a/control-plane/pkg/reconciler/trigger/v2/triggerv2.go b/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
index eade6ff7b..0e19e9b78 100644
--- a/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
+++ b/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
@@ -140,6 +140,7 @@ func (r Reconciler) reconcileConsumerGroup(ctx context.Context, trigger *eventin
 						Filters: trigger.Spec.Filters,
 					},
 					Subscriber: trigger.Spec.Subscriber,
+					Reply:      &internalscg.ReplyStrategy{TopicReply: &internalscg.TopicReply{Enabled: true}},
 				},
 			},
 		},
diff --git a/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index 775825b80..deca8a008 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -41,30 +41,30 @@ Lists of 158 third-party dependencies.
      (The Apache Software License, Version 2.0) CloudEvents - JSON Jackson (io.cloudevents:cloudevents-json-jackson:2.3.0 - https://cloudevents.github.io/sdk-java/cloudevents-json-jackson/)
      (The Apache Software License, Version 2.0) CloudEvents - Kafka Binding (io.cloudevents:cloudevents-kafka:2.3.0 - https://cloudevents.github.io/sdk-java/cloudevents-kafka/)
      (The Apache Software License, Version 2.0) cloudevents-sql (io.cloudevents:cloudevents-sql:2.3.0 - https://cloudevents.github.io/sdk-java/cloudevents-sql/)
-     (Apache Software License 2.0) Debezium API (io.debezium:debezium-api:1.8.0.Final - https://debezium.io/debezium-parent/debezium-api)
-     (Apache Software License 2.0) Debezium Core (io.debezium:debezium-core:1.8.0.Final - https://debezium.io/debezium-parent/debezium-core)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes :: Java Client (io.fabric8:kubernetes-client:5.12.0 - http://fabric8.io/kubernetes-client/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Admission Registration, Authentication and Authorization (io.fabric8:kubernetes-model-admissionregistration:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-admissionregistration/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: API Extensions (io.fabric8:kubernetes-model-apiextensions:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-apiextensions/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Apps (io.fabric8:kubernetes-model-apps:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-apps/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Autoscaling (io.fabric8:kubernetes-model-autoscaling:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-autoscaling/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Batch (io.fabric8:kubernetes-model-batch:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-batch/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Certificates (io.fabric8:kubernetes-model-certificates:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-certificates/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Common (io.fabric8:kubernetes-model-common:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-common/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Coordination (io.fabric8:kubernetes-model-coordination:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-coordination/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Core (io.fabric8:kubernetes-model-core:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-core/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Discovery (io.fabric8:kubernetes-model-discovery:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-discovery/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Events (io.fabric8:kubernetes-model-events:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-events/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Extensions (io.fabric8:kubernetes-model-extensions:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-extensions/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: FlowControl (io.fabric8:kubernetes-model-flowcontrol:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-flowcontrol/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Metrics (io.fabric8:kubernetes-model-metrics:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-metrics/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Networking (io.fabric8:kubernetes-model-networking:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-networking/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Node (io.fabric8:kubernetes-model-node:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-node/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Policy (io.fabric8:kubernetes-model-policy:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-policy/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: RBAC (io.fabric8:kubernetes-model-rbac:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-rbac/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Scheduling (io.fabric8:kubernetes-model-scheduling:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-scheduling/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Storage Class (io.fabric8:kubernetes-model-storageclass:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-storageclass/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes :: Server Mock (io.fabric8:kubernetes-server-mock:5.12.0 - http://fabric8.io/kubernetes-server-mock/)
+     (Apache Software License 2.0) Debezium API (io.debezium:debezium-api:1.8.1.Final - https://debezium.io/debezium-parent/debezium-api)
+     (Apache Software License 2.0) Debezium Core (io.debezium:debezium-core:1.8.1.Final - https://debezium.io/debezium-parent/debezium-core)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes :: Java Client (io.fabric8:kubernetes-client:5.12.1 - http://fabric8.io/kubernetes-client/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Admission Registration, Authentication and Authorization (io.fabric8:kubernetes-model-admissionregistration:5.12.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-admissionregistration/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: API Extensions (io.fabric8:kubernetes-model-apiextensions:5.12.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-apiextensions/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Apps (io.fabric8:kubernetes-model-apps:5.12.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-apps/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Autoscaling (io.fabric8:kubernetes-model-autoscaling:5.12.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-autoscaling/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Batch (io.fabric8:kubernetes-model-batch:5.12.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-batch/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Certificates (io.fabric8:kubernetes-model-certificates:5.12.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-certificates/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Common (io.fabric8:kubernetes-model-common:5.12.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-common/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Coordination (io.fabric8:kubernetes-model-coordination:5.12.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-coordination/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Core (io.fabric8:kubernetes-model-core:5.12.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-core/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Discovery (io.fabric8:kubernetes-model-discovery:5.12.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-discovery/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Events (io.fabric8:kubernetes-model-events:5.12.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-events/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Extensions (io.fabric8:kubernetes-model-extensions:5.12.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-extensions/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: FlowControl (io.fabric8:kubernetes-model-flowcontrol:5.12.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-flowcontrol/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Metrics (io.fabric8:kubernetes-model-metrics:5.12.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-metrics/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Networking (io.fabric8:kubernetes-model-networking:5.12.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-networking/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Node (io.fabric8:kubernetes-model-node:5.12.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-node/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Policy (io.fabric8:kubernetes-model-policy:5.12.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-policy/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: RBAC (io.fabric8:kubernetes-model-rbac:5.12.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-rbac/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Scheduling (io.fabric8:kubernetes-model-scheduling:5.12.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-scheduling/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Storage Class (io.fabric8:kubernetes-model-storageclass:5.12.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-storageclass/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes :: Server Mock (io.fabric8:kubernetes-server-mock:5.12.1 - http://fabric8.io/kubernetes-server-mock/)
      (Apache License, Version 2.0) Fabric8 :: Mock Web Server (io.fabric8:mockwebserver:0.2.2 - http://fabric8.io/)
      (The Apache Software License, Version 2.0) zjsonpatch (io.fabric8:zjsonpatch:0.3.0 - https://github.com/fabric8io/zjsonpatch/)
      (The Apache Software License, Version 2.0) micrometer-core (io.micrometer:micrometer-core:1.8.2 - https://github.com/micrometer-metrics/micrometer)
@@ -85,18 +85,18 @@ Lists of 158 third-party dependencies.
      (Apache License, Version 2.0) Netty/Transport/Classes/Epoll (io.netty:netty-transport-classes-epoll:4.1.73.Final - https://netty.io/netty-transport-classes-epoll/)
      (Apache License, Version 2.0) Netty/Transport/Native/Epoll (io.netty:netty-transport-native-epoll:4.1.73.Final - https://netty.io/netty-transport-native-epoll/)
      (Apache License, Version 2.0) Netty/Transport/Native/Unix/Common (io.netty:netty-transport-native-unix-common:4.1.73.Final - https://netty.io/netty-transport-native-unix-common/)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-api:1.10.1 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-context:1.10.1 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-exporter-logging:1.10.1 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-exporter-zipkin:1.10.1 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-extension-trace-propagators:1.10.1 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk:1.10.1 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-common:1.10.1 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-logs:1.10.1-alpha - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-metrics:1.10.1-alpha - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-testing:1.10.1 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-trace:1.10.1 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-semconv:1.10.1-alpha - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-api:1.11.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-context:1.11.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-exporter-logging:1.11.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-exporter-zipkin:1.11.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-extension-trace-propagators:1.11.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk:1.11.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-common:1.11.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-logs:1.11.0-alpha - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-metrics:1.11.0-alpha - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-testing:1.11.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-trace:1.11.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-semconv:1.11.0-alpha - https://github.com/open-telemetry/opentelemetry-java)
      (The Apache Software License, Version 2.0) Prometheus Java Simpleclient (io.prometheus:simpleclient:0.12.0 - http://github.com/prometheus/client_java/simpleclient)
      (The Apache Software License, Version 2.0) Prometheus Java Simpleclient Common (io.prometheus:simpleclient_common:0.12.0 - http://github.com/prometheus/client_java/simpleclient_common)
      (The Apache Software License, Version 2.0) Prometheus Java Span Context Supplier - Common (io.prometheus:simpleclient_tracer_common:0.12.0 - http://github.com/prometheus/client_java/simpleclient_tracer/simpleclient_tracer_common)
@@ -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.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)
+     (The MIT License) mockito-core (org.mockito:mockito-core:4.3.1 - https://github.com/mockito/mockito)
+     (The MIT License) mockito-junit-jupiter (org.mockito:mockito-junit-jupiter:4.3.1 - 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/)
@@ -154,7 +154,7 @@ Lists of 158 third-party dependencies.
      (Apache-2.0) Scala Compiler (org.scala-lang:scala-reflect:2.12.13 - https://www.scala-lang.org/)
      (Apache-2.0) scala-collection-compat (org.scala-lang.modules:scala-collection-compat_2.12:2.3.0 - http://www.scala-lang.org/)
      (Apache-2.0) scala-java8-compat (org.scala-lang.modules:scala-java8-compat_2.12:0.9.1 - http://www.scala-lang.org/)
-     (MIT License) SLF4J API Module (org.slf4j:slf4j-api:1.7.35 - http://www.slf4j.org)
-     (MIT License) SLF4J NOP Binding (org.slf4j:slf4j-nop:1.7.35 - http://www.slf4j.org)
+     (MIT License) SLF4J API Module (org.slf4j:slf4j-api:1.7.36 - http://www.slf4j.org)
+     (MIT License) SLF4J NOP Binding (org.slf4j:slf4j-nop:1.7.36 - http://www.slf4j.org)
      (Apache-2.0) snappy-java (org.xerial.snappy:snappy-java:1.1.8.1 - https://github.com/xerial/snappy-java)
      (Apache License, Version 2.0) SnakeYAML (org.yaml:snakeyaml:1.28 - http://www.snakeyaml.org)
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/metrics/Metrics.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/metrics/Metrics.java
index 73ef100a6..6cc40379f 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/metrics/Metrics.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/metrics/Metrics.java
@@ -21,7 +21,6 @@ import io.micrometer.core.instrument.DistributionSummary;
 import io.micrometer.core.instrument.MeterRegistry;
 import io.micrometer.core.instrument.binder.BaseUnits;
 import io.micrometer.core.instrument.binder.kafka.KafkaClientMetrics;
-import io.micrometer.core.instrument.config.NamingConvention;
 import io.micrometer.prometheus.PrometheusConfig;
 import io.micrometer.prometheus.PrometheusMeterRegistry;
 import io.vertx.core.http.HttpServerOptions;
@@ -72,6 +71,12 @@ public class Metrics {
    */
   public static final String EVENT_DISPATCH_LATENCY = "event_dispatch_latencies";
 
+  /**
+   * @link https://knative.dev/docs/eventing/observability/metrics/eventing-metrics/
+   * @see Metrics#eventDispatchLatency(io.micrometer.core.instrument.Tags)
+   */
+  public static final String EVENT_PROCESSING_LATENCY = "event_processing_latencies";
+
   /**
    * @link https://knative.dev/docs/eventing/observability/metrics/eventing-metrics/
    */
@@ -79,6 +84,10 @@ public class Metrics {
     public static final String RESPONSE_CODE = "response_code";
     public static final String RESPONSE_CODE_CLASS = "response_code_class";
     public static final String EVENT_TYPE = "event_type";
+
+    public static final String RESOURCE_NAME = "name";
+    public static final String RESOURCE_NAMESPACE = "namespace_name";
+    public static final String CONSUMER_NAME = "consumer_name";
   }
 
   /**
@@ -89,6 +98,8 @@ public class Metrics {
     public static final String DIMENSIONLESS = "1";
   }
 
+  private static final double[] LATENCY_SLOs = new double[]{50, 100, 500, 1000, 5000, 10000};
+
   /**
    * Get metrics options from the given metrics configurations.
    *
@@ -191,6 +202,15 @@ public class Metrics {
       .description("The time spent dispatching an event to Kafka")
       .tags(tags)
       .baseUnit(BaseUnits.MILLISECONDS)
-      .serviceLevelObjectives(1, 2, 5, 10, 20, 50, 100, 500, 1000, 5000, 10000);
+      .serviceLevelObjectives(LATENCY_SLOs);
+  }
+
+  public static DistributionSummary.Builder eventProcessingLatency(final io.micrometer.core.instrument.Tags tags) {
+    return DistributionSummary
+      .builder(EVENT_PROCESSING_LATENCY)
+      .description("The time spent processing an event")
+      .tags(tags)
+      .baseUnit(BaseUnits.MILLISECONDS)
+      .serviceLevelObjectives(LATENCY_SLOs);
   }
 }
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/ConsumerRecordContext.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/ConsumerRecordContext.java
new file mode 100644
index 000000000..74bc14f66
--- /dev/null
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/ConsumerRecordContext.java
@@ -0,0 +1,46 @@
+/*
+ * Copyright © 2018 Knative Authors (knative-dev@googlegroups.com)
+ *
+ * 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 dev.knative.eventing.kafka.broker.dispatcher.impl;
+
+import io.cloudevents.CloudEvent;
+import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
+
+class ConsumerRecordContext {
+
+  private KafkaConsumerRecord<Object, CloudEvent> record;
+  private long receivedAtMs;
+
+  ConsumerRecordContext(KafkaConsumerRecord<Object, CloudEvent> record) {
+    this.record = record;
+    this.resetTimer();
+  }
+
+  void resetTimer() {
+    this.receivedAtMs = System.currentTimeMillis();
+  }
+
+  long performLatency() {
+    return System.currentTimeMillis() - receivedAtMs;
+  }
+
+  KafkaConsumerRecord<Object, CloudEvent> getRecord() {
+    return record;
+  }
+
+  void setRecord(final KafkaConsumerRecord<Object, CloudEvent> record) {
+    this.record = record;
+  }
+}
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java
index 8f45e2faf..ba2582449 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java
@@ -16,6 +16,7 @@
 package dev.knative.eventing.kafka.broker.dispatcher.impl;
 
 import dev.knative.eventing.kafka.broker.core.AsyncCloseable;
+import dev.knative.eventing.kafka.broker.core.metrics.Metrics;
 import dev.knative.eventing.kafka.broker.dispatcher.CloudEventSender;
 import dev.knative.eventing.kafka.broker.dispatcher.Filter;
 import dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcher;
@@ -24,16 +25,21 @@ import dev.knative.eventing.kafka.broker.dispatcher.ResponseHandler;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.CloudEventDeserializer;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.KafkaConsumerRecordUtils;
 import io.cloudevents.CloudEvent;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Tag;
+import io.micrometer.core.instrument.Tags;
 import io.vertx.core.Context;
 import io.vertx.core.Future;
 import io.vertx.core.Promise;
 import io.vertx.core.Vertx;
+import io.vertx.ext.web.client.HttpResponse;
 import io.vertx.kafka.client.common.tracing.ConsumerTracer;
 import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
 import io.vertx.kafka.client.consumer.impl.KafkaConsumerRecordImpl;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.annotation.Nullable;
 import java.util.Objects;
 import java.util.function.Function;
 
@@ -51,12 +57,20 @@ public class RecordDispatcherImpl implements RecordDispatcher {
 
   private static final CloudEventDeserializer cloudEventDeserializer = new CloudEventDeserializer();
 
+  // When we don't have an HTTP response, due to other errors for networking, etc,
+  // we only add the code class tag in the "error" class (5xx).
+  private static final Tag NO_RESPONSE_CODE_CLASS_TAG = Tag.of(Metrics.Tags.RESPONSE_CODE_CLASS, "5xx");
+
   private final Filter filter;
-  private final Function<KafkaConsumerRecord<Object, CloudEvent>, Future<Void>> subscriberSender;
-  private final Function<KafkaConsumerRecord<Object, CloudEvent>, Future<Void>> dlsSender;
+  private final Function<KafkaConsumerRecord<Object, CloudEvent>, Future<HttpResponse<?>>> subscriberSender;
+  private final Function<KafkaConsumerRecord<Object, CloudEvent>, Future<HttpResponse<?>>> dlsSender;
   private final RecordDispatcherListener recordDispatcherListener;
   private final AsyncCloseable closeable;
   private final ConsumerTracer consumerTracer;
+  private final MeterRegistry meterRegistry;
+  private final ResourceContext resourceContext;
+
+  private final Tags noResponseResourceTags;
 
   /**
    * All args constructor.
@@ -69,24 +83,32 @@ public class RecordDispatcherImpl implements RecordDispatcher {
    * @param consumerTracer           consumer tracer
    */
   public RecordDispatcherImpl(
+    final ResourceContext resourceContext,
     final Filter filter,
     final CloudEventSender subscriberSender,
     final CloudEventSender deadLetterSinkSender,
     final ResponseHandler responseHandler,
     final RecordDispatcherListener recordDispatcherListener,
-    final ConsumerTracer consumerTracer) {
+    final ConsumerTracer consumerTracer,
+    final MeterRegistry meterRegistry) {
+    Objects.requireNonNull(resourceContext, "provide resourceContext");
     Objects.requireNonNull(filter, "provide filter");
     Objects.requireNonNull(subscriberSender, "provide subscriberSender");
     Objects.requireNonNull(deadLetterSinkSender, "provide deadLetterSinkSender");
     Objects.requireNonNull(recordDispatcherListener, "provide offsetStrategy");
     Objects.requireNonNull(responseHandler, "provide sinkResponseHandler");
+    Objects.requireNonNull(meterRegistry, "provide meterRegistry");
 
+    this.resourceContext = resourceContext;
     this.filter = filter;
     this.subscriberSender = composeSenderAndSinkHandler(subscriberSender, responseHandler, "subscriber");
     this.dlsSender = composeSenderAndSinkHandler(deadLetterSinkSender, responseHandler, "dead letter sink");
     this.recordDispatcherListener = recordDispatcherListener;
     this.closeable = AsyncCloseable.compose(responseHandler, deadLetterSinkSender, subscriberSender, recordDispatcherListener);
     this.consumerTracer = consumerTracer;
+    this.meterRegistry = meterRegistry;
+
+    this.noResponseResourceTags = resourceContext.getTags().and(NO_RESPONSE_CODE_CLASS_TAG);
   }
 
   /**
@@ -119,8 +141,9 @@ public class RecordDispatcherImpl implements RecordDispatcher {
      */
 
     try {
+      final var recordContext = new ConsumerRecordContext(record);
       Promise<Void> promise = Promise.promise();
-      onRecordReceived(maybeDeserializeValueFromHeaders(record), promise);
+      onRecordReceived(maybeDeserializeValueFromHeaders(recordContext), promise);
       return promise.future();
     } catch (final Exception ex) {
       // This is a fatal exception that shouldn't happen in normal cases.
@@ -137,14 +160,14 @@ public class RecordDispatcherImpl implements RecordDispatcher {
     }
   }
 
-  private void onRecordReceived(final KafkaConsumerRecord<Object, CloudEvent> record, Promise<Void> finalProm) {
-    logDebug("Handling record", record);
+  private void onRecordReceived(final ConsumerRecordContext recordContext, Promise<Void> finalProm) {
+    logDebug("Handling record", recordContext.getRecord());
 
     // Trace record received event
     // This needs to be done manually in order to work properly with both ordered and unordered delivery.
     if (consumerTracer != null) {
       Context context = Vertx.currentContext();
-      ConsumerTracer.StartedSpan span = consumerTracer.prepareMessageReceived(context, record.record());
+      ConsumerTracer.StartedSpan span = consumerTracer.prepareMessageReceived(context, recordContext.getRecord().record());
       if (span != null) {
         finalProm.future()
           .onComplete(ar -> {
@@ -157,60 +180,82 @@ public class RecordDispatcherImpl implements RecordDispatcher {
       }
     }
 
-    recordDispatcherListener.recordReceived(record);
+    recordDispatcherListener.recordReceived(recordContext.getRecord());
     // Execute filtering
-    if (filter.test(record.value())) {
-      onFilterMatching(record, finalProm);
+    final var pass = filter.test(recordContext.getRecord().value());
+
+    Metrics
+      .eventProcessingLatency(this.resourceContext.getTags())
+      .register(meterRegistry)
+      .record(recordContext.performLatency());
+
+    // reset timer to start calculating the dispatch latency.
+    recordContext.resetTimer();
+
+    if (pass) {
+      onFilterMatching(recordContext, finalProm);
     } else {
-      onFilterNotMatching(record, finalProm);
+      onFilterNotMatching(recordContext, finalProm);
     }
   }
 
-  private void onFilterMatching(final KafkaConsumerRecord<Object, CloudEvent> record, final Promise<Void> finalProm) {
-    logDebug("Record match filtering", record);
-    subscriberSender.apply(record)
-      .onSuccess(res -> onSubscriberSuccess(record, finalProm))
-      .onFailure(ex -> onSubscriberFailure(record, finalProm));
+  private void onFilterMatching(final ConsumerRecordContext recordContext, final Promise<Void> finalProm) {
+    logDebug("Record matched filtering", recordContext.getRecord());
+    subscriberSender.apply(recordContext.getRecord())
+      .onSuccess(response -> onSubscriberSuccess(response, recordContext, finalProm))
+      .onFailure(ex -> onSubscriberFailure(ex, recordContext, finalProm));
   }
 
-  private void onFilterNotMatching(final KafkaConsumerRecord<Object, CloudEvent> record,
+  private void onFilterNotMatching(final ConsumerRecordContext recordContext,
                                    final Promise<Void> finalProm) {
-    logDebug("Record doesn't match filtering", record);
-    recordDispatcherListener.recordDiscarded(record);
+    logDebug("Record did not match filtering", recordContext.getRecord());
+    recordDispatcherListener.recordDiscarded(recordContext.getRecord());
     finalProm.complete();
   }
 
-  private void onSubscriberSuccess(final KafkaConsumerRecord<Object, CloudEvent> record,
+  private void onSubscriberSuccess(final HttpResponse<?> response,
+                                   final ConsumerRecordContext recordContext,
                                    final Promise<Void> finalProm) {
-    logDebug("Successfully sent event to subscriber", record);
-    recordDispatcherListener.successfullySentToSubscriber(record);
+    logDebug("Successfully sent event to subscriber", recordContext.getRecord());
+
+    incrementEventCount(response);
+    recordDispatchLatency(response, recordContext);
+
+    recordDispatcherListener.successfullySentToSubscriber(recordContext.getRecord());
     finalProm.complete();
   }
 
-  private void onSubscriberFailure(final KafkaConsumerRecord<Object, CloudEvent> record,
+  private void onSubscriberFailure(final Throwable failure,
+                                   final ConsumerRecordContext recordContext,
                                    final Promise<Void> finalProm) {
-    dlsSender.apply(record)
-      .onSuccess(v -> onDeadLetterSinkSuccess(record, finalProm))
-      .onFailure(ex -> onDeadLetterSinkFailure(record, ex, finalProm));
+
+    final var response = getResponse(failure);
+    incrementEventCount(response);
+    recordDispatchLatency(response, recordContext);
+
+    dlsSender.apply(recordContext.getRecord())
+      .onSuccess(v -> onDeadLetterSinkSuccess(recordContext, finalProm))
+      .onFailure(ex -> onDeadLetterSinkFailure(recordContext, ex, finalProm));
   }
 
-  private void onDeadLetterSinkSuccess(final KafkaConsumerRecord<Object, CloudEvent> record,
+  private void onDeadLetterSinkSuccess(final ConsumerRecordContext recordContext,
                                        final Promise<Void> finalProm) {
-    logDebug("Successfully sent event to the dead letter sink", record);
-    recordDispatcherListener.successfullySentToDeadLetterSink(record);
+    logDebug("Successfully sent event to the dead letter sink", recordContext.getRecord());
+    recordDispatcherListener.successfullySentToDeadLetterSink(recordContext.getRecord());
     finalProm.complete();
   }
 
 
-  private void onDeadLetterSinkFailure(final KafkaConsumerRecord<Object, CloudEvent> record, final Throwable exception,
+  private void onDeadLetterSinkFailure(final ConsumerRecordContext recordContext,
+                                       final Throwable exception,
                                        final Promise<Void> finalProm) {
-    recordDispatcherListener.failedToSendToDeadLetterSink(record, exception);
+    recordDispatcherListener.failedToSendToDeadLetterSink(recordContext.getRecord(), exception);
     finalProm.complete();
   }
 
-  private static KafkaConsumerRecord<Object, CloudEvent> maybeDeserializeValueFromHeaders(KafkaConsumerRecord<Object, CloudEvent> record) {
-    if (record.value() != null) {
-      return record;
+  private static ConsumerRecordContext maybeDeserializeValueFromHeaders(ConsumerRecordContext recordContext) {
+    if (recordContext.getRecord().value() != null) {
+      return recordContext;
     }
     // A valid CloudEvent in the CE binary protocol binding of Kafka
     // might be composed by only Headers.
@@ -220,19 +265,62 @@ public class RecordDispatcherImpl implements RecordDispatcher {
     //
     // That means that we get a record with a null value and some CE
     // headers even though the record is a valid CloudEvent.
-    logDebug("Value is null", record);
-    final var value = cloudEventDeserializer.deserialize(record.record().topic(), record.record().headers(), null);
-    return new KafkaConsumerRecordImpl<>(KafkaConsumerRecordUtils.copyRecordAssigningValue(record.record(), value));
+    logDebug("Value is null", recordContext.getRecord());
+    final var value = cloudEventDeserializer.deserialize(recordContext.getRecord().record().topic(), recordContext.getRecord().record().headers(), null);
+    recordContext.setRecord(new KafkaConsumerRecordImpl<>(KafkaConsumerRecordUtils.copyRecordAssigningValue(recordContext.getRecord().record(), value)));
+    return recordContext;
   }
 
-  private static Function<KafkaConsumerRecord<Object, CloudEvent>, Future<Void>> composeSenderAndSinkHandler(
+  private static Function<KafkaConsumerRecord<Object, CloudEvent>, Future<HttpResponse<?>>> composeSenderAndSinkHandler(
     CloudEventSender sender, ResponseHandler sinkHandler, String senderType) {
     return rec -> sender.send(rec.value())
       .onFailure(ex -> logError("Failed to send event to " + senderType, rec, ex))
       .compose(res ->
         sinkHandler.handle(res)
           .onFailure(ex -> logError("Failed to handle " + senderType + " response", rec, ex))
+          .map(v -> res)
+      );
+  }
+
+  private void incrementEventCount(@Nullable final HttpResponse<?> response) {
+    Metrics
+      .eventCount(getTags(response))
+      .register(meterRegistry)
+      .increment();
+  }
+
+  private void recordDispatchLatency(final HttpResponse<?> response,
+                                     final ConsumerRecordContext recordContext) {
+    final var latency = recordContext.performLatency();
+    logger.debug("Dispatch latency {}", keyValue("latency", latency));
+
+    Metrics
+      .eventDispatchLatency(getTags(response))
+      .register(meterRegistry)
+      .record(latency);
+  }
+
+  private HttpResponse<?> getResponse(final Throwable throwable) {
+    for (Throwable c = throwable; c != null; c = c.getCause()) {
+      if (c instanceof ResponseFailureException) {
+        final var response = (HttpResponse<?>) ((ResponseFailureException) c).getResponse();
+        return response;
+      }
+    }
+    return null;
+  }
+
+  private Tags getTags(HttpResponse<?> response) {
+    Tags tags;
+    if (response == null) {
+      tags = this.noResponseResourceTags;
+    } else {
+      tags = this.resourceContext.getTags().and(
+        Tag.of(Metrics.Tags.RESPONSE_CODE_CLASS, response.statusCode() / 100 + "xx"),
+        Tag.of(Metrics.Tags.RESPONSE_CODE, Integer.toString(response.statusCode()))
       );
+    }
+    return tags;
   }
 
   private static void logError(
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/ResourceContext.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/ResourceContext.java
new file mode 100644
index 000000000..f31a877c7
--- /dev/null
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/ResourceContext.java
@@ -0,0 +1,53 @@
+/*
+ * Copyright © 2018 Knative Authors (knative-dev@googlegroups.com)
+ *
+ * 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 dev.knative.eventing.kafka.broker.dispatcher.impl;
+
+import dev.knative.eventing.kafka.broker.contract.DataPlaneContract;
+import dev.knative.eventing.kafka.broker.core.metrics.Metrics;
+import io.micrometer.core.instrument.Tag;
+import io.micrometer.core.instrument.Tags;
+
+public class ResourceContext {
+
+  private final DataPlaneContract.Resource resource;
+  private final DataPlaneContract.Egress egress;
+  private final Tags tags;
+
+  public ResourceContext(DataPlaneContract.Resource resource, DataPlaneContract.Egress egress) {
+    this.resource = resource;
+    this.egress = egress;
+
+    this.tags = Tags.of(
+      // Resource tags
+      Tag.of(Metrics.Tags.RESOURCE_NAME, resource.getReference().getName()),
+      Tag.of(Metrics.Tags.RESOURCE_NAMESPACE, resource.getReference().getNamespace()),
+      // Egress tags
+      Tag.of(Metrics.Tags.CONSUMER_NAME, egress.getReference().getName())
+    );
+  }
+
+  public DataPlaneContract.Resource getResource() {
+    return resource;
+  }
+
+  public DataPlaneContract.Egress getEgress() {
+    return egress;
+  }
+
+  public Tags getTags() {
+    return tags;
+  }
+}
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/ResponseFailureException.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/ResponseFailureException.java
new file mode 100644
index 000000000..a2a35dc42
--- /dev/null
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/ResponseFailureException.java
@@ -0,0 +1,38 @@
+/*
+ * Copyright © 2018 Knative Authors (knative-dev@googlegroups.com)
+ *
+ * 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 dev.knative.eventing.kafka.broker.dispatcher.impl;
+
+import java.net.http.HttpResponse;
+
+public class ResponseFailureException extends RuntimeException {
+
+  private final HttpResponse<?> response;
+
+  public ResponseFailureException(final HttpResponse response, final String msg) {
+    super(msg);
+    this.response = response;
+  }
+
+  public ResponseFailureException(final HttpResponse<?> response,
+                                  final Throwable ex) {
+    super(ex);
+    this.response = response;
+  }
+
+  public HttpResponse<?> getResponse() {
+    return response;
+  }
+}
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java
index ba9caf8e8..e20477429 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java
@@ -17,6 +17,7 @@ package dev.knative.eventing.kafka.broker.dispatcher.impl.http;
 
 import dev.knative.eventing.kafka.broker.core.tracing.TracingSpan;
 import dev.knative.eventing.kafka.broker.dispatcher.CloudEventSender;
+import dev.knative.eventing.kafka.broker.dispatcher.impl.ResponseFailureException;
 import io.cloudevents.CloudEvent;
 import io.cloudevents.http.vertx.VertxMessageFactory;
 import io.cloudevents.rw.CloudEventRWException;
@@ -80,6 +81,7 @@ public final class WebClientCloudEventSender implements CloudEventSender {
     });
   }
 
+  @SuppressWarnings("rawtypes")
   private void send(final CloudEvent event, final Promise<HttpResponse<Buffer>> breaker) {
     VertxMessageFactory
       .createWriter(client.postAbs(target).putHeader("Prefer", "reply"))
@@ -92,7 +94,10 @@ public final class WebClientCloudEventSender implements CloudEventSender {
 
         if (isRetryableStatusCode(response.statusCode())) {
           logError(event, response);
-          breaker.tryFail("response status code is not 2xx - got: " + response.statusCode());
+          breaker.tryFail(new ResponseFailureException(
+            (java.net.http.HttpResponse) response,
+            "Received failure response, status code: " + response.statusCode())
+          );
           return;
         }
 
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImpl.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImpl.java
index d24a1f636..74126def6 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImpl.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImpl.java
@@ -31,6 +31,7 @@ import dev.knative.eventing.kafka.broker.dispatcher.impl.RecordDispatcherImpl;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.RecordDispatcherMutatorChain;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.ResponseToHttpEndpointHandler;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.ResponseToKafkaTopicHandler;
+import dev.knative.eventing.kafka.broker.dispatcher.impl.ResourceContext;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.BaseConsumerVerticle;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.CloudEventOverridesMutator;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.InvalidCloudEventInterceptor;
@@ -179,6 +180,7 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
 
         final var recordDispatcher = new RecordDispatcherMutatorChain(
           new RecordDispatcherImpl(
+            new ResourceContext(resource, egress),
             filter,
             egressSubscriberSender,
             egressDeadLetterSender,
@@ -190,7 +192,8 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
                 .setConfig(consumerConfigs)
                 // Make sure the policy is propagate for the manually instantiated consumer tracer
                 .setTracingPolicy(TracingPolicy.PROPAGATE)
-            )
+            ),
+            Metrics.getRegistry()
           ),
           new CloudEventOverridesMutator(resource.getCloudEventOverrides())
         );
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index ee60953fc..0c2d2f173 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -31,7 +31,7 @@
     <java.version>15</java.version>
 
     <!-- Maven plugins -->
-    <maven.compiler.plugin.version>3.9.0</maven.compiler.plugin.version>
+    <maven.compiler.plugin.version>3.10.0</maven.compiler.plugin.version>
     <maven.surefire.plugin.version>2.22.2</maven.surefire.plugin.version>
     <maven.shade.plugin.version>3.2.4</maven.shade.plugin.version>
     <maven.editorconfig.plugin.version>0.1.1</maven.editorconfig.plugin.version>
@@ -44,19 +44,19 @@
     <vertx.version>4.2.4</vertx.version>
     <cloudevents.sdk.version>2.3.0</cloudevents.sdk.version>
     <micrometer.version>1.8.2</micrometer.version>
-    <opentelemetry.version>1.10.1</opentelemetry.version>
+    <opentelemetry.version>1.11.0</opentelemetry.version>
     <jackson.version>2.13.1</jackson.version>
     <protobuf.version>3.19.4</protobuf.version>
-    <slf4j.version>1.7.35</slf4j.version>
+    <slf4j.version>1.7.36</slf4j.version>
     <ch.qos.logback.version>1.2.10</ch.qos.logback.version>
     <net.logstash.logback.encoder.version>7.0.1</net.logstash.logback.encoder.version>
     <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.3.0</mokito.junit.jupiter.version>
-    <fabric8.kubernetes.version>5.12.0</fabric8.kubernetes.version>
+    <mokito.junit.jupiter.version>4.3.1</mokito.junit.jupiter.version>
+    <fabric8.kubernetes.version>5.12.1</fabric8.kubernetes.version>
     <kafka.version>2.8.1</kafka.version>
-    <debezium.version>1.8.0.Final</debezium.version>
+    <debezium.version>1.8.1.Final</debezium.version>
   </properties>
 
   <modules>
diff --git a/data-plane/profiler/run.sh b/data-plane/profiler/run.sh
index 31f693929..06726ad9c 100755
--- a/data-plane/profiler/run.sh
+++ b/data-plane/profiler/run.sh
@@ -40,7 +40,7 @@ echo "Profiling event ${EVENT}"
 
 PROJECT_ROOT_DIR=$(dirname $0)/..
 RESOURCES_DIR="$(dirname $0)"/resources
-ASYNC_PROFILER_URL="https://github.com/jvm-profiling-tools/async-profiler/releases/download/v2.0/async-profiler-2.0-linux-x64.tar.gz"
+ASYNC_PROFILER_URL="https://github.com/jvm-profiling-tools/async-profiler/releases/download/v2.6/async-profiler-2.6-linux-x64.tar.gz"
 KAFKA_URL="https://archive.apache.org/dist/kafka/2.6.0/kafka_2.13-2.6.0.tgz"
 LOG_DIR=${LOG_DIR:-"/tmp/eventing-kafka-broker-logs/profiler"}
 
diff --git a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/handler/IngressRequestHandlerImpl.java b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/handler/IngressRequestHandlerImpl.java
index c4d35b474..fd0448664 100644
--- a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/handler/IngressRequestHandlerImpl.java
+++ b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/handler/IngressRequestHandlerImpl.java
@@ -82,8 +82,8 @@ public class IngressRequestHandlerImpl implements IngressRequestHandler {
   public void handle(final RequestContext requestContext, final IngressProducer producer) {
 
     final var resourceTags = Tags.of(
-      Tag.of("name", producer.getReference().getName()),
-      Tag.of("namespace_name", producer.getReference().getNamespace())
+      Tag.of(Metrics.Tags.RESOURCE_NAME, producer.getReference().getName()),
+      Tag.of(Metrics.Tags.RESOURCE_NAMESPACE, producer.getReference().getNamespace())
     );
 
     requestToRecordMapper
diff --git a/hack/artifacts-env.sh b/hack/artifacts-env.sh
index c1da54892..7b0ae2796 100755
--- a/hack/artifacts-env.sh
+++ b/hack/artifacts-env.sh
@@ -15,4 +15,6 @@ export EVENTING_KAFKA_SINK_PROMETHEUS_OPERATOR_ARTIFACT="eventing-kafka-sink-pro
 export EVENTING_KAFKA_CHANNEL_ARTIFACT="eventing-kafka-channel.yaml"
 export EVENTING_KAFKA_CHANNEL_PROMETHEUS_OPERATOR_ARTIFACT="eventing-kafka-channel-prometheus-operator.yaml"
 
+export EVENTING_KAFKA_POST_INSTALL_ARTIFACT="eventing-kafka-post-install.yaml"
+
 export EVENTING_KAFKA_ARTIFACT="eventing-kafka.yaml"
diff --git a/hack/control-plane.sh b/hack/control-plane.sh
index 6c748dad8..52152430a 100644
--- a/hack/control-plane.sh
+++ b/hack/control-plane.sh
@@ -16,10 +16,13 @@
 
 source $(pwd)/hack/label.sh
 
-readonly CONTROL_PLANE_CONFIG_DIR=control-plane/config
+readonly CONTROL_PLANE_CONFIG_DIR=control-plane/config/eventing-kafka-broker
+readonly CONTROL_PLANE_POST_INSTALL_CONFIG_DIR=control-plane/config/post-install
 
 # Note: do not change this function name, it's used during releases.
 function control_plane_setup() {
-  ko resolve ${KO_FLAGS} -Rf "${CONTROL_PLANE_CONFIG_DIR}" | "${LABEL_YAML_CMD[@]}" >>"${EVENTING_KAFKA_CONTROL_PLANE_ARTIFACT}"
+  ko resolve ${KO_FLAGS} -Rf "${CONTROL_PLANE_CONFIG_DIR}" | "${LABEL_YAML_CMD[@]}" >>"${EVENTING_KAFKA_CONTROL_PLANE_ARTIFACT}" &&
+    ko resolve ${KO_FLAGS} -Rf "${CONTROL_PLANE_POST_INSTALL_CONFIG_DIR}" | "${LABEL_YAML_CMD[@]}" >>"${EVENTING_KAFKA_POST_INSTALL_ARTIFACT}"
+
   return $?
 }
diff --git a/hack/release.sh b/hack/release.sh
index e6a82d0ac..b47f0f00d 100755
--- a/hack/release.sh
+++ b/hack/release.sh
@@ -66,6 +66,7 @@ function build_release() {
     "${EVENTING_KAFKA_SINK_PROMETHEUS_OPERATOR_ARTIFACT}"
     "${EVENTING_KAFKA_CHANNEL_ARTIFACT}"
     "${EVENTING_KAFKA_CHANNEL_PROMETHEUS_OPERATOR_ARTIFACT}"
+    "${EVENTING_KAFKA_POST_INSTALL_ARTIFACT}"
   )
 
   # ARTIFACTS_TO_PUBLISH has to be a string, not an array.
diff --git a/hack/update-checksums.sh b/hack/update-checksums.sh
index 54f0d5d0f..df7c14e6c 100755
--- a/hack/update-checksums.sh
+++ b/hack/update-checksums.sh
@@ -26,6 +26,6 @@ fi
 
 source $(dirname $0)/../vendor/knative.dev/hack/library.sh
 
-go run "${REPO_ROOT_DIR}/vendor/knative.dev/pkg/configmap/hash-gen" "${REPO_ROOT_DIR}"/control-plane/config/200-controller/100-config-tracing.yaml
-go run "${REPO_ROOT_DIR}/vendor/knative.dev/pkg/configmap/hash-gen" "${REPO_ROOT_DIR}"/control-plane/config/200-controller/100-config-logging.yaml
-go run "${REPO_ROOT_DIR}/vendor/knative.dev/pkg/configmap/hash-gen" "${REPO_ROOT_DIR}"/control-plane/config/200-controller/100-config-kafka-leader-election.yaml
+go run "${REPO_ROOT_DIR}/vendor/knative.dev/pkg/configmap/hash-gen" "${REPO_ROOT_DIR}"/control-plane/config/eventing-kafka-broker/200-controller/100-config-tracing.yaml
+go run "${REPO_ROOT_DIR}/vendor/knative.dev/pkg/configmap/hash-gen" "${REPO_ROOT_DIR}"/control-plane/config/eventing-kafka-broker/200-controller/100-config-logging.yaml
+go run "${REPO_ROOT_DIR}/vendor/knative.dev/pkg/configmap/hash-gen" "${REPO_ROOT_DIR}"/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-leader-election.yaml


Test patch:
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go
index 3e00b64c7..504c7e9e8 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go
@@ -118,6 +118,45 @@ func TestConsumerGroup_Validate(t *testing.T) {
 			},
 			wantErr: false,
 		},
+		{
+			name: "invalid multiple reply strategies (topic, destination)",
+			ctx:  context.Background(),
+			given: &ConsumerGroup{
+				Spec: ConsumerGroupSpec{
+					Replicas: pointer.Int32Ptr(1),
+					Selector: map[string]string{"app": "app"},
+					Template: ConsumerTemplateSpec{
+						Spec: ConsumerSpec{
+							Topics: []string{"t1"},
+							Configs: ConsumerConfigs{
+								Configs: map[string]string{
+									"group.id":          "g1",
+									"bootstrap.servers": "kafka:9092",
+								},
+							},
+							Reply: &ReplyStrategy{
+								TopicReply: &TopicReply{Enabled: true},
+								URLReply:   &DestinationReply{Enabled: true},
+							},
+							Delivery: &DeliverySpec{
+								DeliverySpec: &eventingduck.DeliverySpec{},
+							},
+							Subscriber: duckv1.Destination{
+								URI: &apis.URL{
+									Scheme: "http",
+									Host:   "127.0.0.1",
+								},
+							},
+							PodBind: &PodBind{
+								PodName:      "p-0",
+								PodNamespace: "ns",
+							},
+						},
+					},
+				},
+			},
+			wantErr: true,
+		},
 		{
 			name: "invalid no pod name",
 			ctx:  context.Background(),
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types_test.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types_test.go
index 4ff7f61a9..6c5d6643f 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types_test.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types_test.go
@@ -17,9 +17,13 @@
 package v1alpha1
 
 import (
+	"io"
+	"sort"
 	"testing"
+	"time"
 
 	"github.com/google/go-cmp/cmp"
+	"github.com/stretchr/testify/require"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
@@ -80,3 +84,55 @@ func TestConsumer_GetConsumerGroup(t *testing.T) {
 		})
 	}
 }
+
+func TestByReadinessAndCreationTime(t *testing.T) {
+
+	consumers := []*Consumer{
+		// Not ready, first created
+		func() *Consumer {
+			c := &Consumer{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:              "c1",
+					Namespace:         "ns",
+					CreationTimestamp: metav1.Time{Time: time.Now().Add(time.Second)},
+				},
+			}
+			c.GetConditionSet().Manage(c.GetStatus()).InitializeConditions()
+			return c
+		}(),
+		// Not ready, second created
+		func() *Consumer {
+			c := &Consumer{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:              "c2",
+					Namespace:         "ns",
+					CreationTimestamp: metav1.Time{Time: time.Now().Add(time.Minute)},
+				},
+			}
+			c.GetConditionSet().Manage(c.GetStatus()).InitializeConditions()
+			_ = c.MarkBindFailed(io.EOF)
+			return c
+		}(),
+		// Ready, third created
+		func() *Consumer {
+			c := &Consumer{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:              "c3",
+					Namespace:         "ns",
+					CreationTimestamp: metav1.Time{Time: time.Now().Add(time.Hour)},
+				},
+			}
+			c.GetConditionSet().Manage(c.GetStatus()).InitializeConditions()
+			c.MarkBindSucceeded()
+			c.MarkReconcileContractSucceeded()
+			return c
+		}(),
+	}
+
+	sort.Stable(ByReadinessAndCreationTime(consumers))
+
+	require.Equal(t, 3, len(consumers))
+	require.Equal(t, "c3", consumers[0].Name)
+	require.Equal(t, "c1", consumers[1].Name)
+	require.Equal(t, "c2", consumers[2].Name)
+}
diff --git a/control-plane/pkg/reconciler/channel/channel_test.go b/control-plane/pkg/reconciler/channel/channel_test.go
index c5e71fdac..a888eee34 100644
--- a/control-plane/pkg/reconciler/channel/channel_test.go
+++ b/control-plane/pkg/reconciler/channel/channel_test.go
@@ -353,6 +353,79 @@ func TestReconcileKind(t *testing.T) {
 				testProber: probertesting.MockProber(prober.StatusUnknown),
 			},
 		},
+		{
+			Name: "Reconciled failed - with single fresh subscriber without URI",
+			Objects: []runtime.Object{
+				NewChannel(WithSubscribers(Subscriber1(WithFreshSubscriber, WithNoSubscriberURI))),
+				NewService(),
+				ChannelReceiverPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				ChannelDispatcherPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
+				}),
+				NewConfigMapWithBinaryData(&env, nil),
+			},
+			Key: testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&env, &contract.Contract{
+					Generation: 1,
+					Resources: []*contract.Resource{
+						{
+							Uid:              ChannelUUID,
+							Topics:           []string{ChannelTopic()},
+							BootstrapServers: ChannelBootstrapServers,
+							Reference:        ChannelReference(),
+							Ingress: &contract.Ingress{
+								IngressType: &contract.Ingress_Path{
+									Path: receiver.Path(ChannelNamespace, ChannelName),
+								},
+							},
+							Egresses: []*contract.Egress{},
+						},
+					},
+				}),
+				ChannelReceiverPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+				ChannelDispatcherPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+			},
+			WantErr:                 true,
+			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewChannel(
+						WithInitKafkaChannelConditions,
+						StatusConfigParsed,
+						StatusConfigMapUpdatedReady(&env),
+						StatusTopicReadyWithName(ChannelTopic()),
+						StatusDataPlaneAvailable,
+						//StatusInitialOffsetsCommitted,
+						WithSubscribers(Subscriber1(WithFreshSubscriber, WithNoSubscriberURI)),
+					),
+				},
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+				Eventf(
+					corev1.EventTypeWarning,
+					"InternalError",
+					"failed to resolve subscriber config: failed to resolve Subscription.Spec.Subscriber: empty subscriber URI",
+				),
+			},
+		},
 		{
 			Name: "Reconciled normal - with single fresh subscriber - no auth",
 			Objects: []runtime.Object{
diff --git a/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go b/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
index 9e9ef95f3..82995e928 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
@@ -131,6 +131,85 @@ func TestReconcileKind(t *testing.T) {
 				},
 			},
 		},
+		{
+			Name: "Consumer update",
+			Objects: []runtime.Object{
+				NewService(),
+				NewConsumer(1,
+					ConsumerSpec(NewConsumerSpec(
+						ConsumerTopics("t1", "t2"),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+						ConsumerVReplicas(1),
+						ConsumerPlacement(kafkainternals.PodBind{PodName: "p1", PodNamespace: systemNamespace}),
+						ConsumerSubscriber(NewSourceSinkReference()),
+					)),
+				),
+				NewConsumerGroup(
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics("t1", "t2"),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+						ConsumerSubscriber(NewSourceSink2Reference()),
+					)),
+					ConsumerGroupReplicas(1),
+					ConsumerForTrigger(),
+				),
+			},
+			Key: ConsumerGroupTestKey,
+			OtherTestData: map[string]interface{}{
+				testSchedulerKey: SchedulerFunc(func(vpod scheduler.VPod) ([]eventingduckv1alpha1.Placement, error) {
+					return []eventingduckv1alpha1.Placement{
+						{PodName: "p1", VReplicas: 1},
+					}, nil
+				}),
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewConsumer(1,
+						ConsumerSpec(NewConsumerSpec(
+							ConsumerTopics("t1", "t2"),
+							ConsumerConfigs(
+								ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+								ConsumerGroupIdConfig("my.group.id"),
+							),
+							ConsumerVReplicas(1),
+							ConsumerPlacement(kafkainternals.PodBind{PodName: "p1", PodNamespace: systemNamespace}),
+							ConsumerSubscriber(NewSourceSink2Reference()),
+						)),
+					),
+				},
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: func() runtime.Object {
+						cg := NewConsumerGroup(
+							ConsumerGroupConsumerSpec(NewConsumerSpec(
+								ConsumerTopics("t1", "t2"),
+								ConsumerConfigs(
+									ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+									ConsumerGroupIdConfig("my.group.id"),
+								),
+								ConsumerSubscriber(NewSourceSink2Reference()),
+							)),
+							ConsumerGroupReplicas(1),
+							ConsumerGroupStatusReplicas(0),
+							ConsumerForTrigger(),
+						)
+						cg.Status.Placements = []eventingduckv1alpha1.Placement{
+							{PodName: "p1", VReplicas: 1},
+						}
+						_ = cg.MarkReconcileConsumersFailed("PropagateSubscriberURI", ErrNoSubscriberURI)
+						cg.MarkScheduleSucceeded()
+						return cg
+					}(),
+				},
+			},
+		},
 		{
 			Name: "Consumers in multiple pods, one exists - not ready",
 			Objects: []runtime.Object{
diff --git a/control-plane/pkg/reconciler/source/v2/sourcev2_test.go b/control-plane/pkg/reconciler/source/v2/sourcev2_test.go
index f920ae1f9..070abf219 100644
--- a/control-plane/pkg/reconciler/source/v2/sourcev2_test.go
+++ b/control-plane/pkg/reconciler/source/v2/sourcev2_test.go
@@ -33,11 +33,12 @@ import (
 	"knative.dev/pkg/logging"
 	. "knative.dev/pkg/reconciler/testing"
 
-	fakeconsumergroupinformer "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client/fake"
 	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
 	fakeeventingkafkaclient "knative.dev/eventing-kafka/pkg/client/injection/client/fake"
 	eventingkafkasourcereconciler "knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource"
 
+	fakeconsumergroupinformer "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client/fake"
+
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/testing"
 )
 
@@ -69,6 +70,7 @@ func TestReconcileKind(t *testing.T) {
 						ConsumerAuth(NewConsumerSpecAuth()),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerReply(ConsumerNoReply()),
 					)),
 					ConsumerGroupReplicas(1),
 				),
@@ -106,6 +108,7 @@ func TestReconcileKind(t *testing.T) {
 						ConsumerAuth(NewConsumerSpecAuth()),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerReply(ConsumerNoReply()),
 						ConsumerCloudEventOverrides(&duckv1.CloudEventOverrides{
 							Extensions: map[string]string{"a": "foo", "b": "foo"},
 						}),
@@ -125,6 +128,64 @@ func TestReconcileKind(t *testing.T) {
 				},
 			},
 		},
+		{
+			Name: "Reconciled normal - existing cg with sink update",
+			Objects: []runtime.Object{
+				NewSource(WithSourceSink(NewSourceSink2Reference())),
+				NewConsumerGroup(
+					WithConsumerGroupName(SourceUUID),
+					WithConsumerGroupNamespace(SourceNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(SourceTopics[0], SourceTopics[1]),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(SourceConsumerGroup),
+							ConsumerBootstrapServersConfig(SourceBootstrapServers),
+						),
+						ConsumerAuth(NewConsumerSpecAuth()),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerReply(ConsumerNoReply()),
+					)),
+					ConsumerGroupReplicas(1),
+					ConsumerGroupReady,
+				),
+			},
+			Key: testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewConsumerGroup(
+						WithConsumerGroupName(SourceUUID),
+						WithConsumerGroupNamespace(SourceNamespace),
+						WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+						WithConsumerGroupLabels(nil),
+						ConsumerGroupConsumerSpec(NewConsumerSpec(
+							ConsumerTopics(SourceTopics[0], SourceTopics[1]),
+							ConsumerConfigs(
+								ConsumerGroupIdConfig(SourceConsumerGroup),
+								ConsumerBootstrapServersConfig(SourceBootstrapServers),
+							),
+							ConsumerAuth(NewConsumerSpecAuth()),
+							ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+							ConsumerSubscriber(NewSourceSink2Reference()),
+							ConsumerReply(ConsumerNoReply()),
+						)),
+						ConsumerGroupReplicas(1),
+						ConsumerGroupReady,
+					),
+				},
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSource(
+						WithSourceSink(NewSourceSink2Reference()),
+						StatusSourceConsumerGroup(),
+						StatusSourceSinkResolved(""),
+					),
+				},
+			},
+		},
 		{
 			Name: "Reconciled normal - existing cg with update",
 			Objects: []runtime.Object{
@@ -138,8 +199,7 @@ func TestReconcileKind(t *testing.T) {
 					ConsumerGroupReady,
 				),
 			},
-			Key:         testKey,
-			WantCreates: []runtime.Object{},
+			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
 				{
 					Object: NewConsumerGroup(
@@ -156,6 +216,7 @@ func TestReconcileKind(t *testing.T) {
 							ConsumerAuth(NewConsumerSpecAuth()),
 							ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 							ConsumerSubscriber(NewSourceSinkReference()),
+							ConsumerReply(ConsumerNoReply()),
 						)),
 						ConsumerGroupReplicas(1),
 						ConsumerGroupReady,
@@ -201,6 +262,7 @@ func TestReconcileKind(t *testing.T) {
 							ConsumerAuth(NewConsumerSpecAuth()),
 							ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 							ConsumerSubscriber(NewSourceSinkReference()),
+							ConsumerReply(ConsumerNoReply()),
 						)),
 						ConsumerGroupReplicas(1),
 					),
@@ -233,13 +295,13 @@ func TestReconcileKind(t *testing.T) {
 						ConsumerAuth(NewConsumerSpecAuth()),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerReply(ConsumerNoReply()),
 					)),
 					ConsumerGroupReplicas(1),
 					ConsumerGroupReady,
 				),
 			},
-			Key:         testKey,
-			WantCreates: []runtime.Object{},
+			Key: testKey,
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
 					Object: NewSource(
@@ -267,6 +329,7 @@ func TestReconcileKind(t *testing.T) {
 						ConsumerAuth(NewConsumerSpecAuth()),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerReply(ConsumerNoReply()),
 					)),
 					ConsumerGroupReplicas(1),
 				),
@@ -300,6 +363,7 @@ func TestReconcileKind(t *testing.T) {
 						ConsumerAuth(NewConsumerSpecAuth()),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerReply(ConsumerNoReply()),
 					)),
 					ConsumerGroupReplicas(1),
 					WithConsumerGroupFailed("failed", "failed"),
@@ -334,6 +398,7 @@ func TestReconcileKind(t *testing.T) {
 						ConsumerAuth(NewConsumerSpecAuth()),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerReply(ConsumerNoReply()),
 					)),
 					ConsumerGroupReplicas(1),
 					ConsumerGroupReplicasStatus(1),
diff --git a/control-plane/pkg/reconciler/testing/objects_channel.go b/control-plane/pkg/reconciler/testing/objects_channel.go
index cdfe2c636..a1147647b 100644
--- a/control-plane/pkg/reconciler/testing/objects_channel.go
+++ b/control-plane/pkg/reconciler/testing/objects_channel.go
@@ -261,6 +261,18 @@ func WithFreshSubscriber(sub *SubscriberInfo) {
 	sub.status = nil
 }
 
+func WithNoSubscriberURI(sub *SubscriberInfo) {
+	sub.spec.SubscriberURI = nil
+	if sub.status == nil {
+		sub.status = &eventingduckv1.SubscriberStatus{
+			UID:                sub.spec.UID,
+			ObservedGeneration: sub.spec.Generation,
+		}
+	}
+	sub.status.Ready = "False"
+	sub.status.Message = "Subscription not ready: failed to resolve subscriber config: failed to resolve Subscription.Spec.Subscriber: empty subscriber URI"
+}
+
 func WithUnreadySubscriber(sub *SubscriberInfo) {
 	sub.status.Ready = "False"
 }
diff --git a/control-plane/pkg/reconciler/testing/objects_common.go b/control-plane/pkg/reconciler/testing/objects_common.go
index 54960b4c1..d676478ee 100644
--- a/control-plane/pkg/reconciler/testing/objects_common.go
+++ b/control-plane/pkg/reconciler/testing/objects_common.go
@@ -46,6 +46,8 @@ const (
 	ServiceNamespace = "test-service-namespace"
 	ServiceName      = "test-service"
 
+	Service2Name = "test-service-2"
+
 	TriggerUUID = "e7185016-5d98-4b54-84e8-3b1cd4acc6b5"
 
 	SecretResourceVersion = "1234"
@@ -75,6 +77,23 @@ func NewService(mutations ...func(*corev1.Service)) *corev1.Service {
 	return s
 }
 
+func NewService2(mutations ...func(*corev1.Service)) *corev1.Service {
+	s := &corev1.Service{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Service",
+			APIVersion: "v1",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      Service2Name,
+			Namespace: ServiceNamespace,
+		},
+	}
+	for _, mut := range mutations {
+		mut(s)
+	}
+	return s
+}
+
 func WithServiceNamespace(ns string) func(s *corev1.Service) {
 	return func(s *corev1.Service) {
 		s.Namespace = ns
diff --git a/control-plane/pkg/reconciler/testing/objects_consumer.go b/control-plane/pkg/reconciler/testing/objects_consumer.go
index bb2a6ca3b..1a99d7f2f 100644
--- a/control-plane/pkg/reconciler/testing/objects_consumer.go
+++ b/control-plane/pkg/reconciler/testing/objects_consumer.go
@@ -226,3 +226,17 @@ func ConsumerReady() ConsumerOption {
 		}
 	}
 }
+
+func ConsumerReply(s *kafkainternals.ReplyStrategy) ConsumerSpecOption {
+	return func(c *kafkainternals.ConsumerSpec) {
+		c.Reply = s
+	}
+}
+
+func ConsumerNoReply() *kafkainternals.ReplyStrategy {
+	return &kafkainternals.ReplyStrategy{NoReply: &kafkainternals.NoReply{Enabled: true}}
+}
+
+func ConsumerTopicReply() *kafkainternals.ReplyStrategy {
+	return &kafkainternals.ReplyStrategy{TopicReply: &kafkainternals.TopicReply{Enabled: true}}
+}
diff --git a/control-plane/pkg/reconciler/testing/objects_consumergroup.go b/control-plane/pkg/reconciler/testing/objects_consumergroup.go
index 79cca14d2..519d34307 100644
--- a/control-plane/pkg/reconciler/testing/objects_consumergroup.go
+++ b/control-plane/pkg/reconciler/testing/objects_consumergroup.go
@@ -70,6 +70,12 @@ func NewConsumerGroup(opts ...ConsumerGroupOption) *kafkainternals.ConsumerGroup
 	return cg
 }
 
+func ConsumerGroupSubscriber(d duckv1.Destination) ConsumerGroupOption {
+	return func(cg *kafkainternals.ConsumerGroup) {
+		cg.Spec.Template.Spec.Subscriber = d
+	}
+}
+
 func ConsumerGroupReady(cg *kafkainternals.ConsumerGroup) {
 	cg.Status.Conditions = duckv1.Conditions{
 		{
diff --git a/control-plane/pkg/reconciler/testing/objects_source.go b/control-plane/pkg/reconciler/testing/objects_source.go
index bb79545db..76e106eb1 100644
--- a/control-plane/pkg/reconciler/testing/objects_source.go
+++ b/control-plane/pkg/reconciler/testing/objects_source.go
@@ -108,6 +108,25 @@ func NewSourceSinkReference() duckv1.Destination {
 	}
 }
 
+func NewSourceSink2Reference() duckv1.Destination {
+	s := NewService2()
+	return duckv1.Destination{
+		Ref: &duckv1.KReference{
+			Kind:       s.Kind,
+			Namespace:  s.Namespace,
+			Name:       s.Name,
+			APIVersion: s.APIVersion,
+		},
+	}
+}
+
+func WithSourceSink(d duckv1.Destination) KRShapedOption {
+	return func(obj duckv1.KRShaped) {
+		s := obj.(*sources.KafkaSource)
+		s.Spec.Sink = d
+	}
+}
+
 func SourceDispatcherPod(namespace string, annotations map[string]string) runtime.Object {
 	return &corev1.Pod{
 		TypeMeta: metav1.TypeMeta{
diff --git a/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go b/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go
index 6e6fe7e24..f723e0743 100644
--- a/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go
+++ b/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go
@@ -91,6 +91,7 @@ func TestReconcileKind(t *testing.T) {
 						),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
 						ConsumerFilters(NewConsumerSpecFilters()),
+						ConsumerReply(ConsumerTopicReply()),
 					)),
 				),
 			},
@@ -128,6 +129,7 @@ func TestReconcileKind(t *testing.T) {
 						),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 						ConsumerFilters(NewConsumerSpecFilters()),
+						ConsumerReply(ConsumerTopicReply()),
 					)),
 				),
 			},
@@ -166,6 +168,7 @@ func TestReconcileKind(t *testing.T) {
 						),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
 						ConsumerFilters(NewConsumerSpecFilters()),
+						ConsumerReply(ConsumerTopicReply()),
 					)),
 				),
 			},
@@ -244,6 +247,7 @@ func TestReconcileKind(t *testing.T) {
 							),
 							ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
 							ConsumerFilters(NewConsumerSpecFilters()),
+							ConsumerReply(ConsumerTopicReply()),
 						)),
 						ConsumerGroupReady,
 					),
@@ -291,6 +295,7 @@ func TestReconcileKind(t *testing.T) {
 							),
 							ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
 							ConsumerFilters(NewConsumerSpecFilters()),
+							ConsumerReply(ConsumerTopicReply()),
 						)),
 					),
 				},
@@ -326,6 +331,7 @@ func TestReconcileKind(t *testing.T) {
 						),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
 						ConsumerFilters(NewConsumerSpecFilters()),
+						ConsumerReply(ConsumerTopicReply()),
 					)),
 					ConsumerGroupReady,
 				),
@@ -362,6 +368,7 @@ func TestReconcileKind(t *testing.T) {
 						),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
 						ConsumerFilters(NewConsumerSpecFilters()),
+						ConsumerReply(ConsumerTopicReply()),
 					)),
 				),
 			},
@@ -397,6 +404,7 @@ func TestReconcileKind(t *testing.T) {
 						),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
 						ConsumerFilters(NewConsumerSpecFilters()),
+						ConsumerReply(ConsumerTopicReply()),
 					)),
 					WithConsumerGroupFailed("failed", "failed"),
 				),
@@ -433,6 +441,7 @@ func TestReconcileKind(t *testing.T) {
 						),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
 						ConsumerFilters(NewConsumerSpecFilters()),
+						ConsumerReply(ConsumerTopicReply()),
 					)),
 					WithDeadLetterSinkURI(url.String()),
 				),
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherTest.java
index 315f3aad5..a297cf662 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherTest.java
@@ -15,6 +15,8 @@
  */
 package dev.knative.eventing.kafka.broker.dispatcher.impl;
 
+import dev.knative.eventing.kafka.broker.contract.DataPlaneContract;
+import dev.knative.eventing.kafka.broker.core.metrics.Metrics;
 import dev.knative.eventing.kafka.broker.core.testing.CoreObjects;
 import dev.knative.eventing.kafka.broker.dispatcher.CloudEventSender;
 import dev.knative.eventing.kafka.broker.dispatcher.CloudEventSenderMock;
@@ -24,17 +26,25 @@ import dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcherListener;
 import dev.knative.eventing.kafka.broker.dispatcher.ResponseHandler;
 import dev.knative.eventing.kafka.broker.dispatcher.ResponseHandlerMock;
 import io.cloudevents.CloudEvent;
+import io.micrometer.core.instrument.search.MeterNotFoundException;
+import io.micrometer.prometheus.PrometheusConfig;
+import io.micrometer.prometheus.PrometheusMeterRegistry;
 import io.vertx.core.Future;
 import io.vertx.junit5.VertxExtension;
 import io.vertx.junit5.VertxTestContext;
 import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
 import io.vertx.kafka.client.consumer.impl.KafkaConsumerRecordImpl;
+import io.vertx.micrometer.MicrometerMetricsOptions;
+import io.vertx.micrometer.backends.BackendRegistries;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 import static org.mockito.ArgumentMatchers.any;
@@ -48,18 +58,38 @@ import static org.mockito.Mockito.when;
 @ExtendWith(VertxExtension.class)
 public class RecordDispatcherTest {
 
+  private static final ResourceContext resourceContext = new ResourceContext(
+    DataPlaneContract.Resource.newBuilder().build(),
+    DataPlaneContract.Egress.newBuilder().build()
+  );
+
+  static {
+    BackendRegistries.setupBackend(new MicrometerMetricsOptions()
+      .setMicrometerRegistry(new PrometheusMeterRegistry(PrometheusConfig.DEFAULT))
+      .setRegistryName(Metrics.METRICS_REGISTRY_NAME));
+  }
+
+  private PrometheusMeterRegistry registry;
+
+  @BeforeEach
+  public void setUp() {
+    registry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
+  }
+
   @Test
   public void shouldNotSendToSubscriberNorToDeadLetterSinkIfValueDoesntMatch() {
 
     final RecordDispatcherListener receiver = offsetManagerMock();
 
     final var dispatcherHandler = new RecordDispatcherImpl(
+      resourceContext,
       value -> false,
       CloudEventSender.noop("subscriber send called"),
       CloudEventSender.noop("DLS send called"),
       new ResponseHandlerMock(),
       receiver,
-      null
+      null,
+      registry
     );
 
     final var record = record();
@@ -70,6 +100,10 @@ public class RecordDispatcherTest {
     verify(receiver, never()).successfullySentToSubscriber(any());
     verify(receiver, never()).successfullySentToDeadLetterSink(any());
     verify(receiver, never()).failedToSendToDeadLetterSink(any(), any());
+
+    assertNoEventCount();
+    assertNoEventDispatchLatency();
+    assertEventProcessingLatency();
   }
 
   @Test
@@ -79,12 +113,14 @@ public class RecordDispatcherTest {
     final RecordDispatcherListener receiver = offsetManagerMock();
 
     final var dispatcherHandler = new RecordDispatcherImpl(
-      value -> true, new CloudEventSenderMock(
-      record -> {
-        sendCalled.set(true);
-        return Future.succeededFuture();
-      }
-    ),
+      resourceContext,
+      value -> true,
+      new CloudEventSenderMock(
+        record -> {
+          sendCalled.set(true);
+          return Future.succeededFuture();
+        }
+      ),
       new CloudEventSenderMock(
         record -> {
           fail("DLS send called");
@@ -93,7 +129,8 @@ public class RecordDispatcherTest {
       ),
       new ResponseHandlerMock(),
       receiver,
-      null
+      null,
+      registry
     );
     final var record = record();
     dispatcherHandler.dispatch(record);
@@ -104,6 +141,10 @@ public class RecordDispatcherTest {
     verify(receiver, never()).successfullySentToDeadLetterSink(any());
     verify(receiver, never()).failedToSendToDeadLetterSink(any(), any());
     verify(receiver, never()).recordDiscarded(any());
+
+    assertEventDispatchLatency();
+    assertEventProcessingLatency();
+    assertEventCount();
   }
 
   @Test
@@ -114,12 +155,14 @@ public class RecordDispatcherTest {
     final RecordDispatcherListener receiver = offsetManagerMock();
 
     final var dispatcherHandler = new RecordDispatcherImpl(
-      value -> true, new CloudEventSenderMock(
-      record -> {
-        subscriberSenderSendCalled.set(true);
-        return Future.failedFuture("");
-      }
-    ),
+      resourceContext,
+      value -> true,
+      new CloudEventSenderMock(
+        record -> {
+          subscriberSenderSendCalled.set(true);
+          return Future.failedFuture("");
+        }
+      ),
       new CloudEventSenderMock(
         record -> {
           dlsSenderSendCalled.set(true);
@@ -127,7 +170,8 @@ public class RecordDispatcherTest {
         }
       ), new ResponseHandlerMock(),
       receiver,
-      null
+      null,
+      registry
     );
     final var record = record();
     dispatcherHandler.dispatch(record);
@@ -139,6 +183,10 @@ public class RecordDispatcherTest {
     verify(receiver, never()).successfullySentToSubscriber(any());
     verify(receiver, never()).failedToSendToDeadLetterSink(any(), any());
     verify(receiver, never()).recordDiscarded(any());
+
+    assertEventDispatchLatency();
+    assertEventProcessingLatency();
+    assertEventCount();
   }
 
   @Test
@@ -149,6 +197,7 @@ public class RecordDispatcherTest {
     final RecordDispatcherListener receiver = offsetManagerMock();
 
     final var dispatcherHandler = new RecordDispatcherImpl(
+      resourceContext,
       value -> true, new CloudEventSenderMock(
       record -> {
         subscriberSenderSendCalled.set(true);
@@ -163,7 +212,8 @@ public class RecordDispatcherTest {
       ),
       new ResponseHandlerMock(),
       receiver,
-      null
+      null,
+      registry
     );
     final var record = record();
     dispatcherHandler.dispatch(record);
@@ -175,6 +225,10 @@ public class RecordDispatcherTest {
     verify(receiver, never()).successfullySentToDeadLetterSink(any());
     verify(receiver, never()).successfullySentToSubscriber(any());
     verify(receiver, never()).recordDiscarded(any());
+
+    assertEventDispatchLatency();
+    assertEventProcessingLatency();
+    assertEventCount();
   }
 
   @Test
@@ -183,6 +237,7 @@ public class RecordDispatcherTest {
     final RecordDispatcherListener receiver = offsetManagerMock();
 
     final var dispatcherHandler = new RecordDispatcherImpl(
+      resourceContext,
       value -> true,
       new CloudEventSenderMock(
         record -> {
@@ -193,7 +248,8 @@ public class RecordDispatcherTest {
       CloudEventSender.noop("No DLS configured"),
       new ResponseHandlerMock(),
       receiver,
-      null
+      null,
+      registry
     );
     final var record = record();
     dispatcherHandler.dispatch(record);
@@ -204,6 +260,10 @@ public class RecordDispatcherTest {
     verify(receiver, never()).successfullySentToDeadLetterSink(any());
     verify(receiver, never()).successfullySentToSubscriber(any());
     verify(receiver, never()).recordDiscarded(any());
+
+    assertEventDispatchLatency();
+    assertEventProcessingLatency();
+    assertEventCount();
   }
 
   @Test
@@ -219,8 +279,14 @@ public class RecordDispatcherTest {
     when(deadLetterSender.close()).thenReturn(Future.succeededFuture());
 
     final RecordDispatcher recordDispatcher = new RecordDispatcherImpl(
-      Filter.noop(), subscriberSender,
-      deadLetterSender, sinkResponseHandler, offsetManagerMock(), null);
+      resourceContext,
+      Filter.noop(),
+      subscriberSender,
+      deadLetterSender,
+      sinkResponseHandler,
+      offsetManagerMock(),
+      null,
+      Metrics.getRegistry());
 
     recordDispatcher.close()
       .onFailure(context::failNow)
@@ -239,4 +305,80 @@ public class RecordDispatcherTest {
   public static RecordDispatcherListener offsetManagerMock() {
     return mock(RecordDispatcherListener.class);
   }
+
+  private void assertNoEventCount() {
+    var counterNotFound = false;
+    try {
+      assertThat(
+        registry
+          .get(Metrics.EVENTS_COUNT)
+          .counters()
+          .stream()
+          .reduce((a, b) -> b)
+          .isEmpty()
+      ).isTrue();
+    } catch (MeterNotFoundException ignored) {
+      counterNotFound = true;
+    }
+    assertThat(counterNotFound).isTrue();
+  }
+
+  private void assertEventCount() {
+    assertThat(
+      registry
+        .get(Metrics.EVENTS_COUNT)
+        .counters()
+        .stream()
+        .reduce((a, b) -> b)
+        .get()
+        .count()
+    ).isGreaterThan(0);
+  }
+
+
+  private void assertNoEventDispatchLatency() {
+    var dispatchLatencyNotFound = false;
+    try {
+      assertThat(
+        registry
+          .get(Metrics.EVENT_DISPATCH_LATENCY)
+          .summaries()
+          .stream()
+          .reduce((a, b) -> b)
+          .isEmpty()
+      ).isTrue();
+    } catch (MeterNotFoundException ignored) {
+      dispatchLatencyNotFound = true;
+    }
+    assertThat(dispatchLatencyNotFound).isTrue();
+  }
+
+  private void assertEventDispatchLatency() {
+    await().untilAsserted(() ->
+      assertThat(
+        registry
+          .get(Metrics.EVENT_DISPATCH_LATENCY)
+          .summaries()
+          .stream()
+          .reduce((a, b) -> b)
+          .get()
+          .max()
+      ).isGreaterThan(0)
+    );
+  }
+
+  private void assertEventProcessingLatency() {
+    await().untilAsserted(() ->
+      assertThat(
+        registry
+          .get(Metrics.EVENT_PROCESSING_LATENCY)
+          .summaries()
+          .stream()
+          .reduce((a, b) -> b)
+          .get()
+          .max()
+      ).isGreaterThan(0)
+    );
+  }
+
 }
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/AbstractConsumerVerticleTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/AbstractConsumerVerticleTest.java
index 4f2051c6c..89c546472 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/AbstractConsumerVerticleTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/AbstractConsumerVerticleTest.java
@@ -15,11 +15,14 @@
  */
 package dev.knative.eventing.kafka.broker.dispatcher.impl.consumer;
 
+import dev.knative.eventing.kafka.broker.contract.DataPlaneContract;
+import dev.knative.eventing.kafka.broker.core.metrics.Metrics;
 import dev.knative.eventing.kafka.broker.dispatcher.CloudEventSender;
 import dev.knative.eventing.kafka.broker.dispatcher.CloudEventSenderMock;
 import dev.knative.eventing.kafka.broker.dispatcher.ResponseHandlerMock;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.RecordDispatcherImpl;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.RecordDispatcherTest;
+import dev.knative.eventing.kafka.broker.dispatcher.impl.ResourceContext;
 import io.cloudevents.CloudEvent;
 import io.vertx.core.AsyncResult;
 import io.vertx.core.Future;
@@ -29,11 +32,8 @@ import io.vertx.core.Vertx;
 import io.vertx.junit5.VertxExtension;
 import io.vertx.junit5.VertxTestContext;
 import io.vertx.kafka.client.consumer.KafkaConsumer;
-import java.time.Duration;
-import java.util.Arrays;
-import java.util.Set;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.stream.Collectors;
+import io.vertx.micrometer.MicrometerMetricsOptions;
+import io.vertx.micrometer.backends.BackendRegistries;
 import org.apache.kafka.clients.consumer.MockConsumer;
 import org.apache.kafka.clients.consumer.OffsetResetStrategy;
 import org.apache.kafka.common.TopicPartition;
@@ -41,6 +41,12 @@ import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
@@ -51,16 +57,27 @@ import static org.mockito.Mockito.when;
 @ExtendWith(VertxExtension.class)
 public abstract class AbstractConsumerVerticleTest {
 
+  private static final ResourceContext resourceContext = new ResourceContext(
+    DataPlaneContract.Resource.newBuilder().build(),
+    DataPlaneContract.Egress.newBuilder().build()
+  );
+
+  static {
+    BackendRegistries.setupBackend(new MicrometerMetricsOptions().setRegistryName(Metrics.METRICS_REGISTRY_NAME));
+  }
+
   @Test
   public void subscribedToTopic(final Vertx vertx, final VertxTestContext context) {
     final var consumer = new MockConsumer<Object, CloudEvent>(OffsetResetStrategy.LATEST);
     final var recordDispatcher = new RecordDispatcherImpl(
+      resourceContext,
       value -> false,
       CloudEventSender.noop("subscriber send called"),
       CloudEventSender.noop("dead letter sink send called"),
       new ResponseHandlerMock(),
       RecordDispatcherTest.offsetManagerMock(),
-      null
+      null,
+      Metrics.getRegistry()
     );
     final var topic = "topic1";
 
@@ -90,12 +107,14 @@ public abstract class AbstractConsumerVerticleTest {
   public void stop(final Vertx vertx, final VertxTestContext context) {
     final var consumer = new MockConsumer<Object, CloudEvent>(OffsetResetStrategy.LATEST);
     final var recordDispatcher = new RecordDispatcherImpl(
+      resourceContext,
       value -> false,
       CloudEventSender.noop("subscriber send called"),
       CloudEventSender.noop("dead letter sink send called"),
       new ResponseHandlerMock(),
       RecordDispatcherTest.offsetManagerMock(),
-      null
+      null,
+      Metrics.getRegistry()
     );
     final var topic = "topic1";
 
@@ -129,7 +148,7 @@ public abstract class AbstractConsumerVerticleTest {
   @Test
   @SuppressWarnings("unchecked")
   public void shouldCloseEverything(final Vertx vertx, final VertxTestContext context) {
-    final var topics = new String[] {"a"};
+    final var topics = new String[]{"a"};
     final KafkaConsumer<Object, CloudEvent> consumer = mock(KafkaConsumer.class);
 
     when(consumer.close()).thenReturn(Future.succeededFuture());
@@ -164,6 +183,7 @@ public abstract class AbstractConsumerVerticleTest {
     final var sinkClosed = new AtomicBoolean(false);
 
     final var recordDispatcher = new RecordDispatcherImpl(
+      resourceContext,
       ce -> true,
       new CloudEventSenderMock(
         record -> Future.succeededFuture(), () -> {
@@ -184,7 +204,8 @@ public abstract class AbstractConsumerVerticleTest {
       }
       ),
       RecordDispatcherTest.offsetManagerMock(),
-      null
+      null,
+      Metrics.getRegistry()
     );
 
     final var verticle = createConsumerVerticle(
diff --git a/test/e2e-common.sh b/test/e2e-common.sh
index 5e3c60d04..5fcbd16b2 100644
--- a/test/e2e-common.sh
+++ b/test/e2e-common.sh
@@ -141,6 +141,7 @@ function install_head() {
   kubectl apply -f "${EVENTING_KAFKA_BROKER_ARTIFACT}" || return $?
   kubectl apply -f "${EVENTING_KAFKA_SINK_ARTIFACT}" || return $?
   kubectl apply -f "${EVENTING_KAFKA_CHANNEL_ARTIFACT}" || return $?
+  kubectl apply -f "${EVENTING_KAFKA_POST_INSTALL_ARTIFACT}" || return $?
 }
 
 function test_setup() {
diff --git a/third_party/eventing-kafka-latest/channel-crds.yaml b/third_party/eventing-kafka-latest/channel-crds.yaml
index 71aafe19a..f8cfc050d 100644
--- a/third_party/eventing-kafka-latest/channel-crds.yaml
+++ b/third_party/eventing-kafka-latest/channel-crds.yaml
@@ -17,7 +17,7 @@ kind: CustomResourceDefinition
 metadata:
   name: kafkachannels.messaging.knative.dev
   labels:
-    kafka.eventing.knative.dev/release: "v20220131-a14b76602"
+    kafka.eventing.knative.dev/release: "v20220208-ce66b2ecd"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
diff --git a/third_party/eventing-kafka-latest/source-crd.yaml b/third_party/eventing-kafka-latest/source-crd.yaml
index 905329876..6c44c3afa 100644
--- a/third_party/eventing-kafka-latest/source-crd.yaml
+++ b/third_party/eventing-kafka-latest/source-crd.yaml
@@ -16,7 +16,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    kafka.eventing.knative.dev/release: "v20220131-a14b76602"
+    kafka.eventing.knative.dev/release: "v20220208-ce66b2ecd"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
diff --git a/third_party/eventing-latest/eventing-core.yaml b/third_party/eventing-latest/eventing-core.yaml
index a46bb3de0..0deb19dc4 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: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 
 ---
@@ -41,8 +41,8 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     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: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     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: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     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: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     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: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     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: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     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: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     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: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     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: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     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: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     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: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     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: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     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: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     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: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     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: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 data:
   # Configures the default for any Broker that does not specify a spec.config or Broker class.
@@ -354,8 +354,8 @@ metadata:
   name: default-ch-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 data:
   # Configuration for defaulting channels that do not specify CRD implementations.
@@ -389,10 +389,10 @@ metadata:
   name: config-ping-defaults
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
   annotations:
     knative.dev/example-checksum: "9185c153"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 data:
   _example: |
@@ -436,10 +436,10 @@ metadata:
   name: config-features
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 data:
   # ALPHA feature: The kreference-group allows you to use the Group field in KReferences.
@@ -483,7 +483,7 @@ metadata:
   name: config-kreference-mapping
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
   annotations:
@@ -532,8 +532,8 @@ metadata:
   name: config-leader-election
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
   annotations:
     knative.dev/example-checksum: "f7948630"
@@ -594,10 +594,10 @@ metadata:
   name: config-logging
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 data:
   # Common configuration for all Knative codebase
@@ -648,10 +648,10 @@ metadata:
   name: config-observability
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
   annotations:
     knative.dev/example-checksum: "f46cf09d"
@@ -725,10 +725,10 @@ metadata:
   name: config-tracing
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
   annotations:
     knative.dev/example-checksum: "0492ceb0"
@@ -783,10 +783,10 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     knative.dev/high-availability: "true"
     app.kubernetes.io/component: eventing-controller
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 spec:
   selector:
@@ -796,9 +796,9 @@ spec:
     metadata:
       labels:
         app: eventing-controller
-        eventing.knative.dev/release: "v20220131-7a44b07f2"
-        app.kubernetes.io/component: eventing-autoscaler
-        app.kubernetes.io/version: "20220131-7a44b07f2"
+        eventing.knative.dev/release: "v20220208-20718e264"
+        app.kubernetes.io/component: eventing-controller
+        app.kubernetes.io/version: "20220208-20718e264"
         app.kubernetes.io/name: knative-eventing
     spec:
       # To avoid node becoming SPOF, spread our replicas to different nodes.
@@ -816,7 +816,7 @@ spec:
       containers:
         - name: eventing-controller
           terminationMessagePolicy: FallbackToLogsOnError
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:69d2d30440ed84ebc50be23694ca194646fa08411ddbe93487bc003090a52d9c
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:de448d61329c04a7f8f0ad72f43022cc19dc618c1074c3a24def274b11d73562
           resources:
             requests:
               cpu: 100m
@@ -834,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:72b6206cf15de6cfdb16217a17ec5d8e0184dd3a14c206861ab16ad5d94802ce
+              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:ce98de86265e2072e7a12433374869d9ad526ceb076ced609bc9dabc3609c60b
             - name: POD_NAME
               valueFrom:
                 fieldRef:
@@ -883,9 +883,9 @@ metadata:
   name: pingsource-mt-adapter
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     app.kubernetes.io/component: pingsource-mt-adapter
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     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.
@@ -898,9 +898,9 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20220131-7a44b07f2"
+        eventing.knative.dev/release: "v20220208-20718e264"
         app.kubernetes.io/component: pingsource-mt-adapter
-        app.kubernetes.io/version: "20220131-7a44b07f2"
+        app.kubernetes.io/version: "20220208-20718e264"
         app.kubernetes.io/name: knative-eventing
     spec:
       affinity:
@@ -914,7 +914,7 @@ spec:
       enableServiceLinks: false
       containers:
         - name: dispatcher
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:744060290509d1172fbaaded191bf1b5b631f835de27662c6bfe62358526ebaa
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:f12bfa0b060ea80d2ad2a666894540aca9c661fd3e0484decaebde0dc54623b6
           env:
             - name: SYSTEM_NAMESPACE
               value: ''
@@ -979,9 +979,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 spec:
   scaleTargetRef:
@@ -1005,9 +1005,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 spec:
   minAvailable: 80%
@@ -1036,9 +1036,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 spec:
   selector:
@@ -1049,9 +1049,9 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20220131-7a44b07f2"
+        eventing.knative.dev/release: "v20220208-20718e264"
         app.kubernetes.io/component: eventing-webhook
-        app.kubernetes.io/version: "20220131-7a44b07f2"
+        app.kubernetes.io/version: "20220208-20718e264"
         app.kubernetes.io/name: knative-eventing
     spec:
       # To avoid node becoming SPOF, spread our replicas to different nodes.
@@ -1071,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:cff89e978a3ecb1e6a3af700525a832058feccf5ec769ce111da673d9fa9b80f
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/webhook@sha256:2704f8da50bc9267f8dbe4cdc19a6e8ac64f63528da17118a0a25c7f84273617
           resources:
             requests:
               # taken from serving.
@@ -1140,10 +1140,10 @@ apiVersion: v1
 kind: Service
 metadata:
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     role: eventing-webhook
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
   name: eventing-webhook
   namespace: knative-eventing
@@ -1175,11 +1175,11 @@ kind: CustomResourceDefinition
 metadata:
   creationTimestamp: null
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -1391,10 +1391,10 @@ kind: CustomResourceDefinition
 metadata:
   name: brokers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -1558,11 +1558,11 @@ kind: CustomResourceDefinition
 metadata:
   name: channels.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -1830,11 +1830,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
   name: containersources.sources.knative.dev
 spec:
@@ -1984,9 +1984,9 @@ kind: CustomResourceDefinition
 metadata:
   name: eventtypes.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -2113,10 +2113,10 @@ kind: CustomResourceDefinition
 metadata:
   name: parallels.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: flows.knative.dev
@@ -2368,11 +2368,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -2559,10 +2559,10 @@ kind: CustomResourceDefinition
 metadata:
   name: sequences.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: flows.knative.dev
@@ -2879,12 +2879,12 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
   name: sinkbindings.sources.knative.dev
 spec:
@@ -3071,9 +3071,9 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -3283,9 +3283,9 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -3469,8 +3469,8 @@ kind: ClusterRole
 metadata:
   name: addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -3483,9 +3483,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: service-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3503,9 +3503,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: serving-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3526,9 +3526,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: channel-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3553,9 +3553,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: broker-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3574,9 +3574,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: flows-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3612,8 +3612,8 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-filter
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3639,8 +3639,8 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-ingress
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3657,8 +3657,8 @@ kind: ClusterRole
 metadata:
   name: eventing-config-reader
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3691,8 +3691,8 @@ kind: ClusterRole
 metadata:
   name: channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -3705,9 +3705,9 @@ kind: ClusterRole
 metadata:
   name: meta-channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     duck.knative.dev/channelable: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "channelable-manipulator" role.
 rules:
@@ -3745,9 +3745,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-eventing-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev"]
@@ -3759,9 +3759,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-messaging-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["messaging.knative.dev"]
@@ -3773,9 +3773,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-flows-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["flows.knative.dev"]
@@ -3787,9 +3787,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-sources-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["sources.knative.dev"]
@@ -3801,9 +3801,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-bindings-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["bindings.knative.dev"]
@@ -3816,8 +3816,8 @@ metadata:
   name: knative-eventing-namespaced-edit
   labels:
     rbac.authorization.k8s.io/aggregate-to-edit: "true"
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
@@ -3830,8 +3830,8 @@ metadata:
   name: knative-eventing-namespaced-view
   labels:
     rbac.authorization.k8s.io/aggregate-to-view: "true"
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
@@ -3858,8 +3858,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-controller
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3994,8 +3994,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-pingsource-mt-adapter
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -4062,8 +4062,8 @@ kind: ClusterRole
 metadata:
   name: podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -4076,9 +4076,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: builtin-podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     duck.knative.dev/podspecable: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "podspecable-binding role.
 rules:
@@ -4124,8 +4124,8 @@ kind: ClusterRole
 metadata:
   name: source-observer
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -4138,9 +4138,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: eventing-sources-source-observer
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     duck.knative.dev/source: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "source-observer" role.
 rules:
@@ -4176,8 +4176,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-sources-controller
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -4263,8 +4263,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 rules:
   # For watching logging configuration and getting certs.
@@ -4363,8 +4363,8 @@ metadata:
   namespace: knative-eventing
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 rules:
   # For manipulating certs into secrets.
@@ -4400,8 +4400,8 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: config.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4438,8 +4438,8 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4472,8 +4472,8 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: validation.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4507,8 +4507,8 @@ metadata:
   name: eventing-webhook-certs
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 # The data is populated at install time.
 
@@ -4532,8 +4532,8 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: sinkbindings.webhook.sources.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220131-7a44b07f2"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
+    app.kubernetes.io/version: "20220208-20718e264"
     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 99b7e5a89..86b2e1332 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: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     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: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     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: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     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: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     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: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     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: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     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: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     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: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     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: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     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: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     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: "v20220131-7a44b07f2"
+    eventing.knative.dev/release: "v20220208-20718e264"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220208-20718e264"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev