Repository: knative-sandbox/eventing-kafka-broker
Pull request #1723 (2022-01-11 07:36:57)

Support replies in KafkaChannel
Fixes #

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Support replies in KafkaChannel
- Create new `ResponseHandler` that does it
- Rename `KafkaResponseHandler` to `ResponseToKafkaTopicHandler` because the original name meant "handle the response from Kafka" to me
- Enable e2e tests that use sink replies

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- :gift: Add new feature
- :bug: Fix bug
- :broom: Update or clean up current behavior
- :wastebasket: Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
:page_facing_up: If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note

```

**Docs**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1723#" title="Author self-approved">aliok</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)~~ [aliok]

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

> :exclamation: Current head d906427 differs from pull request most recent head e7b84bd. Consider uploading reports for the commit e7b84bd to get more accurate results
[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1723/graphs/tree.svg?width=650&height=150&src=pr&token=xyPIgGOiwy&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1723?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@             Coverage Diff              @@
##               main    #1723      +/-   ##
============================================
- Coverage     71.86%   65.92%   -5.95%     
============================================
  Files           122       56      -66     
  Lines          4521     2644    -1877     
  Branches        171        0     -171     
============================================
- Hits           3249     1743    -1506     
+ Misses         1008      719     -289     
+ Partials        264      182      -82     
```

| 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/1723?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [control-plane/pkg/reconciler/channel/channel.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1723/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jaGFubmVsL2NoYW5uZWwuZ28=) | `64.85% <100.00%> (+0.51%)` | :arrow_up: |
| [...apis/internals/kafka/eventing/v1alpha1/register.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1723/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9pbnRlcm5hbHMva2Fma2EvZXZlbnRpbmcvdjFhbHBoYTEvcmVnaXN0ZXIuZ28=) | `0.00% <0.00%> (ø)` | |
| [...nternals/kafka/eventing/v1alpha1/consumer\_types.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1723/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9pbnRlcm5hbHMva2Fma2EvZXZlbnRpbmcvdjFhbHBoYTEvY29uc3VtZXJfdHlwZXMuZ28=) | `0.00% <0.00%> (ø)` | |
| [...ls/kafka/eventing/v1alpha1/consumer\_group\_types.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1723/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9pbnRlcm5hbHMva2Fma2EvZXZlbnRpbmcvdjFhbHBoYTEvY29uc3VtZXJfZ3JvdXBfdHlwZXMuZ28=) | `0.00% <0.00%> (ø)` | |
| [...afka/eventing/v1alpha1/consumer\_group\_lifecycle.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1723/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9pbnRlcm5hbHMva2Fma2EvZXZlbnRpbmcvdjFhbHBoYTEvY29uc3VtZXJfZ3JvdXBfbGlmZWN5Y2xlLmdv) | `0.00% <0.00%> (ø)` | |
| [...r/dispatcher/main/ConsumerVerticleFactoryImpl.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1723/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvbWFpbi9Db25zdW1lclZlcnRpY2xlRmFjdG9yeUltcGwuamF2YQ==) | | |
| [...e/eventing/kafka/broker/core/file/FileWatcher.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1723/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvZmlsZS9GaWxlV2F0Y2hlci5qYXZh) | | |
| [...atcher/impl/consumer/KafkaConsumerRecordUtils.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1723/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9LYWZrYUNvbnN1bWVyUmVjb3JkVXRpbHMuamF2YQ==) | | |
| [...venting/kafka/broker/dispatcher/DeliveryOrder.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1723/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvRGVsaXZlcnlPcmRlci5qYXZh) | | |
| [...fka/broker/core/security/CredentialsValidator.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1723/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvc2VjdXJpdHkvQ3JlZGVudGlhbHNWYWxpZGF0b3IuamF2YQ==) | | |
| ... and [63 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1723/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1723?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1723?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [5fcac54...e7b84bd](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1723?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).

knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/channel/channel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1723/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1480806492029325312/artifacts/line-cov.html#file0) | 78.0% | 78.3% | 0.2

knative-prow-robot: @aliok: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-unit-tests | e7b84bdbc66ae16721b4585401f4f49bd7bbdc82 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1723/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1480806490871697408) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-unit-tests | e7b84bdbc66ae16721b4585401f4f49bd7bbdc82 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1723/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1480806490871697408) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-upgrade-tests | e7b84bdbc66ae16721b4585401f4f49bd7bbdc82 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1723/pull-knative-sandbox-eventing-kafka-broker-upgrade-tests/1480806490922029056) | true | `/test pull-knative-sandbox-eventing-kafka-broker-upgrade-tests`

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

<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-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/channel/channel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1723/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1480814214841896960/artifacts/line-cov.html#file0) | 78.0% | 78.3% | 0.2

knative-prow-robot: @aliok: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-upgrade-tests | e7b84bdbc66ae16721b4585401f4f49bd7bbdc82 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1723/pull-knative-sandbox-eventing-kafka-broker-upgrade-tests/1480806490922029056) | true | `/test pull-knative-sandbox-eventing-kafka-broker-upgrade-tests`
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 8e68865e594abdaa1c235587817dc86e8e82bcff | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1723/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1480814214720262144) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

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

<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-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/channel/channel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1723/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1480819135746150400/artifacts/line-cov.html#file0) | 78.0% | 78.3% | 0.2

knative-prow-robot: @aliok: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-upgrade-tests | e7b84bdbc66ae16721b4585401f4f49bd7bbdc82 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1723/pull-knative-sandbox-eventing-kafka-broker-upgrade-tests/1480806490922029056) | true | `/test pull-knative-sandbox-eventing-kafka-broker-upgrade-tests`
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 994cb1184bf59e6ddc07e55fb1158b4b13037c83 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1723/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1480819135603544064) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 994cb1184bf59e6ddc07e55fb1158b4b13037c83 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1723/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1480819135603544064) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-upgrade-tests | 994cb1184bf59e6ddc07e55fb1158b4b13037c83 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1723/pull-knative-sandbox-eventing-kafka-broker-upgrade-tests/1480819135679041536) | true | `/test pull-knative-sandbox-eventing-kafka-broker-upgrade-tests`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 994cb1184bf59e6ddc07e55fb1158b4b13037c83 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1723/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1480819135603544064) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-upgrade-tests | 994cb1184bf59e6ddc07e55fb1158b4b13037c83 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1723/pull-knative-sandbox-eventing-kafka-broker-upgrade-tests/1480819135679041536) | true | `/test pull-knative-sandbox-eventing-kafka-broker-upgrade-tests`
pull-knative-sandbox-eventing-kafka-broker-reconciler-tests | 994cb1184bf59e6ddc07e55fb1158b4b13037c83 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1723/pull-knative-sandbox-eventing-kafka-broker-reconciler-tests/1480819135712595968) | true | `/test pull-knative-sandbox-eventing-kafka-broker-reconciler-tests`

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

