Repository: knative-sandbox/eventing-kafka-broker
Pull request #2164 (2022-05-09 08:45:50)

Use host-based routing in KafkaChannel
The previous implementation of the KafkaChannel uses host-based
routing.
To upgrade from the previous implementation without a big down-time
we need to use host-based routing in this implementation as well.

Fixes #

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

-
-
-

<!--
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[bot]: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2164#" 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":[]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2164?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2164](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2164?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (2ddb0d1) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/82c8f43754629174ece5bf0700bc22741e12ffc7?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (82c8f43) will **decrease** coverage by `5.54%`.
> The diff coverage is `42.42%`.

```diff
@@             Coverage Diff              @@
##               main    #2164      +/-   ##
============================================
- Coverage     66.63%   61.08%   -5.55%     
============================================
  Files           142       70      -72     
  Lines          9158     6895    -2263     
  Branches        196        0     -196     
============================================
- Hits           6102     4212    -1890     
+ Misses         2651     2381     -270     
+ Partials        405      302     -103     
```

| 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/2164?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/receiver/routing.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2164/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjZWl2ZXIvcm91dGluZy5nbw==) | `66.66% <0.00%> (ø)` | |
| [control-plane/pkg/prober/prober.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2164/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcHJvYmVyL3Byb2Jlci5nbw==) | `60.00% <20.00%> (-7.75%)` | :arrow_down: |
| [...ntrol-plane/pkg/reconciler/channel/v2/channelv2.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2164/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jaGFubmVsL3YyL2NoYW5uZWx2Mi5nbw==) | `68.91% <34.28%> (-2.63%)` | :arrow_down: |
| [control-plane/pkg/reconciler/channel/channel.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2164/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jaGFubmVsL2NoYW5uZWwuZ28=) | `70.98% <40.00%> (ø)` | |
| [control-plane/pkg/prober/async\_prober.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2164/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcHJvYmVyL2FzeW5jX3Byb2Jlci5nbw==) | `92.59% <100.00%> (+3.70%)` | :arrow_up: |
| [control-plane/pkg/reconciler/channel/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2164/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jaGFubmVsL2NvbnRyb2xsZXIuZ28=) | `89.02% <100.00%> (ø)` | |
| [...ol-plane/pkg/reconciler/channel/v2/controllerv2.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2164/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jaGFubmVsL3YyL2NvbnRyb2xsZXJ2Mi5nbw==) | `100.00% <100.00%> (ø)` | |
| [...eceiver/impl/IngressProducerReconcilableStore.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2164/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL0luZ3Jlc3NQcm9kdWNlclJlY29uY2lsYWJsZVN0b3JlLmphdmE=) | | |
| [...g/kafka/broker/receiver/impl/ReceiverVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2164/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL1JlY2VpdmVyVmVydGljbGUuamF2YQ==) | | |
| [...ting/kafka/broker/core/utils/CollectionsUtils.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2164/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdXRpbHMvQ29sbGVjdGlvbnNVdGlscy5qYXZh) | | |
| ... and [70 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2164/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/2164?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/2164?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [82c8f43...2ddb0d1](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2164?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).

knative-prow[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
--- | --- | --- | --- | ---
channel-reconciler-tests-ssl_eventing-kafka-broker_main | 2ddb0d11256363f0b2fdc73fedc60e29c68a8a1d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/channel-reconciler-tests-ssl_eventing-kafka-broker_main/1523585024463474688) | 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
--- | --- | --- | --- | ---
channel-reconciler-tests-ssl_eventing-kafka-broker_main | 2ddb0d11256363f0b2fdc73fedc60e29c68a8a1d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/channel-reconciler-tests-ssl_eventing-kafka-broker_main/1523585024463474688) | true | `/test channel-reconciler-tests-ssl_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | 2ddb0d11256363f0b2fdc73fedc60e29c68a8a1d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1523585024492834816) | true | `/test channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main | 2ddb0d11256363f0b2fdc73fedc60e29c68a8a1d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main/1523585025025511424) | 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
--- | --- | --- | --- | ---
channel-reconciler-tests-ssl_eventing-kafka-broker_main | 2ddb0d11256363f0b2fdc73fedc60e29c68a8a1d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/channel-reconciler-tests-ssl_eventing-kafka-broker_main/1523585024463474688) | true | `/test channel-reconciler-tests-ssl_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | 2ddb0d11256363f0b2fdc73fedc60e29c68a8a1d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1523585024492834816) | true | `/test channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main | 2ddb0d11256363f0b2fdc73fedc60e29c68a8a1d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main/1523585025025511424) | true | `/test channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main`
upgrade-tests_eventing-kafka-broker_main | 2ddb0d11256363f0b2fdc73fedc60e29c68a8a1d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/upgrade-tests_eventing-kafka-broker_main/1523585024266342400) | 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
--- | --- | --- | --- | ---
channel-reconciler-tests-ssl_eventing-kafka-broker_main | 2ddb0d11256363f0b2fdc73fedc60e29c68a8a1d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/channel-reconciler-tests-ssl_eventing-kafka-broker_main/1523585024463474688) | true | `/test channel-reconciler-tests-ssl_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | 2ddb0d11256363f0b2fdc73fedc60e29c68a8a1d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1523585024492834816) | true | `/test channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main | 2ddb0d11256363f0b2fdc73fedc60e29c68a8a1d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main/1523585025025511424) | true | `/test channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main`
upgrade-tests_eventing-kafka-broker_main | 2ddb0d11256363f0b2fdc73fedc60e29c68a8a1d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/upgrade-tests_eventing-kafka-broker_main/1523585024266342400) | true | `/test upgrade-tests_eventing-kafka-broker_main`
reconciler-tests_eventing-kafka-broker_main | 2ddb0d11256363f0b2fdc73fedc60e29c68a8a1d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/reconciler-tests_eventing-kafka-broker_main/1523585024308285440) | 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 -->
pierDipi: Blocked by https://github.com/knative-sandbox/eventing-kafka/pull/1172
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
--- | --- | --- | --- | ---
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | 2ddb0d11256363f0b2fdc73fedc60e29c68a8a1d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1523585024492834816) | true | `/test channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main | 2ddb0d11256363f0b2fdc73fedc60e29c68a8a1d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main/1523585025025511424) | true | `/test channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main`
upgrade-tests_eventing-kafka-broker_main | 2ddb0d11256363f0b2fdc73fedc60e29c68a8a1d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/upgrade-tests_eventing-kafka-broker_main/1523585024266342400) | true | `/test upgrade-tests_eventing-kafka-broker_main`
reconciler-tests_eventing-kafka-broker_main | 2ddb0d11256363f0b2fdc73fedc60e29c68a8a1d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/reconciler-tests_eventing-kafka-broker_main/1523585024308285440) | true | `/test reconciler-tests_eventing-kafka-broker_main`
channel-reconciler-tests-ssl_eventing-kafka-broker_main | d4ae7dffd7d9f7724d3e21446df0d11cd1cea3e4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/channel-reconciler-tests-ssl_eventing-kafka-broker_main/1523625764841328640) | 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
--- | --- | --- | --- | ---
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | 2ddb0d11256363f0b2fdc73fedc60e29c68a8a1d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1523585024492834816) | true | `/test channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main`
upgrade-tests_eventing-kafka-broker_main | 2ddb0d11256363f0b2fdc73fedc60e29c68a8a1d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/upgrade-tests_eventing-kafka-broker_main/1523585024266342400) | true | `/test upgrade-tests_eventing-kafka-broker_main`
reconciler-tests_eventing-kafka-broker_main | 2ddb0d11256363f0b2fdc73fedc60e29c68a8a1d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/reconciler-tests_eventing-kafka-broker_main/1523585024308285440) | true | `/test reconciler-tests_eventing-kafka-broker_main`
channel-reconciler-tests-ssl_eventing-kafka-broker_main | d4ae7dffd7d9f7724d3e21446df0d11cd1cea3e4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/channel-reconciler-tests-ssl_eventing-kafka-broker_main/1523625764841328640) | true | `/test channel-reconciler-tests-ssl_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main | d4ae7dffd7d9f7724d3e21446df0d11cd1cea3e4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main/1523625765415948288) | 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
--- | --- | --- | --- | ---
upgrade-tests_eventing-kafka-broker_main | 2ddb0d11256363f0b2fdc73fedc60e29c68a8a1d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/upgrade-tests_eventing-kafka-broker_main/1523585024266342400) | true | `/test upgrade-tests_eventing-kafka-broker_main`
reconciler-tests_eventing-kafka-broker_main | 2ddb0d11256363f0b2fdc73fedc60e29c68a8a1d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/reconciler-tests_eventing-kafka-broker_main/1523585024308285440) | true | `/test reconciler-tests_eventing-kafka-broker_main`
channel-reconciler-tests-ssl_eventing-kafka-broker_main | d4ae7dffd7d9f7724d3e21446df0d11cd1cea3e4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/channel-reconciler-tests-ssl_eventing-kafka-broker_main/1523625764841328640) | true | `/test channel-reconciler-tests-ssl_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main | d4ae7dffd7d9f7724d3e21446df0d11cd1cea3e4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main/1523625765415948288) | true | `/test channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | d4ae7dffd7d9f7724d3e21446df0d11cd1cea3e4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1523625764870688768) | 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
--- | --- | --- | --- | ---
upgrade-tests_eventing-kafka-broker_main | 2ddb0d11256363f0b2fdc73fedc60e29c68a8a1d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/upgrade-tests_eventing-kafka-broker_main/1523585024266342400) | true | `/test upgrade-tests_eventing-kafka-broker_main`
channel-reconciler-tests-ssl_eventing-kafka-broker_main | d4ae7dffd7d9f7724d3e21446df0d11cd1cea3e4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/channel-reconciler-tests-ssl_eventing-kafka-broker_main/1523625764841328640) | true | `/test channel-reconciler-tests-ssl_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main | d4ae7dffd7d9f7724d3e21446df0d11cd1cea3e4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main/1523625765415948288) | true | `/test channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | d4ae7dffd7d9f7724d3e21446df0d11cd1cea3e4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1523625764870688768) | true | `/test channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main`
reconciler-tests_eventing-kafka-broker_main | d4ae7dffd7d9f7724d3e21446df0d11cd1cea3e4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/reconciler-tests_eventing-kafka-broker_main/1523625764681945088) | 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
--- | --- | --- | --- | ---
channel-reconciler-tests-ssl_eventing-kafka-broker_main | d4ae7dffd7d9f7724d3e21446df0d11cd1cea3e4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/channel-reconciler-tests-ssl_eventing-kafka-broker_main/1523625764841328640) | true | `/test channel-reconciler-tests-ssl_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main | d4ae7dffd7d9f7724d3e21446df0d11cd1cea3e4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main/1523625765415948288) | true | `/test channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | d4ae7dffd7d9f7724d3e21446df0d11cd1cea3e4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1523625764870688768) | true | `/test channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main`
reconciler-tests_eventing-kafka-broker_main | d4ae7dffd7d9f7724d3e21446df0d11cd1cea3e4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/reconciler-tests_eventing-kafka-broker_main/1523625764681945088) | true | `/test reconciler-tests_eventing-kafka-broker_main`
upgrade-tests_eventing-kafka-broker_main | d4ae7dffd7d9f7724d3e21446df0d11cd1cea3e4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/upgrade-tests_eventing-kafka-broker_main/1523625764644196352) | 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 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
--- | --- | --- | --- | ---
upgrade-tests_eventing-kafka-broker_main | 1d78a8ae7f11e30f8b2c4e1d9e3da3b403ac8037 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/upgrade-tests_eventing-kafka-broker_main/1523672918851063808) | 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 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
--- | --- | --- | --- | ---
upgrade-tests_eventing-kafka-broker_main | 1d78a8ae7f11e30f8b2c4e1d9e3da3b403ac8037 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/upgrade-tests_eventing-kafka-broker_main/1523728709515743232) | 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]: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2164#pullrequestreview-968771967" title="Approved">matzew</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2164#" 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)~~ [matzew,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":[]} -->
matzew: /cherry-pick release-1.4
knative-prow-robot: @matzew: once the present PR merges, I will cherry-pick it on top of release-1.4 in a new PR and assign it to you.

