Repository: knative-sandbox/eventing-kafka-broker
Pull request #1889 (2022-02-10 11:25:21)

Add KafkaSource migration logic as a post-install job
The existing KafkaSource creates a deployment for each instance
and nothing more.

This is what it's in the namespace where I created a `KafkaSouce`.
Only the
`deployment.apps/kafkasource-kafka-source-2e0b5e79-8547-46e8-b018-aac56c6c8d16`
is created automatically by the
`kafka-controller-manager`, other resources are sink services
that shouldn't be deleted during the migration.
```
pierdipi@pierdipi eventing-kafka-broker (kafka-source-migration-post-install) $ k get all
NAME                                                                  READY   STATUS    RESTARTS   AGE
pod/kafkasource-kafka-source-2e0b5e79-8547-46e8-b018-aac56c6c8bhp5p   1/1     Running   0          106m
pod/label-demo                                                        1/1     Running   0          106m
pod/label-demo-2                                                      1/1     Running   0          106m

NAME                      TYPE        CLUSTER-IP     EXTERNAL-IP   PORT(S)   AGE
service/event-display     ClusterIP   10.96.16.19    <none>        80/TCP    106m
service/event-display-2   ClusterIP   10.96.177.33   <none>        80/TCP    106m
service/kubernetes        ClusterIP   10.96.0.1      <none>        443/TCP   116m

NAME                                                                            READY   UP-TO-DATE   AVAILABLE   AGE
deployment.apps/kafkasource-kafka-source-2e0b5e79-8547-46e8-b018-aac56c6c8d16   1/1     1            1           106m

NAME                                                                                       DESIRED   CURRENT   READY   AGE
replicaset.apps/kafkasource-kafka-source-2e0b5e79-8547-46e8-b018-aac56c6c8d16-695fbb68b8   1         1         1       106m

NAME                                           TOPICS                   BOOTSTRAPSERVERS                            READY   REASON   AGE
kafkasource.sources.knative.dev/kafka-source   ["knative-demo-topic"]   ["my-cluster-kafka-bootstrap.kafka:9092"]   True             106m
```

The plan for post-install logic is to have a generic post-install
job in eventing-kafka-broker (in addition to a storage version
migrator) this is the job where we place general post-install
operations (manifests placed in control-plane/config/post-install).

In addition I moved the manifests for the control plane to
`control-plane/config/eventing-kafka-broker` to easily produce
the `eventing-kafka-broker.yaml` manifest via recursive
resolution.

*Testing*

- Apply `source.yaml` from `eventing-kafka`.
- Create `KafkaSouce`(s)
- Scale down the `kafka-controller-manager` deployment with:
  `k scale deployment -n knative-eventing kafka-controller-manager --replicas=0`
- Create the post-install manifest with:
  `ko resolve -Rf control-plane/config/post-install/ > eventing-kafka-post-install.yaml`
- Run the post-install job with:
  `ko create -f eventing-kafka-post-install.yaml`