<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 -->
aliok: /test pull-knative-sandbox-eventing-kafka-broker-reconciler-tests
/test pull-knative-sandbox-eventing-kafka-broker-upgrade-tests
/test pull-knative-sandbox-eventing-kafka-broker-unit-tests
knative-prow-robot: @aliok: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-upgrade-tests | 994cb1184bf59e6ddc07e55fb1158b4b13037c83 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1723/pull-knative-sandbox-eventing-kafka-broker-upgrade-tests/1480819135679041536) | true | `/test pull-knative-sandbox-eventing-kafka-broker-upgrade-tests`
pull-knative-sandbox-eventing-kafka-broker-reconciler-tests | 994cb1184bf59e6ddc07e55fb1158b4b13037c83 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1723/pull-knative-sandbox-eventing-kafka-broker-reconciler-tests/1480819135712595968) | true | `/test pull-knative-sandbox-eventing-kafka-broker-reconciler-tests`
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 994cb1184bf59e6ddc07e55fb1158b4b13037c83 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1723/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1480831556212756480) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-upgrade-tests | 994cb1184bf59e6ddc07e55fb1158b4b13037c83 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1723/pull-knative-sandbox-eventing-kafka-broker-upgrade-tests/1480819135679041536) | true | `/test pull-knative-sandbox-eventing-kafka-broker-upgrade-tests`
pull-knative-sandbox-eventing-kafka-broker-reconciler-tests | 994cb1184bf59e6ddc07e55fb1158b4b13037c83 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1723/pull-knative-sandbox-eventing-kafka-broker-reconciler-tests/1480819135712595968) | true | `/test pull-knative-sandbox-eventing-kafka-broker-reconciler-tests`
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 994cb1184bf59e6ddc07e55fb1158b4b13037c83 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1723/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1480831556212756480) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 994cb1184bf59e6ddc07e55fb1158b4b13037c83 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1723/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1480819135632904192) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

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

<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-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/channel/channel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1723/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1480838385823125504/artifacts/line-cov.html#file0) | 78.0% | 78.3% | 0.2

aliok: > Have you considered using composition over inheritance?

I thought about it. I prefer leaving this as is for now.
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/channel/channel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1723/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1480863895529721856/artifacts/line-cov.html#file0) | 78.0% | 78.3% | 0.2


Patch:
diff --git a/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index 428d284f5..87b097039 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -469,6 +469,13 @@ func (r *Reconciler) getSubscriberConfig(ctx context.Context, channel *messaging
 		ConsumerGroup: consumerGroup(channel, subscriber),
 		DeliveryOrder: DefaultDeliveryOrder,
 		Uid:           string(subscriber.UID),
+		ReplyStrategy: &contract.Egress_DiscardReply{},
+	}
+
+	if subscriber.ReplyURI != nil {
+		egress.ReplyStrategy = &contract.Egress_ReplyUrl{
+			ReplyUrl: subscriber.ReplyURI.String(),
+		}
 	}
 
 	subscriptionEgressConfig, err := coreconfig.EgressConfigFromDelivery(ctx, r.Resolver, channel, subscriber.Delivery, r.DefaultBackoffDelayMs)
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/KafkaResponseHandler.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/BaseResponseHandler.java
similarity index 60%
rename from data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/KafkaResponseHandler.java
rename to data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/BaseResponseHandler.java
index 1182eda85..577b82e92 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/KafkaResponseHandler.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/BaseResponseHandler.java
@@ -15,59 +15,36 @@
  */
 package dev.knative.eventing.kafka.broker.dispatcher.impl;
 
-import dev.knative.eventing.kafka.broker.core.AsyncCloseable;
-import dev.knative.eventing.kafka.broker.core.metrics.Metrics;
 import dev.knative.eventing.kafka.broker.core.tracing.TracingSpan;
 import dev.knative.eventing.kafka.broker.dispatcher.ResponseHandler;
 import io.cloudevents.CloudEvent;
 import io.cloudevents.http.vertx.VertxMessageFactory;
-import io.vertx.core.CompositeFuture;
 import io.vertx.core.Future;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.ext.web.client.HttpResponse;
-import io.vertx.kafka.client.producer.KafkaProducer;
-import io.vertx.kafka.client.producer.KafkaProducerRecord;
-import java.util.Objects;
 import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import static dev.knative.eventing.kafka.broker.core.utils.Logging.keyValue;
 
