Repository: knative-sandbox/eventing-kafka-broker
Pull request #2433 (2022-07-28 10:50:07)

Fix profiler job
As per title, flamegraphs can be downloaded from
https://github.com/knative-sandbox/eventing-kafka-broker/actions/runs/2753216718

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2433#" 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[bot]: @pierDipi: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/unit-tests_eventing-kafka-broker_main/1552607333199122432) | true | `/test unit-tests_eventing-kafka-broker_main`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow[bot]: @pierDipi: The following 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
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/unit-tests_eventing-kafka-broker_main/1552607333199122432) | true | `/test unit-tests_eventing-kafka-broker_main`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1552607333622747136) | true | `/test channel-integration-tests-sasl-plain_eventing-kafka-broker_main`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/unit-tests_eventing-kafka-broker_main/1552607333199122432) | true | `/test unit-tests_eventing-kafka-broker_main`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1552607333622747136) | true | `/test channel-integration-tests-sasl-plain_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1552607333735993344) | true | `/test channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow[bot]: @pierDipi: The following 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
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/unit-tests_eventing-kafka-broker_main/1552607333199122432) | true | `/test unit-tests_eventing-kafka-broker_main`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1552607333622747136) | true | `/test channel-integration-tests-sasl-plain_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1552607333735993344) | true | `/test channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main`
channel-reconciler-tests-ssl_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-reconciler-tests-ssl_eventing-kafka-broker_main/1552607333677273088) | true | `/test channel-reconciler-tests-ssl_eventing-kafka-broker_main`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow[bot]: @pierDipi: The following 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
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/unit-tests_eventing-kafka-broker_main/1552607333199122432) | true | `/test unit-tests_eventing-kafka-broker_main`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1552607333622747136) | true | `/test channel-integration-tests-sasl-plain_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1552607333735993344) | true | `/test channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main`
channel-reconciler-tests-ssl_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-reconciler-tests-ssl_eventing-kafka-broker_main/1552607333677273088) | true | `/test channel-reconciler-tests-ssl_eventing-kafka-broker_main`
upgrade-tests_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/upgrade-tests_eventing-kafka-broker_main/1552607333341728768) | true | `/test upgrade-tests_eventing-kafka-broker_main`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow[bot]: @pierDipi: The following 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
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/unit-tests_eventing-kafka-broker_main/1552607333199122432) | true | `/test unit-tests_eventing-kafka-broker_main`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1552607333622747136) | true | `/test channel-integration-tests-sasl-plain_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1552607333735993344) | true | `/test channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main`
channel-reconciler-tests-ssl_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-reconciler-tests-ssl_eventing-kafka-broker_main/1552607333677273088) | true | `/test channel-reconciler-tests-ssl_eventing-kafka-broker_main`
upgrade-tests_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/upgrade-tests_eventing-kafka-broker_main/1552607333341728768) | true | `/test upgrade-tests_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main/1552607333941514240) | true | `/test channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/unit-tests_eventing-kafka-broker_main/1552607333199122432) | true | `/test unit-tests_eventing-kafka-broker_main`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1552607333622747136) | true | `/test channel-integration-tests-sasl-plain_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1552607333735993344) | true | `/test channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main`
channel-reconciler-tests-ssl_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-reconciler-tests-ssl_eventing-kafka-broker_main/1552607333677273088) | true | `/test channel-reconciler-tests-ssl_eventing-kafka-broker_main`
upgrade-tests_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/upgrade-tests_eventing-kafka-broker_main/1552607333341728768) | true | `/test upgrade-tests_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main/1552607333941514240) | true | `/test channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main`
channel-integration-tests-ssl_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-integration-tests-ssl_eventing-kafka-broker_main/1552607333475946496) | true | `/test channel-integration-tests-ssl_eventing-kafka-broker_main`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow[bot]: @pierDipi: The following 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
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/unit-tests_eventing-kafka-broker_main/1552607333199122432) | true | `/test unit-tests_eventing-kafka-broker_main`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1552607333622747136) | true | `/test channel-integration-tests-sasl-plain_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1552607333735993344) | true | `/test channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main`
channel-reconciler-tests-ssl_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-reconciler-tests-ssl_eventing-kafka-broker_main/1552607333677273088) | true | `/test channel-reconciler-tests-ssl_eventing-kafka-broker_main`
upgrade-tests_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/upgrade-tests_eventing-kafka-broker_main/1552607333341728768) | true | `/test upgrade-tests_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main/1552607333941514240) | true | `/test channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main`
channel-integration-tests-ssl_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-integration-tests-ssl_eventing-kafka-broker_main/1552607333475946496) | true | `/test channel-integration-tests-ssl_eventing-kafka-broker_main`
reconciler-tests_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/reconciler-tests_eventing-kafka-broker_main/1552607333413031936) | true | `/test reconciler-tests_eventing-kafka-broker_main`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/unit-tests_eventing-kafka-broker_main/1552607333199122432) | true | `/test unit-tests_eventing-kafka-broker_main`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1552607333622747136) | true | `/test channel-integration-tests-sasl-plain_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1552607333735993344) | true | `/test channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main`
channel-reconciler-tests-ssl_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-reconciler-tests-ssl_eventing-kafka-broker_main/1552607333677273088) | true | `/test channel-reconciler-tests-ssl_eventing-kafka-broker_main`
upgrade-tests_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/upgrade-tests_eventing-kafka-broker_main/1552607333341728768) | true | `/test upgrade-tests_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main/1552607333941514240) | true | `/test channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main`
channel-integration-tests-ssl_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-integration-tests-ssl_eventing-kafka-broker_main/1552607333475946496) | true | `/test channel-integration-tests-ssl_eventing-kafka-broker_main`
reconciler-tests_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/reconciler-tests_eventing-kafka-broker_main/1552607333413031936) | true | `/test reconciler-tests_eventing-kafka-broker_main`
channel-integration-tests-sasl-ssl_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-integration-tests-sasl-ssl_eventing-kafka-broker_main/1552607333568221184) | true | `/test channel-integration-tests-sasl-ssl_eventing-kafka-broker_main`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow[bot]: @pierDipi: The following 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
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/unit-tests_eventing-kafka-broker_main/1552607333199122432) | true | `/test unit-tests_eventing-kafka-broker_main`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1552607333622747136) | true | `/test channel-integration-tests-sasl-plain_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1552607333735993344) | true | `/test channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main`
channel-reconciler-tests-ssl_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-reconciler-tests-ssl_eventing-kafka-broker_main/1552607333677273088) | true | `/test channel-reconciler-tests-ssl_eventing-kafka-broker_main`
upgrade-tests_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/upgrade-tests_eventing-kafka-broker_main/1552607333341728768) | true | `/test upgrade-tests_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main/1552607333941514240) | true | `/test channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main`
channel-integration-tests-ssl_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-integration-tests-ssl_eventing-kafka-broker_main/1552607333475946496) | true | `/test channel-integration-tests-ssl_eventing-kafka-broker_main`
reconciler-tests_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/reconciler-tests_eventing-kafka-broker_main/1552607333413031936) | true | `/test reconciler-tests_eventing-kafka-broker_main`
channel-integration-tests-sasl-ssl_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/channel-integration-tests-sasl-ssl_eventing-kafka-broker_main/1552607333568221184) | true | `/test channel-integration-tests-sasl-ssl_eventing-kafka-broker_main`
integration-tests_eventing-kafka-broker_main | 0583e42b5eea7e6d10c35a3320a0474f2a10ccc6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2433/integration-tests_eventing-kafka-broker_main/1552607333278814208) | true | `/test integration-tests_eventing-kafka-broker_main`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2433?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2433](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2433?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (0583e42) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/d536fe10e79e197ee8ca0405d9b876e8e5291631?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (d536fe1) will **decrease** coverage by `5.62%`.
> The diff coverage is `n/a`.

```diff
@@             Coverage Diff              @@
##               main    #2433      +/-   ##
============================================
- Coverage     64.60%   58.97%   -5.63%     
============================================
  Files           146       74      -72     
  Lines          9783     7340    -2443     
  Branches        215        0     -215     
