Repository: knative-sandbox/eventing-kafka-broker
Pull request #1855 (2022-02-02 18:17:41)

Propagate Consumers status to ConsumerGroup 
Propagate subscriberURI, deadLetterSinkURI,
and ready replica count.

Part of #1537 

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/1855#" 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/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1488939701053165568/artifacts/line-cov.html#file0) | 21.1% | 28.6% | 7.5
[control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1488939701053165568/artifacts/line-cov.html#file1) | 30.8% | 35.7% | 4.9
[control-plane/pkg/reconciler/consumergroup/consumergroup.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1488939701053165568/artifacts/line-cov.html#file2) | 88.2% | 86.0% | -2.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 | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1488939701053165568) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1855). [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: /assign @aavarghese
/cc @aliok @devguyio @matzew
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1855?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1855](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1855?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (7d05947) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/b9c37d635d273b27ba2485c6365b50372d800ff3?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (b9c37d6) will **increase** coverage by `15.27%`.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1855/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/1855?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    #1855       +/-   ##
=============================================
+ Coverage     66.74%   82.01%   +15.27%     
+ Complexity      617      614        -3     
=============================================
  Files           135       69       -66     
  Lines          5722     2024     -3698     
  Branches        178      178               
=============================================
- Hits           3819     1660     -2159     
+ Misses         1598      270     -1328     
+ Partials        305       94      -211     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `82.01% <ø> (-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/1855?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/1855/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/1855/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: |
| [...ls/kafka/eventing/v1alpha1/consumer\_group\_types.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1855/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9pbnRlcm5hbHMva2Fma2EvZXZlbnRpbmcvdjFhbHBoYTEvY29uc3VtZXJfZ3JvdXBfdHlwZXMuZ28=) | | |
| [...nternals/kafka/eventing/v1alpha1/consumer\_types.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1855/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9pbnRlcm5hbHMva2Fma2EvZXZlbnRpbmcvdjFhbHBoYTEvY29uc3VtZXJfdHlwZXMuZ28=) | | |
| [...lane/pkg/reconciler/consumergroup/consumergroup.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1855/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2NvbnN1bWVyZ3JvdXAuZ28=) | | |
| [control-plane/pkg/reconciler/consumer/consumer.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1855/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lci9jb25zdW1lci5nbw==) | | |
| [control-plane/pkg/reconciler/channel/channel.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1855/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jaGFubmVsL2NoYW5uZWwuZ28=) | | |
| [...ane/pkg/apis/eventing/v1alpha1/kafka\_sink\_types.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1855/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9rYWZrYV9zaW5rX3R5cGVzLmdv) | | |
| [...ntrol-plane/pkg/reconciler/source/auth\_provider.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1855/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zb3VyY2UvYXV0aF9wcm92aWRlci5nbw==) | | |
| [control-plane/pkg/kafka/consumer\_group\_lag.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1855/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cva2Fma2EvY29uc3VtZXJfZ3JvdXBfbGFnLmdv) | | |
| ... and [58 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1855/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/1855?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/1855?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [b9c37d6...7d05947](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1855?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).

knative-prow-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 | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1488939701053165568) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1488939700939919360) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1855). [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 | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1488939701053165568) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1488939700939919360) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain/1488957465809653760) | 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/1855). [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 | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1488939701053165568) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1488939700939919360) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain/1488957465809653760) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain`
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl/1488957465742544896) | 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/1855). [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 | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1488939701053165568) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1488939700939919360) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain/1488957465809653760) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain`
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl/1488957465742544896) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl`
pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-tls | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-tls/1488957465847402496) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-tls`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1855). [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 | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1488939701053165568) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1488939700939919360) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain/1488957465809653760) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain`
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl/1488957465742544896) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl`
pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-tls | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-tls/1488957465847402496) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-tls`
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-tls | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-tls/1488957465633492992) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-tls`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1855). [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 | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1488939701053165568) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1488939700939919360) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain/1488957465809653760) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain`
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl/1488957465742544896) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl`
pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-tls | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-tls/1488957465847402496) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-tls`
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-tls | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-tls/1488957465633492992) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-tls`
pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-sasl-plain | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-sasl-plain/1488957466271027200) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-sasl-plain`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1855). [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 | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1488939701053165568) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1488939700939919360) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain/1488957465809653760) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain`
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl/1488957465742544896) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl`
pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-tls | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-tls/1488957465847402496) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-tls`
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-tls | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-tls/1488957465633492992) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-tls`
pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-sasl-plain | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-sasl-plain/1488957466271027200) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-sasl-plain`
pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-sasl-ssl | 7d05947160e83c8c737bb3675c742d84e08c2cc9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1855/pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-sasl-ssl/1488957465885151232) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-sasl-ssl`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1855). [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/OWNERS_ALIASES b/OWNERS_ALIASES
index 600ae4aac..01eafc332 100644
--- a/OWNERS_ALIASES
+++ b/OWNERS_ALIASES
@@ -174,6 +174,7 @@ aliases:
   - csantanapr
   - dprotaso
   - evankanderson
+  - itsmurugappan
   - julz
   - knative-prow-releaser-robot
   - knative-prow-robot
@@ -249,6 +250,7 @@ aliases:
   - lionelvillard
   steering-committee:
   - csantanapr
+  - itsmurugappan
   - lance
   - pmorie
   - thisisnotapril
diff --git a/control-plane/config/200-controller/200-controller-cluster-role.yaml b/control-plane/config/200-controller/200-controller-cluster-role.yaml
index dfb85a5d7..c77a7e915 100644
--- a/control-plane/config/200-controller/200-controller-cluster-role.yaml
+++ b/control-plane/config/200-controller/200-controller-cluster-role.yaml
@@ -69,6 +69,50 @@ rules:
       - get
       - watch
 
+  # Scheduler permissions
+  - apiGroups:
+      - "*"
+    resources:
+      - nodes
+    verbs:
+      - get
+      - list
+      - watch
+  - apiGroups:
+      - "apps"
+    resources:
+      - statefulsets
+      - statefulsets/scale
+    verbs:
+      - get
+      - list
+      - watch
+      - update
+      - patch
+
+  # Internal APIs
+  - apiGroups:
+      - "internal.kafka.eventing.knative.dev"
+    resources:
+      - "consumers"
+      - "consumers/status"
+      - "consumergroups"
+      - "consumergroups/status"
+    verbs:
+      - create
+      - get
+      - list
+      - watch
+      - patch
+      - update
+  - apiGroups:
+      - "internal.kafka.eventing.knative.dev"
+    resources:
+      - "consumers/finalizers"
+      - "consumergroups/finalizers"
+    verbs:
+      - update
+
   # Eventing resources and statuses we care about
   - apiGroups:
       - "eventing.knative.dev"
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go
index e6078edab..02b7b1964 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go
@@ -191,3 +191,13 @@ func (cg *ConsumerGroup) IsNotScheduled() bool {
 	cond := cg.Status.GetCondition(ConditionConsumerGroupConsumersScheduled)
 	return cond.IsFalse() || cond.IsUnknown()
 }
+
+func (cg *ConsumerGroup) HasDeadLetterSink() bool {
+	return hasDeadLetterSink(cg.Spec.Template.Spec.Delivery)
+}
+
+func hasDeadLetterSink(d *DeliverySpec) bool {
+	return d != nil && d.DeliverySpec != nil &&
+		d.DeliverySpec.DeadLetterSink != nil &&
+		(d.DeliverySpec.DeadLetterSink.Ref != nil || d.DeliverySpec.DeadLetterSink.URI != nil)
+}
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
index 92d17daca..612d55d57 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
@@ -230,3 +230,7 @@ func (c *Consumer) GetConsumerGroup() *metav1.OwnerReference {
 	}
 	return nil
 }
