Repository: knative-sandbox/eventing-kafka-broker
Pull request #1495 (2021-11-17 09:36:18)

Probe resource ingress for readiness
This patch adds the `Prober` that probes the receiver for readiness.

It uses the cache introduced in https://github.com/knative-sandbox/eventing-kafka-broker/pull/1401 and the data plane
probe handling introduced in https://github.com/knative-sandbox/eventing-kafka-broker/pull/1431.

## Proposed Changes

- Probe resource ingress for readiness

<!--
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
The kafka-controller deployment emits probe requests against the data plane (kafka-sink-receiver and kafka-broker-receiver) to determine Kafka Broker and KafkaSink readiness.
```

**Docs**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->
```
None
```

/kind enhancement

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/1495#" title="Author self-approved">pierDipi</a>*

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

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
knative-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/prober/async_prober.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460904626155950080/artifacts/line-cov.html#file0) | Do not exist | 92.3% | 
[control-plane/pkg/prober/prober.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460904626155950080/artifacts/line-cov.html#file2) | Do not exist | 77.8% | 
[control-plane/pkg/prober/probertesting/mock_prober.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460904626155950080/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[control-plane/pkg/receiver/address.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460904626155950080/artifacts/line-cov.html#file4) | Do not exist | 0.0% | 
[control-plane/pkg/reconciler/broker/broker.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460904626155950080/artifacts/line-cov.html#file7) | 84.3% | 84.9% | 0.7
[control-plane/pkg/reconciler/broker/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460904626155950080/artifacts/line-cov.html#file8) | 81.2% | 82.4% | 1.1
[control-plane/pkg/reconciler/channel/channel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460904626155950080/artifacts/line-cov.html#file9) | 39.8% | 41.8% | 2.0
[control-plane/pkg/reconciler/channel/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460904626155950080/artifacts/line-cov.html#file10) | 90.0% | 90.9% | 0.9
[control-plane/pkg/reconciler/sink/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460904626155950080/artifacts/line-cov.html#file11) | 78.9% | 81.0% | 2.0
[control-plane/pkg/reconciler/sink/kafka_sink.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460904626155950080/artifacts/line-cov.html#file12) | 80.0% | 81.3% | 1.3

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-go-coverage | b732c938d3887800a2149a50c686381e7db414aa | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460904626155950080) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-go-coverage | b732c938d3887800a2149a50c686381e7db414aa | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460904626155950080) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-build-tests | b732c938d3887800a2149a50c686381e7db414aa | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-build-tests/1460904626072064000) | true | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`

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

<details>

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

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1495/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/1495?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    #1495      +/-   ##
============================================
- Coverage     73.43%   66.62%   -6.82%     
============================================
  Files           101       40      -61     
  Lines          3975     2229    -1746     
  Branches        165        0     -165     