============================================
- Hits           6320     4329    -1991     
+ Misses         3038     2706     -332     
+ Partials        425      305     -120     
```

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

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

| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2433?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [.../broker/receiver/main/ReceiverVerticleFactory.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2433/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9tYWluL1JlY2VpdmVyVmVydGljbGVGYWN0b3J5LmphdmE=) | | |
| [...r/dispatcher/impl/ResponseToKafkaTopicHandler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2433/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9SZXNwb25zZVRvS2Fma2FUb3BpY0hhbmRsZXIuamF2YQ==) | | |
| [...her/impl/filter/subscriptionsapi/PrefixFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2433/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvc3Vic2NyaXB0aW9uc2FwaS9QcmVmaXhGaWx0ZXIuamF2YQ==) | | |
| [.../dispatcher/impl/RecordDispatcherMutatorChain.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2433/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9SZWNvcmREaXNwYXRjaGVyTXV0YXRvckNoYWluLmphdmE=) | | |
| [...tive/eventing/kafka/broker/core/utils/BaseEnv.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2433/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdXRpbHMvQmFzZUVudi5qYXZh) | | |
| [.../core/reconciler/impl/ResourcesReconcilerImpl.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2433/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvcmVjb25jaWxlci9pbXBsL1Jlc291cmNlc1JlY29uY2lsZXJJbXBsLmphdmE=) | | |
| [...er/impl/consumer/InvalidCloudEventInterceptor.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2433/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9JbnZhbGlkQ2xvdWRFdmVudEludGVyY2VwdG9yLmphdmE=) | | |
| [...patcher/impl/consumer/OrderedConsumerVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2433/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9PcmRlcmVkQ29uc3VtZXJWZXJ0aWNsZS5qYXZh) | | |
| [...atcher/impl/consumer/KafkaConsumerRecordUtils.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2433/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9LYWZrYUNvbnN1bWVyUmVjb3JkVXRpbHMuamF2YQ==) | | |
| [.../kafka/broker/core/eventbus/ContractPublisher.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2433/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvZXZlbnRidXMvQ29udHJhY3RQdWJsaXNoZXIuamF2YQ==) | | |
| ... and [62 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2433/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).


Patch:
diff --git a/OWNERS_ALIASES b/OWNERS_ALIASES
index 10b84c7cc..2f14d6006 100644
--- a/OWNERS_ALIASES
+++ b/OWNERS_ALIASES
@@ -201,6 +201,7 @@ aliases:
   - knative-prow-updater-robot
   - knative-test-reporter-robot
   kperf-approvers:
+  - chzhyang
   - maximilien
   - nader-ziada
   - psschwei
@@ -256,6 +257,11 @@ aliases:
   - kvmware
   - psschwei
   - upodroid
+  security-guard-approvers:
+  - davidhadas
+  - maximilien
+  - psschwei
+  - rhuss
   security-wg-leads:
   - evankanderson
   - julz
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink.go b/control-plane/pkg/reconciler/sink/kafka_sink.go
index 2235a2b4f..c2286ee0a 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink.go
@@ -172,7 +172,8 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *eventing.KafkaSink)
 		Uid:    string(ks.UID),
 		Topics: []string{ks.Spec.Topic},
 		Ingress: &contract.Ingress{
-			Path: receiver.PathFromObject(ks),
+			Path:        receiver.PathFromObject(ks),
+			ContentMode: coreconfig.ContentModeFromString(*ks.Spec.ContentMode),
 		},
 		BootstrapServers: kafka.BootstrapServersCommaSeparated(ks.Spec.BootstrapServers),
 		Reference: &contract.Reference{
diff --git a/control-plane/pkg/reconciler/source/source.go b/control-plane/pkg/reconciler/source/source.go
index 2e3a7b2d3..c3536ce35 100644
--- a/control-plane/pkg/reconciler/source/source.go
+++ b/control-plane/pkg/reconciler/source/source.go
@@ -126,7 +126,7 @@ func (r Reconciler) reconcileConsumerGroup(ctx context.Context, ks *sources.Kafk
 						DeliverySpec: &eventingduck.DeliverySpec{
 							Retry:         pointer.Int32(10),
 							BackoffPolicy: &backoffPolicy,
-							BackoffDelay:  pointer.String("PT10S"),
+							BackoffDelay:  pointer.String("PT0.3S"),
 							Timeout:       pointer.String("PT600S"),
 						},
 						Ordering: DefaultDeliveryOrder,
diff --git a/data-plane/config/broker/100-config-kafka-broker-data-plane.yaml b/data-plane/config/broker/100-config-kafka-broker-data-plane.yaml
index 787c1be0f..f3cbd9989 100644
--- a/data-plane/config/broker/100-config-kafka-broker-data-plane.yaml
+++ b/data-plane/config/broker/100-config-kafka-broker-data-plane.yaml
@@ -115,7 +115,7 @@ data:
     value.deserializer=io.cloudevents.kafka.CloudEventDeserializer
     fetch.min.bytes=1
     heartbeat.interval.ms=3000
-    max.partition.fetch.bytes=1048576
+    max.partition.fetch.bytes=65536
     session.timeout.ms=10000
     # ssl.key.password=
     # ssl.keystore.location=
@@ -132,7 +132,7 @@ data:
     fetch.max.bytes=52428800
     isolation.level=read_uncommitted
     max.poll.interval.ms=300000
-    max.poll.records=500
+    max.poll.records=50
     partition.assignment.strategy=org.apache.kafka.clients.consumer.StickyAssignor
     receive.buffer.bytes=65536
     request.timeout.ms=2000
@@ -175,5 +175,6 @@ data:
     # ssl.trustmanager.algorithm
   config-kafka-broker-webclient.properties: |
     idleTimeout=10000
+    maxPoolSize=100
   config-kafka-broker-httpserver.properties: |
     idleTimeout=0
diff --git a/data-plane/config/channel/100-config-kafka-channel-data-plane.yaml b/data-plane/config/channel/100-config-kafka-channel-data-plane.yaml
index 86eed38a1..1216f2b20 100644
--- a/data-plane/config/channel/100-config-kafka-channel-data-plane.yaml
+++ b/data-plane/config/channel/100-config-kafka-channel-data-plane.yaml
@@ -115,7 +115,7 @@ data:
     value.deserializer=io.cloudevents.kafka.CloudEventDeserializer
     fetch.min.bytes=1
     heartbeat.interval.ms=3000
-    max.partition.fetch.bytes=1048576
+    max.partition.fetch.bytes=65536
     session.timeout.ms=10000
     # ssl.key.password=
     # ssl.keystore.location=
@@ -131,7 +131,7 @@ data:
     fetch.max.bytes=52428800
     isolation.level=read_uncommitted
     max.poll.interval.ms=300000
-    max.poll.records=500
+    max.poll.records=50
     partition.assignment.strategy=org.apache.kafka.clients.consumer.StickyAssignor
     receive.buffer.bytes=65536
     request.timeout.ms=2000
@@ -174,5 +174,6 @@ data:
     # ssl.trustmanager.algorithm
   config-kafka-channel-webclient.properties: |
     idleTimeout=10000
+    maxPoolSize=100
   config-kafka-channel-httpserver.properties: |
     idleTimeout=0
diff --git a/data-plane/config/source/100-config-kafka-source-data-plane.yaml b/data-plane/config/source/100-config-kafka-source-data-plane.yaml
index ecc39dbbb..27a4f8ef1 100644
--- a/data-plane/config/source/100-config-kafka-source-data-plane.yaml
+++ b/data-plane/config/source/100-config-kafka-source-data-plane.yaml
@@ -97,7 +97,7 @@ data:
     value.deserializer=io.cloudevents.kafka.CloudEventDeserializer
     fetch.min.bytes=1
     heartbeat.interval.ms=3000
-    max.partition.fetch.bytes=1048576
+    max.partition.fetch.bytes=65536
     session.timeout.ms=10000
     # ssl.key.password=
     # ssl.keystore.location=
@@ -114,7 +114,7 @@ data:
     fetch.max.bytes=52428800
     isolation.level=read_uncommitted
     max.poll.interval.ms=300000
-    max.poll.records=500
+    max.poll.records=50
     partition.assignment.strategy=org.apache.kafka.clients.consumer.StickyAssignor
     receive.buffer.bytes=65536
     request.timeout.ms=2000
@@ -157,4 +157,4 @@ data:
     # ssl.trustmanager.algorithm
   config-kafka-source-webclient.properties: |
     idleTimeout=10000
-    # maxPoolSize=50
+    maxPoolSize=100
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 c4c1862a6..f5254cb07 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,6 +21,7 @@ import io.micrometer.core.instrument.Counter;
 import io.micrometer.core.instrument.DistributionSummary;
 import io.micrometer.core.instrument.Gauge;
 import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Tags;
 import io.micrometer.core.instrument.binder.BaseUnits;
 import io.micrometer.core.instrument.binder.kafka.KafkaClientMetrics;
 import io.micrometer.prometheus.PrometheusConfig;
@@ -121,7 +122,7 @@ public class Metrics {
 
   /**
    * @link https://knative.dev/docs/eventing/observability/metrics/eventing-metrics/
-   * @see Metrics#queueLength(io.micrometer.core.instrument.Tags)
+   * @see Metrics#queueLength(io.micrometer.core.instrument.Tags, Supplier)
    */
   public static final String QUEUE_LENGTH = "queue_length";
 
@@ -159,13 +160,9 @@ public class Metrics {
    */
   public static MetricsOptions getOptions(final BaseEnv metricsConfigs) {
     final var registry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
-    return new MicrometerMetricsOptions()
+    final var options = new MicrometerMetricsOptions()
       .setEnabled(true)
-      .addDisabledMetricsCategory(MetricsDomain.HTTP_CLIENT)
-      .addDisabledMetricsCategory(MetricsDomain.HTTP_SERVER)
       .addDisabledMetricsCategory(MetricsDomain.VERTICLES)
-      .addDisabledMetricsCategory(MetricsDomain.NET_CLIENT)
-      .addDisabledMetricsCategory(MetricsDomain.NET_SERVER)
       .addDisabledMetricsCategory(MetricsDomain.EVENT_BUS)
       .addDisabledMetricsCategory(MetricsDomain.DATAGRAM_SOCKET)
       // NAMED_POOL allocates a lot, so disable it.
@@ -188,6 +185,17 @@ public class Metrics {
         .setStartEmbeddedServer(true)
         .setEnabled(true)
       );
+
+    if (!metricsConfigs.isMetricsHTTPClientEnabled()) {
+      options.addDisabledMetricsCategory(MetricsDomain.HTTP_CLIENT);
+      options.addDisabledMetricsCategory(MetricsDomain.NET_CLIENT);
+    }
+    if (!metricsConfigs.isMetricsHTTPServerEnabled()) {
+      options.addDisabledMetricsCategory(MetricsDomain.HTTP_SERVER);
+      options.addDisabledMetricsCategory(MetricsDomain.NET_SERVER);
+    }
+
+    return options;
   }
 
   /**
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java
index 3479c6197..764f3e2bf 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java
@@ -240,8 +240,12 @@ public class ResourcesReconcilerImpl implements ResourcesReconciler {
     return Objects.equals(e1.getUid(), e2.getUid())
       && Objects.equals(e1.getConsumerGroup(), e2.getConsumerGroup())
       && Objects.equals(e1.getDestination(), e2.getDestination())
+      && Objects.equals(e1.hasDiscardReply(), e2.hasDiscardReply())
       && Objects.equals(e1.getReplyUrl(), e2.getReplyUrl())
       && Objects.equals(e1.getReplyToOriginalTopic(), e2.getReplyToOriginalTopic())
+      && Objects.equals(e1.getDeliveryOrder(), e2.getDeliveryOrder())
+      && Objects.equals(e1.getVReplicas(), e2.getVReplicas())
+      && Objects.equals(e1.getFeatureFlags(), e2.getFeatureFlags())
       && Objects.equals(e1.getEgressConfig(), e2.getEgressConfig())
       && Objects.equals(e1.getFilter(), e2.getFilter())
       && Objects.equals(e1.getKeyType(), e2.getKeyType())
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/utils/BaseEnv.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/utils/BaseEnv.java
index 4ac5ab8cd..06045d436 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/utils/BaseEnv.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/utils/BaseEnv.java
@@ -39,6 +39,12 @@ public class BaseEnv {
   public static final String METRICS_JVM_ENABLED = "METRICS_JVM_ENABLED";
   private final boolean metricsJvmEnabled;
 
+  public static final String METRICS_HTTP_CLIENT_ENABLED = "METRICS_HTTP_CLIENT_ENABLED";
+  private final boolean metricsHTTPClientEnabled;
+
+  public static final String METRICS_HTTP_SERVER_ENABLED = "METRICS_HTTP_SERVER_ENABLED";
+  private final boolean metricsHTTPServerEnabled;
+
   public static final String CONFIG_TRACING_PATH = "CONFIG_TRACING_PATH";
   private final String configTracingPath;
 
@@ -50,6 +56,8 @@ public class BaseEnv {
     this.metricsPort = Integer.parseInt(requireNonNull(envProvider.apply(METRICS_PORT)));
     this.metricsPublishQuantiles = Boolean.parseBoolean(envProvider.apply(METRICS_PUBLISH_QUANTILES));
     this.metricsJvmEnabled = Boolean.parseBoolean(envProvider.apply(METRICS_JVM_ENABLED));
+    this.metricsHTTPClientEnabled = Boolean.parseBoolean(envProvider.apply(METRICS_HTTP_CLIENT_ENABLED));
+    this.metricsHTTPServerEnabled = Boolean.parseBoolean(envProvider.apply(METRICS_HTTP_SERVER_ENABLED));
     this.producerConfigFilePath = requireNonNull(envProvider.apply(PRODUCER_CONFIG_FILE_PATH));
     this.dataPlaneConfigFilePath = requireNonNull(envProvider.apply(DATA_PLANE_CONFIG_FILE_PATH));
     this.configTracingPath = requireNonNull(envProvider.apply(CONFIG_TRACING_PATH));
@@ -80,6 +88,14 @@ public class BaseEnv {
     return metricsJvmEnabled;
   }
 
+  public boolean isMetricsHTTPClientEnabled() {
+    return metricsHTTPClientEnabled;
+  }
+
+  public boolean isMetricsHTTPServerEnabled() {
+    return metricsHTTPServerEnabled;
+  }
+
   public String getConfigTracingPath() {
     return configTracingPath;
   }
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OrderedConsumerVerticle.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OrderedConsumerVerticle.java
index da94fc724..f9a166ebd 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OrderedConsumerVerticle.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OrderedConsumerVerticle.java
@@ -34,6 +34,8 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.time.Duration;
+import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
@@ -219,13 +221,26 @@ public class OrderedConsumerVerticle extends BaseConsumerVerticle {
       return false;
     }
 
-    for (OrderedAsyncExecutor executor: recordDispatcherExecutors.values()) {
-      if (executor.isWaitingForTasks()) {
-        return false;
+    var res = true;
+    final var toResume = new HashSet<TopicPartition>();
+    final var toPause = new HashSet<TopicPartition>();
+
+    for (var executor : recordDispatcherExecutors.entrySet()) {
+      if (executor.getValue().isWaitingForTasks()) {
+        toResume.add(executor.getKey());
+        res = false;
+      } else {
+        toPause.add(executor.getKey());
       }
     }
 
-    // All executors are busy
-    return true;
+    if (!toPause.isEmpty()) {
+      this.consumer.pause(toPause);
+    }
+    if (!toResume.isEmpty()) {
+      this.consumer.resume(toResume);
+    }
+
+    return res;
   }
 }
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 edc4fe74a..0bcc355c6 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
@@ -165,7 +165,8 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
            egress.getEgressConfig() :
            resource.getEgressConfig();
 
-       consumerConfigs.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, maxProcessingTimeMs(egressConfig));
+       final var maxProcessingTime = maxProcessingTimeMs(consumerConfigs, egressConfig);
+       consumerConfigs.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, maxProcessingTime);
 
        final var egressSubscriberSender = createConsumerRecordSender(
          vertx,
@@ -339,10 +340,11 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
                                                       final EgressConfig egressConfig,
                                                       final DataPlaneContract.Egress egress) {
     var webClientOptions = this.webClientOptions;
-    if (egress.getVReplicas() > 0) {
-      webClientOptions = new WebClientOptions(this.webClientOptions);
-      webClientOptions.setMaxPoolSize(egress.getVReplicas());
-    }
+    // TODO use numPartitions for ordered delivery or max.poll.records for unordered delivery
+    // if (egress.getVReplicas() > 0) {
+    //   webClientOptions = new WebClientOptions(this.webClientOptions);
+    //   webClientOptions.setMaxPoolSize(egress.getVReplicas());
+    // }
     return new WebClientCloudEventSender(vertx, WebClient.create(vertx, webClientOptions), target, egressConfig);
   }
 
@@ -367,20 +369,23 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
     return resource != null && !resource.getNamespace().isBlank() && !resource.getName().isBlank();
   }
 
-  private static long maxProcessingTimeMs(final EgressConfig egressConfig) {
+  private static long maxProcessingTimeMs(final HashMap<String, Object> consumerConfigs, final EgressConfig egressConfig) {
+    final var mpr = consumerConfigs.get(ConsumerConfig.MAX_POLL_RECORDS_CONFIG);
+    final long maxPollRecords = mpr == null ? /* default max.poll.records */ 500L : Long.parseLong(mpr.toString());
     final var retryPolicy = WebClientCloudEventSender.computeRetryPolicy(egressConfig);
     final var retry = egressConfig.getRetry();
     final var timeout = egressConfig.getTimeout();
 
-    var maxProcessingTime = 0;
+    long maxProcessingTimeForSingleRecord = 0L;
     for (int i = 1; i <= retry; i++) {
-      maxProcessingTime += timeout + retryPolicy.apply(i);
+      maxProcessingTimeForSingleRecord += timeout + retryPolicy.apply(i);
     }
     // In addition, we add some seconds as overhead for each retry.
-    final var overhead = 10_000 * retry;
-    maxProcessingTime += overhead;
-    // 2 times since we consider maximum processing time as the time we take for sending events to
-    // a subscriber and to the dead letter sink (including retries).
-    return 2L * maxProcessingTime + overhead;
+    final long overhead = 1000L * retry;
+    maxProcessingTimeForSingleRecord += overhead;
+    // So far, the max processing time calculated is for one single record,
+    // however, we poll records in batches based on the `max.poll.records`
+    // configuration, so the total max processing time is:
+    return maxPollRecords * maxProcessingTimeForSingleRecord;
   }
 }
diff --git a/data-plane/profiler/run.sh b/data-plane/profiler/run.sh
index 06726ad9c..070682e7f 100755
--- a/data-plane/profiler/run.sh
+++ b/data-plane/profiler/run.sh
@@ -40,7 +40,7 @@ echo "Profiling event ${EVENT}"
 
 PROJECT_ROOT_DIR=$(dirname $0)/..
 RESOURCES_DIR="$(dirname $0)"/resources
-ASYNC_PROFILER_URL="https://github.com/jvm-profiling-tools/async-profiler/releases/download/v2.6/async-profiler-2.6-linux-x64.tar.gz"
+ASYNC_PROFILER_URL="https://github.com/jvm-profiling-tools/async-profiler/releases/download/v2.8.3/async-profiler-2.8.3-linux-x64.tar.gz"
 KAFKA_URL="https://archive.apache.org/dist/kafka/2.6.0/kafka_2.13-2.6.0.tgz"
 LOG_DIR=${LOG_DIR:-"/tmp/eventing-kafka-broker-logs/profiler"}
 
@@ -101,7 +101,7 @@ export WAIT_STARTUP_SECONDS="8"
 export SERVICE_NAME="kafka-broker-receiver"
 export SERVICE_NAMESPACE="knative-eventing"
 export INGRESS_PORT="8080"
-export METRICS_PORT="9090"
+export METRICS_PORT="9098"
 export INSTANCE_ID="receiver"
 
 # Run receiver.
@@ -116,7 +116,7 @@ receiver_pid=$!
 export SERVICE_NAME="kafka-broker-dispatcher"
 export SERVICE_NAMESPACE="knative-eventing"
 export INGRESS_PORT="8080"
-export METRICS_PORT="9089"
+export METRICS_PORT="9099"
 export INSTANCE_ID="dispatcher"
 
 # Run dispatcher.
diff --git a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStore.java b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStore.java
index 20df63368..872df8190 100644
--- a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStore.java
+++ b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStore.java
@@ -89,7 +89,7 @@ public class IngressProducerReconcilableStore implements IngressReconcilerListen
     // That means, we support these modes:
     // - Request coming to "/path" --> path is used for matching
     // - Request coming to "/" --> hostname is used for matching
-    final var p = pathMapper.get(path);
+    final var p = pathMapper.get(removeTrailingSlash(path));
     if (p != null) {
       return p;
     }
@@ -113,6 +113,13 @@ public class IngressProducerReconcilableStore implements IngressReconcilerListen
     return hostMapper.get(host);
   }
 
+  private String removeTrailingSlash(final String path) {
+    if (path.endsWith("/")) {
+      return path.substring(0, path.length() - 1);
+    }
+    return path;
+  }
+
   @Override
   public Future<Void> onNewIngress(
     DataPlaneContract.Resource resource,
diff --git a/go.mod b/go.mod
index 8904e03f3..f1cd2416e 100644
--- a/go.mod
+++ b/go.mod
@@ -14,10 +14,13 @@ require (
 	github.com/klauspost/compress v1.15.4 // indirect
 	github.com/manifestival/client-go-client v0.5.0
 	github.com/manifestival/manifestival v0.7.1
+	github.com/openzipkin/zipkin-go v0.3.0
 	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475
 	github.com/rickb777/date v1.14.1
 	github.com/stretchr/testify v1.7.0
 	github.com/xdg-go/scram v1.1.1
+	go.opentelemetry.io/otel v0.20.0
+	go.opentelemetry.io/otel/trace v0.20.0
 	go.uber.org/atomic v1.9.0
 	go.uber.org/multierr v1.8.0
 	go.uber.org/zap v1.21.0
diff --git a/hack/release.sh b/hack/release.sh
index 25a7cb31a..941c05f9a 100755
--- a/hack/release.sh
+++ b/hack/release.sh
@@ -38,19 +38,15 @@ function build_release() {
   [ -f "${EVENTING_KAFKA_POST_INSTALL_ARTIFACT}" ] && rm "${EVENTING_KAFKA_POST_INSTALL_ARTIFACT}"
   [ -f "${EVENTING_KAFKA_SOURCE_BUNDLE_ARTIFACT}" ] && rm "${EVENTING_KAFKA_SOURCE_BUNDLE_ARTIFACT}"
 
-  control_plane_setup
-  if [[ $? -ne 0 ]]; then
-    fail "failed to setup control plane artifact"
-  fi
 
   data_plane_setup
   if [[ $? -ne 0 ]]; then
     fail "failed to create data plane artifact"
   fi
 
-  control_plane_source_setup
+  control_plane_setup
   if [[ $? -ne 0 ]]; then
-    fail "failed to setup control plane source bundle artifact"
+    fail "failed to setup control plane artifact"
   fi
 
   data_plane_source_setup
@@ -58,6 +54,11 @@ function build_release() {
     fail "failed to create data plane source bundle artifact"
   fi
 
+  control_plane_source_setup
+  if [[ $? -ne 0 ]]; then
+    fail "failed to setup control plane source bundle artifact"
+  fi
+
   {
     cat ${EVENTING_KAFKA_CONTROL_PLANE_ARTIFACT}
     cat ${EVENTING_KAFKA_SOURCE_ARTIFACT}
diff --git a/third_party/VENDOR-LICENSE/go.opentelemetry.io/otel/LICENSE b/third_party/VENDOR-LICENSE/go.opentelemetry.io/otel/LICENSE
new file mode 100644
index 000000000..261eeb9e9
--- /dev/null
+++ b/third_party/VENDOR-LICENSE/go.opentelemetry.io/otel/LICENSE
@@ -0,0 +1,201 @@
+                                 Apache License
+                           Version 2.0, January 2004
+                        http://www.apache.org/licenses/
+
+   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+   1. Definitions.
+
+      "License" shall mean the terms and conditions for use, reproduction,
+      and distribution as defined by Sections 1 through 9 of this document.
+
+      "Licensor" shall mean the copyright owner or entity authorized by
+      the copyright owner that is granting the License.
+
+      "Legal Entity" shall mean the union of the acting entity and all
+      other entities that control, are controlled by, or are under common
+      control with that entity. For the purposes of this definition,
+      "control" means (i) the power, direct or indirect, to cause the
+      direction or management of such entity, whether by contract or
+      otherwise, or (ii) ownership of fifty percent (50%) or more of the
+      outstanding shares, or (iii) beneficial ownership of such entity.
+
+      "You" (or "Your") shall mean an individual or Legal Entity
+      exercising permissions granted by this License.
+
+      "Source" form shall mean the preferred form for making modifications,
+      including but not limited to software source code, documentation
+      source, and configuration files.
+
+      "Object" form shall mean any form resulting from mechanical
+      transformation or translation of a Source form, including but
+      not limited to compiled object code, generated documentation,
+      and conversions to other media types.
+
+      "Work" shall mean the work of authorship, whether in Source or
+      Object form, made available under the License, as indicated by a
+      copyright notice that is included in or attached to the work
+      (an example is provided in the Appendix below).
+
+      "Derivative Works" shall mean any work, whether in Source or Object
+      form, that is based on (or derived from) the Work and for which the
+      editorial revisions, annotations, elaborations, or other modifications
+      represent, as a whole, an original work of authorship. For the purposes
+      of this License, Derivative Works shall not include works that remain
+      separable from, or merely link (or bind by name) to the interfaces of,
+      the Work and Derivative Works thereof.
+
+      "Contribution" shall mean any work of authorship, including
+      the original version of the Work and any modifications or additions
+      to that Work or Derivative Works thereof, that is intentionally
+      submitted to Licensor for inclusion in the Work by the copyright owner
+      or by an individual or Legal Entity authorized to submit on behalf of
+      the copyright owner. For the purposes of this definition, "submitted"
+      means any form of electronic, verbal, or written communication sent
+      to the Licensor or its representatives, including but not limited to
+      communication on electronic mailing lists, source code control systems,
+      and issue tracking systems that are managed by, or on behalf of, the
+      Licensor for the purpose of discussing and improving the Work, but
+      excluding communication that is conspicuously marked or otherwise
+      designated in writing by the copyright owner as "Not a Contribution."
+
+      "Contributor" shall mean Licensor and any individual or Legal Entity
+      on behalf of whom a Contribution has been received by Licensor and
+      subsequently incorporated within the Work.
+
+   2. Grant of Copyright License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      copyright license to reproduce, prepare Derivative Works of,
+      publicly display, publicly perform, sublicense, and distribute the
+      Work and such Derivative Works in Source or Object form.
+
+   3. Grant of Patent License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      (except as stated in this section) patent license to make, have made,
+      use, offer to sell, sell, import, and otherwise transfer the Work,
+      where such license applies only to those patent claims licensable
+      by such Contributor that are necessarily infringed by their
+      Contribution(s) alone or by combination of their Contribution(s)
+      with the Work to which such Contribution(s) was submitted. If You
+      institute patent litigation against any entity (including a
+      cross-claim or counterclaim in a lawsuit) alleging that the Work
+      or a Contribution incorporated within the Work constitutes direct
+      or contributory patent infringement, then any patent licenses
+      granted to You under this License for that Work shall terminate
+      as of the date such litigation is filed.
+
+   4. Redistribution. You may reproduce and distribute copies of the
+      Work or Derivative Works thereof in any medium, with or without
+      modifications, and in Source or Object form, provided that You
+      meet the following conditions:
+
+      (a) You must give any other recipients of the Work or
+          Derivative Works a copy of this License; and
+
+      (b) You must cause any modified files to carry prominent notices
+          stating that You changed the files; and
+
+      (c) You must retain, in the Source form of any Derivative Works
+          that You distribute, all copyright, patent, trademark, and
+          attribution notices from the Source form of the Work,
+          excluding those notices that do not pertain to any part of
+          the Derivative Works; and
+
+      (d) If the Work includes a "NOTICE" text file as part of its
+          distribution, then any Derivative Works that You distribute must
+          include a readable copy of the attribution notices contained
+          within such NOTICE file, excluding those notices that do not
+          pertain to any part of the Derivative Works, in at least one
+          of the following places: within a NOTICE text file distributed
+          as part of the Derivative Works; within the Source form or
+          documentation, if provided along with the Derivative Works; or,
+          within a display generated by the Derivative Works, if and
+          wherever such third-party notices normally appear. The contents
+          of the NOTICE file are for informational purposes only and
+          do not modify the License. You may add Your own attribution
+          notices within Derivative Works that You distribute, alongside
+          or as an addendum to the NOTICE text from the Work, provided
+          that such additional attribution notices cannot be construed
+          as modifying the License.
+
+      You may add Your own copyright statement to Your modifications and
+      may provide additional or different license terms and conditions
+      for use, reproduction, or distribution of Your modifications, or
+      for any such Derivative Works as a whole, provided Your use,
+      reproduction, and distribution of the Work otherwise complies with
+      the conditions stated in this License.
+
+   5. Submission of Contributions. Unless You explicitly state otherwise,
+      any Contribution intentionally submitted for inclusion in the Work
+      by You to the Licensor shall be under the terms and conditions of
+      this License, without any additional terms or conditions.
+      Notwithstanding the above, nothing herein shall supersede or modify
+      the terms of any separate license agreement you may have executed
+      with Licensor regarding such Contributions.
+
+   6. Trademarks. This License does not grant permission to use the trade
+      names, trademarks, service marks, or product names of the Licensor,
+      except as required for reasonable and customary use in describing the
+      origin of the Work and reproducing the content of the NOTICE file.
+
+   7. Disclaimer of Warranty. Unless required by applicable law or
+      agreed to in writing, Licensor provides the Work (and each
+      Contributor provides its Contributions) on an "AS IS" BASIS,
+      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+      implied, including, without limitation, any warranties or conditions
+      of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+      PARTICULAR PURPOSE. You are solely responsible for determining the
+      appropriateness of using or redistributing the Work and assume any
+      risks associated with Your exercise of permissions under this License.
+
+   8. Limitation of Liability. In no event and under no legal theory,
+      whether in tort (including negligence), contract, or otherwise,
+      unless required by applicable law (such as deliberate and grossly
+      negligent acts) or agreed to in writing, shall any Contributor be
+      liable to You for damages, including any direct, indirect, special,
+      incidental, or consequential damages of any character arising as a
+      result of this License or out of the use or inability to use the
+      Work (including but not limited to damages for loss of goodwill,
+      work stoppage, computer failure or malfunction, or any and all
+      other commercial damages or losses), even if such Contributor
+      has been advised of the possibility of such damages.
+
+   9. Accepting Warranty or Additional Liability. While redistributing
+      the Work or Derivative Works thereof, You may choose to offer,
+      and charge a fee for, acceptance of support, warranty, indemnity,
+      or other liability obligations and/or rights consistent with this
+      License. However, in accepting such obligations, You may act only
+      on Your own behalf and on Your sole responsibility, not on behalf
+      of any other Contributor, and only if You agree to indemnify,
+      defend, and hold each Contributor harmless for any liability
+      incurred by, or claims asserted against, such Contributor by reason
+      of your accepting any such warranty or additional liability.
+
+   END OF TERMS AND CONDITIONS
+
+   APPENDIX: How to apply the Apache License to your work.
+
+      To apply the Apache License to your work, attach the following
+      boilerplate notice, with the fields enclosed by brackets "[]"
+      replaced with your own identifying information. (Don't include
+      the brackets!)  The text should be enclosed in the appropriate
+      comment syntax for the file format. We also recommend that a
+      file or class name and description of purpose be included on the
+      same "printed page" as the copyright notice for easier
+      identification within third-party archives.
+
+   Copyright [yyyy] [name of copyright owner]
+
+   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.
diff --git a/third_party/VENDOR-LICENSE/go.opentelemetry.io/otel/trace/LICENSE b/third_party/VENDOR-LICENSE/go.opentelemetry.io/otel/trace/LICENSE
new file mode 100644
index 000000000..261eeb9e9
--- /dev/null
+++ b/third_party/VENDOR-LICENSE/go.opentelemetry.io/otel/trace/LICENSE
@@ -0,0 +1,201 @@
+                                 Apache License
+                           Version 2.0, January 2004
+                        http://www.apache.org/licenses/
+
+   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+   1. Definitions.
+
+      "License" shall mean the terms and conditions for use, reproduction,
+      and distribution as defined by Sections 1 through 9 of this document.
+
+      "Licensor" shall mean the copyright owner or entity authorized by
+      the copyright owner that is granting the License.
+
+      "Legal Entity" shall mean the union of the acting entity and all
+      other entities that control, are controlled by, or are under common
+      control with that entity. For the purposes of this definition,
+      "control" means (i) the power, direct or indirect, to cause the
+      direction or management of such entity, whether by contract or
+      otherwise, or (ii) ownership of fifty percent (50%) or more of the
+      outstanding shares, or (iii) beneficial ownership of such entity.
+
+      "You" (or "Your") shall mean an individual or Legal Entity
+      exercising permissions granted by this License.
+
+      "Source" form shall mean the preferred form for making modifications,
+      including but not limited to software source code, documentation
+      source, and configuration files.
+
+      "Object" form shall mean any form resulting from mechanical
+      transformation or translation of a Source form, including but
+      not limited to compiled object code, generated documentation,
+      and conversions to other media types.
+
+      "Work" shall mean the work of authorship, whether in Source or
+      Object form, made available under the License, as indicated by a
+      copyright notice that is included in or attached to the work
+      (an example is provided in the Appendix below).
+
+      "Derivative Works" shall mean any work, whether in Source or Object
+      form, that is based on (or derived from) the Work and for which the
+      editorial revisions, annotations, elaborations, or other modifications
+      represent, as a whole, an original work of authorship. For the purposes
+      of this License, Derivative Works shall not include works that remain
+      separable from, or merely link (or bind by name) to the interfaces of,
+      the Work and Derivative Works thereof.
+
+      "Contribution" shall mean any work of authorship, including
+      the original version of the Work and any modifications or additions
+      to that Work or Derivative Works thereof, that is intentionally
+      submitted to Licensor for inclusion in the Work by the copyright owner
+      or by an individual or Legal Entity authorized to submit on behalf of
+      the copyright owner. For the purposes of this definition, "submitted"
+      means any form of electronic, verbal, or written communication sent
+      to the Licensor or its representatives, including but not limited to
+      communication on electronic mailing lists, source code control systems,
+      and issue tracking systems that are managed by, or on behalf of, the
+      Licensor for the purpose of discussing and improving the Work, but
+      excluding communication that is conspicuously marked or otherwise
+      designated in writing by the copyright owner as "Not a Contribution."
+
+      "Contributor" shall mean Licensor and any individual or Legal Entity
+      on behalf of whom a Contribution has been received by Licensor and
+      subsequently incorporated within the Work.
+
+   2. Grant of Copyright License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      copyright license to reproduce, prepare Derivative Works of,
+      publicly display, publicly perform, sublicense, and distribute the
+      Work and such Derivative Works in Source or Object form.
+
+   3. Grant of Patent License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      (except as stated in this section) patent license to make, have made,
+      use, offer to sell, sell, import, and otherwise transfer the Work,
+      where such license applies only to those patent claims licensable
+      by such Contributor that are necessarily infringed by their
+      Contribution(s) alone or by combination of their Contribution(s)
+      with the Work to which such Contribution(s) was submitted. If You
+      institute patent litigation against any entity (including a
+      cross-claim or counterclaim in a lawsuit) alleging that the Work
+      or a Contribution incorporated within the Work constitutes direct
+      or contributory patent infringement, then any patent licenses
+      granted to You under this License for that Work shall terminate
+      as of the date such litigation is filed.
+
+   4. Redistribution. You may reproduce and distribute copies of the
+      Work or Derivative Works thereof in any medium, with or without
+      modifications, and in Source or Object form, provided that You
+      meet the following conditions:
+
+      (a) You must give any other recipients of the Work or
+          Derivative Works a copy of this License; and
+
+      (b) You must cause any modified files to carry prominent notices
+          stating that You changed the files; and
+
+      (c) You must retain, in the Source form of any Derivative Works
+          that You distribute, all copyright, patent, trademark, and
+          attribution notices from the Source form of the Work,
+          excluding those notices that do not pertain to any part of
+          the Derivative Works; and
+
+      (d) If the Work includes a "NOTICE" text file as part of its
+          distribution, then any Derivative Works that You distribute must
+          include a readable copy of the attribution notices contained
+          within such NOTICE file, excluding those notices that do not
+          pertain to any part of the Derivative Works, in at least one
+          of the following places: within a NOTICE text file distributed
+          as part of the Derivative Works; within the Source form or
+          documentation, if provided along with the Derivative Works; or,
+          within a display generated by the Derivative Works, if and
+          wherever such third-party notices normally appear. The contents
+          of the NOTICE file are for informational purposes only and
+          do not modify the License. You may add Your own attribution
+          notices within Derivative Works that You distribute, alongside
+          or as an addendum to the NOTICE text from the Work, provided
+          that such additional attribution notices cannot be construed
+          as modifying the License.
+
+      You may add Your own copyright statement to Your modifications and
+      may provide additional or different license terms and conditions
+      for use, reproduction, or distribution of Your modifications, or
+      for any such Derivative Works as a whole, provided Your use,
+      reproduction, and distribution of the Work otherwise complies with
+      the conditions stated in this License.
+
+   5. Submission of Contributions. Unless You explicitly state otherwise,
+      any Contribution intentionally submitted for inclusion in the Work
+      by You to the Licensor shall be under the terms and conditions of
+      this License, without any additional terms or conditions.
+      Notwithstanding the above, nothing herein shall supersede or modify
+      the terms of any separate license agreement you may have executed
+      with Licensor regarding such Contributions.
+
+   6. Trademarks. This License does not grant permission to use the trade
+      names, trademarks, service marks, or product names of the Licensor,
+      except as required for reasonable and customary use in describing the
+      origin of the Work and reproducing the content of the NOTICE file.
+
+   7. Disclaimer of Warranty. Unless required by applicable law or
+      agreed to in writing, Licensor provides the Work (and each
+      Contributor provides its Contributions) on an "AS IS" BASIS,
+      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+      implied, including, without limitation, any warranties or conditions
+      of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+      PARTICULAR PURPOSE. You are solely responsible for determining the
+      appropriateness of using or redistributing the Work and assume any
+      risks associated with Your exercise of permissions under this License.
+
+   8. Limitation of Liability. In no event and under no legal theory,
+      whether in tort (including negligence), contract, or otherwise,
+      unless required by applicable law (such as deliberate and grossly
+      negligent acts) or agreed to in writing, shall any Contributor be
+      liable to You for damages, including any direct, indirect, special,
+      incidental, or consequential damages of any character arising as a
+      result of this License or out of the use or inability to use the
+      Work (including but not limited to damages for loss of goodwill,
+      work stoppage, computer failure or malfunction, or any and all
+      other commercial damages or losses), even if such Contributor
+      has been advised of the possibility of such damages.
+
+   9. Accepting Warranty or Additional Liability. While redistributing
+      the Work or Derivative Works thereof, You may choose to offer,
+      and charge a fee for, acceptance of support, warranty, indemnity,
+      or other liability obligations and/or rights consistent with this
+      License. However, in accepting such obligations, You may act only
+      on Your own behalf and on Your sole responsibility, not on behalf
+      of any other Contributor, and only if You agree to indemnify,
+      defend, and hold each Contributor harmless for any liability
+      incurred by, or claims asserted against, such Contributor by reason
+      of your accepting any such warranty or additional liability.
+
+   END OF TERMS AND CONDITIONS
+
+   APPENDIX: How to apply the Apache License to your work.
+
+      To apply the Apache License to your work, attach the following
+      boilerplate notice, with the fields enclosed by brackets "[]"
+      replaced with your own identifying information. (Don't include
+      the brackets!)  The text should be enclosed in the appropriate
+      comment syntax for the file format. We also recommend that a
+      file or class name and description of purpose be included on the
+      same "printed page" as the copyright notice for easier
+      identification within third-party archives.
+
+   Copyright [yyyy] [name of copyright owner]
+
+   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.
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 0054c67be..1c9b175b3 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -323,6 +323,7 @@ github.com/modern-go/concurrent
 # github.com/modern-go/reflect2 v1.0.2
 github.com/modern-go/reflect2
 # github.com/openzipkin/zipkin-go v0.3.0
+## explicit
 github.com/openzipkin/zipkin-go
 github.com/openzipkin/zipkin-go/idgenerator
 github.com/openzipkin/zipkin-go/model
@@ -408,12 +409,14 @@ go.opencensus.io/trace/internal
 go.opencensus.io/trace/propagation
 go.opencensus.io/trace/tracestate
 # go.opentelemetry.io/otel v0.20.0
+## explicit
 go.opentelemetry.io/otel/attribute
 go.opentelemetry.io/otel/codes
 go.opentelemetry.io/otel/internal
 go.opentelemetry.io/otel/internal/baggage
 go.opentelemetry.io/otel/propagation
 # go.opentelemetry.io/otel/trace v0.20.0
+## explicit
 go.opentelemetry.io/otel/trace
 # go.uber.org/atomic v1.9.0
 ## explicit
@@ -1295,6 +1298,7 @@ knative.dev/eventing/test/rekt/resources/flaker
 knative.dev/eventing/test/rekt/resources/pingsource
 knative.dev/eventing/test/rekt/resources/source
 knative.dev/eventing/test/rekt/resources/subscription
+knative.dev/eventing/test/rekt/resources/svc
 knative.dev/eventing/test/rekt/resources/trigger
 knative.dev/eventing/test/test_images
 knative.dev/eventing/test/test_images/event-flaker


Test patch:
diff --git a/.github/workflows/kind-e2e.yaml b/.github/workflows/kind-e2e.yaml
index 322378c79..71fd44d20 100644
--- a/.github/workflows/kind-e2e.yaml
+++ b/.github/workflows/kind-e2e.yaml
@@ -24,14 +24,14 @@ jobs:
         eventing-config:
           - "./third_party/eventing-latest/"
 
-        channel-auth-scenario:
-          - ""
-          - "SSL"
-          - "SASL_SSL"
-          - "SASL_PLAIN"
+        broker-class:
+          - Kafka
 
         test-suite:
           - ./test/e2e
+          - ./test/e2e_broker
+          - ./test/e2e_sink
+          - ./test/e2e_source
           - ./test/e2e_new
           - ./test/e2e_channel
           - ./test/e2e_new_channel
@@ -50,27 +50,6 @@ jobs:
           - test-suite: ./test/e2e_channel
             extra-test-flags: -channels=messaging.knative.dev/v1beta1:KafkaChannel
 
-        # don't run e2e for tests other than KafkaChannel when channel-auth-scenario is set
-        exclude:
-          - test-suite: ./test/e2e
-            channel-auth-scenario: "SSL"
-          - test-suite: ./test/e2e
-            channel-auth-scenario: "SASL_SSL"
-          - test-suite: ./test/e2e
-            channel-auth-scenario: "SASL_PLAIN"
-          - test-suite: ./test/e2e_new
-            channel-auth-scenario: "SSL"
-          - test-suite: ./test/e2e_new
-            channel-auth-scenario: "SASL_SSL"
-          - test-suite: ./test/e2e_new
-            channel-auth-scenario: "SASL_PLAIN"
-          - test-suite: ./test/experimental
-            channel-auth-scenario: "SSL"
-          - test-suite: ./test/experimental
-            channel-auth-scenario: "SASL_SSL"
-          - test-suite: ./test/experimental
-            channel-auth-scenario: "SASL_PLAIN"
-
     env:
       GOPATH: ${{ github.workspace }}
       KO_DOCKER_REPO: localhost:5001
@@ -86,7 +65,7 @@ jobs:
       NODE_VERSION: ${{ matrix.k8s-version }}
       NODE_SHA: ${{ matrix.kind-image-sha }}
 
-      EVENTING_KAFKA_BROKER_CHANNEL_AUTH_SCENARIO: ${{ matrix.channel-auth-scenario }}
+      BROKER_CLASS: ${{ matrix.broker-class }}
 
     steps:
       - name: Install Dependencies
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink_test.go b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
index 4c7c06d80..6ef1e2180 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink_test.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
@@ -162,6 +162,60 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 				},
 			},
 		},
+		{
+			Name: "Reconciled normal - content mode structured",
+			Objects: []runtime.Object{
+				NewSink(
+					StatusControllerOwnsTopic(reconciler.ControllerTopicOwner),
+					SinkContentMode(v1alpha1.ModeStructured),
+				),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				SinkReceiverPod(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve": "value_to_preserve",
+				}),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Uid:              SinkUUID,
+							Topics:           []string{SinkTopic()},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, Path: receiver.Path(SinkNamespace, SinkName)},
+							BootstrapServers: bootstrapServers,
+							Reference:        SinkReference(),
+						},
+					},
+					Generation: 1,
+				}),
+				SinkReceiverPodUpdate(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "1",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSink(
+						StatusControllerOwnsTopic(reconciler.ControllerTopicOwner),
+						SinkContentMode(v1alpha1.ModeStructured),
+						InitSinkConditions,
+						StatusDataPlaneAvailable,
+						StatusConfigParsed,
+						BootstrapServers(bootstrapServersArr),
+						StatusConfigMapUpdatedReady(&env),
+						StatusTopicReadyWithOwner(SinkTopic(), sink.ControllerTopicOwner),
+						SinkAddressable(&env),
+						StatusProbeSucceeded,
+					),
+				},
+			},
+		},
 		{
 			Name: "Reconciled normal - with auth config",
 			Objects: []runtime.Object{
diff --git a/control-plane/pkg/reconciler/source/source_test.go b/control-plane/pkg/reconciler/source/source_test.go
index 09d333442..2ace88d4a 100644
--- a/control-plane/pkg/reconciler/source/source_test.go
+++ b/control-plane/pkg/reconciler/source/source_test.go
@@ -112,7 +112,7 @@ func TestReconcileKind(t *testing.T) {
 								internals.Ordered,
 								NewConsumerTimeout("PT600S"),
 								NewConsumerRetry(10),
-								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
 							),
 						),
@@ -158,7 +158,7 @@ func TestReconcileKind(t *testing.T) {
 								internals.Ordered,
 								NewConsumerTimeout("PT600S"),
 								NewConsumerRetry(10),
-								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
 							),
 						),
@@ -286,7 +286,7 @@ func TestReconcileKind(t *testing.T) {
 								internals.Ordered,
 								NewConsumerTimeout("PT600S"),
 								NewConsumerRetry(10),
-								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
 							),
 						),
@@ -398,7 +398,7 @@ func TestReconcileKind(t *testing.T) {
 								internals.Ordered,
 								NewConsumerTimeout("PT600S"),
 								NewConsumerRetry(10),
-								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
 							),
 						),
@@ -447,7 +447,7 @@ func TestReconcileKind(t *testing.T) {
 								internals.Ordered,
 								NewConsumerTimeout("PT600S"),
 								NewConsumerRetry(10),
-								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
 							),
 						),
@@ -495,7 +495,7 @@ func TestReconcileKind(t *testing.T) {
 								internals.Ordered,
 								NewConsumerTimeout("PT600S"),
 								NewConsumerRetry(10),
-								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
 							),
 						),
@@ -527,7 +527,7 @@ func TestReconcileKind(t *testing.T) {
 									internals.Ordered,
 									NewConsumerTimeout("PT600S"),
 									NewConsumerRetry(10),
-									NewConsumerBackoffDelay("PT10S"),
+									NewConsumerBackoffDelay("PT0.3S"),
 									NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
 								),
 							),
@@ -585,7 +585,7 @@ func TestReconcileKind(t *testing.T) {
 									internals.Ordered,
 									NewConsumerTimeout("PT600S"),
 									NewConsumerRetry(10),
-									NewConsumerBackoffDelay("PT10S"),
+									NewConsumerBackoffDelay("PT0.3S"),
 									NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
 								),
 							),
@@ -642,7 +642,7 @@ func TestReconcileKind(t *testing.T) {
 									internals.Ordered,
 									NewConsumerTimeout("PT600S"),
 									NewConsumerRetry(10),
-									NewConsumerBackoffDelay("PT10S"),
+									NewConsumerBackoffDelay("PT0.3S"),
 									NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
 								),
 							),
@@ -685,7 +685,7 @@ func TestReconcileKind(t *testing.T) {
 								internals.Ordered,
 								NewConsumerTimeout("PT600S"),
 								NewConsumerRetry(10),
-								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
 							),
 						),
@@ -729,7 +729,7 @@ func TestReconcileKind(t *testing.T) {
 								internals.Ordered,
 								NewConsumerTimeout("PT600S"),
 								NewConsumerRetry(10),
-								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
 							),
 						),
@@ -773,7 +773,7 @@ func TestReconcileKind(t *testing.T) {
 								internals.Ordered,
 								NewConsumerTimeout("PT600S"),
 								NewConsumerRetry(10),
-								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
 							),
 						),
@@ -818,7 +818,7 @@ func TestReconcileKind(t *testing.T) {
 								internals.Ordered,
 								NewConsumerTimeout("PT600S"),
 								NewConsumerRetry(10),
-								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
 							),
 						),
diff --git a/control-plane/pkg/reconciler/testing/objects_sink.go b/control-plane/pkg/reconciler/testing/objects_sink.go
index 7aaea0294..d2cb723b0 100644
--- a/control-plane/pkg/reconciler/testing/objects_sink.go
+++ b/control-plane/pkg/reconciler/testing/objects_sink.go
@@ -166,3 +166,10 @@ func SinkReceiverPodUpdate(namespace string, annotations map[string]string) clie
 		SinkReceiverPod(namespace, annotations),
 	)
 }
+
+func SinkContentMode(cm string) KRShapedOption {
+	return func(obj duckv1.KRShaped) {
+		ks := obj.(*eventing.KafkaSink)
+		ks.Spec.ContentMode = &cm
+	}
+}
diff --git a/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/utils/BaseEnvTest.java b/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/utils/BaseEnvTest.java
index ede8b0e27..7e1f4f7da 100644
--- a/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/utils/BaseEnvTest.java
+++ b/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/utils/BaseEnvTest.java
@@ -31,7 +31,7 @@ public class BaseEnvTest {
     case "CONFIG_TRACING_PATH" -> "/etc/tracing";
     case "METRICS_JVM_ENABLED" -> "false";
     case "WAIT_STARTUP_SECONDS" -> "1";
-    default -> throw new IllegalArgumentException("unknown " + s);
+    default -> null;
   };
 
   @Test
@@ -73,4 +73,16 @@ public class BaseEnvTest {
     final var metricsConfigs = new BaseEnv(provider);
     assertThat(metricsConfigs.isMetricsJvmEnabled()).isFalse();
   }
+
+  @Test
+  public void shouldGetHttpClientMetricsDisabled() {
+    final var metricsConfigs = new BaseEnv(provider);
+    assertThat(metricsConfigs.isMetricsHTTPClientEnabled()).isFalse();
+  }
+
+  @Test
+  public void shouldGetHttpServerMetricsDisabled() {
+    final var metricsConfigs = new BaseEnv(provider);
+    assertThat(metricsConfigs.isMetricsHTTPServerEnabled()).isFalse();
+  }
 }
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 eecda4cca..9f549995c 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
@@ -422,9 +422,8 @@ public class RecordDispatcherTest {
           .summaries()
           .stream()
           .reduce((a, b) -> b)
-          .get()
-          .max()
-      ).isGreaterThan(0)
+          .isPresent()
+      ).isTrue()
     );
   }
 
diff --git a/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStoreTest.java b/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStoreTest.java
index 6674aca69..b0321cd94 100644
--- a/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStoreTest.java
+++ b/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStoreTest.java
@@ -406,6 +406,7 @@ public class IngressProducerReconcilableStoreTest {
 
           // only use path when the path is registered
           assertThat(store.resolve("http://host1", "/hello1").getKafkaProducer()).isSameAs(producer1);
+          assertThat(store.resolve("http://host1", "/hello1/").getKafkaProducer()).isSameAs(producer1);
           assertThat(store.resolve("http://host2", "/hello1").getKafkaProducer()).isSameAs(producer1);
           assertThat(store.resolve("http://host1", "/hello3").getKafkaProducer()).isSameAs(producer3);
           assertThat(store.resolve("http://host2", "/hello3").getKafkaProducer()).isSameAs(producer3);
diff --git a/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/main/ReceiverEnvTest.java b/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/main/ReceiverEnvTest.java
index 61ca0ce2f..e7468d661 100644
--- a/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/main/ReceiverEnvTest.java
+++ b/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/main/ReceiverEnvTest.java
@@ -48,7 +48,7 @@ class ReceiverEnvTest {
         case BaseEnv.CONFIG_TRACING_PATH -> TRACING_CONFIG_PATH;
         case BaseEnv.METRICS_JVM_ENABLED -> METRICS_JVM_ENABLED;
         case BaseEnv.WAIT_STARTUP_SECONDS -> Integer.valueOf(WAIT_STARTUP_SECONDS).toString();
-        default -> throw new IllegalArgumentException(key);
+        default -> null;
       }
     );
 
diff --git a/test/e2e-tests.sh b/test/e2e-tests.sh
index 748b3ba61..1ae285108 100755
--- a/test/e2e-tests.sh
+++ b/test/e2e-tests.sh
@@ -27,14 +27,19 @@ if [ "${EVENTING_KAFKA_BROKER_CHANNEL_AUTH_SCENARIO:-""}" != "" ]; then
   success
 fi
 
-go_test_e2e -timeout=1h ./test/e2e/... || fail_test "E2E suite failed"
+export BROKER_CLASS="Kafka"
 
-go_test_e2e -timeout=1h ./test/e2e_channel/... -channels=messaging.knative.dev/v1beta1:KafkaChannel || fail_test "E2E suite (KafkaChannel) failed"
+go_test_e2e -timeout=1h ./test/e2e/...        || fail_test "E2E suite failed (directory: ./e2e/...)"
+go_test_e2e -timeout=1h ./test/e2e_broker/... || fail_test "E2E suite failed (directory: ./e2e_broker/...)"
+go_test_e2e -timeout=1h ./test/e2e_sink/...   || fail_test "E2E suite failed (directory: ./e2e_sink/...)"
+go_test_e2e -timeout=1h ./test/e2e_source/... || fail_test "E2E suite failed (directory: ./e2e_source/...)"
 
-go_test_e2e -tags=deletecm ./test/e2e/... || fail_test "E2E (deletecm) suite failed"
+go_test_e2e -timeout=1h ./test/e2e_channel/... -channels=messaging.knative.dev/v1beta1:KafkaChannel || fail_test "E2E suite (KafkaChannel) failed (directory: ./e2e_channel/...)"
+
+go_test_e2e -tags=deletecm ./test/e2e/... || fail_test "E2E (deletecm) suite failed (directory: ./e2e/...)"
 
 if ! ${LOCAL_DEVELOPMENT}; then
-  go_test_e2e -tags=sacura -timeout=40m ./test/e2e/... || fail_test "E2E (sacura) suite failed"
+  go_test_e2e -tags=sacura -timeout=40m ./test/e2e/... || fail_test "E2E (sacura) suite failed (directory: ./e2e/...)"
 fi
 
 success
diff --git a/test/e2e/conformance/data_plane_conformance_test.go b/test/e2e/conformance/data_plane_conformance_test.go
index 4459b818c..481539df2 100644
--- a/test/e2e/conformance/data_plane_conformance_test.go
+++ b/test/e2e/conformance/data_plane_conformance_test.go
@@ -380,6 +380,7 @@ func TestBrokerConsumer(t *testing.T) {
 
 func createBroker(client *testlib.Client) *eventing.Broker {
 
+	// We don't care about the namespaced broker in this dataplane conformance tests
 	broker := client.CreateBrokerOrFail("broker",
 		resources.WithBrokerClassForBroker(kafka.BrokerClass),
 	)
diff --git a/test/e2e/broker_event_trasformation_test.go b/test/e2e_broker/broker_event_trasformation_test.go
similarity index 98%
rename from test/e2e/broker_event_trasformation_test.go
rename to test/e2e_broker/broker_event_trasformation_test.go
index 28e26faf6..9710b185a 100644
--- a/test/e2e/broker_event_trasformation_test.go
+++ b/test/e2e_broker/broker_event_trasformation_test.go
@@ -17,7 +17,7 @@
  * limitations under the License.
  */
 
-package e2e
+package e2e_broker
 
 import (
 	"context"
diff --git a/test/e2e/broker_redelivery_test.go b/test/e2e_broker/broker_redelivery_test.go
similarity index 82%
rename from test/e2e/broker_redelivery_test.go
rename to test/e2e_broker/broker_redelivery_test.go
index 937781e2d..f5d7c0b91 100644
--- a/test/e2e/broker_redelivery_test.go
+++ b/test/e2e_broker/broker_redelivery_test.go
@@ -17,7 +17,7 @@
  * limitations under the License.
  */
 
-package e2e
+package e2e_broker
 
 import (
 	"context"
@@ -29,7 +29,7 @@ import (
 	testlib "knative.dev/eventing/test/lib"
 	"knative.dev/eventing/test/lib/resources"
 
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
+	"knative.dev/eventing-kafka-broker/test/pkg/broker"
 	kafkatesting "knative.dev/eventing-kafka-broker/test/pkg/testing"
 )
 
@@ -38,6 +38,10 @@ const (
 )
 
 func TestBrokerRedeliveryBrokerV1BackoffLinear(t *testing.T) {
+	class, err := broker.GetKafkaClassFromEnv()
+	if err != nil {
+		t.Fatalf("error getting KafkaBroker class from env '%v'", err)
+	}
 
 	kafkatesting.RunMultiple(t, func(t *testing.T) {
 
@@ -46,7 +50,7 @@ func TestBrokerRedeliveryBrokerV1BackoffLinear(t *testing.T) {
 			backoff := eventingduck.BackoffPolicyLinear
 
 			client.CreateBrokerOrFail(brokerName,
-				resources.WithBrokerClassForBroker(kafka.BrokerClass),
+				resources.WithBrokerClassForBroker(class),
 				resources.WithDeliveryForBroker(&eventingduck.DeliverySpec{
 					Retry:         &numRetries,
 					BackoffPolicy: &backoff,
@@ -60,6 +64,10 @@ func TestBrokerRedeliveryBrokerV1BackoffLinear(t *testing.T) {
 }
 
 func TestBrokerRedeliveryBrokerV1BackoffExponential(t *testing.T) {
+	class, err := broker.GetKafkaClassFromEnv()
+	if err != nil {
+		t.Fatalf("error getting KafkaBroker class from env '%v'", err)
+	}
 
 	kafkatesting.RunMultiple(t, func(t *testing.T) {
 
@@ -68,7 +76,7 @@ func TestBrokerRedeliveryBrokerV1BackoffExponential(t *testing.T) {
 			backoff := eventingduck.BackoffPolicyExponential
 
 			client.CreateBrokerOrFail(brokerName,
-				resources.WithBrokerClassForBroker(kafka.BrokerClass),
+				resources.WithBrokerClassForBroker(class),
 				resources.WithDeliveryForBroker(&eventingduck.DeliverySpec{
 					Retry:         &numRetries,
 					BackoffPolicy: &backoff,
diff --git a/test/e2e/broker_sasl_ssl_test.go b/test/e2e_broker/broker_sasl_ssl_test.go
similarity index 95%
rename from test/e2e/broker_sasl_ssl_test.go
rename to test/e2e_broker/broker_sasl_ssl_test.go
index efa3ee71f..9ae083951 100644
--- a/test/e2e/broker_sasl_ssl_test.go
+++ b/test/e2e_broker/broker_sasl_ssl_test.go
@@ -17,7 +17,7 @@
  * limitations under the License.
  */
 
-package e2e
+package e2e_broker
 
 import (
 	"context"
@@ -38,12 +38,17 @@ import (
 	"knative.dev/eventing/test/lib/sender"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
+	"knative.dev/eventing-kafka-broker/test/pkg/broker"
 	. "knative.dev/eventing-kafka-broker/test/pkg/testing"
 )
 
 func brokerAuth(t *testing.T, secretProvider SecretProvider, configProvider ConfigProvider) {
 
+	class, err := broker.GetKafkaClassFromEnv()
+	if err != nil {
+		t.Fatalf("error getting KafkaBroker class from env '%v'", err)
+	}
+
 	RunMultiple(t, func(t *testing.T) {
 
 		ctx := context.Background()
@@ -72,7 +77,7 @@ func brokerAuth(t *testing.T, secretProvider SecretProvider, configProvider Conf
 
 		client.CreateBrokerOrFail(
 			brokerName,
-			resources.WithBrokerClassForBroker(kafka.BrokerClass),
+			resources.WithBrokerClassForBroker(class),
 			resources.WithConfigForBroker(&duckv1.KReference{
 				APIVersion: "v1",
 				Kind:       "ConfigMap",
diff --git a/test/e2e/broker_trigger_sink_test.go b/test/e2e_broker/broker_trigger_sink_test.go
similarity index 94%
rename from test/e2e/broker_trigger_sink_test.go
rename to test/e2e_broker/broker_trigger_sink_test.go
index 27fb01cdd..1f2c43db6 100644
--- a/test/e2e/broker_trigger_sink_test.go
+++ b/test/e2e_broker/broker_trigger_sink_test.go
@@ -17,7 +17,7 @@
  * limitations under the License.
  */
 
-package e2e
+package e2e_broker
 
 import (
 	"context"
@@ -34,8 +34,8 @@ import (
 
 	eventingv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
 	eventingv1alpha1clientset "knative.dev/eventing-kafka-broker/control-plane/pkg/client/clientset/versioned/typed/eventing/v1alpha1"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	"knative.dev/eventing-kafka-broker/test/pkg/addressable"
+	"knative.dev/eventing-kafka-broker/test/pkg/broker"
 	"knative.dev/eventing-kafka-broker/test/pkg/sink"
 	testingpkg "knative.dev/eventing-kafka-broker/test/pkg/testing"
 )
@@ -50,6 +50,11 @@ import (
                 +--------+
 */
 func TestBrokerV1TriggersV1SinkV1Alpha1(t *testing.T) {
+	class, err := broker.GetKafkaClassFromEnv()
+	if err != nil {
+		t.Fatalf("error getting KafkaBroker class from env '%v'", err)
+	}
+
 	testingpkg.RunMultipleN(t, 10, func(t *testing.T) {
 
 		ctx := context.Background()
@@ -81,7 +86,7 @@ func TestBrokerV1TriggersV1SinkV1Alpha1(t *testing.T) {
 		// Create a Kafka Broker
 		br := client.CreateBrokerOrFail(
 			"broker",
-			resources.WithBrokerClassForBroker(kafka.BrokerClass),
+			resources.WithBrokerClassForBroker(class),
 		)
 
 		// Create 2 Triggers with the same subscriber
diff --git a/test/e2e/broker_trigger_test.go b/test/e2e_broker/broker_trigger_test.go
similarity index 94%
rename from test/e2e/broker_trigger_test.go
rename to test/e2e_broker/broker_trigger_test.go
index 9c1b3198d..0e0f2eef4 100644
--- a/test/e2e/broker_trigger_test.go
+++ b/test/e2e_broker/broker_trigger_test.go
@@ -17,7 +17,7 @@
  * limitations under the License.
  */
 
-package e2e
+package e2e_broker
 
 import (
 	"context"
@@ -36,11 +36,17 @@ import (
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
+	brokertest "knative.dev/eventing-kafka-broker/test/pkg/broker"
 	kafkatest "knative.dev/eventing-kafka-broker/test/pkg/kafka"
 	testingpkg "knative.dev/eventing-kafka-broker/test/pkg/testing"
 )
 
 func TestBrokerTrigger(t *testing.T) {
+	class, err := brokertest.GetKafkaClassFromEnv()
+	if err != nil {
+		t.Fatalf("error getting KafkaBroker class from env '%v'", err)
+	}
+
 	testingpkg.RunMultiple(t, func(t *testing.T) {
 
 		ctx := context.Background()
@@ -69,7 +75,7 @@ func TestBrokerTrigger(t *testing.T) {
 
 		br := client.CreateBrokerOrFail(
 			brokerName,
-			resources.WithBrokerClassForBroker(kafka.BrokerClass),
+			resources.WithBrokerClassForBroker(class),
 		)
 
 		eventTracker, _ := recordevents.StartEventRecordOrFail(ctx, client, subscriber)
@@ -159,6 +165,11 @@ func TestBrokerTrigger(t *testing.T) {
 }
 
 func TestBrokerWithConfig(t *testing.T) {
+	class, err := brokertest.GetKafkaClassFromEnv()
+	if err != nil {
+		t.Fatalf("error getting KafkaBroker class from env '%v'", err)
+	}
+
 	testingpkg.RunMultiple(t, func(t *testing.T) {
 
 		ctx := context.Background()
@@ -193,7 +204,7 @@ func TestBrokerWithConfig(t *testing.T) {
 
 		br := client.CreateBrokerOrFail(
 			brokerName,
-			resources.WithBrokerClassForBroker(kafka.BrokerClass),
+			resources.WithBrokerClassForBroker(class),
 			resources.WithConfigForBroker(&duckv1.KReference{
 				Kind:       "ConfigMap",
 				Namespace:  cm.Namespace,
diff --git a/test/e2e/broker_with_many_triggers_test.go b/test/e2e_broker/broker_with_many_triggers_test.go
similarity index 98%
rename from test/e2e/broker_with_many_triggers_test.go
rename to test/e2e_broker/broker_with_many_triggers_test.go
index 3f9eb6670..06b116e34 100644
--- a/test/e2e/broker_with_many_triggers_test.go
+++ b/test/e2e_broker/broker_with_many_triggers_test.go
@@ -17,7 +17,7 @@
  * limitations under the License.
  */
 
-package e2e
+package e2e_broker
 
 import (
 	"context"
diff --git a/test/e2e/conformance/control_plane_conformance_test.go b/test/e2e_broker/conformance/control_plane_conformance_test.go
similarity index 75%
rename from test/e2e/conformance/control_plane_conformance_test.go
rename to test/e2e_broker/conformance/control_plane_conformance_test.go
index c4950297c..f436a0346 100644
--- a/test/e2e/conformance/control_plane_conformance_test.go
+++ b/test/e2e_broker/conformance/control_plane_conformance_test.go
@@ -20,23 +20,29 @@
 package conformance
 
 import (
+	"fmt"
 	"testing"
 
 	conformance "knative.dev/eventing/test/conformance/helpers"
 	testlib "knative.dev/eventing/test/lib"
 	"knative.dev/eventing/test/lib/resources"
 
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
+	"knative.dev/eventing-kafka-broker/test/pkg/broker"
 	pkgtesting "knative.dev/eventing-kafka-broker/test/pkg/testing"
 )
 
 func brokerCreator(client *testlib.Client, name string) {
 
-	broker := client.CreateBrokerOrFail(name,
-		resources.WithBrokerClassForBroker(kafka.BrokerClass),
+	class, err := broker.GetKafkaClassFromEnv()
+	if err != nil {
+		panic(fmt.Sprintf("error getting KafkaBroker class from env '%v'", err))
+	}
+
+	b := client.CreateBrokerOrFail(name,
+		resources.WithBrokerClassForBroker(class),
 	)
 
-	client.WaitForResourceReadyOrFail(broker.Name, testlib.BrokerTypeMeta)
+	client.WaitForResourceReadyOrFail(b.Name, testlib.BrokerTypeMeta)
 }
 
 func TestBrokerControlPlane(t *testing.T) {
diff --git a/test/e2e_broker/conformance/main_test.go b/test/e2e_broker/conformance/main_test.go
new file mode 100644
index 000000000..b5daffbf5
--- /dev/null
+++ b/test/e2e_broker/conformance/main_test.go
@@ -0,0 +1,43 @@
+//go:build e2e
+// +build e2e
+
+/*
+ * Copyright 2020 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package conformance
+
+import (
+	"log"
+	"os"
+	"testing"
+
+	testlib "knative.dev/eventing/test/lib"
+	"knative.dev/pkg/system"
+	"knative.dev/pkg/test/zipkin"
+)
+
+func TestMain(m *testing.M) {
+
+	os.Exit(func() int {
+		// Any tests may SetupZipkinTracing, it will only actually be done once. This should be the ONLY
+		// place that cleans it up. If an individual test calls this instead, then it will break other
+		// tests that need the tracing in place.
+		defer zipkin.CleanupZipkinTracingSetup(log.Printf)
+		defer testlib.ExportLogs(testlib.SystemLogsDir, system.Namespace())
+
+		return m.Run()
+	}())
+}
diff --git a/test/e2e_broker/main_test.go b/test/e2e_broker/main_test.go
new file mode 100644
index 000000000..f23fb50fc
--- /dev/null
+++ b/test/e2e_broker/main_test.go
@@ -0,0 +1,36 @@
+//go:build e2e
+// +build e2e
+
+/*
+ * Copyright 2020 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package e2e_broker
+
+import (
+	"os"
+	"testing"
+
+	testlib "knative.dev/eventing/test/lib"
+	"knative.dev/pkg/system"
+)
+
+func TestMain(m *testing.M) {
+	os.Exit(func() int {
+		defer testlib.ExportLogs(testlib.SystemLogsDir, system.Namespace())
+
+		return m.Run()
+	}())
+}
diff --git a/test/e2e_new/broker_conformance_test.go b/test/e2e_new/broker_conformance_test.go
index c99a223b6..47680aff5 100644
--- a/test/e2e_new/broker_conformance_test.go
+++ b/test/e2e_new/broker_conformance_test.go
@@ -33,8 +33,6 @@ import (
 	"knative.dev/reconciler-test/pkg/environment"
 	"knative.dev/reconciler-test/pkg/k8s"
 	"knative.dev/reconciler-test/pkg/knative"
-
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 )
 
 func TestBrokerConformance(t *testing.T) {
@@ -47,7 +45,7 @@ func TestBrokerConformance(t *testing.T) {
 	)
 
 	// Install and wait for a Ready Broker.
-	env.Prerequisite(ctx, t, brokerfeatures.GoesReady("default", broker.WithBrokerClass(kafka.BrokerClass)))
+	env.Prerequisite(ctx, t, brokerfeatures.GoesReady("default", broker.WithBrokerClass(broker.EnvCfg.BrokerClass)))
 
 	env.TestSet(ctx, t, brokerfeatures.ControlPlaneConformance("default"))
 	env.TestSet(ctx, t, brokerfeatures.DataPlaneConformance("default"))
diff --git a/test/e2e_new/broker_delete_config_map_test.go b/test/e2e_new/broker_delete_config_map_test.go
index 177a28416..152c188c4 100644
--- a/test/e2e_new/broker_delete_config_map_test.go
+++ b/test/e2e_new/broker_delete_config_map_test.go
@@ -34,7 +34,6 @@ import (
 	"knative.dev/reconciler-test/pkg/knative"
 	"knative.dev/reconciler-test/resources/svc"
 
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	"knative.dev/eventing-kafka-broker/test/e2e_new/resources/configmap"
 )
 
@@ -65,7 +64,7 @@ func BrokerDeleteContractConfigMap() *feature.Feature {
 	// Create the broker
 	f.Setup("install broker", broker.Install(
 		brokerName,
-		broker.WithBrokerClass(kafka.BrokerClass),
+		broker.WithBrokerClass(broker.EnvCfg.BrokerClass),
 	))
 	f.Setup("broker is ready", broker.IsReady(brokerName))
 	f.Setup("broker is addressable", broker.IsAddressable(brokerName))
diff --git a/test/e2e_new/broker_not_ready_test.go b/test/e2e_new/broker_not_ready_test.go
index 42948656a..7b13dab4d 100644
--- a/test/e2e_new/broker_not_ready_test.go
+++ b/test/e2e_new/broker_not_ready_test.go
@@ -29,7 +29,6 @@ import (
 	"knative.dev/reconciler-test/pkg/k8s"
 	"knative.dev/reconciler-test/pkg/knative"
 
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	brokerconfigmap "knative.dev/eventing-kafka-broker/test/e2e_new/resources/configmap/broker"
 )
 
@@ -63,7 +62,7 @@ func BrokerNotReadyAfterBeingReady() *feature.Feature {
 
 	f.Setup("install broker", broker.Install(
 		brokerName,
-		broker.WithBrokerClass(kafka.BrokerClass),
+		broker.WithBrokerClass(broker.EnvCfg.BrokerClass),
 		broker.WithConfig(configName),
 	))
 
diff --git a/test/e2e_new/dls_test.go b/test/e2e_new/dls_test.go
index 5afaa0595..23a828981 100644
--- a/test/e2e_new/dls_test.go
+++ b/test/e2e_new/dls_test.go
@@ -35,8 +35,6 @@ import (
 	"knative.dev/reconciler-test/resources/svc"
 
 	. "knative.dev/reconciler-test/pkg/eventshub/assert"
-
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 )
 
 func TestDeadLetterSink(t *testing.T) {
@@ -67,7 +65,7 @@ func SendsEventWithRetries() *feature.Feature {
 
 	f.Setup("install broker", broker.Install(
 		brokerName,
-		broker.WithBrokerClass(kafka.BrokerClass),
+		broker.WithBrokerClass(broker.EnvCfg.BrokerClass),
 	))
 
 	f.Setup("broker is ready", broker.IsReady(brokerName))
@@ -127,7 +125,7 @@ func SendsEventNoRetries() *feature.Feature {
 
 	f.Setup("install broker", broker.Install(
 		brokerName,
-		broker.WithBrokerClass(kafka.BrokerClass),
+		broker.WithBrokerClass(broker.EnvCfg.BrokerClass),
 	))
 
 	f.Setup("broker is ready", broker.IsReady(brokerName))
diff --git a/test/e2e_new/main_test.go b/test/e2e_new/main_test.go
index 7d02eae8c..86df30ac4 100644
--- a/test/e2e_new/main_test.go
+++ b/test/e2e_new/main_test.go
@@ -21,6 +21,8 @@ package e2e_new
 
 import (
 	"flag"
+	"fmt"
+	"log"
 	"os"
 	"testing"
 
@@ -29,6 +31,7 @@ import (
 
 	"knative.dev/pkg/injection"
 	_ "knative.dev/pkg/system/testing"
+	"knative.dev/pkg/test/zipkin"
 
 	"knative.dev/eventing/test/rekt/resources/broker"
 	"knative.dev/reconciler-test/pkg/environment"
@@ -46,7 +49,9 @@ func init() {
 	environment.InitFlags(flag.CommandLine)
 
 	// Force the broker class to be configured properly
-	broker.EnvCfg.BrokerClass = kafka.BrokerClass
+	if broker.EnvCfg.BrokerClass != kafka.BrokerClass && broker.EnvCfg.BrokerClass != kafka.NamespacedBrokerClass {
+		panic(fmt.Errorf("KafkaBroker class '%s' is unknown. Specify 'BROKER_CLASS' env var", broker.EnvCfg.BrokerClass))
+	}
 }
 
 // TestMain is the first entry point for `go test`.
@@ -67,5 +72,11 @@ func TestMain(m *testing.M) {
 	global = environment.NewGlobalEnvironment(ctx)
 
 	// Run the tests.
-	os.Exit(m.Run())
+	os.Exit(func() int {
+		// Any tests may SetupZipkinTracing, it will only actually be done once. This should be the ONLY
+		// place that cleans it up. If an individual test calls this instead, then it will break other
+		// tests that need the tracing in place.
+		defer zipkin.CleanupZipkinTracingSetup(log.Printf)
+		return m.Run()
+	}())
 }
diff --git a/test/e2e_new/ordered_test.go b/test/e2e_new/ordered_test.go
index a815f001f..1383faee4 100644
--- a/test/e2e_new/ordered_test.go
+++ b/test/e2e_new/ordered_test.go
@@ -40,7 +40,6 @@ import (
 	"knative.dev/reconciler-test/pkg/knative"
 	"knative.dev/reconciler-test/resources/svc"
 
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	"knative.dev/eventing-kafka-broker/test/e2e_new/multiple_partition_config"
 	"knative.dev/eventing-kafka-broker/test/e2e_new/single_partition_config"
 
@@ -77,7 +76,7 @@ func SinglePartitionOrderedDelivery() *feature.Feature {
 	f.Setup("install one partition configuration", single_partition_config.Install)
 	f.Setup("install broker", broker.Install(
 		brokerName,
-		broker.WithBrokerClass(kafka.BrokerClass),
+		broker.WithBrokerClass(broker.EnvCfg.BrokerClass),
 		broker.WithConfig(single_partition_config.ConfigMapName),
 	))
 	f.Setup("broker is ready", broker.IsReady(brokerName))
@@ -133,7 +132,7 @@ func MultiplePartitionOrderedDelivery() *feature.Feature {
 	f.Setup("install multiple partition configuration", multiple_partition_config.Install)
 	f.Setup("install broker", broker.Install(
 		brokerName,
-		broker.WithBrokerClass(kafka.BrokerClass),
+		broker.WithBrokerClass(broker.EnvCfg.BrokerClass),
 		broker.WithConfig(multiple_partition_config.ConfigMapName),
 	))
 	f.Setup("broker is ready", broker.IsReady(brokerName))
diff --git a/test/e2e_new/tracing_test.go b/test/e2e_new/tracing_test.go
index 95b921629..dd7242c72 100644
--- a/test/e2e_new/tracing_test.go
+++ b/test/e2e_new/tracing_test.go
@@ -20,11 +20,14 @@
 package e2e_new
 
 import (
+	"context"
 	"fmt"
 	"testing"
 	"time"
 
 	cetest "github.com/cloudevents/sdk-go/v2/test"
+	"github.com/openzipkin/zipkin-go/model"
+	tracinghelper "knative.dev/eventing/test/conformance/helpers/tracing"
 	"knative.dev/eventing/test/rekt/resources/broker"
 	"knative.dev/eventing/test/rekt/resources/trigger"
 	"knative.dev/pkg/system"
@@ -37,7 +40,7 @@ import (
 
 	. "knative.dev/reconciler-test/pkg/eventshub/assert"
 
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
+	"knative.dev/eventing-kafka-broker/test/pkg/tracing"
 )
 
 func TestTracingHeaders(t *testing.T) {
@@ -48,15 +51,16 @@ func TestTracingHeaders(t *testing.T) {
 		knative.WithLoggingConfig,
 		knative.WithTracingConfig,
 		k8s.WithEventListener,
+		tracing.WithZipkin,
 		environment.Managed(t),
 	)
 
-	env.Test(ctx, t, TracingHeadersUsingOrderedDelivery())
+	env.Test(ctx, t, TracingHeadersUsingOrderedDeliveryWithTraceExported())
 	env.Test(ctx, t, TracingHeadersUsingUnorderedDelivery())
 	env.Test(ctx, t, TracingHeadersUsingUnorderedDeliveryWithMultipleTriggers())
 }
 
-func TracingHeadersUsingOrderedDelivery() *feature.Feature {
+func TracingHeadersUsingOrderedDeliveryWithTraceExported() *feature.Feature {
 	f := feature.NewFeature()
 
 	sourceName := feature.MakeRandomK8sName("source")
@@ -66,10 +70,11 @@ func TracingHeadersUsingOrderedDelivery() *feature.Feature {
 
 	ev := cetest.FullEvent()
 	ev.SetID("full-event-ordered")
+	ev.SetSource(sourceName)
 
 	f.Setup("install broker", broker.Install(
 		brokerName,
-		broker.WithBrokerClass(kafka.BrokerClass),
+		broker.WithBrokerClass(broker.EnvCfg.BrokerClass),
 	))
 	f.Setup("broker is ready", broker.IsReady(brokerName))
 	f.Setup("broker is addressable", broker.IsAddressable(brokerName))
@@ -91,17 +96,102 @@ func TracingHeadersUsingOrderedDelivery() *feature.Feature {
 		eventshub.StartSenderToResource(broker.GVR(), brokerName),
 		eventshub.InputEvent(ev),
 		eventshub.AddTracing,
+		// Send at least two events to workaround https://github.com/knative/pkg/issues/2475.
+		// There's some time needed for exporting the trace to Zipkin. Sending two events with
+		// some delay gives the exporter time to export the trace for the first event. The sender
+		// is shutdown immediately after sending the last event so the trace for the last
+		// event will probably not be exported.
+		eventshub.SendMultipleEvents(2, 3*time.Second),
 	))
 
 	f.Assert("received event has traceparent header",
 		OnStore(sinkName).
 			Match(MatchKind(EventReceived), hasTraceparentHeader).
-			Exact(1),
+			AtLeast(1),
 	)
 
+	f.Assert("event trace exported", brokerHasMatchingTraceTree(sourceName, sinkName, brokerName, ev.ID()))
+
 	return f
 }
 
+func brokerHasMatchingTraceTree(sourceName, sinkName, brokerName, eventID string) func(ctx context.Context, t feature.T) {
+	return func(ctx context.Context, t feature.T) {
+		testNS := environment.FromContext(ctx).Namespace()
+		systemNS := knative.KnativeNamespaceFromContext(ctx)
+		expectedTree := tracinghelper.TestSpanTree{
+			Note: "1. Sender pod sends event to the Broker Ingress",
+			Span: tracinghelper.MatchHTTPSpanNoReply(
+				model.Client,
+				tracinghelper.WithHTTPURL(
+					fmt.Sprintf("kafka-broker-ingress.%s.svc", systemNS),
+					fmt.Sprintf("/%s/%s", testNS, brokerName),
+				),
+				tracinghelper.WithLocalEndpointServiceName(sourceName),
+			),
+			Children: []tracinghelper.TestSpanTree{
+				{
+					Note: "2. Kafka Broker Receiver getting the message",
+					Span: tracinghelper.MatchHTTPSpanNoReply(
+						model.Server,
+						tracinghelper.WithLocalEndpointServiceName("kafka-broker-receiver"),
+						tracing.WithMessageIDSource(eventID, sourceName),
+					),
+					Children: []tracinghelper.TestSpanTree{
+						{
+							Note: "3. Kafka Broker Receiver storing message to Kafka",
+							Span: tracinghelper.MatchSpan(
+								model.Producer,
+								tracinghelper.WithLocalEndpointServiceName("kafka-broker-receiver"),
+							),
+							Children: []tracinghelper.TestSpanTree{
+								{
+									Note: "4. Kafka Broker Dispatcher reading message from Kafka",
+									Span: tracinghelper.MatchSpan(
+										model.Consumer,
+										tracinghelper.WithLocalEndpointServiceName("kafka-broker-dispatcher"),
+										tracing.WithMessageIDSource(eventID, sourceName),
+									),
+									Children: []tracinghelper.TestSpanTree{
+										{
+											Note: "5. Kafka Broker Dispatcher sending message to sink",
+											Span: tracinghelper.MatchHTTPSpanNoReply(
+												model.Client,
+												tracinghelper.WithHTTPURL(
+													fmt.Sprintf("%s.%s.svc", sinkName, testNS),
+													"/",
+												),
+												tracinghelper.WithLocalEndpointServiceName("kafka-broker-dispatcher"),
+											),
+											Children: []tracinghelper.TestSpanTree{
+												{
+													Note: "6. The target Pod receiving message",
+													Span: tracinghelper.MatchHTTPSpanNoReply(
+														model.Server,
+														tracinghelper.WithHTTPHostAndPath(
+															fmt.Sprintf("%s.%s.svc", sinkName, testNS),
+															"/",
+														),
+														tracinghelper.WithLocalEndpointServiceName(sinkName),
+													),
+												},
+											},
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		}
+		eventshub.StoreFromContext(ctx, sinkName).AssertAtLeast(t, 1,
+			MatchKind(EventReceived),
+			tracing.TraceTreeMatches(sourceName, eventID, expectedTree),
+		)
+	}
+}
+
 func TracingHeadersUsingUnorderedDelivery() *feature.Feature {
 	f := feature.NewFeature()
 
@@ -115,7 +205,7 @@ func TracingHeadersUsingUnorderedDelivery() *feature.Feature {
 
 	f.Setup("install broker", broker.Install(
 		brokerName,
-		broker.WithBrokerClass(kafka.BrokerClass),
+		broker.WithBrokerClass(broker.EnvCfg.BrokerClass),
 	))
 	f.Setup("broker is ready", broker.IsReady(brokerName))
 	f.Setup("broker is addressable", broker.IsAddressable(brokerName))
@@ -161,7 +251,7 @@ func TracingHeadersUsingUnorderedDeliveryWithMultipleTriggers() *feature.Feature
 
 	f.Setup("install broker", broker.Install(
 		brokerName,
-		broker.WithBrokerClass(kafka.BrokerClass),
+		broker.WithBrokerClass(broker.EnvCfg.BrokerClass),
 	))
 	f.Setup("broker is ready", broker.IsReady(brokerName))
 	f.Setup("broker is addressable", broker.IsAddressable(brokerName))
diff --git a/test/e2e_new_channel/main_test.go b/test/e2e_new_channel/main_test.go
index 8e4d6e5de..324ce6c13 100644
--- a/test/e2e_new_channel/main_test.go
+++ b/test/e2e_new_channel/main_test.go
@@ -21,6 +21,7 @@ package e2e_new_channel
 
 import (
 	"flag"
+	"log"
 	"os"
 	"testing"
 
@@ -31,6 +32,7 @@ import (
 
 	"knative.dev/pkg/injection"
 	_ "knative.dev/pkg/system/testing"
+	"knative.dev/pkg/test/zipkin"
 
 	"knative.dev/reconciler-test/pkg/environment"
 )
@@ -67,5 +69,11 @@ func TestMain(m *testing.M) {
 	global = environment.NewGlobalEnvironment(ctx)
 
 	// Run the tests.
-	os.Exit(m.Run())
+	os.Exit(func() int {
+		// Any tests may SetupZipkinTracing, it will only actually be done once. This should be the ONLY
+		// place that cleans it up. If an individual test calls this instead, then it will break other
+		// tests that need the tracing in place.
+		defer zipkin.CleanupZipkinTracingSetup(log.Printf)
+		return m.Run()
+	}())
 }
diff --git a/test/e2e_new_channel/tracing_test.go b/test/e2e_new_channel/tracing_test.go
new file mode 100644
index 000000000..c21184665
--- /dev/null
+++ b/test/e2e_new_channel/tracing_test.go
@@ -0,0 +1,178 @@
+//go:build e2e
+// +build e2e
+
+/*
+ * 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 e2e_new_channel
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	cetest "github.com/cloudevents/sdk-go/v2/test"
+	"github.com/openzipkin/zipkin-go/model"
+	tracinghelper "knative.dev/eventing/test/conformance/helpers/tracing"
+	"knative.dev/eventing/test/rekt/resources/channel_impl"
+	"knative.dev/eventing/test/rekt/resources/subscription"
+	"knative.dev/eventing/test/rekt/resources/svc"
+	"knative.dev/pkg/system"
+	_ "knative.dev/pkg/system/testing"
+	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/eventshub"
+	. "knative.dev/reconciler-test/pkg/eventshub/assert"
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/k8s"
+	"knative.dev/reconciler-test/pkg/knative"
+
+	"knative.dev/eventing-kafka-broker/test/pkg/tracing"
+)
+
+func TestChannelTracing(t *testing.T) {
+	t.Parallel()
+
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+		tracing.WithZipkin,
+		environment.Managed(t),
+	)
+
+	env.Test(ctx, t, eventWithTraceExported())
+}
+
+func eventWithTraceExported() *feature.Feature {
+	f := feature.NewFeature()
+
+	sourceName := feature.MakeRandomK8sName("source")
+	sinkName := feature.MakeRandomK8sName("sink")
+	channelName := feature.MakeRandomK8sName("channel")
+	subName := feature.MakeRandomK8sName("subscription")
+
+	f.Setup("install sink", eventshub.Install(
+		sinkName,
+		eventshub.StartReceiver,
+	))
+	f.Setup("install channel", channel_impl.Install(channelName))
+	f.Setup("install subscription", subscription.Install(subName,
+		subscription.WithChannel(channel_impl.AsRef(channelName)),
+		subscription.WithSubscriber(svc.AsRef(sinkName), ""),
+	))
+
+	f.Setup("subscription is ready", subscription.IsReady(subName))
+	f.Setup("channel is ready", channel_impl.IsReady(channelName))
+
+	ev := cetest.FullEvent()
+	ev.SetID("full-event-kafka-channel")
+	ev.SetSource(sourceName)
+
+	f.Requirement("install source", eventshub.Install(
+		sourceName,
+		eventshub.StartSenderToResource(channel_impl.GVR(), channelName),
+		eventshub.InputEvent(ev),
+		eventshub.AddTracing,
+		// Send at least two events to workaround https://github.com/knative/pkg/issues/2475.
+		// There's some time needed for exporting the trace to Zipkin. Sending two events with
+		// some delay gives the exporter time to export the trace for the first event. The sender
+		// is shutdown immediately after sending the last event so the trace for the last
+		// event will probably not be exported.
+		eventshub.SendMultipleEvents(2, 3*time.Second),
+	))
+
+	f.Assert("event trace exported", channelHasMatchingTraceTree(sourceName, sinkName, channelName, ev.ID()))
+
+	return f
+}
+
+func channelHasMatchingTraceTree(sourceName, sinkName, channelName, eventID string) func(ctx context.Context, t feature.T) {
+	return func(ctx context.Context, t feature.T) {
+		testNS := environment.FromContext(ctx).Namespace()
+		expectedTree := tracinghelper.TestSpanTree{
+			Note: "1. Sender pod sends event to the Kafka Channel",
+			Span: tracinghelper.MatchHTTPSpanNoReply(
+				model.Client,
+				tracinghelper.WithHTTPURL(
+					fmt.Sprintf("%s-kn-channel.%s.svc", channelName, testNS),
+					"",
+				),
+				tracinghelper.WithLocalEndpointServiceName(sourceName),
+			),
+			Children: []tracinghelper.TestSpanTree{
+				{
+					Note: "2. Kafka Channel Receiver getting the message",
+					Span: tracinghelper.MatchHTTPSpanNoReply(
+						model.Server,
+						tracinghelper.WithLocalEndpointServiceName("kafka-channel-receiver"),
+						tracing.WithMessageIDSource(eventID, sourceName),
+					),
+					Children: []tracinghelper.TestSpanTree{
+						{
+							Note: "3. Kafka Channel Receiver storing message to Kafka",
+							Span: tracinghelper.MatchSpan(
+								model.Producer,
+								tracinghelper.WithLocalEndpointServiceName("kafka-channel-receiver"),
+							),
+							Children: []tracinghelper.TestSpanTree{
+								{
+									Note: "4. Kafka Channel Dispatcher reading message from Kafka",
+									Span: tracinghelper.MatchSpan(
+										model.Consumer,
+										tracinghelper.WithLocalEndpointServiceName("kafka-channel-dispatcher"),
+										tracing.WithMessageIDSource(eventID, sourceName),
+									),
+									Children: []tracinghelper.TestSpanTree{
+										{
+											Note: "5. Kafka Channel Dispatcher sending message to sink",
+											Span: tracinghelper.MatchHTTPSpanNoReply(
+												model.Client,
+												tracinghelper.WithHTTPURL(
+													fmt.Sprintf("%s.%s.svc", sinkName, testNS),
+													"/",
+												),
+												tracinghelper.WithLocalEndpointServiceName("kafka-channel-dispatcher"),
+											),
+											Children: []tracinghelper.TestSpanTree{
+												{
+													Note: "6. The target Pod receiving message",
+													Span: tracinghelper.MatchHTTPSpanNoReply(
+														model.Server,
+														tracinghelper.WithHTTPHostAndPath(
+															fmt.Sprintf("%s.%s.svc", sinkName, testNS),
+															"/",
+														),
+														tracinghelper.WithLocalEndpointServiceName(sinkName),
+													),
+												},
+											},
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		}
+		eventshub.StoreFromContext(ctx, sinkName).AssertAtLeast(t, 1,
+			MatchKind(EventReceived),
+			tracing.TraceTreeMatches(sourceName, eventID, expectedTree),
+		)
+	}
+}
diff --git a/test/e2e/kafka_sink.go b/test/e2e_sink/kafka_sink.go
similarity index 99%
rename from test/e2e/kafka_sink.go
rename to test/e2e_sink/kafka_sink.go
index 3760f60f8..5c607f68b 100644
--- a/test/e2e/kafka_sink.go
+++ b/test/e2e_sink/kafka_sink.go
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package e2e
+package e2e_sink
 
 import (
 	"context"
diff --git a/test/e2e/kafka_sink_test.go b/test/e2e_sink/kafka_sink_test.go
similarity index 99%
rename from test/e2e/kafka_sink_test.go
rename to test/e2e_sink/kafka_sink_test.go
index 735fc6054..0284f9090 100644
--- a/test/e2e/kafka_sink_test.go
+++ b/test/e2e_sink/kafka_sink_test.go
@@ -17,7 +17,7 @@
  * limitations under the License.
  */
 
-package e2e
+package e2e_sink
 
 import (
 	"testing"
diff --git a/test/e2e_sink/main_test.go b/test/e2e_sink/main_test.go
new file mode 100644
index 000000000..8b9727265
--- /dev/null
+++ b/test/e2e_sink/main_test.go
@@ -0,0 +1,36 @@
+//go:build e2e
+// +build e2e
+
+/*
+ * Copyright 2020 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package e2e_sink
+
+import (
+	"os"
+	"testing"
+
+	testlib "knative.dev/eventing/test/lib"
+	"knative.dev/pkg/system"
+)
+
+func TestMain(m *testing.M) {
+	os.Exit(func() int {
+		defer testlib.ExportLogs(testlib.SystemLogsDir, system.Namespace())
+
+		return m.Run()
+	}())
+}
diff --git a/test/e2e/kafka_source_test.go b/test/e2e_source/kafka_source_test.go
similarity index 98%
rename from test/e2e/kafka_source_test.go
rename to test/e2e_source/kafka_source_test.go
index 06e5b9518..99f29b8fa 100644
--- a/test/e2e/kafka_source_test.go
+++ b/test/e2e_source/kafka_source_test.go
@@ -17,7 +17,7 @@
  * limitations under the License.
  */
 
-package e2e
+package e2e_source
 
 import (
 	"testing"
diff --git a/test/e2e_source/main_test.go b/test/e2e_source/main_test.go
new file mode 100644
index 000000000..e728de12f
--- /dev/null
+++ b/test/e2e_source/main_test.go
@@ -0,0 +1,36 @@
+//go:build e2e
+// +build e2e
+
+/*
+ * Copyright 2020 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package e2e_source
+
+import (
+	"os"
+	"testing"
+
+	testlib "knative.dev/eventing/test/lib"
+	"knative.dev/pkg/system"
+)
+
+func TestMain(m *testing.M) {
+	os.Exit(func() int {
+		defer testlib.ExportLogs(testlib.SystemLogsDir, system.Namespace())
+
+		return m.Run()
+	}())
+}
diff --git a/test/pkg/broker/broker.go b/test/pkg/broker/broker.go
index 6472d9d90..8248c92bc 100644
--- a/test/pkg/broker/broker.go
+++ b/test/pkg/broker/broker.go
@@ -19,6 +19,7 @@ package broker
 import (
 	"context"
 	"fmt"
+	"os"
 	"strings"
 
 	corev1 "k8s.io/api/core/v1"
@@ -32,7 +33,29 @@ import (
 	testingpkg "knative.dev/eventing-kafka-broker/test/pkg/testing"
 )
 
+const BrokerClassEnvVarKey = "BROKER_CLASS"
+
+func GetKafkaClassFromEnv() (string, error) {
+	val := ""
+	exists := false
+
+	if val, exists = os.LookupEnv(BrokerClassEnvVarKey); !exists {
+		return "", fmt.Errorf("unable to determine KafkaBroker class. Specify '%s' env var", BrokerClassEnvVarKey)
+	}
+
+	if val != kafka.BrokerClass && val != kafka.NamespacedBrokerClass {
+		return "", fmt.Errorf("KafkaBroker class '%s' is unknown. Specify '%s' env var", val, BrokerClassEnvVarKey)
+	}
+
+	return val, nil
+}
+
 func Creator(client *eventingtestlib.Client, version string) string {
+	class, err := GetKafkaClassFromEnv()
+	if err != nil {
+		panic(fmt.Sprintf("error getting KafkaBroker class from env '%v'", err))
+	}
+
 	name := "broker"
 
 	version = strings.ToLower(version)
@@ -59,7 +82,7 @@ func Creator(client *eventingtestlib.Client, version string) string {
 		}
 		client.CreateBrokerOrFail(
 			name,
-			resources.WithBrokerClassForBroker(kafka.BrokerClass),
+			resources.WithBrokerClassForBroker(class),
 			resources.WithConfigForBroker(&duckv1.KReference{
 				Kind:       "ConfigMap",
 				Namespace:  namespace,
diff --git a/test/pkg/tracing/tracing.go b/test/pkg/tracing/tracing.go
new file mode 100644
index 000000000..40934c299
--- /dev/null
+++ b/test/pkg/tracing/tracing.go
@@ -0,0 +1,84 @@
+/*
+ * 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 tracing
+
+import (
+	"context"
+	"fmt"
+	"regexp"
+	"time"
+
+	cetest "github.com/cloudevents/sdk-go/v2/test"
+	"github.com/openzipkin/zipkin-go/model"
+	"go.opentelemetry.io/otel/propagation"
+	"go.opentelemetry.io/otel/trace"
+	tracinghelper "knative.dev/eventing/test/conformance/helpers/tracing"
+	pkgzipkin "knative.dev/pkg/test/zipkin"
+	"knative.dev/reconciler-test/pkg/eventshub"
+)
+
+func TraceTreeMatches(sourceName, eventID string, expectedTraceTree tracinghelper.TestSpanTree) eventshub.EventInfoMatcher {
+	return func(info eventshub.EventInfo) error {
+		if err := cetest.AllOf(
+			cetest.HasSource(sourceName),
+			cetest.HasId(eventID))(*info.Event); err != nil {
+			return err
+		}
+		traceID, err := getTraceIDHeader(info)
+		if err != nil {
+			return err
+		}
+		trace, err := pkgzipkin.JSONTracePred(traceID, 5*time.Second, func(trace []model.SpanModel) bool {
+			tree, err := tracinghelper.GetTraceTree(trace)
+			if err != nil {
+				return false
+			}
+			return len(expectedTraceTree.MatchesSubtree(nil, tree)) > 0
+		})
+		if err != nil {
+			tree, err := tracinghelper.GetTraceTree(trace)
+			if err != nil {
+				return err
+			}
+			if len(expectedTraceTree.MatchesSubtree(nil, tree)) == 0 {
+				return fmt.Errorf("no matching subtree. want: %v got: %v", expectedTraceTree, tree)
+			}
+		}
+		return nil
+	}
+}
+
+// getTraceIDHeader gets the TraceID from the passed event. It returns an error
+// if trace id is not present in that message.
+func getTraceIDHeader(info eventshub.EventInfo) (string, error) {
+	if info.HTTPHeaders != nil {
+		sc := trace.SpanContextFromContext(propagation.TraceContext{}.Extract(context.TODO(), propagation.HeaderCarrier(info.HTTPHeaders)))
+		if sc.HasTraceID() {
+			return sc.TraceID().String(), nil
+		}
+	}
+	return "", fmt.Errorf("no traceid in info: (%v)", info)
+}
+
+func WithMessageIDSource(eventID, sourceName string) tracinghelper.SpanMatcherOption {
+	return func(m *tracinghelper.SpanMatcher) {
+		m.Tags = map[string]*regexp.Regexp{
+			"messaging.message_id":     regexp.MustCompile("^" + eventID + "$"),
+			"messaging.message_source": regexp.MustCompile("^" + sourceName + "$"),
+		}
+	}
+}
diff --git a/test/pkg/tracing/zipkin.go b/test/pkg/tracing/zipkin.go
new file mode 100644
index 000000000..5bf507e05
--- /dev/null
+++ b/test/pkg/tracing/zipkin.go
@@ -0,0 +1,36 @@
+/*
+ * 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 tracing
+
+import (
+	"context"
+
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	"knative.dev/pkg/logging"
+	"knative.dev/pkg/test/zipkin"
+	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/knative"
+)
+
+func WithZipkin(ctx context.Context, env environment.Environment) (context.Context, error) {
+	err := zipkin.SetupZipkinTracingFromConfigTracing(ctx,
+		kubeclient.Get(ctx),
+		logging.FromContext(ctx).Infof,
+		knative.KnativeNamespaceFromContext(ctx))
+	zipkin.ZipkinTracingEnabled = true
+	return ctx, err
+}
diff --git a/test/reconciler-tests.sh b/test/reconciler-tests.sh
index be8205f0b..c44fefa61 100755
--- a/test/reconciler-tests.sh
+++ b/test/reconciler-tests.sh
@@ -27,6 +27,8 @@ if [ "${EVENTING_KAFKA_BROKER_CHANNEL_AUTH_SCENARIO:-""}" != "" ]; then
   success
 fi
 
+export BROKER_CLASS="Kafka"
+
 go_test_e2e -timeout=1h ./test/e2e_new/... || fail_test "E2E (new) suite failed"
 
 go_test_e2e -tags=e2e,cloudevents -timeout=1h ./test/e2e_new_channel/... || fail_test "E2E (new - KafkaChannel) suite failed"
diff --git a/test/upgrade-tests.sh b/test/upgrade-tests.sh
index 4fef2a0c8..310c6cdb2 100755
--- a/test/upgrade-tests.sh
+++ b/test/upgrade-tests.sh
@@ -42,6 +42,8 @@ set -Eeuo pipefail
 TIMEOUT=${TIMEOUT:-100m}
 GO_TEST_VERBOSITY="${GO_TEST_VERBOSITY:-standard-verbose}"
 
+export BROKER_CLASS="Kafka"
+
 EVENTING_KAFKA_BROKER_UPGRADE_TESTS_FINISHEDSLEEP="5m" \
 go_test_e2e -v \
   -tags=upgrade \
diff --git a/test/upgrade/postdowngrade.go b/test/upgrade/postdowngrade.go
index 997f7edf5..c24d666a5 100644
--- a/test/upgrade/postdowngrade.go
+++ b/test/upgrade/postdowngrade.go
@@ -21,7 +21,7 @@ import (
 	pkgupgrade "knative.dev/pkg/test/upgrade"
 
 	eventing "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
-	"knative.dev/eventing-kafka-broker/test/e2e"
+	"knative.dev/eventing-kafka-broker/test/e2e_sink"
 )
 
 // BrokerPostDowngradeTest tests channel basic channel operations after
@@ -41,6 +41,6 @@ func ChannelPostDowngradeTest() pkgupgrade.Operation {
 // SinkPostDowngradeTest tests sink basic operations after downgrade.
 func SinkPostDowngradeTest() pkgupgrade.Operation {
 	return pkgupgrade.NewOperation("SinkPostDowngradeTest", func(c pkgupgrade.Context) {
-		e2e.RunTestKafkaSink(c.T, eventing.ModeBinary, nil)
+		e2e_sink.RunTestKafkaSink(c.T, eventing.ModeBinary, nil)
 	})
 }
diff --git a/test/upgrade/postupgrade.go b/test/upgrade/postupgrade.go
index bc58ab1ef..ae30bf46d 100644
--- a/test/upgrade/postupgrade.go
+++ b/test/upgrade/postupgrade.go
@@ -31,7 +31,7 @@ import (
 	pkgupgrade "knative.dev/pkg/test/upgrade"
 
 	eventing "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
-	"knative.dev/eventing-kafka-broker/test/e2e"
+	"knative.dev/eventing-kafka-broker/test/e2e_sink"
 	eventingkafkaupgrade "knative.dev/eventing-kafka/test/upgrade"
 )
 
@@ -61,7 +61,7 @@ func SinkPostUpgradeTest() pkgupgrade.Operation {
 			verifyPostInstall(t)
 		})
 		c.T.Run("tests", func(t *testing.T) {
-			e2e.RunTestKafkaSink(t, eventing.ModeBinary, nil)
+			e2e_sink.RunTestKafkaSink(t, eventing.ModeBinary, nil)
 		})
 	})
 }
diff --git a/test/upgrade/preupgrade.go b/test/upgrade/preupgrade.go
index 0ed44eb9b..1fabfb5c6 100644
--- a/test/upgrade/preupgrade.go
+++ b/test/upgrade/preupgrade.go
@@ -20,8 +20,7 @@ import (
 	pkgupgrade "knative.dev/pkg/test/upgrade"
 
 	eventing "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
-	"knative.dev/eventing-kafka-broker/test/e2e"
-
+	"knative.dev/eventing-kafka-broker/test/e2e_sink"
 	eventingkafkaupgrade "knative.dev/eventing-kafka/test/upgrade"
 )
 
@@ -40,6 +39,6 @@ func ChannelPreUpgradeTest() pkgupgrade.Operation {
 // SinkPreUpgradeTest tests sink basic operations pre upgrade.
 func SinkPreUpgradeTest() pkgupgrade.Operation {
 	return pkgupgrade.NewOperation("SinkPreUpgradeTest", func(c pkgupgrade.Context) {
-		e2e.RunTestKafkaSink(c.T, eventing.ModeBinary, nil)
+		e2e_sink.RunTestKafkaSink(c.T, eventing.ModeBinary, nil)
 	})
 }
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/svc/service.go b/vendor/knative.dev/eventing/test/rekt/resources/svc/service.go
new file mode 100644
index 000000000..f453045d8
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/rekt/resources/svc/service.go
@@ -0,0 +1,88 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package svc
+
+import (
+	"context"
+	"embed"
+
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/tracker"
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/k8s"
+	"knative.dev/reconciler-test/pkg/manifest"
+	"knative.dev/reconciler-test/resources/svc"
+)
+
+//go:embed *.yaml
+var yaml embed.FS
+
+// Deprecated, use reconciler-test/resources/svc
+func Gvr() schema.GroupVersionResource {
+	return schema.GroupVersionResource{Group: "", Version: "v1", Resource: "services"}
+}
+
+// Install will create a Service resource mapping :80 to :8080 on the provided
+// selector for pods.
+// Deprecated, use reconciler-test/resources/svc
+func Install(name, selectorKey, selectorValue string) feature.StepFn {
+	cfg := map[string]interface{}{
+		"name":          name,
+		"selectorKey":   selectorKey,
+		"selectorValue": selectorValue,
+	}
+
+	return func(ctx context.Context, t feature.T) {
+		if _, err := manifest.InstallYamlFS(ctx, yaml, cfg); err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
+// AsRef returns a KRef for a Service without namespace.
+// Deprecated, use reconciler-test/resources/svc
+func AsRef(name string) *duckv1.KReference {
+	return &duckv1.KReference{
+		Kind:       "Service",
+		Name:       name,
+		APIVersion: "v1",
+	}
+}
+
+// Deprecated, use reconciler-test/resources/svc
+func AsTrackerReference(name string) *tracker.Reference {
+	return &tracker.Reference{
+		Kind:       "Service",
+		Name:       name,
+		APIVersion: "v1",
+	}
+}
+
+// Deprecated, use reconciler-test/resources/svc
+func AsDestinationRef(name string) *duckv1.Destination {
+	return &duckv1.Destination{
+		Ref: AsRef(name),
+	}
+}
+
+// Address
+// Deprecated, use reconciler-test/resources/svc
+func Address(ctx context.Context, name string) (*apis.URL, error) {
+	return k8s.Address(ctx, svc.GVR(), name)
+}
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/svc/service.yaml b/vendor/knative.dev/eventing/test/rekt/resources/svc/service.yaml
new file mode 100644
index 000000000..e61cd1bfb
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/rekt/resources/svc/service.yaml
@@ -0,0 +1,26 @@
+# Copyright 2020 The Knative Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+apiVersion: v1
+kind: Service
+metadata:
+  name: {{ .name }}
+  namespace: {{ .namespace }}
+spec:
+  selector:
+    {{ .selectorKey }}: {{ .selectorValue }}
+  ports:
+    - protocol: TCP
+      port: 80
+      targetPort: 8080