+
+func (c Consumer) HasDeadLetterSink() bool {
+	return hasDeadLetterSink(c.Spec.Delivery)
+}
diff --git a/control-plane/pkg/reconciler/base/reconciler.go b/control-plane/pkg/reconciler/base/reconciler.go
index 5b4b8f154..2fe24b639 100644
--- a/control-plane/pkg/reconciler/base/reconciler.go
+++ b/control-plane/pkg/reconciler/base/reconciler.go
@@ -199,7 +199,10 @@ func (r *Reconciler) UpdateDataPlaneConfigMap(ctx context.Context, contract *con
 		return fmt.Errorf("failed to marshal contract: %w", err)
 	}
 
-	// Update config map data. TODO is it safe to update this config map? do we need to copy it?
+	// Update config map data.
+	if configMap.BinaryData == nil {
+		configMap.BinaryData = make(map[string][]byte, 1)
+	}
 	configMap.BinaryData[ConfigMapDataKey] = data
 
 	_, err = r.KubeClient.CoreV1().ConfigMaps(configMap.Namespace).Update(ctx, configMap, metav1.UpdateOptions{})
diff --git a/control-plane/pkg/reconciler/consumer/consumer.go b/control-plane/pkg/reconciler/consumer/consumer.go
index 062e08a2f..65b7a6078 100644
--- a/control-plane/pkg/reconciler/consumer/consumer.go
+++ b/control-plane/pkg/reconciler/consumer/consumer.go
@@ -320,11 +320,6 @@ func (r Reconciler) schedule(ctx context.Context, logger *zap.Logger, c *kafkain
 		return false, err
 	}
 