<details>

In response to [this](https://github.com/knative-sandbox/eventing-kafka-broker/pull/2164#issuecomment-1123258104):

>/cherry-pick release-1.4


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>
matzew: /retest
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
--- | --- | --- | --- | ---
upgrade-tests_eventing-kafka-broker_main | 1d78a8ae7f11e30f8b2c4e1d9e3da3b403ac8037 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/upgrade-tests_eventing-kafka-broker_main/1531913445102850048) | 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 -->
matzew: I tested a channel, and got this `address`:

* `http://<channel-name>-kn-channel.default.svc.cluster.local`

and I could curl a CE against it (got 202)


matzew: Is this the way to go forward? Using HOST based path?

I guess it has benefits (expect for the bizzare routing table @pierDipi mentioned on the java code), as we have a _service_ for it (e.g. `<channel-name>-kn-channel`)

I think that maybe... nice for broker too ? 

(e.g. I think the serving domain mapper makes things easier with a "proper" service, per obj)
pierDipi: > Is this the way to go forward? Using HOST based path?
> 
> I guess it has benefits (expect for the bizzare routing table @pierDipi mentioned on the java code), as we have a _service_ for it (e.g. `<channel-name>-kn-channel`)
> 
> I think that maybe... nice for broker too ?
> 

Yes, it might be nice but for now we don't an easy migration paths for changing Broker addresses (and integrations with service meshes or ingresses are gonna break with these address changes)