Output:
```
pierdipi@pierdipi eventing-kafka-broker (kafka-source-migration-post-install) $ k get all
NAME               READY   STATUS    RESTARTS   AGE
pod/label-demo     1/1     Running   0          124m
pod/label-demo-2   1/1     Running   0          124m

NAME                      TYPE        CLUSTER-IP     EXTERNAL-IP   PORT(S)   AGE
service/event-display     ClusterIP   10.96.16.19    <none>        80/TCP    124m
service/event-display-2   ClusterIP   10.96.177.33   <none>        80/TCP    124m
service/kubernetes        ClusterIP   10.96.0.1      <none>        443/TCP   134m

NAME                                           TOPICS                   BOOTSTRAPSERVERS                            READY   REASON   AGE
kafkasource.sources.knative.dev/kafka-source   ["knative-demo-topic"]   ["my-cluster-kafka-bootstrap.kafka:9092"]   True             124m
```

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Add KafkaSource migration logic as a post-install job

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- :gift: Add new feature
- :bug: Fix bug
- :broom: Update or clean up current behavior
- :wastebasket: Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
:page_facing_up: If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note
Add KafkaSource migration logic as a post-install job (`eventing-kafka-post-install.yaml`).
```

Next: left-over global resources deletion

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

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

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

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
knative-prow-robot: @pierDipi: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-build-tests | 78d67bc21eacaee8f4dd277c0f0a94ca9b1ac085 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1889/pull-knative-sandbox-eventing-kafka-broker-build-tests/1491735292934098944) | true | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain | 6dadacd97834ba4b5c5dbd742d298dcdc9957dc4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1889/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain/1491736818146611200) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain | 6dadacd97834ba4b5c5dbd742d298dcdc9957dc4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1889/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain/1491736818146611200) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain`
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 6dadacd97834ba4b5c5dbd742d298dcdc9957dc4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1889/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1491736808344522752) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain | 6dadacd97834ba4b5c5dbd742d298dcdc9957dc4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1889/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain/1491736818146611200) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain`
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 6dadacd97834ba4b5c5dbd742d298dcdc9957dc4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1889/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1491736808344522752) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-ssl | 6dadacd97834ba4b5c5dbd742d298dcdc9957dc4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1889/pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-ssl/1491736819362959360) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-ssl`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain | 6dadacd97834ba4b5c5dbd742d298dcdc9957dc4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1889/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain/1491736818146611200) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain`
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 6dadacd97834ba4b5c5dbd742d298dcdc9957dc4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1889/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1491736808344522752) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-ssl | 6dadacd97834ba4b5c5dbd742d298dcdc9957dc4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1889/pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-ssl/1491736819362959360) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-ssl`
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl | 6dadacd97834ba4b5c5dbd742d298dcdc9957dc4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1889/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl/1491736816833794048) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain | 6dadacd97834ba4b5c5dbd742d298dcdc9957dc4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1889/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain/1491736818146611200) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain`
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 6dadacd97834ba4b5c5dbd742d298dcdc9957dc4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1889/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1491736808344522752) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-ssl | 6dadacd97834ba4b5c5dbd742d298dcdc9957dc4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1889/pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-ssl/1491736819362959360) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-ssl`
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl | 6dadacd97834ba4b5c5dbd742d298dcdc9957dc4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1889/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl/1491736816833794048) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl`
pull-knative-sandbox-eventing-kafka-broker-reconciler-tests | 6dadacd97834ba4b5c5dbd742d298dcdc9957dc4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1889/pull-knative-sandbox-eventing-kafka-broker-reconciler-tests/1491736812379443200) | true | `/test pull-knative-sandbox-eventing-kafka-broker-reconciler-tests`
pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-sasl-ssl | 6dadacd97834ba4b5c5dbd742d298dcdc9957dc4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1889/pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-sasl-ssl/1491736820642222080) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-sasl-ssl`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain | 6dadacd97834ba4b5c5dbd742d298dcdc9957dc4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1889/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain/1491736818146611200) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain`
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 6dadacd97834ba4b5c5dbd742d298dcdc9957dc4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1889/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1491736808344522752) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-ssl | 6dadacd97834ba4b5c5dbd742d298dcdc9957dc4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1889/pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-ssl/1491736819362959360) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-ssl`
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl | 6dadacd97834ba4b5c5dbd742d298dcdc9957dc4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1889/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl/1491736816833794048) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl`
pull-knative-sandbox-eventing-kafka-broker-reconciler-tests | 6dadacd97834ba4b5c5dbd742d298dcdc9957dc4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1889/pull-knative-sandbox-eventing-kafka-broker-reconciler-tests/1491736812379443200) | true | `/test pull-knative-sandbox-eventing-kafka-broker-reconciler-tests`
pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-sasl-ssl | 6dadacd97834ba4b5c5dbd742d298dcdc9957dc4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1889/pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-sasl-ssl/1491736820642222080) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-sasl-ssl`
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl | 6dadacd97834ba4b5c5dbd742d298dcdc9957dc4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1889/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl/1491736815160266752) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl`

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

<details>

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

Patch:
diff --git a/.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..d28346249 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
@@ -197,21 +197,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 +219,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/zz_generated.deepcopy.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go
index b9e527115..42bfddaa6 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
diff --git a/control-plane/pkg/reconciler/consumer/consumer.go b/control-plane/pkg/reconciler/consumer/consumer.go
index 65b7a6078..d225b5966 100644
--- a/control-plane/pkg/reconciler/consumer/consumer.go
+++ b/control-plane/pkg/reconciler/consumer/consumer.go
@@ -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 0c6bd29cb..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 {
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/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/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_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/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() {