Repository: knative-sandbox/eventing-kafka-broker
Pull request #1901 (2022-02-15 10:24:05)

Add Consumer reply validation
Add enabled flag to distinguish between the enabled
reply strategy.

Update KafkaSource reconciler to set the reply strategy
to `NoReply`.

Update Trigger reconciler to set the reply strategy
to `TopicReply`.

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

Part of #1537

## Proposed Changes

- Add Consumer reply validation

<!--
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.
-->


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

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1901). [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/1901?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1901](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1901?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (bdf2547) 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/1901/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/1901?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    #1901       +/-   ##
=============================================
+ 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/1901?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/1901/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/1901/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: |
| [...nternals/kafka/eventing/v1alpha1/consumer\_types.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1901/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9pbnRlcm5hbHMva2Fma2EvZXZlbnRpbmcvdjFhbHBoYTEvY29uc3VtZXJfdHlwZXMuZ28=) | | |
| [...als/kafka/eventing/v1alpha1/consumer\_validation.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1901/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9pbnRlcm5hbHMva2Fma2EvZXZlbnRpbmcvdjFhbHBoYTEvY29uc3VtZXJfdmFsaWRhdGlvbi5nbw==) | | |
| [control-plane/pkg/reconciler/consumer/consumer.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1901/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/broker/broker.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1901/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9icm9rZXIvYnJva2VyLmdv) | | |
| [...ontrol-plane/pkg/reconciler/consumer/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1901/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lci9jb250cm9sbGVyLmdv) | | |
| [control-plane/pkg/config/env\_config.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1901/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvY29uZmlnL2Vudl9jb25maWcuZ28=) | | |
| [control-plane/pkg/prober/cache.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1901/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcHJvYmVyL2NhY2hlLmdv) | | |
| [...rol-plane/pkg/reconciler/source/v2/controllerv2.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1901/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zb3VyY2UvdjIvY29udHJvbGxlcnYyLmdv) | | |
| ... and [58 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1901/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/1901?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/1901?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [8c0b9a8...bdf2547](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1901?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-unit-tests | bdf25477b29d3c45485b2a6e78cd7197e9ada935 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1901/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1493531557896392704) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-build-tests | bdf25477b29d3c45485b2a6e78cd7197e9ada935 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1901/pull-knative-sandbox-eventing-kafka-broker-build-tests/1493531557867032576) | true | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1901). [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-build-tests | bdf25477b29d3c45485b2a6e78cd7197e9ada935 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1901/pull-knative-sandbox-eventing-kafka-broker-build-tests/1493531557867032576) | true | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`
pull-knative-sandbox-eventing-kafka-broker-unit-tests | e82a2e471f45914e0667c22a80de8ae18e1eba25 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1901/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1493535981242945536) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1901). [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-unit-tests | 61f95c0e7fbd0bbec647763600059c6d6252342a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1901/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1493538766822838272) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1901). [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-unit-tests | 61f95c0e7fbd0bbec647763600059c6d6252342a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1901/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1493538766822838272) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl | 61f95c0e7fbd0bbec647763600059c6d6252342a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1901/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl/1493538776050307072) | 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/1901). [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-unit-tests | 61f95c0e7fbd0bbec647763600059c6d6252342a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1901/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1493538766822838272) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl | 61f95c0e7fbd0bbec647763600059c6d6252342a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1901/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl/1493538776050307072) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl`
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl | 61f95c0e7fbd0bbec647763600059c6d6252342a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1901/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl/1493538777732222976) | 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/1901). [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-unit-tests | 61f95c0e7fbd0bbec647763600059c6d6252342a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1901/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1493538766822838272) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl | 61f95c0e7fbd0bbec647763600059c6d6252342a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1901/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl/1493538776050307072) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl`
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl | 61f95c0e7fbd0bbec647763600059c6d6252342a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1901/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl/1493538777732222976) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl`
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain | 61f95c0e7fbd0bbec647763600059c6d6252342a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1901/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain/1493538779409944576) | 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/1901). [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/apis/internals/kafka/eventing/v1alpha1/consumer_validation.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1901/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1493577804149690368/artifacts/line-cov.html#file2) | 84.8% | 84.4% | -0.4

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-sasl-ssl | 61f95c0e7fbd0bbec647763600059c6d6252342a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1901/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl/1493538777732222976) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl`
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain | 3f8facfc35c7b74ff943a6e51ae1e8b915a2e653 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1901/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-plain/1493577804434903040) | 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/1901). [Your PR dashboard](https://gubernator.knative.dev/pr/pierDipi).

<details>

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

Patch:
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
index d28346249..7beb3bd2d 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
@@ -98,22 +98,28 @@ type ConsumerSpec struct {
 	CloudEventOverrides *duckv1.CloudEventOverrides `json:"ceOverrides,omitempty"`
 
 	// VReplicas is the number of virtual replicas for a consumer.
-	VReplicas *int32
+	VReplicas *int32 `json:"vReplicas"`
 
 	// PodBind represents a reference to the pod in which the consumer should be placed.
 	PodBind *PodBind `json:"podBind"`
 }
 
 type ReplyStrategy struct {
-	TopicReply *TopicReply
-	URLReply   *DestinationReply
-	NoReply    *struct{}
+	TopicReply *TopicReply       `json:"topicReply,omitempty"`
+	URLReply   *DestinationReply `json:"URLReply,omitempty"`
+	NoReply    *NoReply          `json:"noReply,omitempty"`
+}
+
+type NoReply struct {
+	Enabled bool `json:"enabled"`
 }
 
 type TopicReply struct {
+	Enabled bool `json:"enabled"`
 }
 
 type DestinationReply struct {
+	Enabled     bool `json:"enabled"`
 	Destination duckv1.Destination
 }
 
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_validation.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_validation.go
index f8dd77591..db02d7e4b 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_validation.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_validation.go
@@ -45,6 +45,7 @@ func (cs *ConsumerSpec) Validate(ctx context.Context) *apis.FieldError {
 		cs.Subscriber.Validate(ctx).ViaField("subscriber"),
 		cs.PodBind.Validate(ctx).ViaField("podBind"),
 		cs.CloudEventOverrides.Validate(ctx).ViaField("ceOverrides"),
+		cs.Reply.Validate(ctx).ViaField("reply"),
 	)
 	return err
 }
@@ -73,6 +74,27 @@ func (f *Filters) Validate(ctx context.Context) *apis.FieldError {
 	return nil
 }
 
+func (in *ReplyStrategy) Validate(ctx context.Context) *apis.FieldError {
+	if in == nil {
+		return nil
+	}
+	setCounter := 0
+	if in.TopicReply != nil && in.TopicReply.Enabled {
+		setCounter++
+	}
+	if in.URLReply != nil && in.URLReply.Enabled {
+		setCounter++
+	}
+	if in.NoReply != nil && in.NoReply.Enabled {
+		setCounter++
+	}
+
+	if setCounter > 1 {
+		return apis.ErrMultipleOneOf("topicReply", "URLReply", "NoReply")
+	}
+	return nil
+}
+
 func (p *PodBind) Validate(ctx context.Context) *apis.FieldError {
 	if p == nil {
 		return apis.ErrMissingField("")
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go
index 42bfddaa6..2498b1e2d 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go
@@ -451,6 +451,22 @@ func (in *Filters) DeepCopy() *Filters {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *NoReply) DeepCopyInto(out *NoReply) {
+	*out = *in
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new NoReply.
+func (in *NoReply) DeepCopy() *NoReply {
+	if in == nil {
+		return nil
+	}
+	out := new(NoReply)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *PodBind) DeepCopyInto(out *PodBind) {
 	*out = *in
@@ -482,7 +498,7 @@ func (in *ReplyStrategy) DeepCopyInto(out *ReplyStrategy) {
 	}
 	if in.NoReply != nil {
 		in, out := &in.NoReply, &out.NoReply
-		*out = new(struct{})
+		*out = new(NoReply)
 		**out = **in
 	}
 	return
diff --git a/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index 57fb269c2..d63fee73b 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -430,7 +430,7 @@ func (r *Reconciler) reconcileSubscribers(ctx context.Context, kafkaClient saram
 				UID:                s.UID,
 				ObservedGeneration: s.Generation,
 				Ready:              corev1.ConditionFalse,
-				Message:            fmt.Sprint("Subscription not ready", err),
+				Message:            fmt.Sprintf("Subscription not ready: %v", err),
 			})
 			globalErr = multierr.Append(globalErr, err)
 		} else {
@@ -479,8 +479,13 @@ func (r *Reconciler) reconcileInitialOffset(ctx context.Context, channel *messag
 }
 
 func (r *Reconciler) getSubscriberConfig(ctx context.Context, channel *messagingv1beta1.KafkaChannel, subscriber *v1.SubscriberSpec) (*contract.Egress, error) {
+	subscriberURI := subscriber.SubscriberURI.String()
+	if subscriberURI == "" {
+		return nil, fmt.Errorf("failed to resolve Subscription.Spec.Subscriber: empty subscriber URI")
+	}
+
 	egress := &contract.Egress{
-		Destination:   subscriber.SubscriberURI.String(),
+		Destination:   subscriberURI,
 		ConsumerGroup: consumerGroup(channel, subscriber),
 		DeliveryOrder: DefaultDeliveryOrder,
 		Uid:           string(subscriber.UID),
diff --git a/control-plane/pkg/reconciler/consumer/consumer.go b/control-plane/pkg/reconciler/consumer/consumer.go
index d225b5966..b4b58c264 100644
--- a/control-plane/pkg/reconciler/consumer/consumer.go
+++ b/control-plane/pkg/reconciler/consumer/consumer.go
@@ -252,11 +252,11 @@ func (r Reconciler) reconcileReplyStrategy(ctx context.Context, c *kafkainternal
 	if c.Spec.Reply == nil {
 		return nil
 	}
-	if c.Spec.Reply.NoReply != nil {
+	if c.Spec.Reply.NoReply != nil && c.Spec.Reply.NoReply.Enabled {
 		egress.ReplyStrategy = &contract.Egress_DiscardReply{}
 		return nil
 	}
-	if c.Spec.Reply.URLReply != nil {
+	if c.Spec.Reply.URLReply != nil && c.Spec.Reply.URLReply.Enabled {
 		destination, err := r.Resolver.URIFromDestinationV1(ctx, c.Spec.Reply.URLReply.Destination, c)
 		if err != nil {
 			return fmt.Errorf("failed to resolve reply destination: %w", err)
@@ -266,7 +266,7 @@ func (r Reconciler) reconcileReplyStrategy(ctx context.Context, c *kafkainternal
 		}
 		return nil
 	}
-	if c.Spec.Reply.TopicReply != nil {
+	if c.Spec.Reply.TopicReply != nil && c.Spec.Reply.TopicReply.Enabled {
 		egress.ReplyStrategy = &contract.Egress_ReplyToOriginalTopic{}
 		return nil
 	}
diff --git a/control-plane/pkg/reconciler/source/v2/sourcev2.go b/control-plane/pkg/reconciler/source/v2/sourcev2.go
index 88e1e1e20..6558ccf0a 100644
--- a/control-plane/pkg/reconciler/source/v2/sourcev2.go
+++ b/control-plane/pkg/reconciler/source/v2/sourcev2.go
@@ -114,6 +114,7 @@ func (r Reconciler) reconcileConsumerGroup(ctx context.Context, ks *sources.Kafk
 						Ordering: DefaultDeliveryOrder,
 					},
 					Subscriber: ks.Spec.Sink,
+					Reply:      &internalscg.ReplyStrategy{NoReply: &internalscg.NoReply{Enabled: true}},
 				},
 			},
 			Replicas: ks.Spec.Consumers,
diff --git a/control-plane/pkg/reconciler/trigger/v2/triggerv2.go b/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
index eade6ff7b..0e19e9b78 100644
--- a/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
+++ b/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
@@ -140,6 +140,7 @@ func (r Reconciler) reconcileConsumerGroup(ctx context.Context, trigger *eventin
 						Filters: trigger.Spec.Filters,
 					},
 					Subscriber: trigger.Spec.Subscriber,
+					Reply:      &internalscg.ReplyStrategy{TopicReply: &internalscg.TopicReply{Enabled: true}},
 				},
 			},
 		},
diff --git a/hack/release.sh b/hack/release.sh
index e6a82d0ac..b47f0f00d 100755
--- a/hack/release.sh
+++ b/hack/release.sh
@@ -66,6 +66,7 @@ function build_release() {
     "${EVENTING_KAFKA_SINK_PROMETHEUS_OPERATOR_ARTIFACT}"
     "${EVENTING_KAFKA_CHANNEL_ARTIFACT}"
     "${EVENTING_KAFKA_CHANNEL_PROMETHEUS_OPERATOR_ARTIFACT}"
+    "${EVENTING_KAFKA_POST_INSTALL_ARTIFACT}"
   )
 
   # ARTIFACTS_TO_PUBLISH has to be a string, not an array.


Test patch:
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go
index 3e00b64c7..504c7e9e8 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go
@@ -118,6 +118,45 @@ func TestConsumerGroup_Validate(t *testing.T) {
 			},
 			wantErr: false,
 		},
+		{
+			name: "invalid multiple reply strategies (topic, destination)",
+			ctx:  context.Background(),
+			given: &ConsumerGroup{
+				Spec: ConsumerGroupSpec{
+					Replicas: pointer.Int32Ptr(1),
+					Selector: map[string]string{"app": "app"},
+					Template: ConsumerTemplateSpec{
+						Spec: ConsumerSpec{
+							Topics: []string{"t1"},
+							Configs: ConsumerConfigs{
+								Configs: map[string]string{
+									"group.id":          "g1",
+									"bootstrap.servers": "kafka:9092",
+								},
+							},
+							Reply: &ReplyStrategy{
+								TopicReply: &TopicReply{Enabled: true},
+								URLReply:   &DestinationReply{Enabled: true},
+							},
+							Delivery: &DeliverySpec{
+								DeliverySpec: &eventingduck.DeliverySpec{},
+							},
+							Subscriber: duckv1.Destination{
+								URI: &apis.URL{
+									Scheme: "http",
+									Host:   "127.0.0.1",
+								},
+							},
+							PodBind: &PodBind{
+								PodName:      "p-0",
+								PodNamespace: "ns",
+							},
+						},
+					},
+				},
+			},
+			wantErr: true,
+		},
 		{
 			name: "invalid no pod name",
 			ctx:  context.Background(),
diff --git a/control-plane/pkg/reconciler/channel/channel_test.go b/control-plane/pkg/reconciler/channel/channel_test.go
index c5e71fdac..a888eee34 100644
--- a/control-plane/pkg/reconciler/channel/channel_test.go
+++ b/control-plane/pkg/reconciler/channel/channel_test.go
@@ -353,6 +353,79 @@ func TestReconcileKind(t *testing.T) {
 				testProber: probertesting.MockProber(prober.StatusUnknown),
 			},
 		},
+		{
+			Name: "Reconciled failed - with single fresh subscriber without URI",
+			Objects: []runtime.Object{
+				NewChannel(WithSubscribers(Subscriber1(WithFreshSubscriber, WithNoSubscriberURI))),
+				NewService(),
+				ChannelReceiverPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				ChannelDispatcherPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
+				}),
+				NewConfigMapWithBinaryData(&env, nil),
+			},
+			Key: testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&env, &contract.Contract{
+					Generation: 1,
+					Resources: []*contract.Resource{
+						{
+							Uid:              ChannelUUID,
+							Topics:           []string{ChannelTopic()},
+							BootstrapServers: ChannelBootstrapServers,
+							Reference:        ChannelReference(),
+							Ingress: &contract.Ingress{
+								IngressType: &contract.Ingress_Path{
+									Path: receiver.Path(ChannelNamespace, ChannelName),
+								},
+							},
+							Egresses: []*contract.Egress{},
+						},
+					},
+				}),
+				ChannelReceiverPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+				ChannelDispatcherPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+			},
+			WantErr:                 true,
+			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewChannel(
+						WithInitKafkaChannelConditions,
+						StatusConfigParsed,
+						StatusConfigMapUpdatedReady(&env),
+						StatusTopicReadyWithName(ChannelTopic()),
+						StatusDataPlaneAvailable,
+						//StatusInitialOffsetsCommitted,
+						WithSubscribers(Subscriber1(WithFreshSubscriber, WithNoSubscriberURI)),
+					),
+				},
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+				Eventf(
+					corev1.EventTypeWarning,
+					"InternalError",
+					"failed to resolve subscriber config: failed to resolve Subscription.Spec.Subscriber: empty subscriber URI",
+				),
+			},
+		},
 		{
 			Name: "Reconciled normal - with single fresh subscriber - no auth",
 			Objects: []runtime.Object{
diff --git a/control-plane/pkg/reconciler/source/v2/sourcev2_test.go b/control-plane/pkg/reconciler/source/v2/sourcev2_test.go
index eec524aeb..070abf219 100644
--- a/control-plane/pkg/reconciler/source/v2/sourcev2_test.go
+++ b/control-plane/pkg/reconciler/source/v2/sourcev2_test.go
@@ -70,6 +70,7 @@ func TestReconcileKind(t *testing.T) {
 						ConsumerAuth(NewConsumerSpecAuth()),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerReply(ConsumerNoReply()),
 					)),
 					ConsumerGroupReplicas(1),
 				),
@@ -107,6 +108,7 @@ func TestReconcileKind(t *testing.T) {
 						ConsumerAuth(NewConsumerSpecAuth()),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerReply(ConsumerNoReply()),
 						ConsumerCloudEventOverrides(&duckv1.CloudEventOverrides{
 							Extensions: map[string]string{"a": "foo", "b": "foo"},
 						}),
@@ -144,6 +146,7 @@ func TestReconcileKind(t *testing.T) {
 						ConsumerAuth(NewConsumerSpecAuth()),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerReply(ConsumerNoReply()),
 					)),
 					ConsumerGroupReplicas(1),
 					ConsumerGroupReady,
@@ -166,6 +169,7 @@ func TestReconcileKind(t *testing.T) {
 							ConsumerAuth(NewConsumerSpecAuth()),
 							ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 							ConsumerSubscriber(NewSourceSink2Reference()),
+							ConsumerReply(ConsumerNoReply()),
 						)),
 						ConsumerGroupReplicas(1),
 						ConsumerGroupReady,
@@ -212,6 +216,7 @@ func TestReconcileKind(t *testing.T) {
 							ConsumerAuth(NewConsumerSpecAuth()),
 							ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 							ConsumerSubscriber(NewSourceSinkReference()),
+							ConsumerReply(ConsumerNoReply()),
 						)),
 						ConsumerGroupReplicas(1),
 						ConsumerGroupReady,
@@ -257,6 +262,7 @@ func TestReconcileKind(t *testing.T) {
 							ConsumerAuth(NewConsumerSpecAuth()),
 							ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 							ConsumerSubscriber(NewSourceSinkReference()),
+							ConsumerReply(ConsumerNoReply()),
 						)),
 						ConsumerGroupReplicas(1),
 					),
@@ -289,6 +295,7 @@ func TestReconcileKind(t *testing.T) {
 						ConsumerAuth(NewConsumerSpecAuth()),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerReply(ConsumerNoReply()),
 					)),
 					ConsumerGroupReplicas(1),
 					ConsumerGroupReady,
@@ -322,6 +329,7 @@ func TestReconcileKind(t *testing.T) {
 						ConsumerAuth(NewConsumerSpecAuth()),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerReply(ConsumerNoReply()),
 					)),
 					ConsumerGroupReplicas(1),
 				),
@@ -355,6 +363,7 @@ func TestReconcileKind(t *testing.T) {
 						ConsumerAuth(NewConsumerSpecAuth()),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerReply(ConsumerNoReply()),
 					)),
 					ConsumerGroupReplicas(1),
 					WithConsumerGroupFailed("failed", "failed"),
@@ -389,6 +398,7 @@ func TestReconcileKind(t *testing.T) {
 						ConsumerAuth(NewConsumerSpecAuth()),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerReply(ConsumerNoReply()),
 					)),
 					ConsumerGroupReplicas(1),
 					ConsumerGroupReplicasStatus(1),
diff --git a/control-plane/pkg/reconciler/testing/objects_channel.go b/control-plane/pkg/reconciler/testing/objects_channel.go
index cdfe2c636..a1147647b 100644
--- a/control-plane/pkg/reconciler/testing/objects_channel.go
+++ b/control-plane/pkg/reconciler/testing/objects_channel.go
@@ -261,6 +261,18 @@ func WithFreshSubscriber(sub *SubscriberInfo) {
 	sub.status = nil
 }
 
+func WithNoSubscriberURI(sub *SubscriberInfo) {
+	sub.spec.SubscriberURI = nil
+	if sub.status == nil {
+		sub.status = &eventingduckv1.SubscriberStatus{
+			UID:                sub.spec.UID,
+			ObservedGeneration: sub.spec.Generation,
+		}
+	}
+	sub.status.Ready = "False"
+	sub.status.Message = "Subscription not ready: failed to resolve subscriber config: failed to resolve Subscription.Spec.Subscriber: empty subscriber URI"
+}
+
 func WithUnreadySubscriber(sub *SubscriberInfo) {
 	sub.status.Ready = "False"
 }
diff --git a/control-plane/pkg/reconciler/testing/objects_consumer.go b/control-plane/pkg/reconciler/testing/objects_consumer.go
index bb2a6ca3b..1a99d7f2f 100644
--- a/control-plane/pkg/reconciler/testing/objects_consumer.go
+++ b/control-plane/pkg/reconciler/testing/objects_consumer.go
@@ -226,3 +226,17 @@ func ConsumerReady() ConsumerOption {
 		}
 	}
 }
+
+func ConsumerReply(s *kafkainternals.ReplyStrategy) ConsumerSpecOption {
+	return func(c *kafkainternals.ConsumerSpec) {
+		c.Reply = s
+	}
+}
+
+func ConsumerNoReply() *kafkainternals.ReplyStrategy {
+	return &kafkainternals.ReplyStrategy{NoReply: &kafkainternals.NoReply{Enabled: true}}
+}
+
+func ConsumerTopicReply() *kafkainternals.ReplyStrategy {
+	return &kafkainternals.ReplyStrategy{TopicReply: &kafkainternals.TopicReply{Enabled: true}}
+}
diff --git a/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go b/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go
index 6e6fe7e24..f723e0743 100644
--- a/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go
+++ b/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go
@@ -91,6 +91,7 @@ func TestReconcileKind(t *testing.T) {
 						),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
 						ConsumerFilters(NewConsumerSpecFilters()),
+						ConsumerReply(ConsumerTopicReply()),
 					)),
 				),
 			},
@@ -128,6 +129,7 @@ func TestReconcileKind(t *testing.T) {
 						),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 						ConsumerFilters(NewConsumerSpecFilters()),
+						ConsumerReply(ConsumerTopicReply()),
 					)),
 				),
 			},
@@ -166,6 +168,7 @@ func TestReconcileKind(t *testing.T) {
 						),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
 						ConsumerFilters(NewConsumerSpecFilters()),
+						ConsumerReply(ConsumerTopicReply()),
 					)),
 				),
 			},
@@ -244,6 +247,7 @@ func TestReconcileKind(t *testing.T) {
 							),
 							ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
 							ConsumerFilters(NewConsumerSpecFilters()),
+							ConsumerReply(ConsumerTopicReply()),
 						)),
 						ConsumerGroupReady,
 					),
@@ -291,6 +295,7 @@ func TestReconcileKind(t *testing.T) {
 							),
 							ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
 							ConsumerFilters(NewConsumerSpecFilters()),
+							ConsumerReply(ConsumerTopicReply()),
 						)),
 					),
 				},
@@ -326,6 +331,7 @@ func TestReconcileKind(t *testing.T) {
 						),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
 						ConsumerFilters(NewConsumerSpecFilters()),
+						ConsumerReply(ConsumerTopicReply()),
 					)),
 					ConsumerGroupReady,
 				),
@@ -362,6 +368,7 @@ func TestReconcileKind(t *testing.T) {
 						),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
 						ConsumerFilters(NewConsumerSpecFilters()),
+						ConsumerReply(ConsumerTopicReply()),
 					)),
 				),
 			},
@@ -397,6 +404,7 @@ func TestReconcileKind(t *testing.T) {
 						),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
 						ConsumerFilters(NewConsumerSpecFilters()),
+						ConsumerReply(ConsumerTopicReply()),
 					)),
 					WithConsumerGroupFailed("failed", "failed"),
 				),
@@ -433,6 +441,7 @@ func TestReconcileKind(t *testing.T) {
 						),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
 						ConsumerFilters(NewConsumerSpecFilters()),
+						ConsumerReply(ConsumerTopicReply()),
 					)),
 					WithDeadLetterSinkURI(url.String()),
 				),