> (e.g. I think the serving domain mapper makes things easier with a "proper" service, per obj)


matzew: > Yes, it might be nice but for now we don't an easy migration path for changing Broker addresses (and integrations with service meshes or ingresses are gonna break with these address changes)

fair point
aliok: /test upgrade-tests_eventing-kafka-broker_main
aliok: ```
prober.go:106: event #3304 had unavailable period 1m42.26562673s which is over down time limit of 1m0s
```

This is the error.

@matzew I remember you were increasing some timeout somewhere. Was it the same thing?
matzew: @aliok see: https://github.com/knative/eventing/pull/6381

/cc @mgencur 
aliok: This PR blocked by https://github.com/knative/eventing/pull/6381
mgencur: Why is it blocked by https://github.com/knative/eventing/pull/6381 ? It will only have a little better reporting in tests. Should not be blocking anything.
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
--- | --- | --- | --- | ---
upgrade-tests_eventing-kafka-broker_main | 1d78a8ae7f11e30f8b2c4e1d9e3da3b403ac8037 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/upgrade-tests_eventing-kafka-broker_main/1532666889241628672) | 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 -->
matzew: /test upgrade-tests_eventing-kafka-broker_main
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
--- | --- | --- | --- | ---
upgrade-tests_eventing-kafka-broker_main | 1d78a8ae7f11e30f8b2c4e1d9e3da3b403ac8037 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/upgrade-tests_eventing-kafka-broker_main/1534053994152857600) | 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 -->
matzew: @pierDipi could you rebase yours, and squeeze in this https://github.com/knative-sandbox/eventing-kafka-broker/pull/2281/commits/26a6593f2a911bada33ec94d463e8afc3d9f5393 ? 
knative-prow[bot]: New changes are detected. LGTM label has been removed.
matzew: @pierDipi I was allowing myself to do the rebase and the cherry-pick :cherries:  :sweat_smile:  
matzew: @pierDipi I hope to get https://github.com/knative-sandbox/eventing-kafka-broker/pull/2282  merged, than this
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
--- | --- | --- | --- | ---
upgrade-tests_eventing-kafka-broker_main | 1d78a8ae7f11e30f8b2c4e1d9e3da3b403ac8037 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/upgrade-tests_eventing-kafka-broker_main/1534053994152857600) | true | `/test upgrade-tests_eventing-kafka-broker_main`
unit-tests_eventing-kafka-broker_main | 95d26a5cd2f7fb0510cf50520f0171dfdd1a1fbe | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2164/unit-tests_eventing-kafka-broker_main/1534491064008708096) | 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 -->

Patch:
diff --git a/control-plane/pkg/prober/async_prober.go b/control-plane/pkg/prober/async_prober.go
index c4f211c09..3caa1c6cd 100644
--- a/control-plane/pkg/prober/async_prober.go
+++ b/control-plane/pkg/prober/async_prober.go
@@ -26,7 +26,7 @@ import (
 	"knative.dev/pkg/logging"
 )
 