-	// Check if the pod is running.
-	if p.Status.Phase != corev1.PodRunning {
-		return false, nil
-	}
-
 	// Get contract associated with the pod.
 	cmName, err := cmNameFromPod(p, c)
 	if err != nil {
@@ -338,6 +333,14 @@ func (r Reconciler) schedule(ctx context.Context, logger *zap.Logger, c *kafkain
 		return false, fmt.Errorf("failed to get or create data plane ConfigMap %s/%s: %w", p.GetNamespace(), cmName, err)
 	}
 
+	// Check if the pod is running after trying to
+	// get or create the associated ConfigMap, since
+	// it won't become ready until we have created the
+	// ConfigMap
+	if p.Status.Phase != corev1.PodRunning {
+		return false, nil
+	}
+
 	ct, err := b.GetDataPlaneConfigMapData(logger, cm)
 	if err != nil {
 		return false, fmt.Errorf("failed to get contract from ConfigMap %s/%s: %w", p.GetNamespace(), cmName, err)
@@ -375,7 +378,7 @@ func (r Reconciler) commonReconciler(p *corev1.Pod, cmName string) base.Reconcil
 func cmNameFromPod(p *corev1.Pod, c *kafkainternals.Consumer) (string, error) {
 	var vDp *corev1.Volume
 	for i, v := range p.Spec.Volumes {
-		if v.Name == eventing.ConfigMapVolumeName && v.ConfigMap != nil && v.ConfigMap.Name != "" {
+		if v.Name == kafkainternals.DispatcherVolumeName && v.ConfigMap != nil && v.ConfigMap.Name != "" {
 			vDp = &p.Spec.Volumes[i]
 			break
 		}
diff --git a/control-plane/pkg/reconciler/consumergroup/consumergroup.go b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
index 6445c606f..401de505b 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
@@ -18,6 +18,7 @@ package consumergroup
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"math"
 	"sort"
@@ -39,6 +40,11 @@ import (
 	kafkainternalslisters "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/listers/eventing/v1alpha1"
 )
 
+var (
+	ErrNoSubscriberURI     = errors.New("no subscriber URI resolved")
+	ErrNoDeadLetterSinkURI = errors.New("no dead letter sink URI resolved")
+)
+
 type schedulerFunc func(s string) scheduler.Scheduler
 
 type Reconciler struct {
@@ -60,6 +66,20 @@ func (r Reconciler) ReconcileKind(ctx context.Context, cg *kafkainternals.Consum
 	if err := r.reconcileConsumers(ctx, cg); err != nil {
 		return err
 	}
+
+	if err := r.propagateStatus(cg); err != nil {
+		return cg.MarkReconcileConsumersFailed("PropagateConsumerStatus", err)
+	}
+
+	if cg.Status.SubscriberURI == nil {
+		_ = cg.MarkReconcileConsumersFailed("PropagateSubscriberURI", ErrNoSubscriberURI)
+		return nil
+	}
+	if cg.HasDeadLetterSink() && cg.Status.DeadLetterSinkURI == nil {
+		_ = cg.MarkReconcileConsumersFailed("PropagateDeadLetterSinkURI", ErrNoDeadLetterSinkURI)
+		return nil
+	}
+
 	cg.MarkReconcileConsumersSucceeded()
 
 	return nil
@@ -241,6 +261,30 @@ func (r Reconciler) joinConsumersByPlacement(placements []eventingduckv1alpha1.P
 	return placementConsumers
 }
 
+func (r Reconciler) propagateStatus(cg *kafkainternals.ConsumerGroup) error {
+	consumers, err := r.ConsumerLister.Consumers(cg.GetNamespace()).List(labels.SelectorFromSet(cg.Spec.Selector))
+	if err != nil {
+		return fmt.Errorf("failed to list consumers for selector %+v: %w", cg.Spec.Selector, err)
+	}
+	count := int32(0)
+	for _, c := range consumers {
+		if c.IsReady() {
+			if c.Spec.VReplicas != nil {
+				count += *c.Spec.VReplicas
+			}
+			if c.Status.SubscriberURI != nil {
+				cg.Status.SubscriberURI = c.Status.SubscriberURI
+			}
+			if c.Status.DeliveryStatus.DeadLetterSinkURI != nil {
+				cg.Status.DeliveryStatus.DeadLetterSinkURI = c.Status.DeadLetterSinkURI
+			}
+		}
+	}
+	cg.Status.Replicas = pointer.Int32(count)
+
+	return nil
+}
+
 var (
 	_ consumergroup.Interface = &Reconciler{}
 )
diff --git a/control-plane/pkg/reconciler/consumergroup/evictor.go b/control-plane/pkg/reconciler/consumergroup/evictor.go
index 28b2149d6..8e5d10675 100644
--- a/control-plane/pkg/reconciler/consumergroup/evictor.go
+++ b/control-plane/pkg/reconciler/consumergroup/evictor.go
@@ -58,7 +58,7 @@ func newEvictor(ctx context.Context, fields ...zap.Field) evictor {
 	}
 }
 
-func (e *evictor) evict(pod *corev1.Pod, vpod scheduler.VPod, from *eventingduckv1alpha1.Placement) error {
+func (e evictor) evict(pod *corev1.Pod, vpod scheduler.VPod, from *eventingduckv1alpha1.Placement) error {
 	key := vpod.GetKey()
 
 	logger := e.logger.
diff --git a/control-plane/pkg/reconciler/source/v2/controllerv2.go b/control-plane/pkg/reconciler/source/v2/controllerv2.go
index b9b072de1..7bb4b69f3 100644
--- a/control-plane/pkg/reconciler/source/v2/controllerv2.go
+++ b/control-plane/pkg/reconciler/source/v2/controllerv2.go
@@ -20,9 +20,10 @@ import (
 	"context"
 
 	"k8s.io/client-go/tools/cache"
+	"knative.dev/pkg/controller"
+
 	consumergroupclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client"
 	consumergroupinformer "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumergroup"
-	"knative.dev/pkg/controller"
 
 	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
 
@@ -47,9 +48,7 @@ func NewController(ctx context.Context, configs *config.Env) *controller.Impl {
 
 	impl := kafkasource.NewImpl(ctx, r)
 
-	kafkaInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
-		Handler: controller.HandleAll(impl.Enqueue),
-	})
+	kafkaInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
 
 	// ConsumerGroup changes and enqueue associated KafkaSource
 	consumerGroupInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{


Test patch:
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types_test.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types_test.go
index f44f42c35..8a0e88a4d 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types_test.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types_test.go
@@ -17,10 +17,14 @@
 package v1alpha1
 
 import (
+	"context"
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
 )
 
 func TestConsumerGroup_GetUserFacingResourceRef(t *testing.T) {
@@ -124,3 +128,58 @@ func TestConsumerGroup_GetUserFacingResourceRef(t *testing.T) {
 		})
 	}
 }
+
+func TestHasDeadLetterSink(t *testing.T) {
+
+	type HasDeadLetterSink interface {
+		HasDeadLetterSink() bool
+		apis.Defaultable
+	}
+
+	tt := []struct {
+		name              string
+		object            HasDeadLetterSink
+		hasDeadLetterSink bool
+	}{
+		{
+			name: "consumer group has dead letter sink",
+			object: &ConsumerGroup{Spec: ConsumerGroupSpec{Template: ConsumerTemplateSpec{Spec: ConsumerSpec{
+				Delivery: &DeliverySpec{DeliverySpec: &eventingduck.DeliverySpec{
+					DeadLetterSink: &duckv1.Destination{URI: apis.HTTP("localhost")},
+				}},
+			}}}},
+			hasDeadLetterSink: true,
+		},
+		{
+			name:              "consumer group has no dead letter sink",
+			object:            &ConsumerGroup{},
+			hasDeadLetterSink: false,
+		},
+		{
+			name: "consumer has dead letter sink",
+			object: &Consumer{Spec: ConsumerSpec{
+				Delivery: &DeliverySpec{DeliverySpec: &eventingduck.DeliverySpec{
+					DeadLetterSink: &duckv1.Destination{URI: apis.HTTP("localhost")},
+				}},
+			}},
+			hasDeadLetterSink: true,
+		},
+		{
+			name:              "consumer has no dead letter sink",
+			object:            &Consumer{},
+			hasDeadLetterSink: false,
+		},
+	}
+
+	for _, tc := range tt {
+		t.Run(tc.name, func(t *testing.T) {
+			tc.object.SetDefaults(context.Background())
+			got := tc.object.HasDeadLetterSink()
+
+			if tc.hasDeadLetterSink != got {
+				t.Errorf("want %v got %v", tc.hasDeadLetterSink, got)
+			}
+		})
+	}
+
+}
diff --git a/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go b/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
index 81f9e3ec3..9e9ef95f3 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
@@ -25,6 +25,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	clientgotesting "k8s.io/client-go/testing"
+	"k8s.io/utils/pointer"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
 	. "knative.dev/pkg/reconciler/testing"
@@ -32,6 +33,7 @@ import (
 	eventingduckv1alpha1 "knative.dev/eventing/pkg/apis/duck/v1alpha1"
 	"knative.dev/eventing/pkg/scheduler"
 
+	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"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/reconciler/eventing/v1alpha1/consumergroup"
@@ -115,13 +117,91 @@ func TestReconcileKind(t *testing.T) {
 								),
 							)),
 							ConsumerGroupReplicas(2),
+							ConsumerGroupStatusReplicas(0),
 							ConsumerForTrigger(),
 						)
 						cg.Status.Placements = []eventingduckv1alpha1.Placement{
 							{PodName: "p1", VReplicas: 1},
 							{PodName: "p2", VReplicas: 1},
 						}
-						cg.MarkReconcileConsumersSucceeded()
+						_ = cg.MarkReconcileConsumersFailed("PropagateSubscriberURI", ErrNoSubscriberURI)
+						cg.MarkScheduleSucceeded()
+						return cg
+					}(),
+				},
+			},
+		},
+		{
+			Name: "Consumers in multiple pods, one exists - not ready",
+			Objects: []runtime.Object{
+				NewConsumer(2,
+					ConsumerSpec(NewConsumerSpec(
+						ConsumerTopics("t1", "t2"),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+						ConsumerVReplicas(1),
+						ConsumerPlacement(kafkainternals.PodBind{
+							PodName:      "p2",
+							PodNamespace: systemNamespace,
+						}),
+					)),
+				),
+				NewConsumerGroup(
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics("t1", "t2"),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+					)),
+					ConsumerForTrigger(),
+					ConsumerGroupReplicas(2),
+				),
+			},
+			Key: ConsumerGroupTestKey,
+			OtherTestData: map[string]interface{}{
+				testSchedulerKey: SchedulerFunc(func(vpod scheduler.VPod) ([]eventingduckv1alpha1.Placement, error) {
+					return []eventingduckv1alpha1.Placement{
+						{PodName: "p1", VReplicas: 1},
+						{PodName: "p2", VReplicas: 1},
+					}, nil
+				}),
+			},
+			WantCreates: []runtime.Object{
+				NewConsumer(1,
+					ConsumerSpec(NewConsumerSpec(
+						ConsumerTopics("t1", "t2"),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+						ConsumerVReplicas(1),
+						ConsumerPlacement(kafkainternals.PodBind{PodName: "p1", PodNamespace: systemNamespace}),
+					)),
+				),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: func() runtime.Object {
+						cg := NewConsumerGroup(
+							ConsumerGroupConsumerSpec(NewConsumerSpec(
+								ConsumerTopics("t1", "t2"),
+								ConsumerConfigs(
+									ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+									ConsumerGroupIdConfig("my.group.id"),
+								),
+							)),
+							ConsumerGroupReplicas(2),
+							ConsumerGroupStatusReplicas(0),
+							ConsumerForTrigger(),
+						)
+						cg.Status.Placements = []eventingduckv1alpha1.Placement{
+							{PodName: "p1", VReplicas: 1},
+							{PodName: "p2", VReplicas: 1},
+						}
+						_ = cg.MarkReconcileConsumersFailed("PropagateSubscriberURI", ErrNoSubscriberURI)
 						cg.MarkScheduleSucceeded()
 						return cg
 					}(),
@@ -129,7 +209,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 		},
 		{
-			Name: "Consumers in multiple pods, one exists",
+			Name: "Consumers in multiple pods, one exists - ready",
 			Objects: []runtime.Object{
 				NewConsumer(2,
 					ConsumerSpec(NewConsumerSpec(
@@ -144,6 +224,7 @@ func TestReconcileKind(t *testing.T) {
 							PodNamespace: systemNamespace,
 						}),
 					)),
+					ConsumerReady(),
 				),
 				NewConsumerGroup(
 					ConsumerGroupConsumerSpec(NewConsumerSpec(
@@ -191,6 +272,7 @@ func TestReconcileKind(t *testing.T) {
 								),
 							)),
 							ConsumerGroupReplicas(2),
+							ConsumerGroupStatusReplicas(1),
 							ConsumerForTrigger(),
 						)
 						cg.Status.Placements = []eventingduckv1alpha1.Placement{
@@ -199,13 +281,131 @@ func TestReconcileKind(t *testing.T) {
 						}
 						cg.MarkReconcileConsumersSucceeded()
 						cg.MarkScheduleSucceeded()
+						cg.Status.SubscriberURI = ConsumerSubscriberURI
+						return cg
+					}(),
+				},
+			},
+		},
+		{
+			Name: "Consumers in multiple pods, one exists - ready, increase replicas, propagate dead letter sink URI",
+			Objects: []runtime.Object{
+				NewConsumer(2,
+					ConsumerSpec(NewConsumerSpec(
+						ConsumerTopics("t1", "t2"),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered,
+							NewConsumerSpecDeliveryDeadLetterSink(),
+						)),
+						ConsumerVReplicas(1),
+						ConsumerPlacement(kafkainternals.PodBind{
+							PodName:      "p2",
+							PodNamespace: systemNamespace,
+						}),
+					)),
+					ConsumerReady(),
+				),
+				NewConsumerGroup(
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics("t1", "t2"),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered,
+							NewConsumerSpecDeliveryDeadLetterSink(),
+						)),
+					)),
+					ConsumerGroupReplicas(3),
+					ConsumerForTrigger(),
+				),
+			},
+			Key: ConsumerGroupTestKey,
+			OtherTestData: map[string]interface{}{
+				testSchedulerKey: SchedulerFunc(func(vpod scheduler.VPod) ([]eventingduckv1alpha1.Placement, error) {
+					return []eventingduckv1alpha1.Placement{
+						{PodName: "p1", VReplicas: 1},
+						{PodName: "p2", VReplicas: 2},
+					}, nil
+				}),
+			},
+			WantCreates: []runtime.Object{
+				NewConsumer(1,
+					ConsumerSpec(NewConsumerSpec(
+						ConsumerTopics("t1", "t2"),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered,
+							NewConsumerSpecDeliveryDeadLetterSink(),
+						)),
+						ConsumerVReplicas(1),
+						ConsumerPlacement(kafkainternals.PodBind{PodName: "p1", PodNamespace: systemNamespace}),
+					)),
+				),
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				clientgotesting.NewUpdateAction(
+					schema.GroupVersionResource{
+						Group:    kafkainternals.SchemeGroupVersion.Group,
+						Version:  kafkainternals.SchemeGroupVersion.Version,
+						Resource: "consumers",
+					},
+					ConsumerNamespace,
+					NewConsumer(2,
+						ConsumerSpec(NewConsumerSpec(
+							ConsumerTopics("t1", "t2"),
+							ConsumerConfigs(
+								ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+								ConsumerGroupIdConfig("my.group.id"),
+							),
+							ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered,
+								NewConsumerSpecDeliveryDeadLetterSink(),
+							)),
+							ConsumerVReplicas(2),
+							ConsumerPlacement(kafkainternals.PodBind{PodName: "p2", PodNamespace: systemNamespace}),
+						)),
+						ConsumerReady(),
+					),
+				),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: func() runtime.Object {
+						cg := NewConsumerGroup(
+							ConsumerGroupConsumerSpec(NewConsumerSpec(
+								ConsumerTopics("t1", "t2"),
+								ConsumerConfigs(
+									ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+									ConsumerGroupIdConfig("my.group.id"),
+								),
+								ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered,
+									NewConsumerSpecDeliveryDeadLetterSink(),
+								)),
+							)),
+							ConsumerGroupReplicas(3),
+							ConsumerForTrigger(),
+						)
+						cg.Status.Placements = []eventingduckv1alpha1.Placement{
+							{PodName: "p1", VReplicas: 1},
+							{PodName: "p2", VReplicas: 2},
+						}
+						cg.MarkReconcileConsumersSucceeded()
+						cg.MarkScheduleSucceeded()
+						cg.Status.SubscriberURI = ConsumerSubscriberURI
+						cg.Status.DeadLetterSinkURI = ConsumerDeadLetterSinkURI
+						cg.Status.Replicas = pointer.Int32(1)
 						return cg
 					}(),
 				},
 			},
 		},
 		{
-			Name: "Consumers in multiple pods, one exists, increase replicas",
+			Name: "Consumers in multiple pods, one exists - ready, increase replicas",
 			Objects: []runtime.Object{
 				NewConsumer(2,
 					ConsumerSpec(NewConsumerSpec(
@@ -220,6 +420,7 @@ func TestReconcileKind(t *testing.T) {
 							PodNamespace: systemNamespace,
 						}),
 					)),
+					ConsumerReady(),
 				),
 				NewConsumerGroup(
 					ConsumerGroupConsumerSpec(NewConsumerSpec(
@@ -273,6 +474,7 @@ func TestReconcileKind(t *testing.T) {
 							ConsumerVReplicas(2),
 							ConsumerPlacement(kafkainternals.PodBind{PodName: "p2", PodNamespace: systemNamespace}),
 						)),
+						ConsumerReady(),
 					),
 				),
 			},
