Repository: knative-sandbox/eventing-kafka-broker
Pull request #1900 (2022-02-15 09:20:38)

Add Consumer reconciler tests
Part of #1537 

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Add Consumer reconciler tests

<!--
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.
-->

TODO:
- finalizer tests
- more reconciler tests

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

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

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

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

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

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

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

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1900). [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/1900?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1900](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1900?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (fe8640b) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/8c0b9a83fa9c71ddf54054f9e3ccf1a915eeecc0?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (8c0b9a8) will **increase** coverage by `14.47%`.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1900/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/1900?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    #1900       +/-   ##
=============================================
+ Coverage     67.60%   82.08%   +14.47%     
+ Complexity      632      629        -3     
=============================================
  Files           138       72       -66     
  Lines          5840     2104     -3736     
  Branches        179      179               
=============================================
- Hits           3948     1727     -2221     
+ Misses         1577      282     -1295     
+ Partials        315       95      -220     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `82.08% <ø> (-0.15%)` | :arrow_down: |

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/1900?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...dispatcher/impl/consumer/BaseConsumerVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1900/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9CYXNlQ29uc3VtZXJWZXJ0aWNsZS5qYXZh) | `87.50% <0.00%> (-6.25%)` | :arrow_down: |
| [...tcher/impl/consumer/UnorderedConsumerVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1900/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9Vbm9yZGVyZWRDb25zdW1lclZlcnRpY2xlLmphdmE=) | `86.95% <0.00%> (-4.35%)` | :arrow_down: |
| [control-plane/pkg/core/config/resource.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1900/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvY29yZS9jb25maWcvcmVzb3VyY2UuZ28=) | | |
| [...lane/pkg/reconciler/base/receiver\_condition\_set.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1900/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9iYXNlL3JlY2VpdmVyX2NvbmRpdGlvbl9zZXQuZ28=) | | |
| [control-plane/pkg/prober/prober.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1900/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcHJvYmVyL3Byb2Jlci5nbw==) | | |
| [control-plane/pkg/reconciler/source/source.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1900/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zb3VyY2Uvc291cmNlLmdv) | | |
| [...-plane/pkg/reconciler/trigger/trigger\_lifecycle.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1900/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci90cmlnZ2VyL3RyaWdnZXJfbGlmZWN5Y2xlLmdv) | | |
| [control-plane/pkg/reconciler/sink/kafka\_sink.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1900/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zaW5rL2thZmthX3NpbmsuZ28=) | | |
| [...trol-plane/pkg/reconciler/consumergroup/evictor.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1900/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2V2aWN0b3IuZ28=) | | |
| [control-plane/pkg/reconciler/kafka/topic.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1900/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9rYWZrYS90b3BpYy5nbw==) | | |
| ... and [58 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1900/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/1900?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/1900?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [8c0b9a8...fe8640b](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1900?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/consumer/consumer.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1900/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1493515589820354560/artifacts/line-cov.html#file0) | 0.0% | 49.3% | 49.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-unit-tests | fe8640b2a7f3542b60f3edb9fe190b59ee6f93f9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1900/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1493515589698719744) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-go-coverage | fe8640b2a7f3542b60f3edb9fe190b59ee6f93f9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1900/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1493515589820354560) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1900). [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/reconciler/consumer/consumer.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1900/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1493518015130505216/artifacts/line-cov.html#file0) | 0.0% | 49.3% | 49.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-unit-tests | fe8640b2a7f3542b60f3edb9fe190b59ee6f93f9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1900/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1493515589698719744) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 680757a9c331da0b080d1d2b4ec79a5edbe72ace | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1900/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1493518015130505216) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1900). [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/reconciler/consumer/consumer.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1900/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1493524754676584448/artifacts/line-cov.html#file0) | 0.0% | 54.6% | 54.6

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-unit-tests | 4ab95cda8a657a7781b29f904dd0a95f7c7fa98d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1900/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1493524751182729216) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl | 4ab95cda8a657a7781b29f904dd0a95f7c7fa98d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1900/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl/1493524757243498496) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl | 4ab95cda8a657a7781b29f904dd0a95f7c7fa98d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1900/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl/1493524757243498496) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl`
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain | 4ab95cda8a657a7781b29f904dd0a95f7c7fa98d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1900/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain/1493524760124985344) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl | 4ab95cda8a657a7781b29f904dd0a95f7c7fa98d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1900/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl/1493524757243498496) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl`
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain | 4ab95cda8a657a7781b29f904dd0a95f7c7fa98d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1900/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain/1493524760124985344) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain`
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl | 4ab95cda8a657a7781b29f904dd0a95f7c7fa98d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1900/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl/1493524758434680832) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1900). [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: /retest-required
pierDipi: Thanks @aavarghese !
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/1900#pullrequestreview-883445881" title="LGTM">aavarghese</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1900#" 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/reconciler/consumer/consumer.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1900/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1493651726643761152/artifacts/line-cov.html#file0) | 0.0% | 54.6% | 54.6

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-unit-tests | f1758f799f0e24a8774412e9178118f8b69ba26a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1900/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1493651726492766208) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1900). [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: PR needs rebase.

<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.
</details>
devguyio: Looks good! Need to rebase

/lgtm
matzew: needs a rebase, @pierDipi 
knative-prow-robot: New changes are detected. LGTM label has been removed.
pierDipi: Rebased!
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/consumer/consumer.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1900/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1493904029854470144/artifacts/line-cov.html#file0) | 0.0% | 54.6% | 54.6


Patch:
diff --git a/control-plane/pkg/prober/async_prober.go b/control-plane/pkg/prober/async_prober.go
index 56daf37f5..1d6d2ed6d 100644
--- a/control-plane/pkg/prober/async_prober.go
+++ b/control-plane/pkg/prober/async_prober.go
@@ -59,7 +59,7 @@ func NewAsync(ctx context.Context, client httpClient, port string, podsLabelsSel
 	}
 }
 
-func (a *asyncProber) Probe(ctx context.Context, addressable Addressable) Status {
+func (a *asyncProber) Probe(ctx context.Context, addressable Addressable, expected Status) Status {
 	address := addressable.Address
 	pods, err := a.podLister()
 	if err != nil {
@@ -98,8 +98,8 @@ func (a *asyncProber) Probe(ctx context.Context, addressable Addressable) Status
 			With(zap.String("address", address))
 
 		currentStatus := a.cache.GetStatus(address)
-		if currentStatus == StatusReady {
-			logger.Debug("Skip probing", zap.String("status", "ready"))
+		if currentStatus == expected {
+			logger.Debug("Skip probing, status in cache is equal to the expected status", zap.String("status", currentStatus.String()))
 			wg.Done()
 			continue
 		}
@@ -112,7 +112,7 @@ func (a *asyncProber) Probe(ctx context.Context, addressable Addressable) Status
 		// We want to requeue the resource only once when we have all probe request
 		// results.
 		enqueueOnce.Do(func() {
-			// Wait in a separate gorouting.
+			// Wait in a separate goroutine.
 			go func() {
 				// Wait for all the prober request results and then requeue the
 				// resource.
@@ -133,7 +133,7 @@ func (a *asyncProber) Probe(ctx context.Context, addressable Addressable) Status
 
 	if aggregatedCurrentKnownStatus == nil {
 		// Every status is ready, return ready.
-		return StatusReady
+		return expected
 	}
 	return *aggregatedCurrentKnownStatus
 }
diff --git a/control-plane/pkg/prober/prober.go b/control-plane/pkg/prober/prober.go
index 59679808b..857693d9c 100644
--- a/control-plane/pkg/prober/prober.go
+++ b/control-plane/pkg/prober/prober.go
@@ -40,18 +40,18 @@ 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
+	Probe(ctx context.Context, addressable Addressable, expected Status) 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
+type Func func(ctx context.Context, addressable Addressable, expected Status) Status
 
 // Probe implements the Prober interface for Func.
-func (p Func) Probe(ctx context.Context, addressable Addressable) Status {
-	return p(ctx, addressable)
+func (p Func) Probe(ctx context.Context, addressable Addressable, expected Status) Status {
+	return p(ctx, addressable, expected)
 }
 
 // httpClient interface is an interface for an HTTP client.
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index 48f4aa396..3b9c20473 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -232,7 +232,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 		},
 	}
 
-	if status := r.Prober.Probe(ctx, proberAddressable); status != prober.StatusReady {
+	if status := r.Prober.Probe(ctx, proberAddressable, prober.StatusReady); status != prober.StatusReady {
 		statusConditionManager.ProbesStatusNotReady(status)
 		return nil // Object will get re-queued once probe status changes.
 	}
@@ -299,7 +299,7 @@ func (r *Reconciler) finalizeKind(ctx context.Context, broker *eventing.Broker)
 			Name:      broker.GetName(),
 		},
 	}
-	if status := r.Prober.Probe(ctx, proberAddressable); status != prober.StatusNotReady {
+	if status := r.Prober.Probe(ctx, proberAddressable, prober.StatusNotReady); status != prober.StatusNotReady {
 		// Return a requeueKeyError that doesn't generate an event and it re-queues the object
 		// for a new reconciliation.
 		return controller.NewRequeueAfter(5 * time.Second)
diff --git a/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index d63fee73b..b79fa344e 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -275,7 +275,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, channel *messagingv1beta
 		},
 	}
 
-	if status := r.Prober.Probe(ctx, proberAddressable); status != prober.StatusReady {
+	if status := r.Prober.Probe(ctx, proberAddressable, prober.StatusReady); status != prober.StatusReady {
 		statusConditionManager.ProbesStatusNotReady(status)
 		return nil // Object will get re-queued once probe status changes.
 	}
@@ -352,7 +352,7 @@ func (r *Reconciler) finalizeKind(ctx context.Context, channel *messagingv1beta1
 			Name:      channel.GetName(),
 		},
 	}
-	if status := r.Prober.Probe(ctx, proberAddressable); status != prober.StatusNotReady {
+	if status := r.Prober.Probe(ctx, proberAddressable, prober.StatusNotReady); status != prober.StatusNotReady {
 		// Return a requeueKeyError that doesn't generate an event and it re-queues the object
 		// for a new reconciliation.
 		return controller.NewRequeueAfter(5 * time.Second)
diff --git a/control-plane/pkg/reconciler/consumer/consumer.go b/control-plane/pkg/reconciler/consumer/consumer.go
index b4b58c264..a06ab9b23 100644
--- a/control-plane/pkg/reconciler/consumer/consumer.go
+++ b/control-plane/pkg/reconciler/consumer/consumer.go
@@ -112,6 +112,8 @@ func (r Reconciler) reconcileContractResource(ctx context.Context, c *kafkainter
 		return nil, nil
 	}
 
+	egress.Reference = userFacingResourceRef
+
 	resource := &contract.Resource{
 		Uid:                 string(c.UID),
 		Topics:              c.Spec.Topics,
@@ -219,12 +221,12 @@ func reconcileCEOverrides(c *kafkainternals.Consumer) *contract.CloudEventOverri
 func (r Reconciler) reconcileUserFacingResourceRef(c *kafkainternals.Consumer) (*contract.Reference, error) {
 
 	cg, err := r.ConsumerGroupLister.ConsumerGroups(c.GetNamespace()).Get(c.GetConsumerGroup().Name)
-	if err != nil && !apierrors.IsNotFound(err) {
-		return nil, fmt.Errorf("failed to get %s: %w", kafkainternals.ConsumerGroupGroupVersionKind.Kind, err)
-	}
 	if apierrors.IsNotFound(err) {
 		return nil, nil
 	}
+	if err != nil {
+		return nil, fmt.Errorf("failed to get %s: %w", kafkainternals.ConsumerGroupGroupVersionKind.Kind, err)
+	}
 
 	userFacingResource := cg.GetUserFacingResourceRef()
 	ref := &contract.Reference{
@@ -308,14 +310,14 @@ func removeResource(_ *zap.Logger, ct *contract.Contract, c *kafkainternals.Cons
 func (r Reconciler) schedule(ctx context.Context, logger *zap.Logger, c *kafkainternals.Consumer, mutatorFunc contractMutatorFunc) (bool, error) {
 	// Get the data plane pod when the Consumer should be scheduled.
 	p, err := r.PodLister.Pods(c.Spec.PodBind.PodNamespace).Get(c.Spec.PodBind.PodName)
-	if err != nil {
-		return false, fmt.Errorf("failed to get pod %s/%s: %w", c.Spec.PodBind.PodNamespace, c.Spec.PodBind.PodName, err)
-	}
 	if apierrors.IsNotFound(err) {
 		// Pod not found, return no error since the Consumer
 		// will get re-queued when the pod is added.
 		return false, nil
 	}
+	if err != nil {
+		return false, fmt.Errorf("failed to get pod %s/%s: %w", c.Spec.PodBind.PodNamespace, c.Spec.PodBind.PodName, err)
+	}
 
 	// Get contract associated with the pod.
 	cmName, err := cmNameFromPod(p, c)
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink.go b/control-plane/pkg/reconciler/sink/kafka_sink.go
index 9ca81b778..14a96a558 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink.go
@@ -237,7 +237,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *eventing.KafkaSink)
 		},
 	}
 
-	if status := r.Prober.Probe(ctx, proberAddressable); status != prober.StatusReady {
+	if status := r.Prober.Probe(ctx, proberAddressable, prober.StatusReady); status != prober.StatusReady {
 		statusConditionManager.ProbesStatusNotReady(status)
 		return nil // Object will get re-queued once probe status changes.
 	}
@@ -303,7 +303,7 @@ func (r *Reconciler) finalizeKind(ctx context.Context, ks *eventing.KafkaSink) e
 			Name:      ks.GetName(),
 		},
 	}
-	if status := r.Prober.Probe(ctx, proberAddressable); status != prober.StatusNotReady {
+	if status := r.Prober.Probe(ctx, proberAddressable, prober.StatusNotReady); status != prober.StatusNotReady {
 		// Return a requeueKeyError that doesn't generate an event and it re-queues the object
 		// for a new reconciliation.
 		return controller.NewRequeueAfter(5 * time.Second)


Test patch:
diff --git a/control-plane/pkg/prober/async_prober_test.go b/control-plane/pkg/prober/async_prober_test.go
index e27b94c56..d68fe4481 100644
--- a/control-plane/pkg/prober/async_prober_test.go
+++ b/control-plane/pkg/prober/async_prober_test.go
@@ -36,6 +36,7 @@ import (
 )
 
 func TestAsyncProber(t *testing.T) {
+	t.Parallel()
 
 	tt := []struct {
 		name                string
@@ -106,12 +107,13 @@ func TestAsyncProber(t *testing.T) {
 	}
 
 	for _, tc := range tt {
+		tc := tc
 		t.Run(tc.name, func(t *testing.T) {
 			ctx, _ := reconcilertesting.SetupFakeContext(t)
-			ctx, cacel := context.WithCancel(ctx)
+			ctx, cancel := context.WithCancel(ctx)
 			defer func() {
 				time.Sleep(time.Second)
-				cacel()
+				cancel()
 			}()
 			wantRequestCountMin := atomic.NewInt64(int64(tc.wantRequestCountMin))
 			h := http.HandlerFunc(func(writer http.ResponseWriter, r *http.Request) {
@@ -121,7 +123,6 @@ func TestAsyncProber(t *testing.T) {
 			})
 			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 {
@@ -134,14 +135,15 @@ func TestAsyncProber(t *testing.T) {
 			prober := NewAsync(ctx, s.Client(), u.Port(), tc.podsLabelsSelector, func(key types.NamespacedName) {
 				wantRequeueCountMin.Dec()
 			})
+
 			probeFunc := func() bool {
-				status := prober.Probe(ctx, tc.addressable)
+				status := prober.Probe(ctx, tc.addressable, tc.wantStatus)
 				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())
+			require.GreaterOrEqual(t, int64(0), wantRequestCountMin.Load(), "")
+			require.GreaterOrEqual(t, int64(0), wantRequeueCountMin.Load(), "")
 		})
 	}
 }
diff --git a/control-plane/pkg/prober/prober_test.go b/control-plane/pkg/prober/prober_test.go
index 850bd1192..bb9bdf30b 100644
--- a/control-plane/pkg/prober/prober_test.go
+++ b/control-plane/pkg/prober/prober_test.go
@@ -28,12 +28,12 @@ func TestFuncProbe(t *testing.T) {
 
 	calls := atomic.NewInt32(0)
 	status := StatusReady
-	p := Func(func(ctx context.Context, addressable Addressable) Status {
+	p := Func(func(ctx context.Context, addressable Addressable, expected Status) Status {
 		calls.Inc()
 		return status
 	})
 
-	s := p.Probe(context.Background(), Addressable{})
+	s := p.Probe(context.Background(), Addressable{}, status)
 
 	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
index 0111c4850..c13110a7d 100644
--- a/control-plane/pkg/prober/probertesting/mock_prober.go
+++ b/control-plane/pkg/prober/probertesting/mock_prober.go
@@ -24,7 +24,7 @@ import (
 
 // 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 prober.Func(func(ctx context.Context, addressable prober.Addressable, expected prober.Status) 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
index 560afebb0..c2c6aba91 100644
--- a/control-plane/pkg/prober/probertesting/mock_prober_test.go
+++ b/control-plane/pkg/prober/probertesting/mock_prober_test.go
@@ -53,7 +53,7 @@ func TestMockProber(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got := MockProber(tt.status).Probe(tt.ctx, tt.addressable)
+			got := MockProber(tt.status).Probe(tt.ctx, tt.addressable, tt.status)
 			if diff := cmp.Diff(tt.status, got); diff != "" {
 				t.Error("(-want, got)", diff)
 			}
diff --git a/control-plane/pkg/reconciler/consumer/consumer_test.go b/control-plane/pkg/reconciler/consumer/consumer_test.go
new file mode 100644
index 000000000..4d8c05f4d
--- /dev/null
+++ b/control-plane/pkg/reconciler/consumer/consumer_test.go
@@ -0,0 +1,379 @@
+/*
+ * 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 consumer
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	clientgotesting "k8s.io/client-go/testing"
+	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
+	"knative.dev/pkg/apis"
+	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/logging"
+	. "knative.dev/pkg/reconciler/testing"
+	"knative.dev/pkg/resolver"
+	"knative.dev/pkg/tracker"
+
+	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
+	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
+	fakekafkainternalsclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client/fake"
+	creconciler "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/reconciler/eventing/v1alpha1/consumer"
+	"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/reconciler/base"
+	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/testing"
+)
+
+var (
+	finalizerUpdatedEvent = Eventf(
+		corev1.EventTypeNormal,
+		"FinalizerUpdate",
+		fmt.Sprintf(`Updated %q finalizers`, ConsumerName),
+	)
+)
+
+func TestReconcileKind(t *testing.T) {
+
+	testKey := fmt.Sprintf("%s/%s", ConsumerNamespace, ConsumerName)
+
+	table := TableTest{
+		{
+			Name: "Reconciled normal",
+			Objects: []runtime.Object{
+				NewService(),
+				NewConsumerGroup(ConsumerGroupOwnerRef(SourceAsOwnerReference())),
+				NewDispatcherPod("p1", PodRunning()),
+				NewConsumer(1,
+					ConsumerUID(ConsumerUUID),
+					ConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(SourceTopics...),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(SourceBootstrapServers),
+							ConsumerGroupIdConfig(SourceConsumerGroup),
+						),
+						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerVReplicas(1),
+						ConsumerPlacement(kafkainternals.PodBind{PodName: "p1", PodNamespace: SystemNamespace}),
+					)),
+					ConsumerOwnerRef(ConsumerGroupAsOwnerRef()),
+				),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantCreates: []runtime.Object{
+				NewConfigMapWithBinaryData(&config.Env{
+					DataPlaneConfigFormat:       base.Json,
+					DataPlaneConfigMapNamespace: SystemNamespace,
+					DataPlaneConfigMapName:      "p1",
+				}, []byte("")),
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&config.Env{
+					DataPlaneConfigFormat:       base.Json,
+					DataPlaneConfigMapNamespace: SystemNamespace,
+					DataPlaneConfigMapName:      "p1",
+				}, &contract.Contract{
+					Generation: 1,
+					Resources: []*contract.Resource{
+						{
+							Uid:              ConsumerUUID,
+							Topics:           SourceTopics,
+							BootstrapServers: SourceBootstrapServers,
+							Egresses: []*contract.Egress{{
+								ConsumerGroup: SourceConsumerGroup,
+								Destination:   ServiceURL,
+								ReplyStrategy: nil,
+								Filter:        nil,
+								Uid:           ConsumerUUID,
+								EgressConfig:  &contract.EgressConfig{},
+								DeliveryOrder: contract.DeliveryOrder_UNORDERED,
+								KeyType:       0,
+								Reference: &contract.Reference{
+									Uuid:      SourceUUID,
+									Namespace: ConsumerNamespace,
+									Name:      SourceName,
+								},
+							}},
+							Auth:                nil,
+							CloudEventOverrides: nil,
+							Reference: &contract.Reference{
+								Uuid:      SourceUUID,
+								Namespace: ConsumerNamespace,
+								Name:      SourceName,
+							},
+						},
+					},
+				}),
+				{Object: NewDispatcherPod("p1",
+					PodRunning(),
+					PodAnnotations(map[string]string{
+						base.VolumeGenerationAnnotationKey: "1",
+					}),
+				)},
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: func() runtime.Object {
+						c := NewConsumer(1,
+							ConsumerUID(ConsumerUUID),
+							ConsumerSpec(NewConsumerSpec(
+								ConsumerTopics(SourceTopics...),
+								ConsumerConfigs(
+									ConsumerBootstrapServersConfig(SourceBootstrapServers),
+									ConsumerGroupIdConfig(SourceConsumerGroup),
+								),
+								ConsumerSubscriber(NewSourceSinkReference()),
+								ConsumerVReplicas(1),
+								ConsumerPlacement(kafkainternals.PodBind{PodName: "p1", PodNamespace: SystemNamespace}),
+							)),
+							ConsumerOwnerRef(ConsumerGroupAsOwnerRef()),
+						)
+						c.GetConditionSet().Manage(c.GetStatus()).InitializeConditions()
+						c.MarkReconcileContractSucceeded()
+						c.MarkBindSucceeded()
+						c.Status.SubscriberURI, _ = apis.ParseURL(ServiceURL)
+						return c
+					}(),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - consumer delivery",
+			Objects: []runtime.Object{
+				NewService(),
+				NewConsumerGroup(ConsumerGroupOwnerRef(SourceAsOwnerReference())),
+				NewDispatcherPod("p1", PodRunning()),
+				NewConsumer(1,
+					ConsumerUID(ConsumerUUID),
+					ConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(SourceTopics...),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(SourceBootstrapServers),
+							ConsumerGroupIdConfig(SourceConsumerGroup),
+						),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered,
+							NewConsumerSpecDeliveryDeadLetterSink(),
+							NewConsumerRetry(10),
+							NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+							NewConsumerBackoffDelay("PT0.2S"),
+							NewConsumerTimeout("PT51S"),
+						)),
+						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerVReplicas(1),
+						ConsumerPlacement(kafkainternals.PodBind{PodName: "p1", PodNamespace: SystemNamespace}),
+					)),
+					ConsumerOwnerRef(ConsumerGroupAsOwnerRef()),
+				),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantCreates: []runtime.Object{
+				NewConfigMapWithBinaryData(&config.Env{
+					DataPlaneConfigFormat:       base.Json,
+					DataPlaneConfigMapNamespace: SystemNamespace,
+					DataPlaneConfigMapName:      "p1",
+				}, []byte("")),
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&config.Env{
+					DataPlaneConfigFormat:       base.Json,
+					DataPlaneConfigMapNamespace: SystemNamespace,
+					DataPlaneConfigMapName:      "p1",
+				}, &contract.Contract{
+					Generation: 1,
+					Resources: []*contract.Resource{
+						{
+							Uid:              ConsumerUUID,
+							Topics:           SourceTopics,
+							BootstrapServers: SourceBootstrapServers,
+							Egresses: []*contract.Egress{{
+								ConsumerGroup: SourceConsumerGroup,
+								Destination:   ServiceURL,
+								ReplyStrategy: nil,
+								Filter:        nil,
+								Uid:           ConsumerUUID,
+								EgressConfig: &contract.EgressConfig{
+									DeadLetter:    ConsumerDeadLetterSinkURI.String(),
+									Retry:         10,
+									BackoffPolicy: contract.BackoffPolicy_Exponential,
+									BackoffDelay:  200,
+									Timeout:       51000,
+								},
+								DeliveryOrder: contract.DeliveryOrder_ORDERED,
+								KeyType:       0,
+								Reference: &contract.Reference{
+									Uuid:      SourceUUID,
+									Namespace: ConsumerNamespace,
+									Name:      SourceName,
+								},
+							}},
+							Auth:                nil,
+							CloudEventOverrides: nil,
+							Reference: &contract.Reference{
+								Uuid:      SourceUUID,
+								Namespace: ConsumerNamespace,
+								Name:      SourceName,
+							},
+						},
+					},
+				}),
+				{Object: NewDispatcherPod("p1",
+					PodRunning(),
+					PodAnnotations(map[string]string{
+						base.VolumeGenerationAnnotationKey: "1",
+					}),
+				)},
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: func() runtime.Object {
+						c := NewConsumer(1,
+							ConsumerUID(ConsumerUUID),
+							ConsumerSpec(NewConsumerSpec(
+								ConsumerTopics(SourceTopics...),
+								ConsumerConfigs(
+									ConsumerBootstrapServersConfig(SourceBootstrapServers),
+									ConsumerGroupIdConfig(SourceConsumerGroup),
+								),
+								ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered,
+									NewConsumerSpecDeliveryDeadLetterSink(),
+									NewConsumerRetry(10),
+									NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+									NewConsumerBackoffDelay("PT0.2S"),
+									NewConsumerTimeout("PT51S"),
+								)),
+								ConsumerSubscriber(NewSourceSinkReference()),
+								ConsumerVReplicas(1),
+								ConsumerPlacement(kafkainternals.PodBind{PodName: "p1", PodNamespace: SystemNamespace}),
+							)),
+							ConsumerOwnerRef(ConsumerGroupAsOwnerRef()),
+						)
+						c.GetConditionSet().Manage(c.GetStatus()).InitializeConditions()
+						c.MarkReconcileContractSucceeded()
+						c.MarkBindSucceeded()
+						c.Status.SubscriberURI, _ = apis.ParseURL(ServiceURL)
+						c.Status.DeliveryStatus.DeadLetterSinkURI = ConsumerDeadLetterSinkURI
+						return c
+					}(),
+				},
+			},
+		},
+		{
+			Name: "Pod not found",
+			Objects: []runtime.Object{
+				NewService(),
+				NewConsumerGroup(ConsumerGroupOwnerRef(SourceAsOwnerReference())),
+				NewConsumer(1,
+					ConsumerUID(ConsumerUUID),
+					ConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(SourceTopics...),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(SourceBootstrapServers),
+							ConsumerGroupIdConfig(SourceConsumerGroup),
+						),
+						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerVReplicas(1),
+						ConsumerPlacement(kafkainternals.PodBind{PodName: "p1", PodNamespace: SystemNamespace}),
+					)),
+					ConsumerOwnerRef(ConsumerGroupAsOwnerRef()),
+				),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: func() runtime.Object {
+						c := NewConsumer(1,
+							ConsumerUID(ConsumerUUID),
+							ConsumerSpec(NewConsumerSpec(
+								ConsumerTopics(SourceTopics...),
+								ConsumerConfigs(
+									ConsumerBootstrapServersConfig(SourceBootstrapServers),
+									ConsumerGroupIdConfig(SourceConsumerGroup),
+								),
+								ConsumerSubscriber(NewSourceSinkReference()),
+								ConsumerVReplicas(1),
+								ConsumerPlacement(kafkainternals.PodBind{PodName: "p1", PodNamespace: SystemNamespace}),
+							)),
+							ConsumerOwnerRef(ConsumerGroupAsOwnerRef()),
+						)
+						c.GetConditionSet().Manage(c.GetStatus()).InitializeConditions()
+						c.MarkReconcileContractSucceeded()
+						c.MarkBindInProgress()
+						c.Status.SubscriberURI, _ = apis.ParseURL(ServiceURL)
+						return c
+					}(),
+				},
+			},
+		},
+	}
+
+	table.Test(t, NewFactory(nil, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
+
+		r := &Reconciler{
+			SerDe:               contract.FormatSerDe{Format: contract.Json},
+			Resolver:            resolver.NewURIResolverFromTracker(ctx, tracker.New(func(name types.NamespacedName) {}, 0)),
+			Tracker:             &FakeTracker{},
+			ConsumerGroupLister: listers.GetConsumerGroupLister(),
+			SecretLister:        listers.GetSecretLister(),
+			PodLister:           listers.GetPodLister(),
+			KubeClient:          kubeclient.Get(ctx),
+		}
+
+		return creconciler.NewReconciler(
+			ctx,
+			logging.FromContext(ctx),
+			fakekafkainternalsclient.Get(ctx),
+			listers.GetConsumerLister(),
+			controller.GetEventRecorder(ctx),
+			r,
+		)
+	}))
+}
+
+func patchFinalizers() clientgotesting.PatchActionImpl {
+	action := clientgotesting.PatchActionImpl{}
+	action.Name = ConsumerName
+	action.Namespace = ConsumerNamespace
+	patch := `{"metadata":{"finalizers":["consumers.internal.kafka.eventing.knative.dev"],"resourceVersion":""}}`
+	action.Patch = []byte(patch)
+	return action
+}
diff --git a/control-plane/pkg/reconciler/testing/objects_common.go b/control-plane/pkg/reconciler/testing/objects_common.go
index d676478ee..5030042f0 100644
--- a/control-plane/pkg/reconciler/testing/objects_common.go
+++ b/control-plane/pkg/reconciler/testing/objects_common.go
@@ -31,6 +31,7 @@ import (
 	reconcilertesting "knative.dev/eventing/pkg/reconciler/testing/v1"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
+	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 	"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"
@@ -52,6 +53,8 @@ const (
 
 	SecretResourceVersion = "1234"
 	SecretUUID            = "a7185016-5d98-4b54-84e8-3b1cd4acc6b6"
+
+	SystemNamespace = "knative-eventing"
 )
 
 var (
@@ -316,3 +319,44 @@ func allocateStatusAnnotations(obj duckv1.KRShaped) {
 		obj.GetStatus().Annotations = make(map[string]string, 1)
 	}
 }
+
+type PodOption func(pod *corev1.Pod)
+
+func NewDispatcherPod(name string, options ...PodOption) *corev1.Pod {
+	p := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: SystemNamespace,
+		},
+		Spec: corev1.PodSpec{
+			Volumes: []corev1.Volume{{
+				Name: kafkainternals.DispatcherVolumeName,
+				VolumeSource: corev1.VolumeSource{
+					ConfigMap: &corev1.ConfigMapVolumeSource{
+						LocalObjectReference: corev1.LocalObjectReference{
+							Name: name,
+						},
+					},
+				},
+			}},
+		},
+	}
+
+	for _, opt := range options {
+		opt(p)
+	}
+
+	return p
+}
+
+func PodRunning() PodOption {
+	return func(pod *corev1.Pod) {
+		pod.Status.Phase = corev1.PodRunning
+	}
+}
+
+func PodAnnotations(annotations map[string]string) PodOption {
+	return func(pod *corev1.Pod) {
+		pod.Annotations = annotations
+	}
+}
diff --git a/control-plane/pkg/reconciler/testing/objects_consumer.go b/control-plane/pkg/reconciler/testing/objects_consumer.go
index 1a99d7f2f..dd185f368 100644
--- a/control-plane/pkg/reconciler/testing/objects_consumer.go
+++ b/control-plane/pkg/reconciler/testing/objects_consumer.go
@@ -20,6 +20,7 @@ import (
 	"fmt"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/utils/pointer"
 	"knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
@@ -37,6 +38,9 @@ const (
 )
 
 var (
+	ConsumerUUID = "c1234567-8901-2345-6789-123456789102"
+	ConsumerName = fmt.Sprintf("%s-%d", ConsumerNamePrefix, 1)
+
 	ConsumerSubscriberURI     = apis.HTTP("localhost")
 	ConsumerDeadLetterSinkURI = apis.HTTP("dls.com")
 )
@@ -91,6 +95,12 @@ func NewConsumerSpec(opts ...ConsumerSpecOption) kafkainternals.ConsumerSpec {
 	return *spec
 }
 
+func ConsumerUID(uid string) ConsumerOption {
+	return func(c *kafkainternals.Consumer) {
+		c.UID = types.UID(uid)
+	}
+}
+
 type DeliverySpecOption func(spec *kafkainternals.DeliverySpec)
 
 func NewConsumerSpecDelivery(order internals.DeliveryOrdering, options ...DeliverySpecOption) *kafkainternals.DeliverySpec {
@@ -112,6 +122,42 @@ func NewConsumerSpecDeliveryDeadLetterSink() DeliverySpecOption {
 	}
 }
 
+func NewConsumerRetry(r int32) DeliverySpecOption {
+	return func(spec *kafkainternals.DeliverySpec) {
+		if spec.DeliverySpec == nil {
+			spec.DeliverySpec = &eventingduck.DeliverySpec{}
+		}
+		spec.DeliverySpec.Retry = &r
+	}
+}
+
+func NewConsumerBackoffPolicy(policy eventingduck.BackoffPolicyType) DeliverySpecOption {
+	return func(spec *kafkainternals.DeliverySpec) {
+		if spec.DeliverySpec == nil {
+			spec.DeliverySpec = &eventingduck.DeliverySpec{}
+		}
+		spec.DeliverySpec.BackoffPolicy = &policy
+	}
+}
+
+func NewConsumerBackoffDelay(delay string) DeliverySpecOption {
+	return func(spec *kafkainternals.DeliverySpec) {
+		if spec.DeliverySpec == nil {
+			spec.DeliverySpec = &eventingduck.DeliverySpec{}
+		}
+		spec.DeliverySpec.BackoffDelay = &delay
+	}
+}
+
+func NewConsumerTimeout(timeout string) DeliverySpecOption {
+	return func(spec *kafkainternals.DeliverySpec) {
+		if spec.DeliverySpec == nil {
+			spec.DeliverySpec = &eventingduck.DeliverySpec{}
+		}
+		spec.DeliverySpec.Timeout = &timeout
+	}
+}
+
 func NewConsumerSpecAuth() *kafkainternals.Auth {
 	return &kafkainternals.Auth{
 		NetSpec: &v1beta1.KafkaNetSpec{},
@@ -240,3 +286,19 @@ func ConsumerNoReply() *kafkainternals.ReplyStrategy {
 func ConsumerTopicReply() *kafkainternals.ReplyStrategy {
 	return &kafkainternals.ReplyStrategy{TopicReply: &kafkainternals.TopicReply{Enabled: true}}
 }
+
+func ConsumerOwnerRef(reference metav1.OwnerReference) ConsumerOption {
+	return func(cg *kafkainternals.Consumer) {
+		cg.OwnerReferences = append(cg.OwnerReferences, reference)
+	}
+}
+
+func ConsumerGroupAsOwnerRef() metav1.OwnerReference {
+	return metav1.OwnerReference{
+		APIVersion:         kafkainternals.SchemeGroupVersion.String(),
+		Kind:               kafkainternals.ConsumerGroupGroupVersionKind.Kind,
+		Name:               ConsumerGroupName,
+		Controller:         pointer.BoolPtr(true),
+		BlockOwnerDeletion: pointer.BoolPtr(true),
+	}
+}
diff --git a/control-plane/pkg/reconciler/testing/objects_consumergroup.go b/control-plane/pkg/reconciler/testing/objects_consumergroup.go
index 519d34307..d6a133c75 100644
--- a/control-plane/pkg/reconciler/testing/objects_consumergroup.go
+++ b/control-plane/pkg/reconciler/testing/objects_consumergroup.go
@@ -154,3 +154,9 @@ func WithDeadLetterSinkURI(uri string) func(cg *kafkainternals.ConsumerGroup) {
 		cg.Status.DeadLetterSinkURI = u
 	}
 }
+
+func ConsumerGroupOwnerRef(reference metav1.OwnerReference) ConsumerGroupOption {
+	return func(cg *kafkainternals.ConsumerGroup) {
+		cg.OwnerReferences = append(cg.OwnerReferences, reference)
+	}
+}
diff --git a/control-plane/pkg/reconciler/testing/objects_source.go b/control-plane/pkg/reconciler/testing/objects_source.go
index 76e106eb1..f3ee2ac14 100644
--- a/control-plane/pkg/reconciler/testing/objects_source.go
+++ b/control-plane/pkg/reconciler/testing/objects_source.go
@@ -20,6 +20,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/utils/pointer"
 	"knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
 	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
 	"knative.dev/pkg/apis"
@@ -184,3 +185,14 @@ func SourceReference() *contract.Reference {
 		Uuid:      SourceUUID,
 	}
 }
+
+func SourceAsOwnerReference() metav1.OwnerReference {
+	return metav1.OwnerReference{
+		APIVersion:         sources.SchemeGroupVersion.String(),
+		Kind:               "KafkaSource",
+		Name:               SourceName,
+		UID:                SourceUUID,
+		Controller:         pointer.Bool(true),
+		BlockOwnerDeletion: pointer.Bool(true),
+	}
+}