============================================
- Hits           2919     1485    -1434     
+ Misses          818      590     -228     
+ Partials        238      154      -84     
```

| 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/1495?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/address.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1495/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjZWl2ZXIvYWRkcmVzcy5nbw==) | `0.00% <0.00%> (ø)` | |
| [...lane/pkg/reconciler/base/receiver\_condition\_set.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1495/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9iYXNlL3JlY2VpdmVyX2NvbmRpdGlvbl9zZXQuZ28=) | `0.00% <0.00%> (ø)` | |
| [control-plane/pkg/prober/prober.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1495/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcHJvYmVyL3Byb2Jlci5nbw==) | `70.00% <70.00%> (ø)` | |
| [control-plane/pkg/prober/async\_prober.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1495/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.85% <92.85%> (ø)` | |
| [control-plane/pkg/prober/cache.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1495/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcHJvYmVyL2NhY2hlLmdv) | `100.00% <100.00%> (ø)` | |
| [...trol-plane/pkg/prober/probertesting/mock\_prober.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1495/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcHJvYmVyL3Byb2JlcnRlc3RpbmcvbW9ja19wcm9iZXIuZ28=) | `100.00% <100.00%> (ø)` | |
| [control-plane/pkg/reconciler/base/reconciler.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1495/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9iYXNlL3JlY29uY2lsZXIuZ28=) | `67.22% <100.00%> (ø)` | |
| [control-plane/pkg/reconciler/broker/broker.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1495/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9icm9rZXIvYnJva2VyLmdv) | `72.09% <100.00%> (+1.64%)` | :arrow_up: |
| [control-plane/pkg/reconciler/broker/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1495/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9icm9rZXIvY29udHJvbGxlci5nbw==) | `76.71% <100.00%> (+0.65%)` | :arrow_up: |
| [control-plane/pkg/reconciler/channel/channel.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1495/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jaGFubmVsL2NoYW5uZWwuZ28=) | `33.08% <100.00%> (+3.12%)` | :arrow_up: |
| ... and [69 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1495/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/1495?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/1495?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [09d6a05...b732c93](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1495?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/prober/async_prober.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460906403706179584/artifacts/line-cov.html#file0) | Do not exist | 92.3% | 
[control-plane/pkg/prober/prober.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460906403706179584/artifacts/line-cov.html#file2) | Do not exist | 77.8% | 
[control-plane/pkg/prober/probertesting/mock_prober.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460906403706179584/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[control-plane/pkg/receiver/address.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460906403706179584/artifacts/line-cov.html#file4) | Do not exist | 0.0% | 
[control-plane/pkg/reconciler/broker/broker.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460906403706179584/artifacts/line-cov.html#file7) | 84.3% | 84.9% | 0.7
[control-plane/pkg/reconciler/broker/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460906403706179584/artifacts/line-cov.html#file8) | 81.2% | 82.4% | 1.1
[control-plane/pkg/reconciler/channel/channel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460906403706179584/artifacts/line-cov.html#file9) | 39.8% | 41.8% | 2.0
[control-plane/pkg/reconciler/channel/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460906403706179584/artifacts/line-cov.html#file10) | 90.0% | 90.9% | 0.9
[control-plane/pkg/reconciler/sink/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460906403706179584/artifacts/line-cov.html#file11) | 78.9% | 81.0% | 2.0
[control-plane/pkg/reconciler/sink/kafka_sink.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460906403706179584/artifacts/line-cov.html#file12) | 80.0% | 81.3% | 1.3

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-build-tests | b732c938d3887800a2149a50c686381e7db414aa | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-build-tests/1460904626072064000) | true | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 65d8abce1527de947b09b6f060b9547f24800e04 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460906403706179584) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 65d8abce1527de947b09b6f060b9547f24800e04 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460906403706179584) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 65d8abce1527de947b09b6f060b9547f24800e04 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1460906403672625152) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-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/prober/async_prober.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460921229190893568/artifacts/line-cov.html#file0) | Do not exist | 82.1% | 
[control-plane/pkg/prober/cache.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460921229190893568/artifacts/line-cov.html#file1) | 96.9% | 94.4% | -2.4
[control-plane/pkg/prober/prober.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460921229190893568/artifacts/line-cov.html#file2) | Do not exist | 55.6% | 
[control-plane/pkg/prober/probertesting/mock_prober.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460921229190893568/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[control-plane/pkg/receiver/address.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460921229190893568/artifacts/line-cov.html#file4) | Do not exist | 0.0% | 
[control-plane/pkg/reconciler/broker/broker.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460921229190893568/artifacts/line-cov.html#file7) | 84.3% | 84.9% | 0.7
[control-plane/pkg/reconciler/broker/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460921229190893568/artifacts/line-cov.html#file8) | 81.2% | 82.4% | 1.1
[control-plane/pkg/reconciler/channel/channel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460921229190893568/artifacts/line-cov.html#file9) | 39.8% | 41.8% | 2.0
[control-plane/pkg/reconciler/channel/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460921229190893568/artifacts/line-cov.html#file10) | 90.0% | 90.9% | 0.9
[control-plane/pkg/reconciler/sink/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460921229190893568/artifacts/line-cov.html#file11) | 78.9% | 81.0% | 2.0
[control-plane/pkg/reconciler/sink/kafka_sink.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460921229190893568/artifacts/line-cov.html#file12) | 80.0% | 81.3% | 1.3

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 65d8abce1527de947b09b6f060b9547f24800e04 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460906403706179584) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 65d8abce1527de947b09b6f060b9547f24800e04 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1460906403672625152) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-unit-tests | bd1b9acdc3acb58c2c0247a19abc821ec4feec82 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1460921229136367616) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 65d8abce1527de947b09b6f060b9547f24800e04 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1460906403672625152) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-unit-tests | bd1b9acdc3acb58c2c0247a19abc821ec4feec82 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1460921229136367616) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-go-coverage | bd1b9acdc3acb58c2c0247a19abc821ec4feec82 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460921229190893568) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-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/prober/async_prober.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460924878398951424/artifacts/line-cov.html#file1) | Do not exist | 90.5% | 
[control-plane/pkg/prober/cache.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460924878398951424/artifacts/line-cov.html#file2) | 96.9% | 100.0% | 3.1
[control-plane/pkg/prober/prober.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460924878398951424/artifacts/line-cov.html#file3) | Do not exist | 77.8% | 
[control-plane/pkg/prober/probertesting/mock_prober.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460924878398951424/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[control-plane/pkg/receiver/address.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460924878398951424/artifacts/line-cov.html#file5) | Do not exist | 0.0% | 

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 65d8abce1527de947b09b6f060b9547f24800e04 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1460906403672625152) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-go-coverage | bd1b9acdc3acb58c2c0247a19abc821ec4feec82 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460921229190893568) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 36a36cb1dca7e0aa1e7636a4b6dbf3db3c3c302a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1460924878336036864) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 65d8abce1527de947b09b6f060b9547f24800e04 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1460906403672625152) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 36a36cb1dca7e0aa1e7636a4b6dbf3db3c3c302a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1460924878336036864) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 36a36cb1dca7e0aa1e7636a4b6dbf3db3c3c302a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460924878398951424) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-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/prober/async_prober.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460927685877305344/artifacts/line-cov.html#file1) | Do not exist | 90.5% | 
[control-plane/pkg/prober/cache.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460927685877305344/artifacts/line-cov.html#file2) | 96.9% | 100.0% | 3.1
[control-plane/pkg/prober/prober.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460927685877305344/artifacts/line-cov.html#file3) | Do not exist | 77.8% | 
[control-plane/pkg/prober/probertesting/mock_prober.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460927685877305344/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[control-plane/pkg/receiver/address.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460927685877305344/artifacts/line-cov.html#file5) | Do not exist | 0.0% | 
[control-plane/pkg/reconciler/broker/broker.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460927685877305344/artifacts/line-cov.html#file8) | 84.3% | 84.9% | 0.7
[control-plane/pkg/reconciler/broker/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460927685877305344/artifacts/line-cov.html#file9) | 81.2% | 82.4% | 1.1
[control-plane/pkg/reconciler/channel/channel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460927685877305344/artifacts/line-cov.html#file10) | 39.8% | 41.8% | 2.0
[control-plane/pkg/reconciler/channel/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460927685877305344/artifacts/line-cov.html#file11) | 90.0% | 90.9% | 0.9
[control-plane/pkg/reconciler/sink/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460927685877305344/artifacts/line-cov.html#file12) | 78.9% | 81.0% | 2.0
[control-plane/pkg/reconciler/sink/kafka_sink.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460927685877305344/artifacts/line-cov.html#file13) | 80.0% | 81.3% | 1.3

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/prober/async_prober.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460938031786627072/artifacts/line-cov.html#file1) | Do not exist | 90.5% | 
[control-plane/pkg/prober/cache.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460938031786627072/artifacts/line-cov.html#file2) | 96.9% | 100.0% | 3.1
[control-plane/pkg/prober/prober.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460938031786627072/artifacts/line-cov.html#file3) | Do not exist | 77.8% | 
[control-plane/pkg/prober/probertesting/mock_prober.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460938031786627072/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[control-plane/pkg/receiver/address.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460938031786627072/artifacts/line-cov.html#file5) | Do not exist | 0.0% | 
[control-plane/pkg/reconciler/broker/broker.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460938031786627072/artifacts/line-cov.html#file8) | 84.3% | 84.9% | 0.7
[control-plane/pkg/reconciler/broker/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460938031786627072/artifacts/line-cov.html#file9) | 81.2% | 82.4% | 1.1
[control-plane/pkg/reconciler/channel/channel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460938031786627072/artifacts/line-cov.html#file10) | 39.8% | 41.8% | 2.0
[control-plane/pkg/reconciler/channel/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460938031786627072/artifacts/line-cov.html#file11) | 90.0% | 90.9% | 0.9
[control-plane/pkg/reconciler/sink/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460938031786627072/artifacts/line-cov.html#file12) | 78.9% | 81.0% | 2.0
[control-plane/pkg/reconciler/sink/kafka_sink.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460938031786627072/artifacts/line-cov.html#file13) | 80.0% | 81.3% | 1.3

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 65d8abce1527de947b09b6f060b9547f24800e04 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1460906403672625152) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 6393591383ce4aab795c9fd3297554fbc92d7f99 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460938031786627072) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 6393591383ce4aab795c9fd3297554fbc92d7f99 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460938031786627072) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 6393591383ce4aab795c9fd3297554fbc92d7f99 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1460938031761461248) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-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/prober/async_prober.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460958183789957120/artifacts/line-cov.html#file1) | Do not exist | 90.5% | 
[control-plane/pkg/prober/prober.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460958183789957120/artifacts/line-cov.html#file3) | Do not exist | 77.8% | 
[control-plane/pkg/prober/probertesting/mock_prober.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460958183789957120/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[control-plane/pkg/receiver/address.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460958183789957120/artifacts/line-cov.html#file5) | Do not exist | 0.0% | 
[control-plane/pkg/reconciler/broker/broker.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460958183789957120/artifacts/line-cov.html#file8) | 84.3% | 84.8% | 0.6
[control-plane/pkg/reconciler/broker/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460958183789957120/artifacts/line-cov.html#file9) | 81.2% | 82.4% | 1.1
[control-plane/pkg/reconciler/channel/channel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460958183789957120/artifacts/line-cov.html#file10) | 39.8% | 41.5% | 1.7
[control-plane/pkg/reconciler/channel/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460958183789957120/artifacts/line-cov.html#file11) | 90.0% | 90.9% | 0.9
[control-plane/pkg/reconciler/sink/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460958183789957120/artifacts/line-cov.html#file12) | 78.9% | 81.0% | 2.0
[control-plane/pkg/reconciler/sink/kafka_sink.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460958183789957120/artifacts/line-cov.html#file13) | 80.0% | 81.1% | 1.1

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 6393591383ce4aab795c9fd3297554fbc92d7f99 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1460938031761461248) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 79c7ccd7cbf2a66705c659bbd3fbc7c4a06eddff | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460958183789957120) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 79c7ccd7cbf2a66705c659bbd3fbc7c4a06eddff | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1460958183789957120) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 79c7ccd7cbf2a66705c659bbd3fbc7c4a06eddff | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1460958183760596992) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
pierDipi: 1/1 Worked, retesting
pull-knative-sandbox-eventing-kafka-broker-integration-tests 
pierDipi: 2/2 Worked, retesting
pull-knative-sandbox-eventing-kafka-broker-integration-tests
pierDipi: 1/1 Worked, retesting
/test pull-knative-sandbox-eventing-kafka-broker-integration-tests
pierDipi: 2/2 Worked, retesting
/test pull-knative-sandbox-eventing-kafka-broker-integration-tests
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/prober/probertesting/mock_prober.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1461233002368471040/artifacts/line-cov.html#file1) | Do not exist | 100.0% | 
[control-plane/pkg/receiver/address.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1461233002368471040/artifacts/line-cov.html#file2) | Do not exist | 0.0% | 

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/prober/async_prober.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1461233924637200384/artifacts/line-cov.html#file1) | Do not exist | 90.5% | 
[control-plane/pkg/prober/cache.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1461233924637200384/artifacts/line-cov.html#file2) | 96.9% | 100.0% | 3.1
[control-plane/pkg/prober/prober.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1461233924637200384/artifacts/line-cov.html#file3) | Do not exist | 83.3% | 
[control-plane/pkg/prober/probertesting/mock_prober.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1461233924637200384/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[control-plane/pkg/receiver/address.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1461233924637200384/artifacts/line-cov.html#file5) | Do not exist | 0.0% | 
[control-plane/pkg/reconciler/broker/broker.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1461233924637200384/artifacts/line-cov.html#file8) | 84.3% | 84.8% | 0.6
[control-plane/pkg/reconciler/broker/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1461233924637200384/artifacts/line-cov.html#file9) | 81.2% | 82.4% | 1.1
[control-plane/pkg/reconciler/channel/channel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1461233924637200384/artifacts/line-cov.html#file10) | 39.8% | 41.5% | 1.7
[control-plane/pkg/reconciler/channel/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1461233924637200384/artifacts/line-cov.html#file11) | 90.0% | 90.9% | 0.9
[control-plane/pkg/reconciler/sink/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1461233924637200384/artifacts/line-cov.html#file12) | 78.9% | 81.0% | 2.0
[control-plane/pkg/reconciler/sink/kafka_sink.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1461233924637200384/artifacts/line-cov.html#file13) | 80.0% | 81.1% | 1.1

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/prober/async_prober.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1461234673718923264/artifacts/line-cov.html#file1) | Do not exist | 90.5% | 
[control-plane/pkg/prober/cache.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1461234673718923264/artifacts/line-cov.html#file2) | 96.9% | 100.0% | 3.1
[control-plane/pkg/prober/prober.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1461234673718923264/artifacts/line-cov.html#file3) | Do not exist | 83.3% | 
[control-plane/pkg/prober/probertesting/mock_prober.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1461234673718923264/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[control-plane/pkg/receiver/address.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1461234673718923264/artifacts/line-cov.html#file5) | Do not exist | 0.0% | 
[control-plane/pkg/reconciler/broker/broker.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1461234673718923264/artifacts/line-cov.html#file8) | 84.3% | 84.8% | 0.6
[control-plane/pkg/reconciler/broker/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1461234673718923264/artifacts/line-cov.html#file9) | 81.2% | 82.4% | 1.1
[control-plane/pkg/reconciler/channel/channel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1461234673718923264/artifacts/line-cov.html#file10) | 39.8% | 41.5% | 1.7
[control-plane/pkg/reconciler/channel/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1461234673718923264/artifacts/line-cov.html#file11) | 90.0% | 90.9% | 0.9
[control-plane/pkg/reconciler/sink/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1461234673718923264/artifacts/line-cov.html#file12) | 78.9% | 81.0% | 2.0
[control-plane/pkg/reconciler/sink/kafka_sink.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1461234673718923264/artifacts/line-cov.html#file13) | 80.0% | 81.1% | 1.1

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 1a4c96fda4cb32d61c48f7dacb4ca2f484d48fe8 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1461234673718923264) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 1a4c96fda4cb32d61c48f7dacb4ca2f484d48fe8 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1461234673718923264) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 1a4c96fda4cb32d61c48f7dacb4ca2f484d48fe8 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1495/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1461234673689563136) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

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

<details>

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

Patch:
diff --git a/control-plane/config/200-controller/500-controller.yaml b/control-plane/config/200-controller/500-controller.yaml
index 404cc1f37..41270a6ce 100644
--- a/control-plane/config/200-controller/500-controller.yaml
+++ b/control-plane/config/200-controller/500-controller.yaml
@@ -98,6 +98,13 @@ spec:
             - name: SOURCE_GENERAL_CONFIG_MAP_NAME
               value: kafka-broker-config
 
+            - name: BROKER_INGRESS_POD_PORT
+              value: "8080"
+            - name: CHANNEL_INGRESS_POD_PORT
+              value: "8080"
+            - name: SINK_INGRESS_POD_PORT
+              value: "8080"
+
             - name: BROKER_SYSTEM_NAMESPACE
               valueFrom:
                 fieldRef:
diff --git a/control-plane/pkg/config/env_config.go b/control-plane/pkg/config/env_config.go
index 29ab75a63..51a587a58 100644
--- a/control-plane/pkg/config/env_config.go
+++ b/control-plane/pkg/config/env_config.go
@@ -27,6 +27,7 @@ type Env struct {
 	DataPlaneConfigMapName      string `required:"true" split_words:"true"`
 	GeneralConfigMapName        string `required:"true" split_words:"true"`
 	IngressName                 string `required:"true" split_words:"true"`
+	IngressPodPort              string `required:"false" split_words:"true"`
 	SystemNamespace             string `required:"true" split_words:"true"`
 	DataPlaneConfigFormat       string `required:"true" split_words:"true"`
 	DefaultBackoffDelayMs       uint64 `required:"false" split_words:"true"`
diff --git a/control-plane/pkg/prober/async_prober.go b/control-plane/pkg/prober/async_prober.go
new file mode 100644
index 000000000..56daf37f5
--- /dev/null
+++ b/control-plane/pkg/prober/async_prober.go
@@ -0,0 +1,143 @@
+/*
+ * Copyright 2021 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 prober
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"go.uber.org/zap"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/types"
+	podinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/pod"
+	"knative.dev/pkg/logging"
+)
+
+type asyncProber struct {
+	client    httpClient
+	enqueue   EnqueueFunc
+	logger    *zap.Logger
+	cache     Cache
+	podLister func() ([]*corev1.Pod, error)
+	port      string
+}
+
+// NewAsync creates an async Prober.
+//
+// It reports status changes using the provided EnqueueFunc.
+func NewAsync(ctx context.Context, client httpClient, port string, podsLabelsSelector labels.Selector, enqueue EnqueueFunc) Prober {
+	logger := logging.FromContext(ctx).Desugar().
+		With(zap.String("scope", "prober"))
+
+	if len(port) == 0 {
+		logger.Fatal("Port is required")
+	}
+	podLister := podinformer.Get(ctx).Lister()
+	return &asyncProber{
+		client:    client,
+		enqueue:   enqueue,
+		logger:    logger,
+		cache:     NewLocalExpiringCache(ctx, 30*time.Minute),
+		podLister: func() ([]*corev1.Pod, error) { return podLister.List(podsLabelsSelector) },
+		port:      port,
+	}
+}
+
+func (a *asyncProber) Probe(ctx context.Context, addressable Addressable) Status {
+	address := addressable.Address
+	pods, err := a.podLister()
+	if err != nil {
+		a.logger.Error("Failed to list pods", zap.Error(err))
+		return StatusUnknown
+	}
+	// Return `StatusNotReady` when there are no pods.
+	if len(pods) == 0 {
+		return StatusNotReady
+	}
+
+	// aggregatedCurrentKnownStatus keeps track of the current status in the cache excluding `StatusReady`
+	// since we just skip pods that have `StatusReady`.
+	//
+	// If there is a status that is `StatusUnknown` the final status  we want to return is `StatusUnknown`,
+	// while we return `StatusNotReady` when the status is known and all probes returned `StatusNotReady`.
+	var aggregatedCurrentKnownStatus *Status
+
+	// wg keeps track of each request probe status.
+	//
+	// It goes to done once we have all probe request results regardless of whether they are coming from
+	// the cache or from an actual request.
+	var wg sync.WaitGroup
+	wg.Add(len(pods))
+
+	// enqueueOnce allows requeuing the resource only once, when we have all probe request results.
+	var enqueueOnce sync.Once
+
+	for _, p := range pods {
+		podUrl := *address
+		podUrl.Host = p.Status.PodIP + ":" + a.port
+		address := podUrl.String()
+
+		logger := a.logger.
+			With(zap.String("pod.metadata.name", p.Name)).
+			With(zap.String("address", address))
+
+		currentStatus := a.cache.GetStatus(address)
+		if currentStatus == StatusReady {
+			logger.Debug("Skip probing", zap.String("status", "ready"))
+			wg.Done()
+			continue
+		}
+		if aggregatedCurrentKnownStatus == nil || *aggregatedCurrentKnownStatus > currentStatus {
+			aggregatedCurrentKnownStatus = &currentStatus
+		}
+
+		resourceKey := addressable.ResourceKey
+
+		// We want to requeue the resource only once when we have all probe request
+		// results.
+		enqueueOnce.Do(func() {
+			// Wait in a separate gorouting.
+			go func() {
+				// Wait for all the prober request results and then requeue the
+				// resource.
+				wg.Wait()
+				a.enqueue(resourceKey)
+			}()
+		})
+
+		go func() {
+			defer wg.Done()
+			// Probe the pod.
+			status := probe(ctx, a.client, logger, address)
+			logger.Debug("Probe status", zap.Stringer("status", status))
+			// Update the status in the cache.
+			a.cache.UpsertStatus(address, status, resourceKey, a.enqueueArg)
+		}()
+	}
+
+	if aggregatedCurrentKnownStatus == nil {
+		// Every status is ready, return ready.
+		return StatusReady
+	}
+	return *aggregatedCurrentKnownStatus
+}
+
+func (a *asyncProber) enqueueArg(_ string, arg interface{}) {
+	a.enqueue(arg.(types.NamespacedName))
+}
diff --git a/control-plane/pkg/prober/cache.go b/control-plane/pkg/prober/cache.go
index e8e28d312..a2a7e16f5 100644
--- a/control-plane/pkg/prober/cache.go
+++ b/control-plane/pkg/prober/cache.go
@@ -35,6 +35,17 @@ const (
 // Status represents the resource status.
 type Status int
 
+func (s Status) String() string {
+	switch s {
+	case StatusReady:
+		return "Ready"
+	case StatusNotReady:
+		return "NotReady"
+	default:
+		return "Unknown"
+	}
+}
+
 // Cache is a key-status store.
 type Cache interface {
 	// GetStatus retries the status associated with the given key.
diff --git a/control-plane/pkg/prober/prober.go b/control-plane/pkg/prober/prober.go
new file mode 100644
index 000000000..59679808b
--- /dev/null
+++ b/control-plane/pkg/prober/prober.go
@@ -0,0 +1,91 @@
+/*
+ * Copyright 2021 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 prober
+
+import (
+	"context"
+	"net/http"
+	"net/url"
+
+	"go.uber.org/zap"
+	"k8s.io/apimachinery/pkg/types"
+	"knative.dev/pkg/network"
+)
+
+// Addressable contains addressable resource data for the prober.
+type Addressable struct {
+	// Addressable address.
+	Address *url.URL
+	// Resource key.
+	ResourceKey types.NamespacedName
+}
+
+// EnqueueFunc enqueues the given provided resource key.
+type EnqueueFunc func(key types.NamespacedName)
+
+// Prober probes an addressable resource.
+type Prober interface {
+	// Probe probes the provided Addressable resource and returns its Status.
+	Probe(ctx context.Context, addressable Addressable) Status
+}
+
+// Func type is an adapter to allow the use of
+// ordinary functions as Prober. If f is a function
+// with the appropriate signature, Func(f) is a
+// Prober that calls f.
+type Func func(ctx context.Context, addressable Addressable) Status
+
+// Probe implements the Prober interface for Func.
+func (p Func) Probe(ctx context.Context, addressable Addressable) Status {
+	return p(ctx, addressable)
+}
+
+// httpClient interface is an interface for an HTTP client.
+type httpClient interface {
+	Do(r *http.Request) (*http.Response, error)
+}
+
+func probe(ctx context.Context, client httpClient, logger *zap.Logger, address string) Status {
+	logger.Debug("Sending probe request")
+
+	r, err := http.NewRequestWithContext(ctx, http.MethodGet, address, nil)
+	if err != nil {
+		logger.Error("Failed to create request", zap.Error(err))
+		return StatusUnknown
+	}
+	r.Header.Add(network.ProbeHeaderName, network.ProbeHeaderValue)
+	r.Header.Add(network.HashHeaderName, "probe")
+
+	select {
+	case <-ctx.Done():
+		return StatusUnknown
+	default:
+	}
+
+	response, err := client.Do(r)
+	if err != nil {
+		logger.Error("Failed probe", zap.Error(err))
+		return StatusUnknown
+	}
+
+	if response.StatusCode != http.StatusOK {
+		logger.Info("Resource not ready", zap.Int("statusCode", response.StatusCode))
+		return StatusNotReady
+	}
+
+	return StatusReady
+}
diff --git a/control-plane/pkg/receiver/address.go b/control-plane/pkg/receiver/address.go
new file mode 100644
index 000000000..dc6966b9f
--- /dev/null
+++ b/control-plane/pkg/receiver/address.go
@@ -0,0 +1,32 @@
+/*
+ * Copyright 2021 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 receiver
+
+import (
+	"fmt"
+	"net/url"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func Address(host string, object metav1.Object) *url.URL {
+	return &url.URL{
+		Scheme: "http",
+		Host:   host,
+		Path:   fmt.Sprintf("/%s/%s", object.GetNamespace(), object.GetName()),
+	}
+}
diff --git a/control-plane/pkg/reconciler/base/receiver_condition_set.go b/control-plane/pkg/reconciler/base/receiver_condition_set.go
index e904d8710..df9794c36 100644
--- a/control-plane/pkg/reconciler/base/receiver_condition_set.go
+++ b/control-plane/pkg/reconciler/base/receiver_condition_set.go
@@ -19,18 +19,19 @@ package base
 
 import (
 	"fmt"
+	"net/url"
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/tools/record"
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
-	"knative.dev/pkg/network"
 	"knative.dev/pkg/reconciler"
 
 	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
 )
 
 const (
@@ -40,6 +41,7 @@ const (
 	ConditionConfigMapUpdated        apis.ConditionType = "ConfigMapUpdated"
 	ConditionConfigParsed            apis.ConditionType = "ConfigParsed"
 	ConditionInitialOffsetsCommitted apis.ConditionType = "InitialOffsetsCommitted"
+	ConditionProbeSucceeded          apis.ConditionType = "ProbeSucceeded"
 )
 
 var IngressConditionSet = apis.NewLivingConditionSet(
@@ -48,6 +50,7 @@ var IngressConditionSet = apis.NewLivingConditionSet(
 	ConditionTopicReady,
 	ConditionConfigMapUpdated,
 	ConditionConfigParsed,
+	ConditionProbeSucceeded,
 )
 
 var EgressConditionSet = apis.NewLivingConditionSet(
@@ -180,20 +183,21 @@ func (manager *StatusConditionManager) TopicReady(topic string) {
 	)
 }
 
-func (manager *StatusConditionManager) Reconciled() reconciler.Event {
-
-	if manager.SetAddress != nil {
-		object := manager.Object
-
-		manager.SetAddress(&apis.URL{
-			Scheme: "http",
-			Host:   network.GetServiceHostname(manager.Env.IngressName, manager.Env.SystemNamespace),
-			Path:   fmt.Sprintf("/%s/%s", object.GetNamespace(), object.GetName()),
-		})
-		object.GetConditionSet().Manage(object.GetStatus()).MarkTrue(ConditionAddressable)
-	}
-
-	return nil
+func (manager *StatusConditionManager) Addressable(address *url.URL) {
+	manager.SetAddress(&apis.URL{
+		Scheme:      address.Scheme,
+		Opaque:      address.Opaque,
+		User:        address.User,
+		Host:        address.Host,
+		Path:        address.Path,
+		RawPath:     address.RawPath,
+		ForceQuery:  address.ForceQuery,
+		RawQuery:    address.RawQuery,
+		Fragment:    address.Fragment,
+		RawFragment: address.RawFragment,
+	})
+	manager.Object.GetConditionSet().Manage(manager.Object.GetStatus()).MarkTrue(ConditionAddressable)
+	manager.ProbesStatusReady()
 }
 
 func (manager *StatusConditionManager) FailedToUpdateDispatcherPodsAnnotation(err error) {
@@ -282,3 +286,15 @@ func (manager *StatusConditionManager) FailedToResolveSink(err error) error {
 	)
 	return fmt.Errorf("failed to resolve sink: %w", err)
 }
+
+func (manager *StatusConditionManager) ProbesStatusNotReady(status prober.Status) {
+	manager.Object.GetConditionSet().Manage(manager.Object.GetStatus()).MarkFalse(
+		ConditionProbeSucceeded,
+		"ProbeStatus",
+		fmt.Sprintf("status: %s", status.String()),
+	)
+}
+
+func (manager *StatusConditionManager) ProbesStatusReady() {
+	manager.Object.GetConditionSet().Manage(manager.Object.GetStatus()).MarkTrue(ConditionProbeSucceeded)
+}
diff --git a/control-plane/pkg/reconciler/base/reconciler.go b/control-plane/pkg/reconciler/base/reconciler.go
index 7b9e4ba42..663002d4a 100644
--- a/control-plane/pkg/reconciler/base/reconciler.go
+++ b/control-plane/pkg/reconciler/base/reconciler.go
@@ -88,7 +88,7 @@ type Reconciler struct {
 }
 
 func (r *Reconciler) IsReceiverRunning() bool {
-	pods, err := r.PodLister.List(r.receiverSelector())
+	pods, err := r.PodLister.List(r.ReceiverSelector())
 	return err == nil && len(pods) > 0 && isAtLeastOneRunning(pods)
 }
 
@@ -215,7 +215,7 @@ func (r *Reconciler) UpdateDispatcherPodsAnnotation(ctx context.Context, logger
 }
 
 func (r *Reconciler) UpdateReceiverPodsAnnotation(ctx context.Context, logger *zap.Logger, volumeGeneration uint64) error {
-	pods, errors := r.PodLister.Pods(r.SystemNamespace).List(r.receiverSelector())
+	pods, errors := r.PodLister.Pods(r.SystemNamespace).List(r.ReceiverSelector())
 	if errors != nil {
 		return fmt.Errorf("failed to list receiver pods in namespace %s: %w", r.SystemNamespace, errors)
 	}
@@ -262,7 +262,7 @@ func (r *Reconciler) updatePodsAnnotation(ctx context.Context, logger *zap.Logge
 	return errors
 }
 
-func (r *Reconciler) receiverSelector() labels.Selector {
+func (r *Reconciler) ReceiverSelector() labels.Selector {
 	return labels.SelectorFromSet(map[string]string{"app": r.ReceiverLabel})
 }
 
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index 3cbe2842d..29b7035e4 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -25,6 +25,7 @@ import (
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
 	corelisters "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/util/retry"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
@@ -37,6 +38,7 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 	coreconfig "knative.dev/eventing-kafka-broker/control-plane/pkg/core/config"
 	kafkalogging "knative.dev/eventing-kafka-broker/control-plane/pkg/logging"
+	"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"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
@@ -65,6 +67,10 @@ type Reconciler struct {
 	NewKafkaClusterAdminClient kafka.NewClusterAdminClientFunc
 
 	BootstrapServers string
+
+	Prober prober.Prober
+
+	IngressHost string
 }
 
 func (r *Reconciler) ReconcileKind(ctx context.Context, broker *eventing.Broker) reconciler.Event {
@@ -221,7 +227,22 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 		logger.Debug("Updated dispatcher pod annotation")
 	}
 
-	return statusConditionManager.Reconciled()
+	address := receiver.Address(r.IngressHost, broker)
+	proberAddressable := prober.Addressable{
+		Address: address,
+		ResourceKey: types.NamespacedName{
+			Namespace: broker.GetNamespace(),
+			Name:      broker.GetName(),
+		},
+	}
+
+	if status := r.Prober.Probe(ctx, proberAddressable); status != prober.StatusReady {
+		statusConditionManager.ProbesStatusNotReady(status)
+		return nil // Object will get re-queued once probe status changes.
+	}
+	statusConditionManager.Addressable(address)
+
+	return nil
 }
 
 func (r *Reconciler) FinalizeKind(ctx context.Context, broker *eventing.Broker) reconciler.Event {
diff --git a/control-plane/pkg/reconciler/broker/controller.go b/control-plane/pkg/reconciler/broker/controller.go
index a292d5898..c3f173d39 100644
--- a/control-plane/pkg/reconciler/broker/controller.go
+++ b/control-plane/pkg/reconciler/broker/controller.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"net/http"
 
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
@@ -31,6 +32,7 @@ import (
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
+	"knative.dev/pkg/network"
 	"knative.dev/pkg/resolver"
 
 	brokerinformer "knative.dev/eventing/pkg/client/injection/informers/eventing/v1/broker"
@@ -40,6 +42,7 @@ import (
 	secretinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/secret"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
 )
@@ -97,6 +100,8 @@ func NewController(ctx context.Context, watcher configmap.Watcher, env *config.E
 	impl := brokerreconciler.NewImpl(ctx, reconciler, kafka.BrokerClass)
 
 	reconciler.Resolver = resolver.NewURIResolverFromTracker(ctx, impl.Tracker)
+	reconciler.Prober = prober.NewAsync(ctx, http.DefaultClient, env.IngressPodPort, reconciler.ReceiverSelector(), impl.EnqueueKey)
+	reconciler.IngressHost = network.GetServiceHostname(env.IngressName, env.SystemNamespace)
 
 	brokerInformer := brokerinformer.Get(ctx)
 
diff --git a/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index 8d05efdd7..ff39bfd90 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -48,6 +48,7 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 	coreconfig "knative.dev/eventing-kafka-broker/control-plane/pkg/core/config"
 	kafkalogging "knative.dev/eventing-kafka-broker/control-plane/pkg/logging"
+	"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"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
@@ -80,6 +81,10 @@ type Reconciler struct {
 	InitOffsetsFunc kafka.InitOffsetsFunc
 
 	ConfigMapLister corelisters.ConfigMapLister
+
+	Prober prober.Prober
+
+	IngressHost string
 }
 
 func (r *Reconciler) ReconcileKind(ctx context.Context, channel *messagingv1beta1.KafkaChannel) reconciler.Event {
@@ -267,7 +272,22 @@ func (r *Reconciler) reconcileKind(ctx context.Context, channel *messagingv1beta
 		return subscriptionError
 	}
 
-	return statusConditionManager.Reconciled()
+	address := receiver.Address(r.IngressHost, channel)
+	proberAddressable := prober.Addressable{
+		Address: address,
+		ResourceKey: types.NamespacedName{
+			Namespace: channel.GetNamespace(),
+			Name:      channel.GetName(),
+		},
+	}
+
+	if status := r.Prober.Probe(ctx, proberAddressable); status != prober.StatusReady {
+		statusConditionManager.ProbesStatusNotReady(status)
+		return nil // Object will get re-queued once probe status changes.
+	}
+	statusConditionManager.Addressable(address)
+
+	return nil
 }
 
 func (r *Reconciler) FinalizeKind(ctx context.Context, channel *messagingv1beta1.KafkaChannel) reconciler.Event {
diff --git a/control-plane/pkg/reconciler/channel/controller.go b/control-plane/pkg/reconciler/channel/controller.go
index 40e8e19c2..8ccfa800b 100644
--- a/control-plane/pkg/reconciler/channel/controller.go
+++ b/control-plane/pkg/reconciler/channel/controller.go
@@ -18,16 +18,17 @@ package channel
 
 import (
 	"context"
+	"net/http"
 
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/tools/cache"
-	"knative.dev/eventing-kafka/pkg/common/kafka/offset"
 
 	messagingv1beta "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	kafkachannelinformer "knative.dev/eventing-kafka/pkg/client/injection/informers/messaging/v1beta1/kafkachannel"
 	kafkachannelreconciler "knative.dev/eventing-kafka/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel"
+	"knative.dev/eventing-kafka/pkg/common/kafka/offset"
 
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	configmapinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap"
@@ -37,9 +38,11 @@ import (
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
+	"knative.dev/pkg/network"
 	"knative.dev/pkg/resolver"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 )
 
@@ -80,6 +83,8 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *conf
 	}
 
 	impl := kafkachannelreconciler.NewImpl(ctx, reconciler)
+	reconciler.Prober = prober.NewAsync(ctx, http.DefaultClient, configs.IngressPodPort, reconciler.ReceiverSelector(), impl.EnqueueKey)
+	reconciler.IngressHost = network.GetServiceHostname(configs.IngressName, configs.SystemNamespace)
 
 	channelInformer := kafkachannelinformer.Get(ctx)
 
diff --git a/control-plane/pkg/reconciler/sink/controller.go b/control-plane/pkg/reconciler/sink/controller.go
index cd83f8dab..967dcf4b6 100644
--- a/control-plane/pkg/reconciler/sink/controller.go
+++ b/control-plane/pkg/reconciler/sink/controller.go
@@ -18,6 +18,7 @@ package sink
 
 import (
 	"context"
+	"net/http"
 
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
@@ -30,12 +31,14 @@ import (
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
+	"knative.dev/pkg/network"
 	"knative.dev/pkg/tracker"
 
 	eventing "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
 	sinkinformer "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/informers/eventing/v1alpha1/kafkasink"
 	sinkreconciler "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/reconciler/eventing/v1alpha1/kafkasink"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 )
 
@@ -72,6 +75,8 @@ func NewController(ctx context.Context, _ configmap.Watcher, configs *config.Env
 	}
 
 	impl := sinkreconciler.NewImpl(ctx, reconciler)
+	reconciler.Prober = prober.NewAsync(ctx, http.DefaultClient, configs.IngressPodPort, reconciler.ReceiverSelector(), impl.EnqueueKey)
+	reconciler.IngressHost = network.GetServiceHostname(configs.IngressName, configs.SystemNamespace)
 
 	sinkInformer := sinkinformer.Get(ctx)
 
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink.go b/control-plane/pkg/reconciler/sink/kafka_sink.go
index 6ed5ac6fb..3609b1506 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink.go
@@ -22,6 +22,7 @@ import (
 
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
+	"k8s.io/apimachinery/pkg/types"
 	corelisters "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/util/retry"
 	"knative.dev/pkg/controller"
@@ -32,6 +33,7 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 	coreconfig "knative.dev/eventing-kafka-broker/control-plane/pkg/core/config"
 	kafkalogging "knative.dev/eventing-kafka-broker/control-plane/pkg/logging"
+	"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"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
@@ -52,6 +54,10 @@ type Reconciler struct {
 	// NewKafkaClusterAdminClient creates new sarama ClusterAdmin. It's convenient to add this as Reconciler field so that we can
 	// mock the function used during the reconciliation loop.
 	NewKafkaClusterAdminClient kafka.NewClusterAdminClientFunc
+
+	Prober prober.Prober
+
+	IngressHost string
 }
 
 func (r *Reconciler) ReconcileKind(ctx context.Context, ks *eventing.KafkaSink) reconciler.Event {
@@ -216,7 +222,22 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *eventing.KafkaSink)
 
 	logger.Debug("Updated receiver pod annotation")
 
-	return statusConditionManager.Reconciled()
+	address := receiver.Address(r.IngressHost, ks)
+	proberAddressable := prober.Addressable{
+		Address: address,
+		ResourceKey: types.NamespacedName{
+			Namespace: ks.GetNamespace(),
+			Name:      ks.GetName(),
+		},
+	}
+
+	if status := r.Prober.Probe(ctx, proberAddressable); status != prober.StatusReady {
+		statusConditionManager.ProbesStatusNotReady(status)
+		return nil // Object will get re-queued once probe status changes.
+	}
+	statusConditionManager.Addressable(address)
+
+	return nil
 }
 
 func (r *Reconciler) FinalizeKind(ctx context.Context, ks *eventing.KafkaSink) reconciler.Event {
diff --git a/control-plane/pkg/reconciler/source/source.go b/control-plane/pkg/reconciler/source/source.go
index 9e406c49c..e3980959c 100644
--- a/control-plane/pkg/reconciler/source/source.go
+++ b/control-plane/pkg/reconciler/source/source.go
@@ -208,7 +208,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *sources.KafkaSource)
 		logger.Debug("Updated dispatcher pod annotation")
 	}
 
-	return statusConditionManager.Reconciled()
+	return nil
 }
 
 func (r *Reconciler) FinalizeKind(ctx context.Context, ks *sources.KafkaSource) reconciler.Event {
diff --git a/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index ecf8db86b..4b215f723 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -115,7 +115,7 @@ Lists of 157 third-party dependencies.
      (Eclipse Public License 1.0) JUnit (junit:junit:4.12 - http://junit.org)
      (Apache License, Version 2.0) Byte Buddy (without dependencies) (net.bytebuddy:byte-buddy:1.11.19 - https://bytebuddy.net/byte-buddy)
      (Apache License, Version 2.0) Byte Buddy agent (net.bytebuddy:byte-buddy-agent:1.11.19 - https://bytebuddy.net/byte-buddy-agent)
-     (Apache License, Version 2.0) (MIT License) Logstash Logback Encoder (net.logstash.logback:logstash-logback-encoder:7.0 - https://github.com/logstash/logstash-logback-encoder)
+     (Apache License, Version 2.0) (MIT License) Logstash Logback Encoder (net.logstash.logback:logstash-logback-encoder:7.0.1 - https://github.com/logfellow/logstash-logback-encoder)
      (The MIT License) JOpt Simple (net.sf.jopt-simple:jopt-simple:4.6 - http://pholser.github.com/jopt-simple)
      (The MIT License) JOpt Simple (net.sf.jopt-simple:jopt-simple:5.0.4 - http://jopt-simple.github.io/jopt-simple)
      (MIT) argparse4j (net.sourceforge.argparse4j:argparse4j:0.7.0 - http://argparse4j.github.io)
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index 1dee78b74..1844eff98 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -49,7 +49,7 @@
     <protobuf.version>3.19.1</protobuf.version>
     <slf4j.version>1.7.32</slf4j.version>
     <ch.qos.logback.version>1.2.7</ch.qos.logback.version>
-    <net.logstash.logback.encoder.version>7.0</net.logstash.logback.encoder.version>
+    <net.logstash.logback.encoder.version>7.0.1</net.logstash.logback.encoder.version>
     <assertj.version>3.21.0</assertj.version>
     <awaitility.version>4.1.1</awaitility.version>
     <junit.jupiter.version>5.8.1</junit.jupiter.version>
diff --git a/go.mod b/go.mod
index 6d60f4a20..af6d812c5 100644
--- a/go.mod
+++ b/go.mod
@@ -13,6 +13,7 @@ require (
 	github.com/rickb777/date v1.14.1
 	github.com/stretchr/testify v1.7.0
 	github.com/xdg-go/scram v1.0.2
+	go.uber.org/atomic v1.9.0
 	go.uber.org/multierr v1.6.0
 	go.uber.org/zap v1.19.1
 	golang.org/x/crypto v0.0.0-20210921155107-089bfa567519 // indirect
@@ -24,9 +25,9 @@ require (
 	k8s.io/apiserver v0.21.4
 	k8s.io/client-go v0.21.4
 	k8s.io/utils v0.0.0-20210111153108-fddb29f9d009
-	knative.dev/eventing v0.27.1-0.20211117135535-e7452e66b37c
-	knative.dev/eventing-kafka v0.27.1-0.20211117082433-3237bedea284
+	knative.dev/eventing v0.27.1-0.20211119140523-4e372e5242a0
+	knative.dev/eventing-kafka v0.27.1-0.20211119140324-957ed8ba00f1
 	knative.dev/hack v0.0.0-20211117134436-69a2295d54ce
-	knative.dev/pkg v0.0.0-20211117133034-22c0eba0cacf
+	knative.dev/pkg v0.0.0-20211120133512-d016976f2567
 	knative.dev/reconciler-test v0.0.0-20211112132636-ae9e2e21972f
 )
diff --git a/go.sum b/go.sum
index 89c99a909..1e2cb5bad 100644
--- a/go.sum
+++ b/go.sum
@@ -1326,23 +1326,21 @@ k8s.io/utils v0.0.0-20201110183641-67b214c5f920/go.mod h1:jPW/WVKK9YHAvNhRxK0md/
 k8s.io/utils v0.0.0-20210111153108-fddb29f9d009 h1:0T5IaWHO3sJTEmCP6mUlBvMukxPKUQWqiI/YuiBNMiQ=
 k8s.io/utils v0.0.0-20210111153108-fddb29f9d009/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 knative.dev/control-protocol v0.0.0-20211115173555-0910714d1183/go.mod h1:h/Ac2W8Db0pgTo4baEfzRhn+MPpxJnkSQTBTZjthwXY=
-knative.dev/eventing v0.27.1-0.20211115154555-d5201e7b4c2e/go.mod h1:eJGGW9Rbrq585P2Ppc0a+nDLfJzGXe/R8eMXcNwUKlI=
-knative.dev/eventing v0.27.1-0.20211117135535-e7452e66b37c h1:m1WFpIweDVlMXm6829Vp5NI2ajSLd9rFlOAZ98RZNo4=
-knative.dev/eventing v0.27.1-0.20211117135535-e7452e66b37c/go.mod h1:wfjYgRZX2jlPewfsSGrX2OuOdwbx+n7jryLEz7Dak8k=
-knative.dev/eventing-kafka v0.27.1-0.20211117082433-3237bedea284 h1:Jm8m7WDYmo7nqOanv+CsY2UPUK1WtFsGwuCnZFjEEcY=
-knative.dev/eventing-kafka v0.27.1-0.20211117082433-3237bedea284/go.mod h1:pewzUwhWpTybd+lBFbu9ntB1xF8wDIbCQ32SAiRcYpM=
+knative.dev/eventing v0.27.1-0.20211119132322-a5081ac5b9bd/go.mod h1:XPbYrmewuiYE3cHHDUJWRkAqvyNJcIMFJEg7iNs9ZFM=
+knative.dev/eventing v0.27.1-0.20211119140523-4e372e5242a0 h1:uklBDn7Ihkd7w1wEYKtkhJDpbiXjjvNsp7IfhQgBz2o=
+knative.dev/eventing v0.27.1-0.20211119140523-4e372e5242a0/go.mod h1:XPbYrmewuiYE3cHHDUJWRkAqvyNJcIMFJEg7iNs9ZFM=
+knative.dev/eventing-kafka v0.27.1-0.20211119140324-957ed8ba00f1 h1:zJcZfHr9JVnjGg/2LIT9p9swfXMoZF5tNd08v6eSKCg=
+knative.dev/eventing-kafka v0.27.1-0.20211119140324-957ed8ba00f1/go.mod h1:/WtPSbXnaWJs/sm5aeiAEd8lrt4b/i5yr2xSXY+gUq0=
 knative.dev/hack v0.0.0-20211101195839-11d193bf617b/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/hack v0.0.0-20211108170701-96aac1c30be3/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack v0.0.0-20211112192837-128cf0150a69/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack v0.0.0-20211117134436-69a2295d54ce h1:dPARWsX/9nmWg9g09MtGnsn7nZUwjA0UWpKSUozgS6g=
 knative.dev/hack v0.0.0-20211117134436-69a2295d54ce/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/hack/schema v0.0.0-20211112192837-128cf0150a69/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
+knative.dev/hack/schema v0.0.0-20211117134436-69a2295d54ce/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
 knative.dev/pkg v0.0.0-20211101212339-96c0204a70dc/go.mod h1:SkfDk9bWIiNZD7XtILGkG7AKVyF/M6M0bGxLgl0SYL8=
-knative.dev/pkg v0.0.0-20211111114938-0b0c3390a475/go.mod h1:plYIuKk6WNC9LyiG/9xBnOBYuxO0Ro6xN0z7iXWVUtk=
 knative.dev/pkg v0.0.0-20211115071955-517ef0292b53/go.mod h1:VqUp1KWJqpTDNoiSI/heaX3uMdubImslJE2tBkP+Bbw=
-knative.dev/pkg v0.0.0-20211117100634-54071addcdb8/go.mod h1:VqUp1KWJqpTDNoiSI/heaX3uMdubImslJE2tBkP+Bbw=
-knative.dev/pkg v0.0.0-20211117133034-22c0eba0cacf h1:8hLmIbMv9lvpLwjuaVH/rEJ9fKrThEWViyzYZcBPNNc=
-knative.dev/pkg v0.0.0-20211117133034-22c0eba0cacf/go.mod h1:VqUp1KWJqpTDNoiSI/heaX3uMdubImslJE2tBkP+Bbw=
+knative.dev/pkg v0.0.0-20211117215328-5708c4c44232/go.mod h1:VqUp1KWJqpTDNoiSI/heaX3uMdubImslJE2tBkP+Bbw=
+knative.dev/pkg v0.0.0-20211120133512-d016976f2567 h1:IPQ6QKOvMIRX9/UI1cAtSl36+CbGv8HErSvID03CRr4=
+knative.dev/pkg v0.0.0-20211120133512-d016976f2567/go.mod h1:VqUp1KWJqpTDNoiSI/heaX3uMdubImslJE2tBkP+Bbw=
 knative.dev/reconciler-test v0.0.0-20211112132636-ae9e2e21972f h1:bSdsp5av09LqSF4iXo8LCCDEENJ+JEbzzdgHBoJdf2w=
 knative.dev/reconciler-test v0.0.0-20211112132636-ae9e2e21972f/go.mod h1:gTsbLk496j/M9xqMpx/liyCQ0X3bwDpRtcs2Zzws364=
 modernc.org/cc v1.0.0/go.mod h1:1Sk4//wdnYJiUIxnW8ddKpaOJCF37yAdqYnkxUpaYxw=
diff --git a/vendor/knative.dev/pkg/apis/duck/v1/status_types.go b/vendor/knative.dev/pkg/apis/duck/v1/status_types.go
index bf740e065..15e264534 100644
--- a/vendor/knative.dev/pkg/apis/duck/v1/status_types.go
+++ b/vendor/knative.dev/pkg/apis/duck/v1/status_types.go
@@ -21,7 +21,7 @@ import (
 
 	"knative.dev/pkg/apis"
 	"knative.dev/pkg/apis/duck/ducktypes"
-	"knative.dev/pkg/kmeta"
+	"knative.dev/pkg/kmap"
 )
 
 // +genduck
@@ -92,8 +92,7 @@ func (s *Status) GetCondition(t apis.ConditionType) *apis.Condition {
 func (s *Status) ConvertTo(ctx context.Context, sink *Status, predicates ...func(apis.ConditionType) bool) {
 	sink.ObservedGeneration = s.ObservedGeneration
 	if s.Annotations != nil {
-		// This will deep copy the map.
-		sink.Annotations = kmeta.UnionMaps(s.Annotations)
+		sink.Annotations = kmap.Union(s.Annotations)
 	}
 
 	conditions := make(apis.Conditions, 0, len(s.Conditions))
diff --git a/vendor/knative.dev/pkg/apis/duck/v1beta1/status_types.go b/vendor/knative.dev/pkg/apis/duck/v1beta1/status_types.go
index 722902676..e1ae875eb 100644
--- a/vendor/knative.dev/pkg/apis/duck/v1beta1/status_types.go
+++ b/vendor/knative.dev/pkg/apis/duck/v1beta1/status_types.go
@@ -26,7 +26,7 @@ import (
 
 	"knative.dev/pkg/apis"
 	"knative.dev/pkg/apis/duck/ducktypes"
-	"knative.dev/pkg/kmeta"
+	"knative.dev/pkg/kmap"
 )
 
 // +genduck
@@ -109,8 +109,7 @@ func (s *Status) GetCondition(t apis.ConditionType) *apis.Condition {
 func (s *Status) ConvertTo(ctx context.Context, sink *Status) {
 	sink.ObservedGeneration = s.ObservedGeneration
 	if s.Annotations != nil {
-		// This will deep copy the map.
-		sink.Annotations = kmeta.UnionMaps(s.Annotations)
+		sink.Annotations = kmap.Copy(s.Annotations)
 	}
 	for _, c := range s.Conditions {
 		switch c.Type {
diff --git a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/comment_parser.go b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/comment_parser.go
index ada0dff3e..68ac8b4cb 100644
--- a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/comment_parser.go
+++ b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/comment_parser.go
@@ -19,25 +19,41 @@ import "strings"
 
 // Adapted from the k8s.io comment parser https://github.com/kubernetes/gengo/blob/master/types/comments.go
 
+// CommentsTags maps marker prefixes to a set of tags containing keys and values
+type CommentTags map[string]CommentTag
+
+// CommentTags maps keys to a list of values
+type CommentTag map[string][]string
+
 // ExtractCommentTags parses comments for lines of the form:
 //
-//   'marker' + ':' "key=value,key2=value2".
+//   "marker" + "prefix" + ':' + "key=value,key2=value2".
 //
-// Values are optional; empty map is the default.  A tag can be specified more than
+//  In the following example the marker is '+' and the prefix is 'foo':
+//   +foo:key=value1,key2=value2,key=value3
+//
+// Values are optional; empty map is the default. A tag can be specified more than
 // one time and all values are returned.  If the resulting map has an entry for
 // a key, the value (a slice) is guaranteed to have at least 1 element.
 //
 // Example: if you pass "+" for 'marker', and the following lines are in
 // the comments:
-//   +foo:key=value1,key2=value2
+//   +foo:key=value1,key2=value2,key=value3
 //   +bar
 //
 // Then this function will return:
-//   map[string]map[string]string{"foo":{"key":value1","key2":"value2"}, "bar": nil}
+//   map[string]map[string]string{
+//     "foo":{
+//      "key":  []string{"value1", "value3"},
+//      "key2": []string{"value2"}
+//     },
+//     "bar": {},
+//  }
 //
 // Users are not expected to repeat values.
-func ExtractCommentTags(marker string, lines []string) map[string]map[string]string {
-	out := map[string]map[string]string{}
+func ExtractCommentTags(marker string, lines []string) CommentTags {
+	out := CommentTags{}
+
 	for _, line := range lines {
 		line = strings.TrimSpace(line)
 		if len(line) == 0 || !strings.HasPrefix(line, marker) {
@@ -45,29 +61,32 @@ func ExtractCommentTags(marker string, lines []string) map[string]map[string]str
 		}
 
 		options := strings.SplitN(line[len(marker):], ":", 2)
+		prefix := options[0]
+
 		if len(options) == 2 {
 			vals := strings.Split(options[1], ",")
 
-			opts := out[options[0]]
+			opts := out[prefix]
 			if opts == nil {
-				opts = make(map[string]string, len(vals))
+				opts = make(CommentTag, len(vals))
+				out[prefix] = opts
 			}
 
 			for _, pair := range vals {
-				if kv := strings.SplitN(pair, "=", 2); len(kv) == 2 {
-					opts[kv[0]] = kv[1]
-				} else if kv[0] != "" {
-					opts[kv[0]] = ""
+				kv := strings.SplitN(pair, "=", 2)
+				if len(kv) == 1 && kv[0] == "" {
+					continue
+				}
+				if _, ok := opts[kv[0]]; !ok {
+					opts[kv[0]] = []string{}
+				}
+				if len(kv) == 2 {
+					opts[kv[0]] = append(opts[kv[0]], kv[1])
 				}
-			}
-			if len(opts) == 0 {
-				out[options[0]] = nil
-			} else {
-				out[options[0]] = opts
 			}
 		} else if len(options) == 1 && options[0] != "" {
-			if _, has := out[options[0]]; !has {
-				out[options[0]] = nil
+			if _, has := out[prefix]; !has {
+				out[prefix] = CommentTag{}
 			}
 		}
 	}
diff --git a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/packages.go b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/packages.go
index 1900802fa..f79339824 100644
--- a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/packages.go
+++ b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/packages.go
@@ -193,29 +193,36 @@ func MustParseClientGenTags(lines []string) Tags {
 	return ret
 }
 
-func extractCommentTags(t *types.Type) map[string]map[string]string {
+func extractCommentTags(t *types.Type) CommentTags {
 	comments := append(append([]string{}, t.SecondClosestCommentLines...), t.CommentLines...)
 	return ExtractCommentTags("+", comments)
 }
 
-func extractReconcilerClassTag(tags map[string]map[string]string) (string, bool) {
+func extractReconcilerClassesTag(tags CommentTags) ([]string, bool) {
 	vals, ok := tags["genreconciler"]
 	if !ok {
-		return "", false
+		return nil, false
 	}
-	classname, has := vals["class"]
-	return classname, has
+	classnames, has := vals["class"]
+	if has && len(classnames) == 0 {
+		return nil, false
+	}
+	return classnames, has
 }
 
-func isKRShaped(tags map[string]map[string]string) bool {
+func isKRShaped(tags CommentTags) bool {
 	vals, has := tags["genreconciler"]
 	if !has {
 		return false
 	}
-	return vals["krshapedlogic"] != "false"
+	stringVals, has := vals["krshapedlogic"]
+	if !has || len(vals) == 0 {
+		return true // Default is true
+	}
+	return stringVals[0] != "false"
 }
 
-func isNonNamespaced(tags map[string]map[string]string) bool {
+func isNonNamespaced(tags CommentTags) bool {
 	vals, has := tags["genclient"]
 	if !has {
 		return false
@@ -224,7 +231,7 @@ func isNonNamespaced(tags map[string]map[string]string) bool {
 	return has
 }
 
-func stubs(tags map[string]map[string]string) bool {
+func stubs(tags CommentTags) bool {
 	vals, has := tags["genreconciler"]
 	if !has {
 		return false
@@ -545,7 +552,7 @@ func reconcilerPackages(basePackage string, groupPkgName string, gv clientgentyp
 		// Fix for golang iterator bug.
 		t := t
 		extracted := extractCommentTags(t)
-		reconcilerClass, hasReconcilerClass := extractReconcilerClassTag(extracted)
+		reconcilerClasses, hasReconcilerClass := extractReconcilerClassesTag(extracted)
 		nonNamespaced := isNonNamespaced(extracted)
 		isKRShaped := isKRShaped(extracted)
 		stubs := stubs(extracted)
@@ -573,7 +580,7 @@ func reconcilerPackages(basePackage string, groupPkgName string, gv clientgentyp
 					clientPkg:           clientPackagePath,
 					informerPackagePath: informerPackagePath,
 					schemePkg:           filepath.Join(customArgs.VersionedClientSetPackage, "scheme"),
-					reconcilerClass:     reconcilerClass,
+					reconcilerClasses:   reconcilerClasses,
 					hasReconcilerClass:  hasReconcilerClass,
 					hasStatus:           hasStatus(t),
 				})
@@ -602,7 +609,7 @@ func reconcilerPackages(basePackage string, groupPkgName string, gv clientgentyp
 						outputPackage:       filepath.Join(packagePath, "stub"),
 						imports:             generator.NewImportTracker(),
 						informerPackagePath: informerPackagePath,
-						reconcilerClass:     reconcilerClass,
+						reconcilerClasses:   reconcilerClasses,
 						hasReconcilerClass:  hasReconcilerClass,
 					})
 					return generators
@@ -633,7 +640,7 @@ func reconcilerPackages(basePackage string, groupPkgName string, gv clientgentyp
 					listerPkg:          listerPackagePath,
 					groupGoName:        groupGoName,
 					groupVersion:       gv,
-					reconcilerClass:    reconcilerClass,
+					reconcilerClasses:  reconcilerClasses,
 					hasReconcilerClass: hasReconcilerClass,
 					nonNamespaced:      nonNamespaced,
 					isKRShaped:         isKRShaped,
diff --git a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_controller.go b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_controller.go
index a624275a5..63556dec6 100644
--- a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_controller.go
+++ b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_controller.go
@@ -38,7 +38,7 @@ type reconcilerControllerGenerator struct {
 	schemePkg           string
 	informerPackagePath string
 
-	reconcilerClass    string
+	reconcilerClasses  []string
 	hasReconcilerClass bool
 	hasStatus          bool
 }
@@ -69,7 +69,7 @@ func (g *reconcilerControllerGenerator) GenerateType(c *generator.Context, t *ty
 	m := map[string]interface{}{
 		"type":      t,
 		"group":     g.groupName,
-		"class":     g.reconcilerClass,
+		"classes":   g.reconcilerClasses,
 		"hasClass":  g.hasReconcilerClass,
 		"hasStatus": g.hasStatus,
 		"controllerImpl": c.Universe.Type(types.Name{
@@ -195,12 +195,24 @@ var reconcilerControllerNewImpl = `
 const (
 	defaultControllerAgentName = "{{.type|lowercaseSingular}}-controller"
 	defaultFinalizerName       = "{{.type|allLowercasePlural}}.{{.group}}"
-	{{if .hasClass}}
+	{{if .hasClass }}
 	// ClassAnnotationKey points to the annotation for the class of this resource.
-	ClassAnnotationKey = "{{ .class }}"
+	{{if gt (.classes | len) 1 }}// Deprecated: Use ClassAnnotationKeys given multiple keys exist
+	{{end -}}
+	ClassAnnotationKey = "{{ index .classes 0 }}"
 	{{end}}
 )
 
+{{if gt (.classes | len) 1 }}
+var (
+	// ClassAnnotationKeys points to the annotation for the class of this resource.
+	ClassAnnotationKeys = []string{
+		{{range $class := .classes}}"{{$class}}",
+		{{end}}
+	}
+)
+{{end}}
+
 // NewImpl returns a {{.controllerImpl|raw}} that handles queuing and feeding work from
 // the queue through an implementation of {{.controllerReconciler|raw}}, delegating to
 // the provided Interface and optional Finalizer methods. OptionsFn is used to return
diff --git a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_controller_stub.go b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_controller_stub.go
index 37db246c3..13dff4d51 100644
--- a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_controller_stub.go
+++ b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_controller_stub.go
@@ -35,7 +35,7 @@ type reconcilerControllerStubGenerator struct {
 
 	reconcilerPkg       string
 	informerPackagePath string
-	reconcilerClass     string
+	reconcilerClasses   []string
 	hasReconcilerClass  bool
 }
 
@@ -64,7 +64,7 @@ func (g *reconcilerControllerStubGenerator) GenerateType(c *generator.Context, t
 
 	m := map[string]interface{}{
 		"type":     t,
-		"class":    g.reconcilerClass,
+		"classes":  g.reconcilerClasses,
 		"hasClass": g.hasReconcilerClass,
 		"informerGet": c.Universe.Function(types.Name{
 			Package: g.informerPackagePath,
@@ -91,10 +91,18 @@ func (g *reconcilerControllerStubGenerator) GenerateType(c *generator.Context, t
 			Package: g.reconcilerPkg,
 			Name:    "ClassAnnotationKey",
 		}),
+		"classAnnotationKeys": c.Universe.Variable(types.Name{
+			Package: g.reconcilerPkg,
+			Name:    "ClassAnnotationKeys",
+		}),
 		"annotationFilterFunc": c.Universe.Function(types.Name{
 			Package: "knative.dev/pkg/reconciler",
 			Name:    "AnnotationFilterFunc",
 		}),
+		"orFilterFunc": c.Universe.Function(types.Name{
+			Package: "knative.dev/pkg/reconciler",
+			Name:    "Or",
+		}),
 		"filterHandler": c.Universe.Type(types.Name{
 			Package: "k8s.io/client-go/tools/cache",
 			Name:    "FilteringResourceEventHandler",
@@ -120,7 +128,17 @@ func NewController(
 
 	{{if .hasClass}}
 	classValue := "default" // TODO: update this to the appropriate value.
+	{{if len .classes | eq 1 }}
 	classFilter := {{.annotationFilterFunc|raw}}({{.classAnnotationKey|raw}}, classValue, false /*allowUnset*/)
+	{{else if gt (len .classes) 1 }}
+	filterFuncs := make([]func(interface{}) bool, 0, len({{.classAnnotationKeys|raw}}))
+
+	for _, key := range {{.classAnnotationKeys|raw}} {
+		filterFuncs = append(filterFuncs, {{.annotationFilterFunc|raw}}(key, classValue, false /*allowUnset*/))
+	}
+
+	classFilter := {{.orFilterFunc|raw}}(filterFuncs...)
+	{{end}}
 	{{end}}
 
 	// TODO: setup additional informers here.
diff --git a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_reconciler.go b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_reconciler.go
index e787096b3..5e3f23faf 100644
--- a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_reconciler.go
+++ b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_reconciler.go
@@ -37,7 +37,7 @@ type reconcilerReconcilerGenerator struct {
 	listerName     string
 	listerPkg      string
 
-	reconcilerClass    string
+	reconcilerClasses  []string
 	hasReconcilerClass bool
 	nonNamespaced      bool
 	isKRShaped         bool
@@ -74,7 +74,7 @@ func (g *reconcilerReconcilerGenerator) GenerateType(c *generator.Context, t *ty
 		"type":          t,
 		"group":         namer.IC(g.groupGoName),
 		"version":       namer.IC(g.groupVersion.Version.String()),
-		"class":         g.reconcilerClass,
+		"classes":       g.reconcilerClasses,
 		"hasClass":      g.hasReconcilerClass,
 		"isKRShaped":    g.isKRShaped,
 		"hasStatus":     g.hasStatus,
@@ -219,6 +219,9 @@ func (g *reconcilerReconcilerGenerator) GenerateType(c *generator.Context, t *ty
 	sw.Do(reconcilerInterfaceFactory, m)
 	sw.Do(reconcilerNewReconciler, m)
 	sw.Do(reconcilerImplFactory, m)
+	if len(g.reconcilerClasses) > 1 {
+		sw.Do(reconcilerLookupClass, m)
+	}
 	if g.hasStatus {
 		sw.Do(reconcilerStatusFactory, m)
 	}
@@ -227,6 +230,17 @@ func (g *reconcilerReconcilerGenerator) GenerateType(c *generator.Context, t *ty
 	return sw.Error()
 }
 
+var reconcilerLookupClass = `
+func lookupClass(annotations map[string]string) (string, bool) {
+	for _, key := range ClassAnnotationKeys {
+		 if val, ok := annotations[key]; ok {
+		   return val, true
+		 }
+	}
+	return "", false
+}
+`
+
 var reconcilerInterfaceFactory = `
 // Interface defines the strongly typed interfaces to be implemented by a
 // controller reconciling {{.type|raw}}.
@@ -293,8 +307,15 @@ type reconcilerImpl struct {
 	skipStatusUpdates bool
 	{{end}}
 
-	{{if .hasClass}}
-	// classValue is the resource annotation[{{ .class }}] instance value this reconciler instance filters on.
+	{{if len .classes | eq 1 }}
+	// classValue is the resource annotation[{{ index .classes 0 }}] instance value this reconciler instance filters on.
+	classValue string
+	{{else if gt (len .classes) 1 }}
+	// classValue is the resource annotation instance value this reconciler instance filters on.
+	// The annotations key are:
+	{{- range $class := .classes}}
+	//   {{$class}}
+	{{- end}}
 	classValue string
 	{{end}}
 }
@@ -416,14 +437,22 @@ func (r *reconcilerImpl) Reconcile(ctx {{.contextContext|raw}}, key string) erro
 	} else if err != nil {
 		return err
 	}
-	{{if .hasClass}}
+
+{{if len .classes | eq 1 }}
 	if classValue, found := original.GetAnnotations()[ClassAnnotationKey]; !found || classValue != r.classValue {
 		logger.Debugw("Skip reconciling resource, class annotation value does not match reconciler instance value.",
 			zap.String("classKey", ClassAnnotationKey),
 			zap.String("issue", classValue+"!="+r.classValue))
 		return nil
 	}
-	{{end}}
+{{else if gt (len .classes) 1 }}
+	if classValue, found := lookupClass(original.GetAnnotations()); !found || classValue != r.classValue {
+		logger.Debugw("Skip reconciling resource, class annotation value does not match reconciler instance value.",
+			zap.Strings("classKeys", ClassAnnotationKeys),
+			zap.String("issue", classValue+"!="+r.classValue))
+		return nil
+	}
+{{end}}
 
 	// Don't modify the informers copy.
 	resource := original.DeepCopy()
diff --git a/vendor/knative.dev/pkg/kmap/lookup.go b/vendor/knative.dev/pkg/kmap/lookup.go
new file mode 100644
index 000000000..e24041b6e
--- /dev/null
+++ b/vendor/knative.dev/pkg/kmap/lookup.go
@@ -0,0 +1,77 @@
+/*
+Copyright 2021 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 kmap
+
+// KeyPriority is a utility struct for getting values from a map
+// given a list of ordered keys
+//
+// This is to help the migration/renaming of annotations & labels
+type KeyPriority []string
+
+// Key returns the default key that should be used for
+// accessing the map
+func (p KeyPriority) Key() string {
+	// this intentionally panics rather than returning an empty string
+	return p[0]
+}
+
+// Value iterates looks up the ordered keys in the map and returns
+// a string value. An empty string will be returned if the keys
+// are not present in the map
+func (p KeyPriority) Value(m map[string]string) string {
+	_, v, _ := p.Get(m)
+	return v
+}
+
+// Get iterates over the ordered keys and looks up the corresponding
+// values in the map
+//
+// It returns the key, value, and true|false signaling whether the
+// key was present in the map
+//
+// If no key is present the default key (lowest ordinal) is returned
+// with an empty string as the value
+func (p KeyPriority) Get(m map[string]string) (string, string, bool) {
+	var k, v string
+	var ok bool
+	for _, k = range p {
+		v, ok = m[k]
+		if ok {
+			return k, v, ok
+		}
+	}
+
+	return p.Key(), "", false
+}
+
+// UpdateKey will update the map with the KeyPriority's default
+// key iff any of the other synonym keys are present
+func (p KeyPriority) UpdateKey(m map[string]string) {
+	if k, v, ok := p.Get(m); ok && k != p.Key() {
+		delete(m, k)
+		m[p.Key()] = v
+	}
+}
+
+// UpdateKeys iterates over the lookups and updates entries in the map
+// to use the default key
+func UpdateKeys(m map[string]string, keys ...KeyPriority) map[string]string {
+	for _, key := range keys {
+		key.UpdateKey(m)
+	}
+	return m
+}
diff --git a/vendor/knative.dev/pkg/kmap/map.go b/vendor/knative.dev/pkg/kmap/map.go
new file mode 100644
index 000000000..006348d76
--- /dev/null
+++ b/vendor/knative.dev/pkg/kmap/map.go
@@ -0,0 +1,80 @@
+/*
+Copyright 2021 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 kmap
+
+// Copy makes a copy of the map.
+func Copy(a map[string]string) map[string]string {
+	ret := make(map[string]string, len(a))
+	for k, v := range a {
+		ret[k] = v
+	}
+	return ret
+}
+
+// Union returns a map constructed from the union of input maps.
+// where values from latter maps win.
+func Union(maps ...map[string]string) map[string]string {
+	if len(maps) == 0 {
+		return map[string]string{}
+	}
+	out := make(map[string]string, len(maps[0]))
+
+	for _, m := range maps {
+		for k, v := range m {
+			out[k] = v
+		}
+	}
+	return out
+}
+
+// Filter creates a copy of the provided map, filtering out the elements
+// that match `filter`.
+// nil `filter` is accepted.
+func Filter(in map[string]string, filter func(string) bool) map[string]string {
+	ret := make(map[string]string, len(in))
+	for k, v := range in {
+		if filter != nil && filter(k) {
+			continue
+		}
+		ret[k] = v
+	}
+	return ret
+}
+
+// ExcludeKeys creates a copy of the provided map filtering out the excluded `keys`
+func ExcludeKeys(in map[string]string, keys ...string) map[string]string {
+	return ExcludeKeyList(in, keys)
+}
+
+// ExcludeKeyList creates a copy of the provided map filtering out excluded `keys`
+func ExcludeKeyList(in map[string]string, keys []string) map[string]string {
+	ret := make(map[string]string, len(in))
+
+outer:
+	for k, v := range in {
+		// opted to skip memory allocation (creating a set) in favour of
+		// looping since the places Knative will use this we typically
+		// exclude one or two keys
+		for _, excluded := range keys {
+			if k == excluded {
+				continue outer
+			}
+		}
+		ret[k] = v
+	}
+	return ret
+}
diff --git a/vendor/knative.dev/pkg/kmeta/map.go b/vendor/knative.dev/pkg/kmeta/map.go
index 37154cd28..475923320 100644
--- a/vendor/knative.dev/pkg/kmeta/map.go
+++ b/vendor/knative.dev/pkg/kmeta/map.go
@@ -16,41 +16,19 @@ limitations under the License.
 
 package kmeta
 
+import "knative.dev/pkg/kmap"
+
 // CopyMap makes a copy of the map.
-func CopyMap(a map[string]string) map[string]string {
-	ret := make(map[string]string, len(a))
-	for k, v := range a {
-		ret[k] = v
-	}
-	return ret
-}
+// Deprecated: use kmap.Copy
+var CopyMap = kmap.Copy
 
 // UnionMaps returns a map constructed from the union of input maps.
 // where values from latter maps win.
-func UnionMaps(maps ...map[string]string) map[string]string {
-	if len(maps) == 0 {
-		return map[string]string{}
-	}
-	out := make(map[string]string, len(maps[0]))
-
-	for _, m := range maps {
-		for k, v := range m {
-			out[k] = v
-		}
-	}
-	return out
-}
+// Deprecated: use kmap.Union
+var UnionMaps = kmap.Union
 
 // FilterMap creates a copy of the provided map, filtering out the elements
 // that match `filter`.
 // nil `filter` is accepted.
-func FilterMap(in map[string]string, filter func(string) bool) map[string]string {
-	ret := make(map[string]string, len(in))
-	for k, v := range in {
-		if filter != nil && filter(k) {
-			continue
-		}
-		ret[k] = v
-	}
-	return ret
-}
+// Deprecated: use kmap.Filter
+var FilterMap = kmap.Filter
diff --git a/vendor/knative.dev/pkg/reconciler/filter.go b/vendor/knative.dev/pkg/reconciler/filter.go
index 36c86c4d3..9f7d2fa36 100644
--- a/vendor/knative.dev/pkg/reconciler/filter.go
+++ b/vendor/knative.dev/pkg/reconciler/filter.go
@@ -77,6 +77,18 @@ func Not(f func(interface{}) bool) func(interface{}) bool {
 	}
 }
 
+// Or creates a FilterFunc which performs an OR of the passed in FilterFuncs
+func Or(funcs ...func(interface{}) bool) func(interface{}) bool {
+	return func(obj interface{}) bool {
+		for _, f := range funcs {
+			if f(obj) {
+				return true
+			}
+		}
+		return false
+	}
+}
+
 // ChainFilterFuncs creates a FilterFunc which performs an AND of the passed FilterFuncs.
 func ChainFilterFuncs(funcs ...func(interface{}) bool) func(interface{}) bool {
 	return func(obj interface{}) bool {
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 1572e8802..b29d0119a 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -382,6 +382,7 @@ go.opentelemetry.io/otel/label
 go.opentelemetry.io/otel/propagation
 go.opentelemetry.io/otel/trace
 # go.uber.org/atomic v1.9.0
+## explicit
 go.uber.org/atomic
 # go.uber.org/automaxprocs v1.4.0
 go.uber.org/automaxprocs
@@ -1102,7 +1103,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.27.1-0.20211117135535-e7452e66b37c
+# knative.dev/eventing v0.27.1-0.20211119140523-4e372e5242a0
 ## explicit
 knative.dev/eventing/pkg/apis/config
 knative.dev/eventing/pkg/apis/duck
@@ -1211,7 +1212,7 @@ knative.dev/eventing/test/test_images/performance
 knative.dev/eventing/test/test_images/print
 knative.dev/eventing/test/test_images/recordevents
 knative.dev/eventing/test/test_images/request-sender
-# knative.dev/eventing-kafka v0.27.1-0.20211117082433-3237bedea284
+# knative.dev/eventing-kafka v0.27.1-0.20211119140324-957ed8ba00f1
 ## explicit
 knative.dev/eventing-kafka/pkg/apis/bindings
 knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1
@@ -1268,7 +1269,7 @@ knative.dev/eventing-kafka/test/lib/resources
 # knative.dev/hack v0.0.0-20211117134436-69a2295d54ce
 ## explicit
 knative.dev/hack
-# knative.dev/pkg v0.0.0-20211117133034-22c0eba0cacf
+# knative.dev/pkg v0.0.0-20211120133512-d016976f2567
 ## explicit
 knative.dev/pkg/apis
 knative.dev/pkg/apis/duck
@@ -1311,6 +1312,7 @@ knative.dev/pkg/injection/clients/namespacedkube/informers/core/v1/secret
 knative.dev/pkg/injection/clients/namespacedkube/informers/factory
 knative.dev/pkg/injection/sharedmain
 knative.dev/pkg/kflag
+knative.dev/pkg/kmap
 knative.dev/pkg/kmeta
 knative.dev/pkg/kmp
 knative.dev/pkg/leaderelection


Test patch:
diff --git a/control-plane/pkg/prober/async_prober_test.go b/control-plane/pkg/prober/async_prober_test.go
new file mode 100644
index 000000000..e27b94c56
--- /dev/null
+++ b/control-plane/pkg/prober/async_prober_test.go
@@ -0,0 +1,147 @@
+/*
+ * Copyright 2021 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 prober
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"go.uber.org/atomic"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/types"
+	podinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake"
+	"knative.dev/pkg/network"
+	reconcilertesting "knative.dev/pkg/reconciler/testing"
+)
+
+func TestAsyncProber(t *testing.T) {
+
+	tt := []struct {
+		name                string
+		pods                []*corev1.Pod
+		podsLabelsSelector  labels.Selector
+		addressable         Addressable
+		responseStatusCode  int
+		wantStatus          Status
+		wantRequeueCountMin int
+		wantRequestCountMin int
+	}{
+		{
+			name:               "no pods",
+			pods:               []*corev1.Pod{},
+			podsLabelsSelector: labels.SelectorFromSet(map[string]string{"app": "p"}),
+			addressable: Addressable{
+				Address:     &url.URL{Scheme: "http", Path: "/b1/b1"},
+				ResourceKey: types.NamespacedName{Namespace: "b1", Name: "b1"},
+			},
+			responseStatusCode:  http.StatusOK,
+			wantStatus:          StatusNotReady,
+			wantRequeueCountMin: 0,
+		},
+		{
+			name: "single pod",
+			pods: []*corev1.Pod{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "ns",
+						Name:      "p1",
+						Labels:    map[string]string{"app": "p"},
+					},
+					Status: corev1.PodStatus{PodIP: "127.0.0.1"},
+				},
+			},
+			podsLabelsSelector: labels.SelectorFromSet(map[string]string{"app": "p"}),
+			addressable: Addressable{
+				Address:     &url.URL{Scheme: "http", Path: "/b1/b1"},
+				ResourceKey: types.NamespacedName{Namespace: "b1", Name: "b1"},
+			},
+			responseStatusCode:  http.StatusOK,
+			wantStatus:          StatusReady,
+			wantRequeueCountMin: 1,
+			wantRequestCountMin: 1,
+		},
+		{
+			name: "single pod - 404",
+			pods: []*corev1.Pod{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "ns",
+						Name:      "p1",
+						Labels:    map[string]string{"app": "p"},
+					},
+					Status: corev1.PodStatus{PodIP: "127.0.0.1"},
+				},
+			},
+			podsLabelsSelector: labels.SelectorFromSet(map[string]string{"app": "p"}),
+			addressable: Addressable{
+				Address:     &url.URL{Scheme: "http", Path: "/b1/b1"},
+				ResourceKey: types.NamespacedName{Namespace: "b1", Name: "b1"},
+			},
+			responseStatusCode:  http.StatusNotFound,
+			wantStatus:          StatusNotReady,
+			wantRequeueCountMin: 1,
+			wantRequestCountMin: 1,
+		},
+	}
+
+	for _, tc := range tt {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx, _ := reconcilertesting.SetupFakeContext(t)
+			ctx, cacel := context.WithCancel(ctx)
+			defer func() {
+				time.Sleep(time.Second)
+				cacel()
+			}()
+			wantRequestCountMin := atomic.NewInt64(int64(tc.wantRequestCountMin))
+			h := http.HandlerFunc(func(writer http.ResponseWriter, r *http.Request) {
+				wantRequestCountMin.Dec()
+				require.Equal(t, network.ProbeHeaderValue, r.Header.Get(network.ProbeHeaderName))
+				writer.WriteHeader(tc.responseStatusCode)
+			})
+			s := httptest.NewUnstartedServer(h)
+			s.Start()
+			time.Sleep(time.Second) // Wait for the server time to start.
+			defer s.Close()
+
+			for _, p := range tc.pods {
+				_ = podinformer.Get(ctx).Informer().GetStore().Add(p)
+			}
+			tc.addressable.Address.Host = s.URL
+			u, _ := url.Parse(s.URL)
+
+			wantRequeueCountMin := atomic.NewInt64(int64(tc.wantRequeueCountMin))
+			prober := NewAsync(ctx, s.Client(), u.Port(), tc.podsLabelsSelector, func(key types.NamespacedName) {
+				wantRequeueCountMin.Dec()
+			})
+			probeFunc := func() bool {
+				status := prober.Probe(ctx, tc.addressable)
+				return status == tc.wantStatus
+			}
+
+			require.Eventuallyf(t, probeFunc, time.Second, 10*time.Millisecond, "")
+			require.GreaterOrEqual(t, int64(0), wantRequeueCountMin.Load())
+			require.GreaterOrEqual(t, int64(0), wantRequestCountMin.Load())
+		})
+	}
+}
diff --git a/control-plane/pkg/prober/prober_test.go b/control-plane/pkg/prober/prober_test.go
new file mode 100644
index 000000000..850bd1192
--- /dev/null
+++ b/control-plane/pkg/prober/prober_test.go
@@ -0,0 +1,40 @@
+/*
+ * Copyright 2021 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 prober
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"go.uber.org/atomic"
+)
+
+func TestFuncProbe(t *testing.T) {
+
+	calls := atomic.NewInt32(0)
+	status := StatusReady
+	p := Func(func(ctx context.Context, addressable Addressable) Status {
+		calls.Inc()
+		return status
+	})
+
+	s := p.Probe(context.Background(), Addressable{})
+
+	require.Equal(t, status, s, s.String())
+	require.Equal(t, int32(1), calls.Load())
+}
diff --git a/control-plane/pkg/prober/probertesting/mock_prober.go b/control-plane/pkg/prober/probertesting/mock_prober.go
new file mode 100644
index 000000000..0111c4850
--- /dev/null
+++ b/control-plane/pkg/prober/probertesting/mock_prober.go
@@ -0,0 +1,30 @@
+/*
+ * Copyright 2021 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 probertesting
+
+import (
+	"context"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
+)
+
+// MockProber returns a prober that always returns the provided status.
+func MockProber(status prober.Status) prober.Prober {
+	return prober.Func(func(ctx context.Context, addressable prober.Addressable) prober.Status {
+		return status
+	})
+}
diff --git a/control-plane/pkg/prober/probertesting/mock_prober_test.go b/control-plane/pkg/prober/probertesting/mock_prober_test.go
new file mode 100644
index 000000000..560afebb0
--- /dev/null
+++ b/control-plane/pkg/prober/probertesting/mock_prober_test.go
@@ -0,0 +1,62 @@
+/*
+ * Copyright 2021 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 probertesting
+
+import (
+	"context"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
+)
+
+func TestMockProber(t *testing.T) {
+	tests := []struct {
+		name        string
+		status      prober.Status
+		ctx         context.Context
+		addressable prober.Addressable
+	}{
+		{
+			name:        "unknown",
+			status:      prober.StatusUnknown,
+			ctx:         context.Background(),
+			addressable: prober.Addressable{},
+		},
+		{
+			name:        "ready",
+			status:      prober.StatusReady,
+			ctx:         context.Background(),
+			addressable: prober.Addressable{},
+		},
+		{
+			name:        "notReady",
+			status:      prober.StatusNotReady,
+			ctx:         context.Background(),
+			addressable: prober.Addressable{},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := MockProber(tt.status).Probe(tt.ctx, tt.addressable)
+			if diff := cmp.Diff(tt.status, got); diff != "" {
+				t.Error("(-want, got)", diff)
+			}
+		})
+	}
+}
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index 4c47235cd..c1f87df3b 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -25,7 +25,11 @@ import (
 	"time"
 
 	"k8s.io/utils/pointer"
+	"knative.dev/pkg/network"
+
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober/probertesting"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 
@@ -63,6 +67,7 @@ const (
 	wantErrorOnCreateTopic = "wantErrorOnCreateTopic"
 	wantErrorOnDeleteTopic = "wantErrorOnDeleteTopic"
 	ExpectedTopicDetail    = "expectedTopicDetail"
+	testProber             = "testProber"
 )
 
 const (
@@ -164,6 +169,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						StatusBrokerConfigParsed,
 						StatusBrokerTopicReady,
 						BrokerAddressable(&env),
+						StatusBrokerProbeSucceeded,
 					),
 				},
 			},
@@ -171,6 +177,126 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				BootstrapServersConfigMapKey: bootstrapServers,
 			},
 		},
+		{
+			Name: "Reconciled failed - probe " + prober.StatusNotReady.String(),
+			Objects: []runtime.Object{
+				NewBroker(),
+				NewConfigMapWithBinaryData(&env, nil),
+				NewService(),
+				BrokerReceiverPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				BrokerDispatcherPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&env, &contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Uid:              BrokerUUID,
+							Topics:           []string{BrokerTopic()},
+							Ingress:          &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							BootstrapServers: bootstrapServers,
+						},
+					},
+					Generation: 1,
+				}),
+				BrokerReceiverPodUpdate(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "1",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "1",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewBroker(
+						reconcilertesting.WithInitBrokerConditions,
+						StatusBrokerConfigMapUpdatedReady(&env),
+						StatusBrokerDataPlaneAvailable,
+						StatusBrokerConfigParsed,
+						StatusBrokerTopicReady,
+						StatusBrokerProbeFailed(prober.StatusNotReady),
+					),
+				},
+			},
+			OtherTestData: map[string]interface{}{
+				BootstrapServersConfigMapKey: bootstrapServers,
+				testProber:                   probertesting.MockProber(prober.StatusNotReady),
+			},
+		},
+		{
+			Name: "Reconciled failed - probe " + prober.StatusUnknown.String(),
+			Objects: []runtime.Object{
+				NewBroker(),
+				NewConfigMapWithBinaryData(&env, nil),
+				NewService(),
+				BrokerReceiverPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				BrokerDispatcherPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&env, &contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Uid:              BrokerUUID,
+							Topics:           []string{BrokerTopic()},
+							Ingress:          &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							BootstrapServers: bootstrapServers,
+						},
+					},
+					Generation: 1,
+				}),
+				BrokerReceiverPodUpdate(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "1",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "1",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewBroker(
+						reconcilertesting.WithInitBrokerConditions,
+						StatusBrokerConfigMapUpdatedReady(&env),
+						StatusBrokerDataPlaneAvailable,
+						StatusBrokerConfigParsed,
+						StatusBrokerTopicReady,
+						StatusBrokerProbeFailed(prober.StatusUnknown),
+					),
+				},
+			},
+			OtherTestData: map[string]interface{}{
+				BootstrapServersConfigMapKey: bootstrapServers,
+				testProber:                   probertesting.MockProber(prober.StatusUnknown),
+			},
+		},
 		{
 			Name: "Reconciled normal - with DLS",
 			Objects: []runtime.Object{
@@ -221,6 +347,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						StatusBrokerTopicReady,
 						StatusBrokerConfigParsed,
 						BrokerAddressable(&env),
+						StatusBrokerProbeSucceeded,
 						BrokerDLSResolved(ServiceURL),
 					),
 				},
@@ -279,6 +406,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						StatusBrokerTopicReady,
 						StatusBrokerConfigParsed,
 						BrokerAddressable(&env),
+						StatusBrokerProbeSucceeded,
 						BrokerDLSResolved(ServiceURLFrom(BrokerNamespace, ServiceName)),
 					),
 				},
@@ -380,6 +508,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						StatusBrokerConfigMapUpdatedReady(&env),
 						StatusBrokerTopicReady,
 						BrokerAddressable(&env),
+						StatusBrokerProbeSucceeded,
 						BrokerDLSResolved(ServiceURL),
 					),
 				},
@@ -434,6 +563,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						StatusBrokerConfigParsed,
 						StatusBrokerTopicReady,
 						BrokerAddressable(&env),
+						StatusBrokerProbeSucceeded,
 					),
 				},
 			},
@@ -514,6 +644,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						StatusBrokerConfigParsed,
 						StatusBrokerTopicReady,
 						BrokerAddressable(&env),
+						StatusBrokerProbeSucceeded,
 					),
 				},
 			},
@@ -610,6 +741,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						StatusBrokerConfigParsed,
 						StatusBrokerTopicReady,
 						BrokerAddressable(&env),
+						StatusBrokerProbeSucceeded,
 						BrokerDLSResolved("http://www.my-sink.com/api"),
 					),
 				},
@@ -694,6 +826,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						StatusBrokerConfigParsed,
 						StatusBrokerTopicReady,
 						BrokerAddressable(&env),
+						StatusBrokerProbeSucceeded,
 					),
 				},
 			},
@@ -770,6 +903,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						StatusBrokerConfigParsed,
 						StatusBrokerTopicReady,
 						BrokerAddressable(&env),
+						StatusBrokerProbeSucceeded,
 					),
 				},
 			},
@@ -938,6 +1072,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						StatusBrokerConfigParsed,
 						StatusBrokerTopicReady,
 						BrokerAddressable(&env),
+						StatusBrokerProbeSucceeded,
 					),
 				},
 			},
@@ -1016,6 +1151,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						StatusBrokerConfigParsed,
 						StatusBrokerTopicReady,
 						BrokerAddressable(&env),
+						StatusBrokerProbeSucceeded,
 					),
 				},
 			},
@@ -1227,6 +1363,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						StatusBrokerConfigParsed,
 						StatusBrokerTopicReady,
 						BrokerAddressable(&env),
+						StatusBrokerProbeSucceeded,
 						BrokerDLSResolved(ServiceURL),
 					),
 				},
@@ -1319,6 +1456,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						StatusBrokerTopicReady,
 						StatusBrokerConfigParsed,
 						BrokerAddressable(&env),
+						StatusBrokerProbeSucceeded,
 						BrokerDLSResolved(ServiceURL),
 					),
 				},
@@ -1384,6 +1522,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						StatusBrokerTopicReady,
 						StatusBrokerConfigParsed,
 						BrokerAddressable(&env),
+						StatusBrokerProbeSucceeded,
 						BrokerDLSResolved(ServiceURL),
 					),
 				},
@@ -1446,6 +1585,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						StatusBrokerTopicReady,
 						StatusBrokerConfigParsed,
 						BrokerAddressable(&env),
+						StatusBrokerProbeSucceeded,
 						BrokerDLSResolved(ServiceURL),
 					),
 				},
@@ -1511,6 +1651,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						StatusBrokerTopicReady,
 						StatusBrokerConfigParsed,
 						BrokerAddressable(&env),
+						StatusBrokerProbeSucceeded,
 						BrokerDLSResolved(ServiceURL),
 					),
 				},
@@ -1559,6 +1700,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						StatusBrokerTopicReady,
 						StatusBrokerConfigParsed,
 						BrokerAddressable(&env),
+						StatusBrokerProbeSucceeded,
 						BrokerDLSResolved(ServiceURL),
 					),
 				},
@@ -1610,6 +1752,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						StatusBrokerTopicReady,
 						StatusBrokerConfigParsed,
 						BrokerAddressable(&env),
+						StatusBrokerProbeSucceeded,
 						BrokerDLSResolved(ServiceURL),
 					),
 				},
@@ -1883,6 +2026,11 @@ func useTable(t *testing.T, table TableTest, env *config.Env) {
 			expectedTopicDetail = td.(sarama.TopicDetail)
 		}
 
+		proberMock := probertesting.MockProber(prober.StatusReady)
+		if p, ok := row.OtherTestData[testProber]; ok {
+			proberMock = p.(prober.Prober)
+		}
+
 		reconciler := &Reconciler{
 			Reconciler: &base.Reconciler{
 				KubeClient:                  kubeclient.Get(ctx),
@@ -1907,7 +2055,9 @@ func useTable(t *testing.T, table TableTest, env *config.Env) {
 					T:                   t,
 				}, nil
 			},
-			Env: env,
+			Env:         env,
+			Prober:      proberMock,
+			IngressHost: network.GetServiceHostname(env.IngressName, env.SystemNamespace),
 		}
 		reconciler.SetBootstrapServers(bootstrapServers)
 
diff --git a/control-plane/pkg/reconciler/broker/controller_test.go b/control-plane/pkg/reconciler/broker/controller_test.go
index 5a259625c..d33eb2e1d 100644
--- a/control-plane/pkg/reconciler/broker/controller_test.go
+++ b/control-plane/pkg/reconciler/broker/controller_test.go
@@ -43,6 +43,7 @@ func TestNewController(t *testing.T) {
 	env := &config.Env{
 		SystemNamespace:      "cm",
 		GeneralConfigMapName: "cm",
+		IngressPodPort:       "8080",
 	}
 
 	ctx, _ = fakekubeclient.With(
diff --git a/control-plane/pkg/reconciler/channel/channel_test.go b/control-plane/pkg/reconciler/channel/channel_test.go
index f5ee7e165..fb988ead5 100644
--- a/control-plane/pkg/reconciler/channel/channel_test.go
+++ b/control-plane/pkg/reconciler/channel/channel_test.go
@@ -22,6 +22,7 @@ import (
 	"testing"
 
 	"github.com/Shopify/sarama"
+	"knative.dev/pkg/network"
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -38,6 +39,8 @@ import (
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober/probertesting"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/channel"
@@ -79,6 +82,8 @@ kafka:
 }
 
 const (
+	testProber = "testProber"
+
 	finalizerName = "kafkachannels.messaging.knative.dev"
 )
 
@@ -165,6 +170,72 @@ func TestReconcileKind(t *testing.T) {
 						StatusDataPlaneAvailable,
 						//StatusInitialOffsetsCommitted,
 						ChannelAddressable(&env),
+						StatusProbeSucceeded,
+					),
+				},
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+		},
+		{
+			Name: "Reconciled failed - probe " + prober.StatusNotReady.String(),
+			Objects: []runtime.Object{
+				NewChannel(),
+				NewService(),
+				ChannelReceiverPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				ChannelDispatcherPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, configKafka),
+			},
+			Key: testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&env, &contract.Contract{
+					Generation: 1,
+					Resources: []*contract.Resource{
+						{
+							Uid:              ChannelUUID,
+							Topics:           []string{ChannelTopic()},
+							BootstrapServers: ChannelBootstrapServers,
+							Ingress: &contract.Ingress{
+								IngressType: &contract.Ingress_Path{
+									Path: receiver.Path(ChannelNamespace, ChannelName),
+								},
+							},
+						},
+					},
+				}),
+				ChannelReceiverPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+				ChannelDispatcherPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+			},
+			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
+			WantCreates: []runtime.Object{
+				NewConfigMapWithBinaryData(&env, nil),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewChannel(
+						WithInitKafkaChannelConditions,
+						StatusConfigParsed,
+						StatusConfigMapUpdatedReady(&env),
+						StatusTopicReadyWithName(ChannelTopic()),
+						StatusDataPlaneAvailable,
+						//StatusInitialOffsetsCommitted,
+						StatusProbeFailed(prober.StatusNotReady),
 					),
 				},
 			},
@@ -174,6 +245,77 @@ func TestReconcileKind(t *testing.T) {
 			WantEvents: []string{
 				finalizerUpdatedEvent,
 			},
+			OtherTestData: map[string]interface{}{
+				testProber: probertesting.MockProber(prober.StatusNotReady),
+			},
+		},
+		{
+			Name: "Reconciled failed - probe " + prober.StatusUnknown.String(),
+			Objects: []runtime.Object{
+				NewChannel(),
+				NewService(),
+				ChannelReceiverPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				ChannelDispatcherPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, configKafka),
+			},
+			Key: testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&env, &contract.Contract{
+					Generation: 1,
+					Resources: []*contract.Resource{
+						{
+							Uid:              ChannelUUID,
+							Topics:           []string{ChannelTopic()},
+							BootstrapServers: ChannelBootstrapServers,
+							Ingress: &contract.Ingress{
+								IngressType: &contract.Ingress_Path{
+									Path: receiver.Path(ChannelNamespace, ChannelName),
+								},
+							},
+						},
+					},
+				}),
+				ChannelReceiverPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+				ChannelDispatcherPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+			},
+			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
+			WantCreates: []runtime.Object{
+				NewConfigMapWithBinaryData(&env, nil),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewChannel(
+						WithInitKafkaChannelConditions,
+						StatusConfigParsed,
+						StatusConfigMapUpdatedReady(&env),
+						StatusTopicReadyWithName(ChannelTopic()),
+						StatusDataPlaneAvailable,
+						//StatusInitialOffsetsCommitted,
+						StatusProbeFailed(prober.StatusUnknown),
+					),
+				},
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			OtherTestData: map[string]interface{}{
+				testProber: probertesting.MockProber(prober.StatusUnknown),
+			},
 		},
 	}
 
@@ -183,6 +325,11 @@ func TestReconcileKind(t *testing.T) {
 func useTable(t *testing.T, table TableTest, env config.Env) {
 	table.Test(t, NewFactory(&env, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
 
+		proberMock := probertesting.MockProber(prober.StatusReady)
+		if p, ok := row.OtherTestData[testProber]; ok {
+			proberMock = p.(prober.Prober)
+		}
+
 		reconciler := &Reconciler{
 			Reconciler: &base.Reconciler{
 				KubeClient:                  kubeclient.Get(ctx),
@@ -213,6 +360,8 @@ func useTable(t *testing.T, table TableTest, env config.Env) {
 					T: t,
 				}, nil
 			},
+			Prober:      proberMock,
+			IngressHost: network.GetServiceHostname(env.IngressName, env.SystemNamespace),
 		}
 
 		reconciler.ConfigMapTracker = &FakeTracker{}
diff --git a/control-plane/pkg/reconciler/channel/controller_test.go b/control-plane/pkg/reconciler/channel/controller_test.go
index ee8c07e98..cc3ca16ce 100644
--- a/control-plane/pkg/reconciler/channel/controller_test.go
+++ b/control-plane/pkg/reconciler/channel/controller_test.go
@@ -42,6 +42,7 @@ func TestNewController(t *testing.T) {
 	configs := &config.Env{
 		SystemNamespace:      "cm",
 		GeneralConfigMapName: "cm",
+		IngressPodPort:       "8080",
 	}
 
 	ctx, _ = fakekubeclient.With(
diff --git a/control-plane/pkg/reconciler/sink/controller_test.go b/control-plane/pkg/reconciler/sink/controller_test.go
index 7ca57c5b3..4811af5f4 100644
--- a/control-plane/pkg/reconciler/sink/controller_test.go
+++ b/control-plane/pkg/reconciler/sink/controller_test.go
@@ -35,7 +35,9 @@ func TestNewController(t *testing.T) {
 
 	ctx, _ := reconcilertesting.SetupFakeContext(t)
 
-	controller := NewController(ctx, nil, &config.Env{})
+	controller := NewController(ctx, nil, &config.Env{
+		IngressPodPort: "8080",
+	})
 
 	assert.NotNil(t, controller, "controller is nil")
 }
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink_test.go b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
index f41000721..c62a9001c 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink_test.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
@@ -22,9 +22,13 @@ import (
 	"io"
 	"testing"
 
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/network"
+
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
-	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober/probertesting"
 
 	"github.com/Shopify/sarama"
 	corev1 "k8s.io/api/core/v1"
@@ -61,6 +65,7 @@ const (
 	ExpectedTopicsOnDescribeTopics = "expectedTopicsOnDescribeTopics"
 	ExpectedTopicIsPresent         = "expectedTopicIsPresent"
 	ExpectedErrorOnDescribeTopics  = "expectedErrorOnDescribeTopics"
+	testProber                     = "testProber"
 
 	TopicPrefix = "knative-sink-"
 )
@@ -151,6 +156,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						StatusConfigMapUpdatedReady(&env),
 						StatusTopicReadyWithOwner(SinkTopic(), sink.ControllerTopicOwner),
 						SinkAddressable(&env),
+						StatusProbeSucceeded,
 					),
 				},
 			},
@@ -213,6 +219,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						StatusConfigMapUpdatedReady(&env),
 						StatusTopicReadyWithOwner(SinkTopic(), sink.ControllerTopicOwner),
 						SinkAddressable(&env),
+						StatusProbeSucceeded,
 					),
 				},
 			},
@@ -279,6 +286,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						StatusConfigMapUpdatedReady(&env),
 						StatusTopicReadyWithOwner(SinkTopic(), sink.ExternalTopicOwner),
 						SinkAddressable(&env),
+						StatusProbeSucceeded,
 					),
 				},
 			},
@@ -390,6 +398,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						StatusTopicReadyWithOwner("my-topic-1", sink.ControllerTopicOwner),
 						StatusConfigMapUpdatedReady(&env),
 						SinkAddressable(&env),
+						StatusProbeSucceeded,
 					),
 				},
 			},
@@ -489,6 +498,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						StatusConfigMapUpdatedReady(&env),
 						StatusTopicReadyWithOwner(SinkTopic(), sink.ControllerTopicOwner),
 						SinkAddressable(&env),
+						StatusProbeSucceeded,
 					),
 				},
 			},
@@ -537,6 +547,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						StatusConfigMapUpdatedReady(&env),
 						StatusTopicReadyWithOwner(SinkTopic(), sink.ControllerTopicOwner),
 						SinkAddressable(&env),
+						StatusProbeSucceeded,
 					),
 				},
 			},
@@ -608,6 +619,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						StatusConfigMapUpdatedReady(&env),
 						StatusTopicReadyWithOwner(SinkTopic(), sink.ControllerTopicOwner),
 						SinkAddressable(&env),
+						StatusProbeSucceeded,
 					),
 				},
 			},
@@ -677,6 +689,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						StatusConfigMapUpdatedReady(&env),
 						StatusTopicReadyWithOwner(SinkTopic(), sink.ControllerTopicOwner),
 						SinkAddressable(&env),
+						StatusProbeSucceeded,
 					),
 				},
 			},
@@ -757,6 +770,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						StatusConfigMapUpdatedReady(&env),
 						StatusTopicReadyWithOwner(SinkTopic(), sink.ControllerTopicOwner),
 						SinkAddressable(&env),
+						StatusProbeSucceeded,
 					),
 				},
 			},
@@ -815,10 +829,117 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						StatusConfigMapUpdatedReady(&env),
 						StatusTopicReadyWithOwner(SinkTopic(), sink.ControllerTopicOwner),
 						SinkAddressable(&env),
+						StatusProbeSucceeded,
 					),
 				},
 			},
 		},
+		{
+			Name: "Reconciled failed - probe " + prober.StatusNotReady.String(),
+			Objects: []runtime.Object{
+				NewSink(
+					StatusControllerOwnsTopic(reconciler.ControllerTopicOwner),
+				),
+				NewConfigMapWithBinaryData(&env, nil),
+				SinkReceiverPod(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve": "value_to_preserve",
+				}),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&env, &contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Uid:              SinkUUID,
+							Topics:           []string{SinkTopic()},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
+							BootstrapServers: bootstrapServers,
+						},
+					},
+					Generation: 1,
+				}),
+				SinkReceiverPodUpdate(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "1",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSink(
+						StatusControllerOwnsTopic(reconciler.ControllerTopicOwner),
+						InitSinkConditions,
+						StatusDataPlaneAvailable,
+						StatusConfigParsed,
+						BootstrapServers(bootstrapServersArr),
+						StatusConfigMapUpdatedReady(&env),
+						StatusTopicReadyWithOwner(SinkTopic(), sink.ControllerTopicOwner),
+						StatusProbeFailed(prober.StatusNotReady),
+					),
+				},
+			},
+			OtherTestData: map[string]interface{}{
+				testProber: probertesting.MockProber(prober.StatusNotReady),
+			},
+		},
+		{
+			Name: "Reconciled failed - probe " + prober.StatusUnknown.String(),
+			Objects: []runtime.Object{
+				NewSink(
+					StatusControllerOwnsTopic(reconciler.ControllerTopicOwner),
+				),
+				NewConfigMapWithBinaryData(&env, nil),
+				SinkReceiverPod(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve": "value_to_preserve",
+				}),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&env, &contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Uid:              SinkUUID,
+							Topics:           []string{SinkTopic()},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
+							BootstrapServers: bootstrapServers,
+						},
+					},
+					Generation: 1,
+				}),
+				SinkReceiverPodUpdate(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "1",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSink(
+						StatusControllerOwnsTopic(reconciler.ControllerTopicOwner),
+						InitSinkConditions,
+						StatusDataPlaneAvailable,
+						StatusConfigParsed,
+						BootstrapServers(bootstrapServersArr),
+						StatusConfigMapUpdatedReady(&env),
+						StatusTopicReadyWithOwner(SinkTopic(), sink.ControllerTopicOwner),
+						StatusProbeFailed(prober.StatusUnknown),
+					),
+				},
+			},
+			OtherTestData: map[string]interface{}{
+				testProber: probertesting.MockProber(prober.StatusUnknown),
+			},
+		},
 	}
 
 	for i := range table {
@@ -1089,6 +1210,11 @@ func useTable(t *testing.T, table TableTest, env *config.Env) {
 			errorOnDescribeTopics = isPresentError.(error)
 		}
 
+		proberMock := probertesting.MockProber(prober.StatusReady)
+		if p, ok := row.OtherTestData[testProber]; ok {
+			proberMock = p.(prober.Prober)
+		}
+
 		reconciler := &sink.Reconciler{
 			Reconciler: &base.Reconciler{
 				KubeClient:                  kubeclient.Get(ctx),
@@ -1113,7 +1239,9 @@ func useTable(t *testing.T, table TableTest, env *config.Env) {
 					T:                                      t,
 				}, nil
 			},
-			Env: env,
+			Env:         env,
+			Prober:      proberMock,
+			IngressHost: network.GetServiceHostname(env.IngressName, env.SystemNamespace),
 		}
 
 		reconciler.SecretTracker = &FakeTracker{}
diff --git a/control-plane/pkg/reconciler/testing/objects_broker.go b/control-plane/pkg/reconciler/testing/objects_broker.go
index 6b614d132..3690c990e 100644
--- a/control-plane/pkg/reconciler/testing/objects_broker.go
+++ b/control-plane/pkg/reconciler/testing/objects_broker.go
@@ -24,7 +24,6 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	clientgotesting "k8s.io/client-go/testing"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
 	reconcilertesting "knative.dev/eventing/pkg/reconciler/testing/v1"
@@ -32,6 +31,9 @@ import (
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/network"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
+
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
@@ -285,3 +287,13 @@ func BrokerReceiverPodUpdate(namespace string, annotations map[string]string) cl
 		BrokerReceiverPod(namespace, annotations),
 	)
 }
+
+func StatusBrokerProbeSucceeded(broker *eventing.Broker) {
+	StatusProbeSucceeded(broker)
+}
+
+func StatusBrokerProbeFailed(status prober.Status) reconcilertesting.BrokerOption {
+	return func(broker *eventing.Broker) {
+		StatusProbeFailed(status)(broker)
+	}
+}
diff --git a/control-plane/pkg/reconciler/testing/objects_common.go b/control-plane/pkg/reconciler/testing/objects_common.go
index ce0f8af11..3caea87f2 100644
--- a/control-plane/pkg/reconciler/testing/objects_common.go
+++ b/control-plane/pkg/reconciler/testing/objects_common.go
@@ -28,11 +28,13 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	clientgotesting "k8s.io/client-go/testing"
+	reconcilertesting "knative.dev/eventing/pkg/reconciler/testing/v1"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
-	reconcilertesting "knative.dev/eventing/pkg/reconciler/testing/v1"
-	duckv1 "knative.dev/pkg/apis/duck/v1"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 )
@@ -264,6 +266,20 @@ func StatusDataPlaneNotAvailable(obj duckv1.KRShaped) {
 	)
 }
 
+func StatusProbeSucceeded(obj duckv1.KRShaped) {
+	obj.GetConditionSet().Manage(obj.GetStatus()).MarkTrue(base.ConditionProbeSucceeded)
+}
+
+func StatusProbeFailed(status prober.Status) func(obj duckv1.KRShaped) {
+	return func(obj duckv1.KRShaped) {
+		obj.GetConditionSet().Manage(obj.GetStatus()).MarkFalse(
+			base.ConditionProbeSucceeded,
+			"ProbeStatus",
+			fmt.Sprintf("status: %s", status.String()),
+		)
+	}
+}
+
 func allocateStatusAnnotations(obj duckv1.KRShaped) {
 	if obj.GetStatus().Annotations == nil {
 		obj.GetStatus().Annotations = make(map[string]string, 1)
diff --git a/test/e2e_new/tracing_test.go b/test/e2e_new/tracing_test.go
index 54516208e..e4f2c4fe4 100644
--- a/test/e2e_new/tracing_test.go
+++ b/test/e2e_new/tracing_test.go
@@ -86,7 +86,7 @@ func TracingHeadersUsingOrderedDelivery() *feature.Feature {
 	))
 	f.Setup("trigger is ready", trigger.IsReady(triggerName))
 
-	f.Setup("install source", eventshub.Install(
+	f.Requirement("install source", eventshub.Install(
 		sourceName,
 		eventshub.StartSenderToResource(broker.GVR(), brokerName),
 		eventshub.InputEvent(ev),
@@ -131,7 +131,7 @@ func TracingHeadersUsingUnorderedDelivery() *feature.Feature {
 	))
 	f.Setup("trigger is ready", trigger.IsReady(triggerName))
 
-	f.Setup("install source", eventshub.Install(
+	f.Requirement("install source", eventshub.Install(
 		sourceName,
 		eventshub.StartSenderToResource(broker.GVR(), brokerName),
 		eventshub.InputEvent(ev),
@@ -182,7 +182,7 @@ func TracingHeadersUsingUnorderedDeliveryWithMultipleTriggers() *feature.Feature
 	))
 	f.Setup("trigger is ready", trigger.IsReady(triggerAName))
 
-	f.Setup("install source", eventshub.Install(
+	f.Requirement("install source", eventshub.Install(
 		sourceName,
 		eventshub.StartSenderToResource(broker.GVR(), brokerName),
 		eventshub.InputEvent(ev),
diff --git a/third_party/eventing-kafka-latest/channel-crds.yaml b/third_party/eventing-kafka-latest/channel-crds.yaml
index d3ecb3c1e..6bdcb3633 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: "v20211115-03ab63a8a"
+    kafka.eventing.knative.dev/release: "v20211117-3237bedea"
     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 30521bdb3..a8edfd145 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: "v20211115-03ab63a8a"
+    kafka.eventing.knative.dev/release: "v20211117-3237bedea"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
diff --git a/vendor/knative.dev/eventing-kafka/test/e2e/helpers/kafka_helper.go b/vendor/knative.dev/eventing-kafka/test/e2e/helpers/kafka_helper.go
index 71c29179a..c41bf7d3e 100644
--- a/vendor/knative.dev/eventing-kafka/test/e2e/helpers/kafka_helper.go
+++ b/vendor/knative.dev/eventing-kafka/test/e2e/helpers/kafka_helper.go
@@ -19,10 +19,13 @@ package helpers
 import (
 	"context"
 	"fmt"
+	"log"
 	"strings"
 	"time"
 
 	"github.com/google/uuid"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
 
 	"github.com/davecgh/go-spew/spew"
 	appsv1 "k8s.io/api/apps/v1"
@@ -31,6 +34,7 @@ import (
 	apierrs "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/wait"
 	testlib "knative.dev/eventing/test/lib"
@@ -241,6 +245,11 @@ func MustCreateTopic(client *testlib.Client, clusterName, clusterNamespace, topi
 	}
 
 	client.Tracker.Add(topicGVR.Group, topicGVR.Version, topicGVR.Resource, clusterNamespace, topicName)
+
+	// Wait for the topic to be ready
+	if err := WaitForTopicReady(context.Background(), client, clusterNamespace, topicName, topicGVR); err != nil {
+		client.T.Fatalf("Error while creating the topic %s: %v", topicName, err)
+	}
 }
 
 //CheckKafkaSourceState waits for specified kafka source resource state
@@ -287,3 +296,36 @@ func CheckRADeployment(ctx context.Context, c *testlib.Client, name string, inSt
 	}
 	return nil
 }
+
+func WaitForTopicReady(ctx context.Context, client *testlib.Client, namespace, name string, gvr schema.GroupVersionResource) error {
+	like := &duckv1.KResource{}
+	return wait.PollImmediate(interval, timeout, func() (bool, error) {
+		us, err := client.Dynamic.Resource(gvr).Namespace(namespace).Get(ctx, name, metav1.GetOptions{})
+		if err != nil {
+			if apierrs.IsNotFound(err) {
+				log.Println(namespace, name, "not found", err)
+				// keep polling
+				return false, nil
+			}
+			return false, err
+		}
+		obj := like.DeepCopy()
+		if err = runtime.DefaultUnstructuredConverter.FromUnstructured(us.Object, obj); err != nil {
+			log.Fatalf("Error DefaultUnstructured.Dynamiconverter. %v", err)
+		}
+		obj.ResourceVersion = gvr.Version
+		obj.APIVersion = gvr.GroupVersion().String()
+
+		// First see if the resource has conditions.
+		if len(obj.Status.Conditions) == 0 {
+			return false, nil // keep polling
+		}
+
+		ready := obj.Status.GetCondition(apis.ConditionReady)
+		if ready != nil && ready.Status == corev1.ConditionTrue {
+			return true, nil
+		}
+
+		return false, nil
+	})
+}
diff --git a/vendor/knative.dev/eventing/test/conformance/helpers/broker_control_plane_test_helper.go b/vendor/knative.dev/eventing/test/conformance/helpers/broker_control_plane_test_helper.go
index 511bba7ac..b248219b8 100644
--- a/vendor/knative.dev/eventing/test/conformance/helpers/broker_control_plane_test_helper.go
+++ b/vendor/knative.dev/eventing/test/conformance/helpers/broker_control_plane_test_helper.go
@@ -117,20 +117,24 @@ func readyBrokerV1AvailableHelper(t *testing.T, brokerName string, client *testl
 
 func brokerV1ConfigCanNotBeUpdated(t *testing.T, brokerName string, client *testlib.Client) {
 	client.WaitForResourceReadyOrFail(brokerName, testlib.BrokerTypeMeta)
-	broker, err := client.Eventing.EventingV1().Brokers(client.Namespace).Get(context.Background(), brokerName, metav1.GetOptions{})
-	if err != nil {
-		t.Fatalf("Error: Could not get broker %s: %v", brokerName, err)
-	}
-	broker.Spec = eventingv1.BrokerSpec{
-		Config: &v1.KReference{
-			Kind:       "kind",
-			Namespace:  "namespace",
-			Name:       "name",
-			APIVersion: "apiversion",
-		},
-	}
+	err := client.RetryWebhookErrors(func(i int) error {
+		broker, err := client.Eventing.EventingV1().Brokers(client.Namespace).Get(context.Background(), brokerName, metav1.GetOptions{})
+		if err != nil {
+			t.Logf("Error: Could not get broker %s: %v", brokerName, err)
+			return err
+		}
+		broker.Spec = eventingv1.BrokerSpec{
+			Config: &v1.KReference{
+				Kind:       "kind",
+				Namespace:  "namespace",
+				Name:       "name",
+				APIVersion: "apiversion",
+			},
+		}
 
-	_, err = client.Eventing.EventingV1().Brokers(client.Namespace).Update(context.Background(), broker, metav1.UpdateOptions{})
+		_, err = client.Eventing.EventingV1().Brokers(client.Namespace).Update(context.Background(), broker, metav1.UpdateOptions{})
+		return err
+	})
 	if err == nil {
 		t.Fatalf("Error: Was able to update the broker.Spec.Config %s", brokerName)
 	}
@@ -150,14 +154,16 @@ func triggerV1ReadyBrokerReadyHelper(triggerName, brokerName string, client *tes
 }
 
 func triggerV1CanNotUpdateBroker(t *testing.T, triggerName, brokerName string, client *testlib.Client) {
-	err := reconciler.RetryUpdateConflicts(func(attempts int) (err error) {
-		tr, err := client.Eventing.EventingV1().Triggers(client.Namespace).Get(context.Background(), triggerName, metav1.GetOptions{})
-		if err != nil {
-			t.Fatalf("Error: Could not get trigger %s: %v", triggerName, err)
-		}
-		tr.Spec.Broker = brokerName
-		_, e := client.Eventing.EventingV1().Triggers(client.Namespace).Update(context.Background(), tr, metav1.UpdateOptions{})
-		return e
+	err := client.RetryWebhookErrors(func(i int) error {
+		return reconciler.RetryUpdateConflicts(func(attempts int) (err error) {
+			tr, err := client.Eventing.EventingV1().Triggers(client.Namespace).Get(context.Background(), triggerName, metav1.GetOptions{})
+			if err != nil {
+				t.Fatalf("Error: Could not get trigger %s: %v", triggerName, err)
+			}
+			tr.Spec.Broker = brokerName
+			_, e := client.Eventing.EventingV1().Triggers(client.Namespace).Update(context.Background(), tr, metav1.UpdateOptions{})
+			return e
+		})
 	})
 	if err == nil {
 		t.Fatalf("Error: Was able to update the trigger.Spec.Broker %s", triggerName)