@@ -296,6 +498,8 @@ func TestReconcileKind(t *testing.T) {
 						}
 						cg.MarkReconcileConsumersSucceeded()
 						cg.MarkScheduleSucceeded()
+						cg.Status.SubscriberURI = ConsumerSubscriberURI
+						cg.Status.Replicas = pointer.Int32(1)
 						return cg
 					}(),
 				},
@@ -392,8 +596,9 @@ func TestReconcileKind(t *testing.T) {
 							{PodName: "p1", VReplicas: 1},
 							{PodName: "p2", VReplicas: 1},
 						}
-						cg.MarkReconcileConsumersSucceeded()
+						_ = cg.MarkReconcileConsumersFailed("PropagateSubscriberURI", ErrNoSubscriberURI)
 						cg.MarkScheduleSucceeded()
+						cg.Status.Replicas = pointer.Int32(0)
 						return cg
 					}(),
 				},
diff --git a/control-plane/pkg/reconciler/testing/objects_consumer.go b/control-plane/pkg/reconciler/testing/objects_consumer.go
index 4797e71f5..bb2a6ca3b 100644
--- a/control-plane/pkg/reconciler/testing/objects_consumer.go
+++ b/control-plane/pkg/reconciler/testing/objects_consumer.go
@@ -22,7 +22,9 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/pointer"
 	"knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