-/**
- * This class implements a {@link ResponseHandler} that will convert the sink response into a {@link CloudEvent} and push it to Kafka.
- */
-public final class KafkaResponseHandler implements ResponseHandler {
-
-  private static final Logger logger = LoggerFactory.getLogger(KafkaResponseHandler.class);
-
-  private final String topic;
-  private final KafkaProducer<String, CloudEvent> producer;
-  private final AutoCloseable producerMeterBinder;
+public abstract class BaseResponseHandler implements ResponseHandler {
 
-  /**
-   * All args constructor.
-   *
-   * @param producer Kafka producer.
-   * @param topic    topic to produce records.
-   */
-  public KafkaResponseHandler(final KafkaProducer<String, CloudEvent> producer, final String topic) {
+  protected final Logger logger;
 
-    Objects.requireNonNull(topic, "provide topic");
-    Objects.requireNonNull(producer, "provide producer");
-
-    this.topic = topic;
-    this.producer = producer;
-    this.producerMeterBinder = Metrics.register(this.producer.unwrap());
+  public BaseResponseHandler(Logger logger) {
+    this.logger = logger;
   }
 
   /**
-   * Handle the given response.
+   * The {@link BaseResponseHandler} will convert the response from the sink into a CloudEvent. Implementations should
+   * handle the CloudEvent.
    *
-   * @param response response to handle
-   * @return a succeeded or failed future.
+   * @param event CloudEvent parsed from the sink response
+   * @return future for handling
    */
+  protected abstract Future<Void> doHandleEvent(CloudEvent event);
+
   @Override
-  public Future<Void> handle(final HttpResponse<Buffer> response) {
+  public Future<Void> handle(HttpResponse<Buffer> response) {
     CloudEvent event;
 
     try {
@@ -98,9 +75,7 @@ public final class KafkaResponseHandler implements ResponseHandler {
 
     TracingSpan.decorateCurrentWithEvent(event);
 
-    return producer
-      .send(KafkaProducerRecord.create(topic, event))
-      .mapEmpty();
+    return this.doHandleEvent(event);
   }
 
   private static boolean maybeIsNotEvent(final HttpResponse<Buffer> response) {
@@ -108,12 +83,4 @@ public final class KafkaResponseHandler implements ResponseHandler {
     // are valid Cloud Events.
     return response == null || response.body() == null || response.body().length() <= 0;
   }
-
-  @Override
-  public Future<Void> close() {
-    return CompositeFuture.all(
-      this.producer.close(),
-      AsyncCloseable.wrapAutoCloseable(producerMeterBinder).close()
-    ).mapEmpty();
-  }
 }
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/ResponseToHttpEndpointHandler.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/ResponseToHttpEndpointHandler.java
new file mode 100644
index 000000000..3045cee0e
--- /dev/null
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/ResponseToHttpEndpointHandler.java
@@ -0,0 +1,45 @@
+/*
+ * Copyright © 2018 Knative Authors (knative-dev@googlegroups.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package dev.knative.eventing.kafka.broker.dispatcher.impl;
+
+import dev.knative.eventing.kafka.broker.dispatcher.CloudEventSender;
+import dev.knative.eventing.kafka.broker.dispatcher.ResponseHandler;
+import io.cloudevents.CloudEvent;
+import io.vertx.core.Future;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This class implements a {@link ResponseHandler} that will convert the sink response into a {@link CloudEvent} and post it to a URL.
+ */
+public class ResponseToHttpEndpointHandler extends BaseResponseHandler implements ResponseHandler {
+
+  private final CloudEventSender cloudEventSender;
+
+  public ResponseToHttpEndpointHandler(CloudEventSender cloudEventSender) {
+    super(LoggerFactory.getLogger(ResponseToKafkaTopicHandler.class));
+    this.cloudEventSender = cloudEventSender;
+  }
+
+  @Override
+  protected Future<Void> doHandleEvent(CloudEvent event) {
+    return cloudEventSender.send(event).mapEmpty();
+  }
+
+  @Override
+  public Future<Void> close() {
+    return this.cloudEventSender.close().mapEmpty();
+  }
+}
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/ResponseToKafkaTopicHandler.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/ResponseToKafkaTopicHandler.java
new file mode 100644
index 000000000..5fd117209
--- /dev/null
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/ResponseToKafkaTopicHandler.java
@@ -0,0 +1,70 @@
+/*
+ * Copyright © 2018 Knative Authors (knative-dev@googlegroups.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package dev.knative.eventing.kafka.broker.dispatcher.impl;
+
+import dev.knative.eventing.kafka.broker.core.AsyncCloseable;
+import dev.knative.eventing.kafka.broker.core.metrics.Metrics;
+import dev.knative.eventing.kafka.broker.dispatcher.ResponseHandler;
+import io.cloudevents.CloudEvent;
+import io.vertx.core.CompositeFuture;
+import io.vertx.core.Future;
+import io.vertx.kafka.client.producer.KafkaProducer;
+import io.vertx.kafka.client.producer.KafkaProducerRecord;
+import org.slf4j.LoggerFactory;
+
+import java.util.Objects;
+
+/**
+ * This class implements a {@link ResponseHandler} that will convert the sink response into a {@link CloudEvent} and push it to a Kafka topic.
+ */
+public final class ResponseToKafkaTopicHandler extends BaseResponseHandler implements ResponseHandler {
+
+  private final String topic;
+  private final KafkaProducer<String, CloudEvent> producer;
+  private final AutoCloseable producerMeterBinder;
+
+  /**
+   * All args constructor.
+   *
+   * @param producer Kafka producer.
+   * @param topic    topic to produce records.
+   */
+  public ResponseToKafkaTopicHandler(final KafkaProducer<String, CloudEvent> producer, final String topic) {
+    super(LoggerFactory.getLogger(ResponseToKafkaTopicHandler.class));
+
+    Objects.requireNonNull(topic, "provide topic");
+    Objects.requireNonNull(producer, "provide producer");
+
+    this.topic = topic;
+    this.producer = producer;
+    this.producerMeterBinder = Metrics.register(this.producer.unwrap());
+  }
+
+  @Override
+  protected Future<Void> doHandleEvent(CloudEvent event) {
+    return producer
+      .send(KafkaProducerRecord.create(topic, event))
+      .mapEmpty();
+  }
+
+  @Override
+  public Future<Void> close() {
+    return CompositeFuture.all(
+      this.producer.close(),
+      AsyncCloseable.wrapAutoCloseable(producerMeterBinder).close()
+    ).mapEmpty();
+  }
+}
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 af623c6f4..d24a1f636 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
@@ -26,10 +26,11 @@ import dev.knative.eventing.kafka.broker.dispatcher.ConsumerVerticleFactory;
 import dev.knative.eventing.kafka.broker.dispatcher.DeliveryOrder;
 import dev.knative.eventing.kafka.broker.dispatcher.Filter;
 import dev.knative.eventing.kafka.broker.dispatcher.ResponseHandler;
-import dev.knative.eventing.kafka.broker.dispatcher.impl.KafkaResponseHandler;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.NoopResponseHandler;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.RecordDispatcherImpl;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.RecordDispatcherMutatorChain;
+import dev.knative.eventing.kafka.broker.dispatcher.impl.ResponseToHttpEndpointHandler;
+import dev.knative.eventing.kafka.broker.dispatcher.impl.ResponseToKafkaTopicHandler;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.BaseConsumerVerticle;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.CloudEventOverridesMutator;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.InvalidCloudEventInterceptor;
@@ -171,7 +172,9 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
           new AttributesFilter(egress.getFilter().getAttributesMap()) :
           Filter.noop();
 
-        final var responseHandler = getNoopResponseHandlerOrDefault(egress, () -> getKafkaResponseHandler(vertx, producerConfigs, resource));
+        final var responseHandler = getResponseHandler(egress,
+          () -> getResponseToKafkaTopicHandler(vertx, producerConfigs, resource),
+          () -> new ResponseToHttpEndpointHandler(createConsumerRecordSender(vertx, egress.getReplyUrl(), egressConfig)));
         final var commitIntervalMs = Integer.parseInt(String.valueOf(consumerConfigs.get(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG)));
 
         final var recordDispatcher = new RecordDispatcherMutatorChain(
@@ -202,19 +205,25 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
     return getConsumerVerticle(deliveryOrder, initializer, new HashSet<>(resource.getTopicsList()));
   }
 
-  static ResponseHandler getNoopResponseHandlerOrDefault(final DataPlaneContract.Egress egress,
-                                                         final Supplier<ResponseHandler> defaultSupplier) {
-    if (egress.hasDiscardReply()) {
+  static ResponseHandler getResponseHandler(final DataPlaneContract.Egress egress,
+                                            final Supplier<ResponseHandler> kafkaSupplier,
+                                            final Supplier<ResponseHandler> httpSupplier) {
+    if(egress.hasReplyUrl()){
+      return httpSupplier.get();
+    } else if(egress.hasReplyToOriginalTopic()){
+      return kafkaSupplier.get();
+    } else if (egress.hasDiscardReply()) {
       return new NoopResponseHandler();
     }
-    return defaultSupplier.get();
+    //TODO: log
+    return kafkaSupplier.get();
   }
 
-  private KafkaResponseHandler getKafkaResponseHandler(final Vertx vertx,
-                                                       final Map<String, Object> producerConfigs,
-                                                       final DataPlaneContract.Resource resource) {
+  private ResponseToKafkaTopicHandler getResponseToKafkaTopicHandler(final Vertx vertx,
+                                                                     final Map<String, Object> producerConfigs,
+                                                                     final DataPlaneContract.Resource resource) {
     final KafkaProducer<String, CloudEvent> producer = createProducer(vertx, producerConfigs);
-    return new KafkaResponseHandler(producer, resource.getTopics(0));
+    return new ResponseToKafkaTopicHandler(producer, resource.getTopics(0));
   }
 
   protected KafkaProducer<String, CloudEvent> createProducer(final Vertx vertx,


Test patch:
diff --git a/control-plane/pkg/reconciler/channel/channel_test.go b/control-plane/pkg/reconciler/channel/channel_test.go
index eecb34858..64f9e2831 100644
--- a/control-plane/pkg/reconciler/channel/channel_test.go
+++ b/control-plane/pkg/reconciler/channel/channel_test.go
@@ -371,6 +371,7 @@ func TestReconcileKind(t *testing.T) {
 								Destination:   "http://" + Subscription1URI,
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
+								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
 							}},
 						},
 					},
@@ -445,6 +446,7 @@ func TestReconcileKind(t *testing.T) {
 								Destination:   "http://" + Subscription1URI,
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
+								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
 							}},
 						},
 					},
@@ -519,6 +521,7 @@ func TestReconcileKind(t *testing.T) {
 								Destination:   "http://" + Subscription1URI,
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
+								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
 							}},
 						},
 					},
@@ -596,11 +599,13 @@ func TestReconcileKind(t *testing.T) {
 								Destination:   "http://" + Subscription1URI,
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
+								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
 							}, {
 								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription2UUID,
 								Destination:   "http://" + Subscription2URI,
 								Uid:           Subscription2UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
+								ReplyStrategy: &contract.Egress_DiscardReply{},
 							}},
 						},
 					},
@@ -957,6 +962,7 @@ func TestReconcileKind(t *testing.T) {
 								Destination:   "http://" + Subscription1URI,
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
+								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
 							}},
 						},
 					},