-type IPsLister func() ([]string, error)
+type IPsLister func(addressable Addressable) ([]string, error)
 
 type asyncProber struct {
 	client    httpClient
@@ -44,8 +44,8 @@ func NewAsync(ctx context.Context, client httpClient, port string, IPsLister IPs
 	logger := logging.FromContext(ctx).Desugar().
 		With(zap.String("scope", "prober"))
 
-	if len(port) == 0 {
-		logger.Fatal("Port is required")
+	if len(port) > 0 && port[0] != ':' {
+		port = ":" + port
 	}
 	return &asyncProber{
 		client:    client,
@@ -59,7 +59,7 @@ func NewAsync(ctx context.Context, client httpClient, port string, IPsLister IPs
 
 func (a *asyncProber) Probe(ctx context.Context, addressable Addressable, expected Status) Status {
 	address := addressable.Address
-	IPs, err := a.IPsLister()
+	IPs, err := a.IPsLister(addressable)
 	if err != nil {
 		a.logger.Error("Failed to list IPs", zap.Error(err))
 		return StatusUnknown
@@ -88,7 +88,7 @@ func (a *asyncProber) Probe(ctx context.Context, addressable Addressable, expect
 
 	for _, IP := range IPs {
 		podUrl := *address
-		podUrl.Host = IP + ":" + a.port
+		podUrl.Host = IP + a.port
 		address := podUrl.String()
 
 		logger := a.logger.
diff --git a/control-plane/pkg/prober/prober.go b/control-plane/pkg/prober/prober.go
index b0cf2f7d8..fa0f514d1 100644
--- a/control-plane/pkg/prober/prober.go
+++ b/control-plane/pkg/prober/prober.go
@@ -92,7 +92,13 @@ func probe(ctx context.Context, client httpClient, logger *zap.Logger, address s
 }
 
 func IPsListerFromService(svc types.NamespacedName) IPsLister {
-	return func() ([]string, error) {
+	return func(addressable Addressable) ([]string, error) {
 		return []string{fmt.Sprintf("%s.%s.svc", svc.Name, svc.Namespace)}, nil
 	}
 }
+
+func IdentityIPsLister() IPsLister {
+	return func(addressable Addressable) ([]string, error) {
+		return []string{addressable.Address.Host}, nil
+	}
+}
diff --git a/control-plane/pkg/receiver/path.go b/control-plane/pkg/receiver/routing.go
similarity index 74%
rename from control-plane/pkg/receiver/path.go
rename to control-plane/pkg/receiver/routing.go
index 3139b1c26..20f0473f3 100644
--- a/control-plane/pkg/receiver/path.go
+++ b/control-plane/pkg/receiver/routing.go
@@ -20,6 +20,8 @@ import (
 	"fmt"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources"
+	"knative.dev/pkg/network"
 )
 
 // PathFromObject returns an HTTP request path given a generic object.
@@ -31,3 +33,8 @@ func PathFromObject(obj metav1.Object) string {
 func Path(namespace, name string) string {
 	return fmt.Sprintf("/%s/%s", namespace, name)
 }
+
+// Host returns an HTTP host given namespace and name of an object.
+func Host(namespace, name string) string {
+	return fmt.Sprintf("%s.%s.svc.%s", resources.MakeChannelServiceName(name), namespace, network.GetClusterDomainName())
+}
diff --git a/control-plane/pkg/receiver/path_test.go b/control-plane/pkg/receiver/routing_test.go
similarity index 100%
rename from control-plane/pkg/receiver/path_test.go
rename to control-plane/pkg/receiver/routing_test.go
diff --git a/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index 33104e4d4..ff8431866 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -19,12 +19,14 @@ package channel
 import (
 	"context"
 	"fmt"
+	"net/url"
 	"strconv"
 	"time"
 
 	"k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"knative.dev/pkg/network"
 
 	"github.com/Shopify/sarama"
 
@@ -62,7 +64,7 @@ const (
 	// TopicPrefix is the Kafka Channel topic prefix - (topic name: knative-messaging-kafka.<channel-namespace>.<channel-name>).
 	TopicPrefix                  = "knative-messaging-kafka"
 	DefaultDeliveryOrder         = contract.DeliveryOrder_ORDERED
-	NewChannelIngressServiceName = "kafka-channel-dispatcher"
+	NewChannelIngressServiceName = "kafka-channel-ingress"
 )
 
 type Reconciler struct {
@@ -278,7 +280,13 @@ func (r *Reconciler) reconcileKind(ctx context.Context, channel *messagingv1beta
 		return subscriptionError
 	}
 
-	address := receiver.Address(r.IngressHost, channel)
+	channelService, err := r.reconcileChannelService(ctx, channel)
+	if err != nil {
+		logger.Error("Error reconciling the backwards compatibility channel service.", zap.Error(err))
+		return err
+	}
+
+	address, _ := url.Parse(fmt.Sprintf("http://%s.%s.svc.%s", channelService.Name, channelService.Namespace, network.GetClusterDomainName()))
 	proberAddressable := prober.Addressable{
 		Address: address,
 		ResourceKey: types.NamespacedName{
@@ -295,12 +303,6 @@ func (r *Reconciler) reconcileKind(ctx context.Context, channel *messagingv1beta
 	}
 	statusConditionManager.Addressable(address)
 
-	err = r.reconcileChannelService(ctx, channel)
-	if err != nil {
-		logger.Error("Error reconciling the backwards compatibility channel service.", zap.Error(err))
-		return err
-	}
-
 	return nil
 }
 
@@ -580,7 +582,7 @@ func (r *Reconciler) getChannelContractResource(ctx context.Context, topic strin
 		Uid:    string(channel.UID),
 		Topics: []string{topic},
 		Ingress: &contract.Ingress{
-			Path: receiver.PathFromObject(channel),
+			Host: receiver.Host(channel.GetNamespace(), channel.GetName()),
 		},
 		BootstrapServers: config.GetBootstrapServers(),
 		Reference: &contract.Reference{
@@ -605,10 +607,10 @@ func (r *Reconciler) getChannelContractResource(ctx context.Context, topic strin
 	return resource, nil
 }
 
-func (r *Reconciler) reconcileChannelService(ctx context.Context, channel *messagingv1beta1.KafkaChannel) error {
+func (r *Reconciler) reconcileChannelService(ctx context.Context, channel *messagingv1beta1.KafkaChannel) (*corev1.Service, error) {
 	expected, err := resources.MakeK8sService(channel, resources.ExternalService(system.Namespace(), NewChannelIngressServiceName))
 	if err != nil {
-		return fmt.Errorf("failed to create the channel service object: %w", err)
+		return expected, fmt.Errorf("failed to create the channel service object: %w", err)
 	}
 
 	svc, err := r.ServiceLister.Services(channel.Namespace).Get(resources.MakeChannelServiceName(channel.Name))
@@ -616,25 +618,25 @@ func (r *Reconciler) reconcileChannelService(ctx context.Context, channel *messa
 		if apierrors.IsNotFound(err) {
 			_, err = r.KubeClient.CoreV1().Services(channel.Namespace).Create(ctx, expected, metav1.CreateOptions{})
 			if err != nil {
-				return fmt.Errorf("failed to create the channel service object: %w", err)
+				return expected, fmt.Errorf("failed to create the channel service object: %w", err)
 			}
-			return nil
+			return expected, nil
 		}
-		return err
+		return expected, err
 	} else if !equality.Semantic.DeepEqual(svc.Spec, expected.Spec) {
 		svc = svc.DeepCopy()
 		svc.Spec = expected.Spec
 
 		_, err = r.KubeClient.CoreV1().Services(channel.Namespace).Update(ctx, svc, metav1.UpdateOptions{})
 		if err != nil {
-			return fmt.Errorf("failed to update the channel service: %w", err)
+			return expected, fmt.Errorf("failed to update the channel service: %w", err)
 		}
 	}
 	// Check to make sure that the KafkaChannel owns this service and if not, complain.
 	if !metav1.IsControlledBy(svc, channel) {
-		return fmt.Errorf("kafkachannel: %s/%s does not own Service: %q", channel.Namespace, channel.Name, svc.Name)
+		return expected, fmt.Errorf("kafkachannel: %s/%s does not own Service: %q", channel.Namespace, channel.Name, svc.Name)
 	}
-	return nil
+	return expected, nil
 }
 
 // consumerGroup returns a consumerGroup name for the given channel and subscription
diff --git a/control-plane/pkg/reconciler/channel/controller.go b/control-plane/pkg/reconciler/channel/controller.go
index c9831742d..c45d597a1 100644
--- a/control-plane/pkg/reconciler/channel/controller.go
+++ b/control-plane/pkg/reconciler/channel/controller.go
@@ -23,7 +23,6 @@ import (
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/cache"
 
 	messagingv1beta "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
@@ -85,8 +84,8 @@ func NewController(ctx context.Context, configs *config.Env) *controller.Impl {
 	}
 
 	impl := kafkachannelreconciler.NewImpl(ctx, reconciler)
-	IPsLister := prober.IPsListerFromService(types.NamespacedName{Namespace: configs.SystemNamespace, Name: configs.IngressName})
-	reconciler.Prober = prober.NewAsync(ctx, http.DefaultClient, configs.IngressPodPort, IPsLister, impl.EnqueueKey)
+	IPsLister := prober.IdentityIPsLister()
+	reconciler.Prober = prober.NewAsync(ctx, http.DefaultClient, "", IPsLister, impl.EnqueueKey)
 	reconciler.IngressHost = network.GetServiceHostname(configs.IngressName, configs.SystemNamespace)
 
 	channelInformer := kafkachannelinformer.Get(ctx)
diff --git a/control-plane/pkg/reconciler/channel/v2/channelv2.go b/control-plane/pkg/reconciler/channel/v2/channelv2.go
index f78b7fe79..b07024c9f 100644
--- a/control-plane/pkg/reconciler/channel/v2/channelv2.go
+++ b/control-plane/pkg/reconciler/channel/v2/channelv2.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"net/url"
 	"strconv"
 	"strings"
 	"time"
@@ -34,6 +35,8 @@ import (
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
 	corelisters "k8s.io/client-go/listers/core/v1"
+	"knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources"
+	"knative.dev/pkg/network"
 	"knative.dev/pkg/resolver"
 
 	v1 "knative.dev/eventing/pkg/apis/duck/v1"
@@ -53,6 +56,7 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
+	channelreconciler "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/channel"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
 
 	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
@@ -96,6 +100,7 @@ type Reconciler struct {
 	NewKafkaClient kafka.NewClientFunc
 
 	ConfigMapLister corelisters.ConfigMapLister
+	ServiceLister   corelisters.ServiceLister
 
 	Prober prober.Prober
 
@@ -277,7 +282,13 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, channel *messagingv1beta
 		logger.Debug("Updated dispatcher pod annotation")
 	}
 
-	address := receiver.Address(r.IngressHost, channel)
+	channelService, err := r.reconcileChannelService(ctx, channel)
+	if err != nil {
+		logger.Error("Error reconciling the backwards compatibility channel service.", zap.Error(err))
+		return err
+	}
+
+	address, _ := url.Parse(fmt.Sprintf("http://%s.%s.svc.%s", channelService.Name, channelService.Namespace, network.GetClusterDomainName()))
 	proberAddressable := prober.Addressable{
 		Address: address,
 		ResourceKey: types.NamespacedName{
@@ -286,7 +297,9 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, channel *messagingv1beta
 		},
 	}
 
+	logger.Debug("Going to probe address to check ingress readiness for the channel.", zap.Any("proberAddressable", proberAddressable))
 	if status := r.Prober.Probe(ctx, proberAddressable, prober.StatusReady); status != prober.StatusReady {
+		logger.Debug("Ingress is not ready for the channel. Going to requeue.", zap.Any("proberAddressable", proberAddressable))
 		statusConditionManager.ProbesStatusNotReady(status)
 		return nil // Object will get re-queued once probe status changes.
 	}
@@ -295,6 +308,38 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, channel *messagingv1beta
 	return nil
 }
 
+func (r *Reconciler) reconcileChannelService(ctx context.Context, channel *messagingv1beta1.KafkaChannel) (*corev1.Service, error) {
+	expected, err := resources.MakeK8sService(channel, resources.ExternalService(r.Env.SystemNamespace, channelreconciler.NewChannelIngressServiceName))
+	if err != nil {
+		return expected, fmt.Errorf("failed to create the channel service object: %w", err)
+	}
+
+	svc, err := r.ServiceLister.Services(channel.Namespace).Get(resources.MakeChannelServiceName(channel.Name))
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			_, err = r.KubeClient.CoreV1().Services(channel.Namespace).Create(ctx, expected, metav1.CreateOptions{})
+			if err != nil {
+				return expected, fmt.Errorf("failed to create the channel service object: %w", err)
+			}
+			return expected, nil
+		}
+		return expected, err
+	} else if !equality.Semantic.DeepEqual(svc.Spec, expected.Spec) {
+		svc = svc.DeepCopy()
+		svc.Spec = expected.Spec
+
+		_, err = r.KubeClient.CoreV1().Services(channel.Namespace).Update(ctx, svc, metav1.UpdateOptions{})
+		if err != nil {
+			return expected, fmt.Errorf("failed to update the channel service: %w", err)
+		}
+	}
+	// Check to make sure that the KafkaChannel owns this service and if not, complain.
+	if !metav1.IsControlledBy(svc, channel) {
+		return expected, fmt.Errorf("kafkachannel: %s/%s does not own Service: %q", channel.Namespace, channel.Name, svc.Name)
+	}
+	return expected, nil
+}
+
 func (r *Reconciler) FinalizeKind(ctx context.Context, channel *messagingv1beta1.KafkaChannel) reconciler.Event {
 	logger := kafkalogging.CreateFinalizeMethodLogger(ctx, channel)
 
@@ -571,7 +616,7 @@ func (r *Reconciler) getChannelContractResource(ctx context.Context, topic strin
 		Uid:    string(channel.UID),
 		Topics: []string{topic},
 		Ingress: &contract.Ingress{
-			Path: receiver.PathFromObject(channel),
+			Host: receiver.Host(channel.GetNamespace(), channel.GetName()),
 		},
 		BootstrapServers: config.GetBootstrapServers(),
 		Reference: &contract.Reference{
diff --git a/control-plane/pkg/reconciler/channel/v2/controllerv2.go b/control-plane/pkg/reconciler/channel/v2/controllerv2.go
index 2454e5e8f..671dffa3c 100644
--- a/control-plane/pkg/reconciler/channel/v2/controllerv2.go
+++ b/control-plane/pkg/reconciler/channel/v2/controllerv2.go
@@ -22,8 +22,8 @@ import (
 
 	"github.com/Shopify/sarama"
 	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/cache"
+	serviceinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/service"
 	"knative.dev/pkg/resolver"
 
 	messagingv1beta "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
@@ -69,13 +69,14 @@ func NewController(ctx context.Context, configs *config.Env) *controller.Impl {
 		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
 		Env:                        configs,
 		ConfigMapLister:            configmapInformer.Lister(),
+		ServiceLister:              serviceinformer.Get(ctx).Lister(),
 		ConsumerGroupLister:        consumerGroupInformer.Lister(),
 		InternalsClient:            consumergroupclient.Get(ctx),
 	}
 
 	impl := kafkachannelreconciler.NewImpl(ctx, reconciler)
-	IPsLister := prober.IPsListerFromService(types.NamespacedName{Namespace: configs.SystemNamespace, Name: configs.IngressName})
-	reconciler.Prober = prober.NewAsync(ctx, http.DefaultClient, configs.IngressPodPort, IPsLister, impl.EnqueueKey)
+	IPsLister := prober.IdentityIPsLister()
+	reconciler.Prober = prober.NewAsync(ctx, http.DefaultClient, "", IPsLister, impl.EnqueueKey)
 	reconciler.IngressHost = network.GetServiceHostname(configs.IngressName, configs.SystemNamespace)
 	reconciler.Resolver = resolver.NewURIResolverFromTracker(ctx, impl.Tracker)
 
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 88832b4c5..20df63368 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,9 +89,27 @@ 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
-    if (pathMapper.get(path) != null) {
-      return pathMapper.get(path);
+    final var p = pathMapper.get(path);
+    if (p != null) {
+      return p;
     }
+
+    // Host based routing is not as simple as path based routing, since the host header
+    // is variable.
+    // For example an external name service might be hit in multiple ways producing
+    // different host headers:
+    // - my-kafka-channel-kn-channel.default
+    // - my-kafka-channel-kn-channel.default:80
+    // - my-kafka-channel-kn-channel.default.svc
+    // - my-kafka-channel-kn-channel.default.svc:80
+    // - my-kafka-channel-kn-channel.default.svc.cluster.local [1]
+    // - my-kafka-channel-kn-channel.default.svc.cluster.local:80
+    //
+    // Current implementations expects only the long form that contains the cluster
+    // domain without the port [1], however, this is clearly a non-robust implementation,
+    // or it might be wrong with some clients, for now, we keep the same behavior as
+    // the old implementations.
+
     return hostMapper.get(host);
   }
 
diff --git a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/ReceiverVerticle.java b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/ReceiverVerticle.java
index a8a056e07..f817b775d 100644
--- a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/ReceiverVerticle.java
+++ b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/ReceiverVerticle.java
@@ -108,9 +108,9 @@ public class ReceiverVerticle extends AbstractVerticle implements Handler<HttpSe
   @Override
   public void stop(Promise<Void> stopPromise) {
     CompositeFuture.all(
-      server.close().mapEmpty(),
-      messageConsumer.unregister()
-    )
+        server.close().mapEmpty(),
+        messageConsumer.unregister()
+      )
       .<Void>mapEmpty()
       .onComplete(stopPromise);
   }
@@ -124,7 +124,11 @@ public class ReceiverVerticle extends AbstractVerticle implements Handler<HttpSe
     IngressProducer producer = this.ingressProducerStore.resolve(request.host(), request.path());
     if (producer == null) {
       request.response().setStatusCode(NOT_FOUND.code()).end();
-      logger.warn("Resource not found {}", keyValue("path", request.path()));
+      logger.warn("Resource not found {} {} {}",
+        keyValue("path", request.path()),
+        keyValue("host", request.host()),
+        keyValue("hostHeader", request.getHeader("Host"))
+      );
       return;
     }
 


Test patch:
diff --git a/control-plane/pkg/prober/async_prober_test.go b/control-plane/pkg/prober/async_prober_test.go
index 893a3b6c8..7a749f72b 100644
--- a/control-plane/pkg/prober/async_prober_test.go
+++ b/control-plane/pkg/prober/async_prober_test.go
@@ -132,7 +132,7 @@ func TestAsyncProber(t *testing.T) {
 			u, _ := url.Parse(s.URL)
 
 			wantRequeueCountMin := atomic.NewInt64(int64(tc.wantRequeueCountMin))
-			var IPsLister IPsLister = func() ([]string, error) {
+			var IPsLister IPsLister = func(addressable Addressable) ([]string, error) {
 				pods, err := podinformer.Get(ctx).Lister().List(tc.podsLabelsSelector)
 				if err != nil {
 					return nil, err
diff --git a/control-plane/pkg/prober/prober_test.go b/control-plane/pkg/prober/prober_test.go
index 596c8985d..cafb496ae 100644
--- a/control-plane/pkg/prober/prober_test.go
+++ b/control-plane/pkg/prober/prober_test.go
@@ -60,7 +60,7 @@ func TestIPsListerFromService(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := IPsListerFromService(tt.svc)()
+			got, err := IPsListerFromService(tt.svc)(Addressable{})
 			if tt.wantErr != (err != nil) {
 				t.Errorf("Got err %v, wantErr %v", err, tt.wantErr)
 			}
diff --git a/control-plane/pkg/reconciler/channel/channel_test.go b/control-plane/pkg/reconciler/channel/channel_test.go
index 62cd532f0..1414e6dbc 100644
--- a/control-plane/pkg/reconciler/channel/channel_test.go
+++ b/control-plane/pkg/reconciler/channel/channel_test.go
@@ -74,7 +74,7 @@ var DefaultEnv = &config.Env{
 	DataPlaneConfigMapNamespace: "knative-eventing",
 	DataPlaneConfigMapName:      "kafka-channel-channels-subscriptions",
 	GeneralConfigMapName:        "kafka-channel-config",
-	IngressName:                 "kafka-channel-dispatcher",
+	IngressName:                 "kafka-channel-ingress",
 	SystemNamespace:             "knative-eventing",
 	DataPlaneConfigFormat:       base.Json,
 }
@@ -164,7 +164,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -237,7 +237,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							EgressConfig: &contract.EgressConfig{
 								DeadLetter: ServiceURL,
@@ -312,7 +312,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -329,6 +329,7 @@ func TestReconcileKind(t *testing.T) {
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
 				NewConfigMapWithBinaryData(&env, nil),
+				NewPerChannelService(&env),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -380,7 +381,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -397,6 +398,7 @@ func TestReconcileKind(t *testing.T) {
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
 				NewConfigMapWithBinaryData(&env, nil),
+				NewPerChannelService(&env),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -449,7 +451,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{},
 						},
@@ -520,7 +522,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{{
 								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription1UUID,
@@ -593,7 +595,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{{
 								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription1UUID,
@@ -666,7 +668,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{{
 								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription1UUID,
@@ -742,7 +744,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{{
 								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription1UUID,
@@ -1020,7 +1022,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -1112,7 +1114,7 @@ func TestReconcileKind(t *testing.T) {
 								},
 							},
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{{
 								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription1UUID,
@@ -1209,7 +1211,7 @@ func TestReconcileKind(t *testing.T) {
 								},
 							},
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{{
 								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription1UUID,
@@ -1303,7 +1305,7 @@ func TestReconcileKind(t *testing.T) {
 								},
 							},
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{{
 								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription1UUID,
@@ -1375,7 +1377,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -1443,7 +1445,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -1553,7 +1555,7 @@ func TestFinalizeKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
diff --git a/control-plane/pkg/reconciler/channel/v2/channelv2_test.go b/control-plane/pkg/reconciler/channel/v2/channelv2_test.go
index e043e11b6..8c738f201 100644
--- a/control-plane/pkg/reconciler/channel/v2/channelv2_test.go
+++ b/control-plane/pkg/reconciler/channel/v2/channelv2_test.go
@@ -155,7 +155,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -197,7 +197,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -214,6 +214,7 @@ func TestReconcileKind(t *testing.T) {
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
 				NewConfigMapWithBinaryData(&env, nil),
+				NewPerChannelService(&env),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -267,7 +268,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							EgressConfig: &contract.EgressConfig{
 								DeadLetter: ServiceURL,
@@ -288,6 +289,7 @@ func TestReconcileKind(t *testing.T) {
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
 				NewConfigMapWithBinaryData(&env, nil),
+				NewPerChannelService(&env),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -333,7 +335,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -342,6 +344,7 @@ func TestReconcileKind(t *testing.T) {
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
 				NewConfigMapWithBinaryData(&env, nil),
+				NewPerChannelService(&env),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -384,7 +387,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -393,6 +396,7 @@ func TestReconcileKind(t *testing.T) {
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
 				NewConfigMapWithBinaryData(&env, nil),
+				NewPerChannelService(&env),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -427,6 +431,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			Key: testKey,
 			WantCreates: []runtime.Object{
+				NewPerChannelService(&env),
 				NewConsumerGroup(
 					WithConsumerGroupName(Subscription1UUID),
 					WithConsumerGroupNamespace(ChannelNamespace),
@@ -454,7 +459,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -493,6 +498,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			Key: testKey,
 			WantCreates: []runtime.Object{
+				NewPerChannelService(&env),
 				NewConsumerGroup(
 					WithConsumerGroupName(Subscription1UUID),
 					WithConsumerGroupNamespace(ChannelNamespace),
@@ -520,7 +526,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -559,6 +565,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			Key: testKey,
 			WantCreates: []runtime.Object{
+				NewPerChannelService(&env),
 				NewConsumerGroup(
 					WithConsumerGroupName(Subscription1UUID),
 					WithConsumerGroupNamespace(ChannelNamespace),
@@ -586,7 +593,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -631,6 +638,9 @@ func TestReconcileKind(t *testing.T) {
 				),
 			},
 			Key: testKey,
+			WantCreates: []runtime.Object{
+				NewPerChannelService(&env),
+			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
 				ConfigMapUpdate(&env, &contract.Contract{
 					Generation: 1,
@@ -641,7 +651,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -714,6 +724,9 @@ func TestReconcileKind(t *testing.T) {
 				),
 			},
 			Key: testKey,
+			WantCreates: []runtime.Object{
+				NewPerChannelService(&env),
+			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
 					Object: NewChannel(
@@ -738,7 +751,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -765,6 +778,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			Key: testKey,
 			WantCreates: []runtime.Object{
+				NewPerChannelService(&env),
 				NewConsumerGroup(
 					WithConsumerGroupName(Subscription1UUID),
 					WithConsumerGroupNamespace(ChannelNamespace),
@@ -808,7 +822,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -865,6 +879,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			Key: testKey,
 			WantCreates: []runtime.Object{
+				NewPerChannelService(&env),
 				NewConsumerGroup(
 					WithConsumerGroupName(Subscription1UUID),
 					WithConsumerGroupNamespace(ChannelNamespace),
@@ -892,7 +907,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -1049,7 +1064,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -1058,6 +1073,7 @@ func TestReconcileKind(t *testing.T) {
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
 				NewConfigMapWithBinaryData(&env, nil),
+				NewPerChannelService(&env),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -1112,6 +1128,9 @@ func TestReconcileKind(t *testing.T) {
 				),
 			},
 			Key: testKey,
+			WantCreates: []runtime.Object{
+				NewPerChannelService(&env),
+			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
 				ConfigMapUpdate(&env, &contract.Contract{
 					Generation: 1,
@@ -1130,7 +1149,7 @@ func TestReconcileKind(t *testing.T) {
 								},
 							},
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -1177,7 +1196,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -1186,6 +1205,7 @@ func TestReconcileKind(t *testing.T) {
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
 				NewConfigMapWithBinaryData(&env, nil),
+				NewPerChannelService(&env),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -1217,6 +1237,9 @@ func TestReconcileKind(t *testing.T) {
 				NewConfigMapWithBinaryData(&env, nil),
 			},
 			Key: testKey,
+			WantCreates: []runtime.Object{
+				NewPerChannelService(&env),
+			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
 				ConfigMapUpdate(&env, &contract.Contract{
 					Generation: 1,
@@ -1227,7 +1250,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -1336,6 +1359,7 @@ func TestReconcileKind(t *testing.T) {
 				}, nil
 			},
 			ConfigMapLister:     listers.GetConfigMapLister(),
+			ServiceLister:       listers.GetServiceLister(),
 			ConsumerGroupLister: listers.GetConsumerGroupLister(),
 			InternalsClient:     fakeconsumergroupinformer.Get(ctx),
 			Prober:              proberMock,
diff --git a/control-plane/pkg/reconciler/channel/v2/controllerv2_test.go b/control-plane/pkg/reconciler/channel/v2/controllerv2_test.go
index 469317c78..34459582f 100644
--- a/control-plane/pkg/reconciler/channel/v2/controllerv2_test.go
+++ b/control-plane/pkg/reconciler/channel/v2/controllerv2_test.go
@@ -24,16 +24,18 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	fakekubeclientset "k8s.io/client-go/kubernetes/fake"
 
-	_ "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumergroup/fake"
 	_ "knative.dev/eventing-kafka/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/fake"
 	_ "knative.dev/pkg/client/injection/ducks/duck/v1/addressable/fake"
 	fakekubeclient "knative.dev/pkg/client/injection/kube/client/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/secret/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/service/fake"
 	dynamicclient "knative.dev/pkg/injection/clients/dynamicclient/fake"
 	reconcilertesting "knative.dev/pkg/reconciler/testing"
 
+	_ "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumergroup/fake"
+
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 )
 
diff --git a/control-plane/pkg/reconciler/testing/objects_channel.go b/control-plane/pkg/reconciler/testing/objects_channel.go
index f52359054..2982313c9 100644
--- a/control-plane/pkg/reconciler/testing/objects_channel.go
+++ b/control-plane/pkg/reconciler/testing/objects_channel.go
@@ -193,8 +193,7 @@ func ChannelAddressable(env *config.Env) func(obj duckv1.KRShaped) {
 
 		channel.Status.Address.URL = &apis.URL{
 			Scheme: "http",
-			Host:   network.GetServiceHostname(env.IngressName, env.SystemNamespace),
-			Path:   fmt.Sprintf("/%s/%s", channel.Namespace, channel.Name),
+			Host:   fmt.Sprintf("%s.%s.svc.%s", resources.MakeChannelServiceName(channel.Name), channel.Namespace, network.GetClusterDomainName()),
 		}
 
 		channel.GetConditionSet().Manage(&channel.Status).MarkTrue(base.ConditionAddressable)
diff --git a/third_party/eventing-kafka-latest/channel-crds.yaml b/third_party/eventing-kafka-latest/channel-crds.yaml
index 135680307..92c9fd105 100644
--- a/third_party/eventing-kafka-latest/channel-crds.yaml
+++ b/third_party/eventing-kafka-latest/channel-crds.yaml
@@ -17,7 +17,7 @@ kind: CustomResourceDefinition
 metadata:
   name: kafkachannels.messaging.knative.dev
   labels:
-    kafka.eventing.knative.dev/release: "v20220530-7b2963149"
+    kafka.eventing.knative.dev/release: "v20220608-f0ac467d6"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
diff --git a/third_party/eventing-kafka-latest/source-crd.yaml b/third_party/eventing-kafka-latest/source-crd.yaml
index 597f16f6e..cb3a03197 100644
--- a/third_party/eventing-kafka-latest/source-crd.yaml
+++ b/third_party/eventing-kafka-latest/source-crd.yaml
@@ -16,7 +16,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    kafka.eventing.knative.dev/release: "v20220530-7b2963149"
+    kafka.eventing.knative.dev/release: "v20220608-f0ac467d6"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"