+	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	v1 "knative.dev/eventing/pkg/apis/eventing/v1"
+	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
 	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
@@ -34,6 +36,11 @@ const (
 	ConsumerNamespace  = "test-cg-ns"
 )
 
+var (
+	ConsumerSubscriberURI     = apis.HTTP("localhost")
+	ConsumerDeadLetterSinkURI = apis.HTTP("dls.com")
+)
+
 type ConsumerOption func(cg *kafkainternals.Consumer)
 
 type ConsumerSpecOption func(c *kafkainternals.ConsumerSpec)
@@ -84,9 +91,24 @@ func NewConsumerSpec(opts ...ConsumerSpecOption) kafkainternals.ConsumerSpec {
 	return *spec
 }
 
-func NewConsumerSpecDelivery(order internals.DeliveryOrdering) *kafkainternals.DeliverySpec {
-	return &kafkainternals.DeliverySpec{
-		Ordering: order,
+type DeliverySpecOption func(spec *kafkainternals.DeliverySpec)
+
+func NewConsumerSpecDelivery(order internals.DeliveryOrdering, options ...DeliverySpecOption) *kafkainternals.DeliverySpec {
+	d := &kafkainternals.DeliverySpec{Ordering: order}
+
+	for _, opt := range options {
+		opt(d)
+	}
+
+	return d
+}
+
+func NewConsumerSpecDeliveryDeadLetterSink() DeliverySpecOption {
+	return func(spec *kafkainternals.DeliverySpec) {
+		if spec.DeliverySpec == nil {
+			spec.DeliverySpec = &eventingduck.DeliverySpec{}
+		}
+		spec.DeliverySpec.DeadLetterSink = &duckv1.Destination{URI: ConsumerDeadLetterSinkURI}
 	}
 }
 
@@ -192,3 +214,15 @@ func ConsumerForTrigger() ConsumerGroupOption {
 		})
 	}
 }