diff --git a/control-plane/pkg/reconciler/testing/objects_channel.go b/control-plane/pkg/reconciler/testing/objects_channel.go
index 336fc2ece..cdfe2c636 100644
--- a/control-plane/pkg/reconciler/testing/objects_channel.go
+++ b/control-plane/pkg/reconciler/testing/objects_channel.go
@@ -51,7 +51,6 @@ const (
 	Subscription1URI      = "sub-1-uri"
 	Subscription2URI      = "sub-2-uri"
 	Subscription1ReplyURI = "sub-1-reply-uri"
-	Subscription2ReplyURI = "sub-2-reply-uri"
 )
 
 func ChannelTopic() string {
@@ -244,7 +243,7 @@ func Subscriber2(options ...subscriberInfoOption) *SubscriberInfo {
 			UID:           Subscription2UUID,
 			Generation:    1,
 			SubscriberURI: apis.HTTP(Subscription2URI),
-			ReplyURI:      apis.HTTP(Subscription2ReplyURI),
+			// no replies on this one
 		},
 		status: &eventingduckv1.SubscriberStatus{
 			UID:                Subscription2UUID,
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/ResponseToHttpEndpointHandlerTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/ResponseToHttpEndpointHandlerTest.java
new file mode 100644
index 000000000..5032d3e65
--- /dev/null
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/ResponseToHttpEndpointHandlerTest.java
@@ -0,0 +1,143 @@
+/*
+ * Copyright © 2018 Knative Authors (knative-dev@googlegroups.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package dev.knative.eventing.kafka.broker.dispatcher.impl;
+
+import dev.knative.eventing.kafka.broker.dispatcher.CloudEventSender;
+import io.cloudevents.core.provider.EventFormatProvider;
+import io.cloudevents.core.v1.CloudEventBuilder;
+import io.vertx.core.Future;
+import io.vertx.core.MultiMap;
+import io.vertx.core.Vertx;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.http.HttpHeaders;
+import io.vertx.ext.web.client.HttpResponse;
+import io.vertx.junit5.VertxExtension;
+import io.vertx.junit5.VertxTestContext;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
+import org.mockito.Mockito;
+
+import java.net.URI;
+import java.util.concurrent.CountDownLatch;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@Execution(ExecutionMode.CONCURRENT)
+@ExtendWith(VertxExtension.class)
+public class ResponseToHttpEndpointHandlerTest {
+
+  private CloudEventSender cloudEventSender;
+  private ResponseToHttpEndpointHandler handler;
+
+  @BeforeEach
+  void setUp() {
+    cloudEventSender = mock(CloudEventSender.class);
+    handler = new ResponseToHttpEndpointHandler(cloudEventSender);
+  }
+
+  @Test
+  public void shouldSucceedOnUnknownEncodingAndEmptyResponse(final Vertx vertx, final VertxTestContext context) {
+    final HttpResponse<Buffer> response = mock(HttpResponse.class);
+    when(response.statusCode()).thenReturn(202);
+    when(response.body()).thenReturn(Buffer.buffer());
+    when(response.headers()).thenReturn(MultiMap.caseInsensitiveMultiMap());
+
+    context
+      .assertComplete(handler.handle(response))
+      .onSuccess(v -> context.completeNow());
+  }
+
+  @Test
+  public void shouldSucceedOnUnknownEncodingAndNullResponseBody(final Vertx vertx, final VertxTestContext context) {
+    final HttpResponse<Buffer> response = mock(HttpResponse.class);
+    when(response.statusCode()).thenReturn(202);
+    when(response.body()).thenReturn(null);
+    when(response.headers()).thenReturn(MultiMap.caseInsensitiveMultiMap());
+
+    context
+      .assertComplete(handler.handle(response))
+      .onSuccess(v -> context.completeNow());
+  }
+
+  @Test
+  public void shouldFailOnUnknownEncodingAndNonEmptyResponse(final Vertx vertx, final VertxTestContext context) {
+    final HttpResponse<Buffer> response = mock(HttpResponse.class);
+    when(response.statusCode()).thenReturn(202);
+    when(response.body()).thenReturn(Buffer.buffer(new byte[]{'a'}));
+    when(response.headers()).thenReturn(MultiMap.caseInsensitiveMultiMap());
+
+    context
+      .assertFailure(handler.handle(response))
+      .onSuccess(s -> context.failNow(new Exception("expected failed future")))
+      .onFailure(v -> context.completeNow());
+  }
+
+  @Test
+  public void shouldSendRecord(final Vertx vertx, final VertxTestContext context) throws InterruptedException {
+
+    final var event = new CloudEventBuilder()
+      .withId("1234")
+      .withSource(URI.create("/api"))
+      .withSubject("subject")
+      .withType("type")
+      .build();
+
+    final HttpResponse<Buffer> response = mock(HttpResponse.class);
+    when(response.body()).thenReturn(Buffer.buffer(
+      EventFormatProvider.getInstance()
+        .resolveFormat("application/cloudevents+json")
+        .serialize(event)
+    ));
+    when(response.headers()).thenReturn(MultiMap.caseInsensitiveMultiMap()
+      .set(HttpHeaders.CONTENT_TYPE, "application/cloudevents+json")
+    );
+    when(cloudEventSender.send(event)).thenReturn(Future.succeededFuture());
+
+    final var wait = new CountDownLatch(1);
+    handler.handle(response)
+      .onSuccess(ignored -> wait.countDown())
+      .onFailure(context::failNow);
+
+    wait.await();
+
+    Mockito.verify(cloudEventSender).send(event);
+
+    context.completeNow();
+  }
+
+  @Test
+  public void shouldCloseProducer(final Vertx vertx, final VertxTestContext context) {
+    when(cloudEventSender.close()).thenReturn(Future.succeededFuture());
+
+    final var responseHandler = new ResponseToHttpEndpointHandler(cloudEventSender);
+
+    vertx.runOnContext(v -> {
+      responseHandler.close()
+        .onFailure(context::failNow)
+        .onSuccess(r -> context.verify(() -> {
+          verify(cloudEventSender, times(1)).close();
+          context.completeNow();
+        }));
+    });
+  }
+
+}
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/KafkaResponseHandlerTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/ResponseToKafkaTopicHandlerTest.java
similarity index 94%
rename from data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/KafkaResponseHandlerTest.java
rename to data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/ResponseToKafkaTopicHandlerTest.java
index d7c0cfd3e..3a95cadac 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/KafkaResponseHandlerTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/ResponseToKafkaTopicHandlerTest.java
@@ -31,8 +31,6 @@ import io.vertx.junit5.VertxTestContext;
 import io.vertx.kafka.client.producer.KafkaProducer;
 import io.vertx.micrometer.MicrometerMetricsOptions;
 import io.vertx.micrometer.backends.BackendRegistries;
-import java.net.URI;
-import java.util.concurrent.CountDownLatch;
 import org.apache.kafka.clients.producer.MockProducer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.serialization.StringSerializer;
@@ -42,6 +40,9 @@ import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.api.parallel.Execution;
 import org.junit.jupiter.api.parallel.ExecutionMode;
 
+import java.net.URI;
+import java.util.concurrent.CountDownLatch;
+
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -49,7 +50,7 @@ import static org.mockito.Mockito.when;
 
 @Execution(ExecutionMode.CONCURRENT)
 @ExtendWith(VertxExtension.class)
-public class KafkaResponseHandlerTest {
+public class ResponseToKafkaTopicHandlerTest {
 
   static {
     BackendRegistries.setupBackend(new MicrometerMetricsOptions().setRegistryName(Metrics.METRICS_REGISTRY_NAME));
@@ -64,11 +65,10 @@ public class KafkaResponseHandlerTest {
       new StringSerializer(),
       new CloudEventSerializerMock()
     );
-    final var handler = new KafkaResponseHandler(
+    final var handler = new ResponseToKafkaTopicHandler(
       KafkaProducer.create(vertx, producer), TOPIC
     );
 
-    // Empty response
     final HttpResponse<Buffer> response = mock(HttpResponse.class);
     when(response.statusCode()).thenReturn(202);
     when(response.body()).thenReturn(Buffer.buffer());
@@ -86,11 +86,10 @@ public class KafkaResponseHandlerTest {
       new StringSerializer(),
       new CloudEventSerializerMock()
     );
-    final var handler = new KafkaResponseHandler(
+    final var handler = new ResponseToKafkaTopicHandler(
       KafkaProducer.create(vertx, producer), TOPIC
     );
 
-    // Empty response
     final HttpResponse<Buffer> response = mock(HttpResponse.class);
     when(response.statusCode()).thenReturn(202);
     when(response.body()).thenReturn(null);
@@ -110,11 +109,10 @@ public class KafkaResponseHandlerTest {
       new StringSerializer(),
       new CloudEventSerializerMock()
     );
-    final var handler = new KafkaResponseHandler(
+    final var handler = new ResponseToKafkaTopicHandler(
       KafkaProducer.create(vertx, producer), TOPIC
     );
 
-    // Empty response
     final HttpResponse<Buffer> response = mock(HttpResponse.class);
     when(response.statusCode()).thenReturn(202);
     when(response.body()).thenReturn(Buffer.buffer(new byte[] {'a'}));
@@ -134,7 +132,7 @@ public class KafkaResponseHandlerTest {
       new StringSerializer(),
       new CloudEventSerializerMock()
     );
-    final var handler = new KafkaResponseHandler(
+    final var handler = new ResponseToKafkaTopicHandler(
       KafkaProducer.create(vertx, producer), TOPIC
     );
 
@@ -175,7 +173,7 @@ public class KafkaResponseHandlerTest {
     final var mockProducer = new MockProducer<String, CloudEvent>();
     when(producer.unwrap()).thenReturn(mockProducer);
 
-    final var sinkResponseHandler = new KafkaResponseHandler(
+    final var sinkResponseHandler = new ResponseToKafkaTopicHandler(
       producer, "topic"
     );
 
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImplTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImplTest.java
index bfbaac059..406ae4998 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImplTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImplTest.java
@@ -27,14 +27,13 @@ import io.micrometer.core.instrument.MeterRegistry;
 import io.vertx.ext.web.client.WebClientOptions;
 import io.vertx.micrometer.MicrometerMetricsOptions;
 import io.vertx.micrometer.backends.BackendRegistries;
-
-import java.util.Properties;
-import java.util.concurrent.atomic.AtomicInteger;
-
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.junit.jupiter.api.Test;
 
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicInteger;
+
 import static org.apache.kafka.clients.consumer.ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG;
 import static org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG;
 import static org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG;
@@ -235,34 +234,93 @@ public class ConsumerVerticleFactoryImplTest {
 
   @Test
   public void getNoopResponseHandler() {
-    final var counter = new AtomicInteger(0);
+    final var kafkaCounter = new AtomicInteger(0);
+    final var httpCounter = new AtomicInteger(0);
     final var egress = DataPlaneContract.Egress.newBuilder()
       .setDiscardReply(DataPlaneContract.Empty.newBuilder().build())
       .build();
 
-    final var r = ConsumerVerticleFactoryImpl.getNoopResponseHandlerOrDefault(egress, () -> {
-      counter.incrementAndGet();
-      return null;
-    });
+    final var r = ConsumerVerticleFactoryImpl.getResponseHandler(egress,
+      () -> {
+        kafkaCounter.incrementAndGet();
+        return null;
+      },
+      () -> {
+        httpCounter.incrementAndGet();
+        return null;
+      });
 
     assertThat(r).isNotNull();
-    assertThat(counter.get()).isEqualTo(0);
+    assertThat(kafkaCounter.get()).isEqualTo(0);
+    assertThat(httpCounter.get()).isEqualTo(0);
   }
 
   @Test
   public void getKafkaResponseHandler() {
-    final var counter = new AtomicInteger(0);
+    final var kafkaCounter = new AtomicInteger(0);
+    final var httpCounter = new AtomicInteger(0);
 
     final var egress = DataPlaneContract.Egress.newBuilder()
       .setReplyToOriginalTopic(DataPlaneContract.Empty.newBuilder().build())
       .build();
 
-    final var r = ConsumerVerticleFactoryImpl.getNoopResponseHandlerOrDefault(egress, () -> {
-      counter.incrementAndGet();
-      return null;
-    });
+    final var r = ConsumerVerticleFactoryImpl.getResponseHandler(egress,
+      () -> {
+        kafkaCounter.incrementAndGet();
+        return null;
+      },
+      () -> {
+        httpCounter.incrementAndGet();
+        return null;
+      });
+
+    assertThat(r).isNull();
+    assertThat(kafkaCounter.get()).isEqualTo(1);
+    assertThat(httpCounter.get()).isEqualTo(0);
+  }
+
+  @Test
+  public void getHttpResponseHandler() {
+    final var kafkaCounter = new AtomicInteger(0);
+    final var httpCounter = new AtomicInteger(0);
+
+    final var egress = DataPlaneContract.Egress.newBuilder()
+      .setReplyUrl("http://foo.bar")
+      .build();
+
+    final var r = ConsumerVerticleFactoryImpl.getResponseHandler(egress,
+      () -> {
+        kafkaCounter.incrementAndGet();
+        return null;
+      },
+      () -> {
+        httpCounter.incrementAndGet();
+        return null;
+      });
+
+    assertThat(r).isNull();
+    assertThat(kafkaCounter.get()).isEqualTo(0);
+    assertThat(httpCounter.get()).isEqualTo(1);
+  }
+
+  @Test
+  public void getShouldBackToNoopResponseHandlerIfNothingSet() {
+    final var kafkaCounter = new AtomicInteger(0);
+    final var httpCounter = new AtomicInteger(0);
+    final var egress = DataPlaneContract.Egress.newBuilder().build();
+
+    final var r = ConsumerVerticleFactoryImpl.getResponseHandler(egress,
+      () -> {
+        kafkaCounter.incrementAndGet();
+        return null;
+      },
+      () -> {
+        httpCounter.incrementAndGet();
+        return null;
+      });
 
     assertThat(r).isNull();
-    assertThat(counter.get()).isEqualTo(1);
+    assertThat(kafkaCounter.get()).isEqualTo(1);
+    assertThat(httpCounter.get()).isEqualTo(0);
   }
 }
diff --git a/test/e2e-common.sh b/test/e2e-common.sh
index 7a9559a4f..8d9265191 100644
--- a/test/e2e-common.sh
+++ b/test/e2e-common.sh
@@ -123,6 +123,7 @@ function install_latest_release() {
   kubectl apply -f "${PREVIOUS_RELEASE_URL}/${EVENTING_KAFKA_CONTROL_PLANE_ARTIFACT}" || return $?
   kubectl apply -f "${PREVIOUS_RELEASE_URL}/${EVENTING_KAFKA_BROKER_ARTIFACT}" || return $?
   kubectl apply -f "${PREVIOUS_RELEASE_URL}/${EVENTING_KAFKA_SINK_ARTIFACT}" || return $?
+  kubectl apply -f "${PREVIOUS_RELEASE_URL}/${EVENTING_KAFKA_SOURCE_ARTIFACT}" || return $?
   # kubectl apply -f "${PREVIOUS_RELEASE_URL}/${EVENTING_KAFKA_CHANNEL_ARTIFACT}" || return $?
 }
 
diff --git a/test/e2e/channel_event_transformation_test.go b/test/e2e/channel_event_transformation_test.go
new file mode 100644
index 000000000..07e3d9d8f
--- /dev/null
+++ b/test/e2e/channel_event_transformation_test.go
@@ -0,0 +1,31 @@
+//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
+
+import (
+	"context"
+	"testing"
+
+	"knative.dev/eventing/test/e2e/helpers"
+)
+
+func TestEventTransformationForSubscription(t *testing.T) {
+	helpers.EventTransformationForSubscriptionTestHelper(context.Background(), t, helpers.SubscriptionV1, channelTestRunner)
+}
diff --git a/test/e2e/kafka_sink.go b/test/e2e/kafka_sink.go
new file mode 100644
index 000000000..3760f60f8
--- /dev/null
+++ b/test/e2e/kafka_sink.go
@@ -0,0 +1,106 @@
+/*
+ * 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
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/utils/pointer"
+	testlib "knative.dev/eventing/test/lib"
+
+	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/test/pkg/addressable"
+	"knative.dev/eventing-kafka-broker/test/pkg/sink"
+
+	. "knative.dev/eventing-kafka-broker/test/pkg/testing"
+)
+
+const (
+	sinkSecretName = "secret-test"
+)
+
+func RunTestKafkaSink(t *testing.T, mode string, sp SecretProvider, opts ...func(kss *eventingv1alpha1.KafkaSinkSpec) error) {
+	RunMultiple(t, func(t *testing.T) {
+
+		ctx := context.Background()
+
+		const (
+			kafkaSinkName = "kafka-sink"
+		)
+
+		client := testlib.Setup(t, false)
+		defer testlib.TearDown(client)
+
+		clientSet, err := eventingv1alpha1clientset.NewForConfig(client.Config)
+		require.Nil(t, err)
+
+		// Create a KafkaSink with the following spec.
+
+		kss := eventingv1alpha1.KafkaSinkSpec{
+			Topic:             "kafka-sink-" + client.Namespace,
+			NumPartitions:     pointer.Int32Ptr(10),
+			ReplicationFactor: func(rf int16) *int16 { return &rf }(1),
+			BootstrapServers:  BootstrapServersPlaintextArr,
+			ContentMode:       pointer.StringPtr(mode),
+		}
+		for _, opt := range opts {
+			require.Nil(t, opt(&kss))
+		}
+
+		t.Logf("%+v", kss)
+
+		if sp != nil {
+			secretData := sp(t, client)
+			secret := &corev1.Secret{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: client.Namespace,
+					Name:      sinkSecretName,
+				},
+				Data: secretData,
+			}
+			secret, err = client.Kube.CoreV1().Secrets(client.Namespace).Create(ctx, secret, metav1.CreateOptions{})
+			require.Nil(t, err)
+			client.Tracker.Add(corev1.GroupName, "v1", "Secret", secret.Namespace, secret.Name)
+		}
+
+		createFunc := sink.CreatorV1Alpha1(clientSet, kss)
+
+		kafkaSink, err := createFunc(types.NamespacedName{
+			Namespace: client.Namespace,
+			Name:      kafkaSinkName,
+		})
+		require.Nil(t, err)
+
+		ks, err := clientSet.KafkaSinks(client.Namespace).Get(ctx, kafkaSinkName, metav1.GetOptions{})
+		require.Nil(t, err)
+		client.Tracker.AddObj(ks)
+
+		client.WaitForResourceReadyOrFail(kafkaSink.Name, &kafkaSink.TypeMeta)
+
+		// Send events to the KafkaSink.
+		ids := addressable.Send(t, kafkaSink)
+
+		// Read events from the topic.
+		sink.Verify(t, client, mode, kss.Topic, ids)
+	})
+}
diff --git a/test/e2e/kafka_sink_test.go b/test/e2e/kafka_sink_test.go
index e7d5e92b5..735fc6054 100644
--- a/test/e2e/kafka_sink_test.go
+++ b/test/e2e/kafka_sink_test.go
@@ -20,122 +20,43 @@
 package e2e
 
 import (
-	"context"
 	"testing"
 
-	"github.com/stretchr/testify/require"
-	corev1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/utils/pointer"
-	testlib "knative.dev/eventing/test/lib"
 
 	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/test/pkg/addressable"
-	"knative.dev/eventing-kafka-broker/test/pkg/sink"
 	. "knative.dev/eventing-kafka-broker/test/pkg/testing"
 )
 
-const (
-	sinkSecretName = "secret-test"
-)
-
 func TestKafkaSinkV1Alpha1DefaultContentMode(t *testing.T) {
-	testKafkaSink(t, eventingv1alpha1.ModeStructured, nil, func(kss *eventingv1alpha1.KafkaSinkSpec) error {
+	RunTestKafkaSink(t, eventingv1alpha1.ModeStructured, nil, func(kss *eventingv1alpha1.KafkaSinkSpec) error {
 		kss.ContentMode = pointer.StringPtr("")
 		return nil
 	})
 }
 
 func TestKafkaSinkV1Alpha1StructuredContentMode(t *testing.T) {
-	testKafkaSink(t, eventingv1alpha1.ModeStructured, nil)
+	RunTestKafkaSink(t, eventingv1alpha1.ModeStructured, nil)
 }
 
 func TestKafkaSinkV1Alpha1BinaryContentMode(t *testing.T) {
-	testKafkaSink(t, eventingv1alpha1.ModeBinary, nil)
+	RunTestKafkaSink(t, eventingv1alpha1.ModeBinary, nil)
 }
 
 func TestKafkaSinkV1Alpha1AuthPlaintext(t *testing.T) {
-	testKafkaSink(t, eventingv1alpha1.ModeStructured, Plaintext, withBootstrap(BootstrapServersPlaintextArr), withSecret)
+	RunTestKafkaSink(t, eventingv1alpha1.ModeStructured, Plaintext, withBootstrap(BootstrapServersPlaintextArr), withSecret)
 }
 
 func TestKafkaSinkV1Alpha1AuthSsl(t *testing.T) {
-	testKafkaSink(t, eventingv1alpha1.ModeStructured, Ssl, withBootstrap(BootstrapServersSslArr), withSecret)
+	RunTestKafkaSink(t, eventingv1alpha1.ModeStructured, Ssl, withBootstrap(BootstrapServersSslArr), withSecret)
 }
 
 func TestKafkaSinkV1Alpha1AuthSaslPlaintextScram512(t *testing.T) {
-	testKafkaSink(t, eventingv1alpha1.ModeStructured, SaslPlaintextScram512, withBootstrap(BootstrapServersSaslPlaintextArr), withSecret)
+	RunTestKafkaSink(t, eventingv1alpha1.ModeStructured, SaslPlaintextScram512, withBootstrap(BootstrapServersSaslPlaintextArr), withSecret)
 }
 
 func TestKafkaSinkV1Alpha1AuthSslSaslScram512(t *testing.T) {
-	testKafkaSink(t, eventingv1alpha1.ModeStructured, SslSaslScram512, withBootstrap(BootstrapServersSslSaslScramArr), withSecret)
-}
-
-func testKafkaSink(t *testing.T, mode string, sp SecretProvider, opts ...func(kss *eventingv1alpha1.KafkaSinkSpec) error) {
-	RunMultiple(t, func(t *testing.T) {
-
-		ctx := context.Background()
-
-		const (
-			kafkaSinkName = "kafka-sink"
-		)
-
-		client := testlib.Setup(t, false)
-		defer testlib.TearDown(client)
-
-		clientSet, err := eventingv1alpha1clientset.NewForConfig(client.Config)
-		require.Nil(t, err)
-
-		// Create a KafkaSink with the following spec.
-
-		kss := eventingv1alpha1.KafkaSinkSpec{
-			Topic:             "kafka-sink-" + client.Namespace,
-			NumPartitions:     pointer.Int32Ptr(10),
-			ReplicationFactor: func(rf int16) *int16 { return &rf }(1),
-			BootstrapServers:  BootstrapServersPlaintextArr,
-			ContentMode:       pointer.StringPtr(mode),
-		}
-		for _, opt := range opts {
-			require.Nil(t, opt(&kss))
-		}
-
-		t.Log(kss)
-
-		if sp != nil {
-			secretData := sp(t, client)
-			secret := &corev1.Secret{
-				ObjectMeta: metav1.ObjectMeta{
-					Namespace: client.Namespace,
-					Name:      sinkSecretName,
-				},
-				Data: secretData,
-			}
-			secret, err = client.Kube.CoreV1().Secrets(client.Namespace).Create(ctx, secret, metav1.CreateOptions{})
-			require.Nil(t, err)
-			client.Tracker.Add(corev1.GroupName, "v1", "Secret", secret.Namespace, secret.Name)
-		}
-
-		createFunc := sink.CreatorV1Alpha1(clientSet, kss)
-
-		kafkaSink, err := createFunc(types.NamespacedName{
-			Namespace: client.Namespace,
-			Name:      kafkaSinkName,
-		})
-		require.Nil(t, err)
-
-		ks, err := clientSet.KafkaSinks(client.Namespace).Get(ctx, kafkaSinkName, metav1.GetOptions{})
-		require.Nil(t, err)
-		client.Tracker.AddObj(ks)
-
-		client.WaitForResourceReadyOrFail(kafkaSink.Name, &kafkaSink.TypeMeta)
-
-		// Send events to the KafkaSink.
-		ids := addressable.Send(t, kafkaSink)
-
-		// Read events from the topic.
-		sink.Verify(t, client, mode, kss.Topic, ids)
-	})
+	RunTestKafkaSink(t, eventingv1alpha1.ModeStructured, SslSaslScram512, withBootstrap(BootstrapServersSslSaslScramArr), withSecret)
 }
 
 func withSecret(kss *eventingv1alpha1.KafkaSinkSpec) error {
diff --git a/test/upgrade-tests.sh b/test/upgrade-tests.sh
index 5973232ae..7856de479 100755
--- a/test/upgrade-tests.sh
+++ b/test/upgrade-tests.sh
@@ -21,6 +21,15 @@ readonly SKIP_INITIALIZE=${SKIP_INITIALIZE:-false}
 
 source $(dirname $0)/e2e-common.sh
 
+# Override test_setup from e2e-common since we don't want to apply the latest release
+# before running the upgrade test.
+function test_setup() {
+  build_components_from_source || return $?
+
+  # Apply test configurations, and restart data plane components (we don't have hot reload)
+  ko apply -f ./test/config/ || fail_test "Failed to apply test configurations"
+}
+
 if ! ${SKIP_INITIALIZE}; then
   initialize $@ --skip-istio-addon
   save_release_artifacts || fail_test "Failed to save release artifacts"
diff --git a/test/upgrade/continual.go b/test/upgrade/continual.go
index 11443dbd3..0eda11517 100644
--- a/test/upgrade/continual.go
+++ b/test/upgrade/continual.go
@@ -22,6 +22,13 @@ import (
 	"knative.dev/eventing-kafka-broker/test/upgrade/continual"
 )
 
+// ContinualTests returns all continual tests.
+func ContinualTests() []pkgupgrade.BackgroundOperation {
+	c := BrokerContinualTests()
+	c = append(c, SinkContinualTests()...)
+	return c
+}
+
 // BrokerContinualTests returns background operations to test broker
 // functionality in continual manner during the whole upgrade and downgrade
 // process asserting that all events are propagated.
@@ -30,3 +37,12 @@ func BrokerContinualTests() []pkgupgrade.BackgroundOperation {
 		continual.BrokerTest(continual.KafkaBrokerTestOptions{}),
 	}
 }
+
+// SinkContinualTests returns background operations to test KafkaSink
+// functionality in continual manner during the whole upgrade and downgrade
+// process asserting that all events are propagated.
+func SinkContinualTests() []pkgupgrade.BackgroundOperation {
+	return []pkgupgrade.BackgroundOperation{
+		continual.SinkSourceTest(continual.KafkaSinkSourceTestOptions{}),
+	}
+}
diff --git a/test/upgrade/continual/kafka-sink-source-config.toml b/test/upgrade/continual/kafka-sink-source-config.toml
new file mode 100644
index 000000000..99f04dcf1
--- /dev/null
+++ b/test/upgrade/continual/kafka-sink-source-config.toml
@@ -0,0 +1,7 @@
+# logLevel = 'DEBUG'
+[sender]
+address = '{{- .Endpoint -}}'
+interval = {{ .Config.Interval.Nanoseconds }}
+
+[forwarder]
+target = 'http://wathola-receiver.{{- .Namespace -}}.svc.cluster.local'
diff --git a/test/upgrade/continual/sink_source.go b/test/upgrade/continual/sink_source.go
new file mode 100644
index 000000000..a78de876a
--- /dev/null
+++ b/test/upgrade/continual/sink_source.go
@@ -0,0 +1,154 @@
+/*
+ * 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 continual
+
+import (
+	"fmt"
+
+	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/utils/pointer"
+	bindings "knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
+	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	eventingkafkaupgrade "knative.dev/eventing-kafka/test/upgrade/continual"
+	"knative.dev/eventing/test/upgrade/prober/sut"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	pkgupgrade "knative.dev/pkg/test/upgrade"
+
+	eventingkafkatestlib "knative.dev/eventing-kafka/test/lib"
+
+	eventing "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/test/pkg/sink"
+	testingpkg "knative.dev/eventing-kafka-broker/test/pkg/testing"
+)
+
+const (
+	kafkaSinkConfigTemplate = "test/upgrade/continual/kafka-sink-source-config.toml"
+
+	defaultTopic = "kafka-sink-source-continual-test-topic"
+)
+
+type KafkaSinkSourceTestOptions struct {
+	*eventingkafkaupgrade.TestOptions
+	*Sink
+	*Source
+}
+
+func (o *KafkaSinkSourceTestOptions) setDefaults() {
+	if o.TestOptions == nil {
+		o.TestOptions = &eventingkafkaupgrade.TestOptions{}
+	}
+	if o.Sink == nil {
+		rf := int16(3)
+		o.Sink = &Sink{
+			Name: "sink-kafka-sink-source-continual-test",
+			Spec: eventing.KafkaSinkSpec{
+				Topic:             defaultTopic,
+				NumPartitions:     pointer.Int32(10),
+				ReplicationFactor: &rf,
+				BootstrapServers:  []string{testingpkg.BootstrapServersPlaintext},
+			},
+		}
+	}
+	if o.Source == nil {
+		o.Source = &Source{
+			Name: "source-kafka-sink-source-continual-test",
+			Spec: sources.KafkaSourceSpec{
+				Topics:        []string{defaultTopic},
+				ConsumerGroup: "source-kafka-sink-source-continual-test-consumer-group",
+				InitialOffset: sources.OffsetEarliest,
+				KafkaAuthSpec: bindings.KafkaAuthSpec{
+					BootstrapServers: []string{testingpkg.BootstrapServersPlaintext},
+				},
+			},
+		}
+	}
+}
+
+func (o *KafkaSinkSourceTestOptions) validate() error {
+	if len(o.Source.Spec.Topics) != 1 || o.Sink.Spec.Topic != o.Source.Spec.Topics[0] {
+		return fmt.Errorf("expected sink and source topic to be the same topic, source topics %v, sink topic %v", o.Source.Spec.Topics, o.Sink.Spec.Topic)
+	}
+	return nil
+}
+
+type kafkaSinkSourceSut struct {
+	Sink
+	Source
+}
+
+func (k kafkaSinkSourceSut) Deploy(ctx sut.Context, destination duckv1.Destination) interface{} {
+	k.deploySink(ctx)
+	k.deploySource(ctx, destination)
+	url := k.fetchUrl(ctx)
+	return url
+}
+
+func (k kafkaSinkSourceSut) deploySink(ctx sut.Context) {
+	clientSet, err := eventingv1alpha1clientset.NewForConfig(ctx.Config)
+	require.Nil(ctx.T, err)
+
+	s := types.NamespacedName{
+		Namespace: ctx.Client.Namespace,
+		Name:      k.Sink.Name,
+	}
+	if _, err = sink.CreatorV1Alpha1(clientSet, k.Sink.Spec)(s); err != nil {
+		ctx.T.Fatalf("failed to create KafkaSink %+v: %v", s, err)
+	}
+}
+
+func (k kafkaSinkSourceSut) deploySource(ctx sut.Context, destination duckv1.Destination) {
+	ks := &sources.KafkaSource{
+		TypeMeta: metav1.TypeMeta{Kind: "KafkaSource", APIVersion: sources.SchemeGroupVersion.String()},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      k.Source.Name,
+			Namespace: ctx.Client.Namespace,
+		},
+		Spec: *k.Source.Spec.DeepCopy(),
+	}
+	ks.Spec.Sink = destination
+	eventingkafkatestlib.CreateKafkaSourceV1Beta1OrFail(ctx.Client, ks)
+}
+
+func (k kafkaSinkSourceSut) fetchUrl(ctx sut.Context) *apis.URL {
+	tm := &metav1.TypeMeta{
+		Kind:       "KafkaSink",
+		APIVersion: eventing.SchemeGroupVersion.String(),
+	}
+	ctx.Client.WaitForResourceReadyOrFail(k.Sink.Name, tm)
+	address, err := ctx.Client.GetAddressableURI(k.Sink.Name, tm)
+	require.Nil(ctx.T, err, "Failed to get address from sink %s", k.Sink.Name)
+
+	url, _ := apis.ParseURL(address)
+	return url
+}
+
+func SinkSourceTest(opts KafkaSinkSourceTestOptions) pkgupgrade.BackgroundOperation {
+	opts.setDefaults()
+	if err := opts.validate(); err != nil {
+		panic(err)
+	}
+	return continualVerification(
+		"KafkaSinkSourceContinualTest",
+		opts.TestOptions,
+		&kafkaSinkSourceSut{Sink: *opts.Sink, Source: *opts.Source},
+		kafkaSinkConfigTemplate,
+	)
+}
diff --git a/test/upgrade/continual/types.go b/test/upgrade/continual/types.go
index ff92ce4f3..3c9450efa 100644
--- a/test/upgrade/continual/types.go
+++ b/test/upgrade/continual/types.go
@@ -17,10 +17,13 @@
 package continual
 
 import (
+	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
 	eventingkafkaupgrade "knative.dev/eventing-kafka/test/upgrade/continual"
 	"knative.dev/eventing/test/upgrade/prober"
 	"knative.dev/eventing/test/upgrade/prober/sut"
 	"knative.dev/eventing/test/upgrade/prober/wathola/event"
+
+	eventing "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
 )
 
 // TestOptions holds options for EventingKafka continual tests.
@@ -39,6 +42,16 @@ type Broker struct {
 	*eventingkafkaupgrade.RetryOptions
 }
 
+type Sink struct {
+	Name string
+	Spec eventing.KafkaSinkSpec
+}
+
+type Source struct {
+	Name string
+	Spec sources.KafkaSourceSpec
+}
+
 var eventTypes = []string{
 	event.Step{}.Type(),
 	event.Finished{}.Type(),
diff --git a/test/upgrade/postdowngrade.go b/test/upgrade/postdowngrade.go
index dcb3d28cd..6e61863ad 100644
--- a/test/upgrade/postdowngrade.go
+++ b/test/upgrade/postdowngrade.go
@@ -18,6 +18,9 @@ package upgrade
 
 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"
 )
 
 // BrokerPostDowngradeTest tests channel basic channel operations after
@@ -27,3 +30,10 @@ func BrokerPostDowngradeTest() pkgupgrade.Operation {
 		runBrokerSmokeTest(c.T)
 	})
 }
+
+// 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)
+	})
+}
diff --git a/test/upgrade/postupgrade.go b/test/upgrade/postupgrade.go
index 58ebc0d1e..f555ce771 100644
--- a/test/upgrade/postupgrade.go
+++ b/test/upgrade/postupgrade.go
@@ -18,6 +18,9 @@ package upgrade
 
 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"
 )
 
 // BrokerPostUpgradeTest tests channel operations after upgrade.
@@ -26,3 +29,10 @@ func BrokerPostUpgradeTest() pkgupgrade.Operation {
 		runBrokerSmokeTest(c.T)
 	})
 }
+
+// SinkPostUpgradeTest tests sink basic operations post upgrade.
+func SinkPostUpgradeTest() pkgupgrade.Operation {
+	return pkgupgrade.NewOperation("SinkPostDowngradeTest", func(c pkgupgrade.Context) {
+		e2e.RunTestKafkaSink(c.T, eventing.ModeBinary, nil)
+	})
+}
diff --git a/test/upgrade/preupgrade.go b/test/upgrade/preupgrade.go
index 700b9d39d..c46b23f41 100644
--- a/test/upgrade/preupgrade.go
+++ b/test/upgrade/preupgrade.go
@@ -18,6 +18,9 @@ package upgrade
 
 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"
 )
 
 // BrokerPreUpgradeTest tests channel operations before upgrade.
@@ -26,3 +29,10 @@ func BrokerPreUpgradeTest() pkgupgrade.Operation {
 		runBrokerSmokeTest(c.T)
 	})
 }
+
+// SinkPreUpgradeTest tests sink basic operations pre upgrade.
+func SinkPreUpgradeTest() pkgupgrade.Operation {
+	return pkgupgrade.NewOperation("SinkPostDowngradeTest", func(c pkgupgrade.Context) {
+		e2e.RunTestKafkaSink(c.T, eventing.ModeBinary, nil)
+	})
+}
diff --git a/test/upgrade/suite.go b/test/upgrade/suite.go
index 77caa9fd3..ce5da52c9 100644
--- a/test/upgrade/suite.go
+++ b/test/upgrade/suite.go
@@ -28,14 +28,17 @@ func Suite() pkgupgrade.Suite {
 		Tests: pkgupgrade.Tests{
 			PreUpgrade: []pkgupgrade.Operation{
 				BrokerPreUpgradeTest(),
+				SinkPreUpgradeTest(),
 			},
 			PostUpgrade: []pkgupgrade.Operation{
 				BrokerPostUpgradeTest(),
+				SinkPostUpgradeTest(),
 			},
 			PostDowngrade: []pkgupgrade.Operation{
 				BrokerPostDowngradeTest(),
+				SinkPostDowngradeTest(),
 			},
-			Continual: BrokerContinualTests(),
+			Continual: ContinualTests(),
 		},
 		Installations: pkgupgrade.Installations{
 			Base: []pkgupgrade.Operation{