+
+func ConsumerReady() ConsumerOption {
+	return func(c *kafkainternals.Consumer) {
+		c.MarkBindSucceeded()
+		c.MarkReconcileContractSucceeded()
+		c.Status.SubscriberURI = ConsumerSubscriberURI
+
+		if c.HasDeadLetterSink() {
+			c.Status.DeadLetterSinkURI = ConsumerDeadLetterSinkURI
+		}
+	}
+}
diff --git a/control-plane/pkg/reconciler/testing/objects_consumergroup.go b/control-plane/pkg/reconciler/testing/objects_consumergroup.go
index a569f4f9a..79cca14d2 100644
--- a/control-plane/pkg/reconciler/testing/objects_consumergroup.go
+++ b/control-plane/pkg/reconciler/testing/objects_consumergroup.go
@@ -115,6 +115,12 @@ func ConsumerGroupReplicas(replicas int32) ConsumerGroupOption {
 	}
 }
 
+func ConsumerGroupStatusReplicas(replicas int32) ConsumerGroupOption {
+	return func(cg *kafkainternals.ConsumerGroup) {
+		cg.Status.Replicas = pointer.Int32Ptr(replicas)
+	}
+}
+
 func ConsumerGroupReplicasStatus(replicas int32) ConsumerGroupOption {
 	return func(cg *kafkainternals.ConsumerGroup) {
 		cg.Status.Replicas = pointer.Int32Ptr(replicas)