Repository: knative-sandbox/eventing-kafka-broker
Pull request #2720 (2022-10-12 12:35:08)

Set MAX_POLL_INTERVAL_MS_CONFIG before creating consumer
We're setting it after a Consumer is created.

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>


knative-prow[bot]: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2720#" title="Author self-approved">pierDipi</a>*

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

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2720?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2720](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2720?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (a71e6ee) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/c3fa79cd280ab9e172e195e7634cff85ceaf53dc?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (c3fa79c) will **decrease** coverage by `5.87%`.
> The diff coverage is `n/a`.

```diff
@@             Coverage Diff              @@
##               main    #2720      +/-   ##
============================================
- Coverage     61.77%   55.90%   -5.88%     
============================================
  Files           148       75      -73     
  Lines         10434     7892    -2542     
  Branches        229        0     -229     
============================================
- Hits           6446     4412    -2034     
+ Misses         3531     3159     -372     
+ Partials        457      321     -136     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `?` | |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2720?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...lane/pkg/reconciler/consumergroup/consumergroup.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2720/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2NvbnN1bWVyZ3JvdXAuZ28=) | `70.84% <0.00%> (-1.11%)` | :arrow_down: |
| [...r/dispatcher/main/ConsumerVerticleFactoryImpl.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2720/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvbWFpbi9Db25zdW1lclZlcnRpY2xlRmFjdG9yeUltcGwuamF2YQ==) | | |
| [...atcher/impl/consumer/KafkaConsumerRecordUtils.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2720/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9LYWZrYUNvbnN1bWVyUmVjb3JkVXRpbHMuamF2YQ==) | | |
| [...r/dispatcher/impl/ResponseToKafkaTopicHandler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2720/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9SZXNwb25zZVRvS2Fma2FUb3BpY0hhbmRsZXIuamF2YQ==) | | |
| [...tive/eventing/kafka/broker/receiver/main/Main.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2720/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9tYWluL01haW4uamF2YQ==) | | |
| [...g/kafka/broker/core/security/KafkaClientsAuth.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2720/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvc2VjdXJpdHkvS2Fma2FDbGllbnRzQXV0aC5qYXZh) | | |
| [.../kafka/broker/core/eventbus/ContractPublisher.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2720/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvZXZlbnRidXMvQ29udHJhY3RQdWJsaXNoZXIuamF2YQ==) | | |
| [...ive/eventing/kafka/broker/core/utils/Shutdown.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2720/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdXRpbHMvU2h1dGRvd24uamF2YQ==) | | |
| [...ka/broker/core/security/KubernetesCredentials.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2720/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvc2VjdXJpdHkvS3ViZXJuZXRlc0NyZWRlbnRpYWxzLmphdmE=) | | |
| [...ting/kafka/broker/core/utils/ReferenceCounter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2720/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdXRpbHMvUmVmZXJlbmNlQ291bnRlci5qYXZh) | | |
| ... and [64 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2720/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

knative-prow[bot]: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2720#pullrequestreview-1139022473" title="Approved">matzew</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2720#" title="Author self-approved">pierDipi</a>*

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

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

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

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

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/unit-tests_eventing-kafka-broker_main/1580175238413422592) | true | `/test unit-tests`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/unit-tests_eventing-kafka-broker_main/1580175238413422592) | true | `/test unit-tests`
channel-integration-tests-ssl_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-integration-tests-ssl_eventing-kafka-broker_main/1580175238832852992) | true | `/test channel-integration-tests-ssl`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/unit-tests_eventing-kafka-broker_main/1580175238413422592) | true | `/test unit-tests`
channel-integration-tests-ssl_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-integration-tests-ssl_eventing-kafka-broker_main/1580175238832852992) | true | `/test channel-integration-tests-ssl`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1580175238950293504) | true | `/test channel-integration-tests-sasl-plain`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/unit-tests_eventing-kafka-broker_main/1580175238413422592) | true | `/test unit-tests`
channel-integration-tests-ssl_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-integration-tests-ssl_eventing-kafka-broker_main/1580175238832852992) | true | `/test channel-integration-tests-ssl`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1580175238950293504) | true | `/test channel-integration-tests-sasl-plain`
channel-integration-tests-sasl-ssl_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-integration-tests-sasl-ssl_eventing-kafka-broker_main/1580175238883184640) | true | `/test channel-integration-tests-sasl-ssl`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
channel-integration-tests-ssl_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-integration-tests-ssl_eventing-kafka-broker_main/1580175238832852992) | true | `/test channel-integration-tests-ssl`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1580175238950293504) | true | `/test channel-integration-tests-sasl-plain`
channel-integration-tests-sasl-ssl_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-integration-tests-sasl-ssl_eventing-kafka-broker_main/1580175238883184640) | true | `/test channel-integration-tests-sasl-ssl`
unit-tests_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/unit-tests_eventing-kafka-broker_main/1580180304637726720) | true | `/test unit-tests`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
channel-integration-tests-ssl_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-integration-tests-ssl_eventing-kafka-broker_main/1580175238832852992) | true | `/test channel-integration-tests-ssl`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1580175238950293504) | true | `/test channel-integration-tests-sasl-plain`
channel-integration-tests-sasl-ssl_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-integration-tests-sasl-ssl_eventing-kafka-broker_main/1580175238883184640) | true | `/test channel-integration-tests-sasl-ssl`
unit-tests_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/unit-tests_eventing-kafka-broker_main/1580180304637726720) | true | `/test unit-tests`
upgrade-tests_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/upgrade-tests_eventing-kafka-broker_main/1580175238623137792) | true | `/test upgrade-tests`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
channel-integration-tests-ssl_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-integration-tests-ssl_eventing-kafka-broker_main/1580175238832852992) | true | `/test channel-integration-tests-ssl`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1580175238950293504) | true | `/test channel-integration-tests-sasl-plain`
channel-integration-tests-sasl-ssl_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-integration-tests-sasl-ssl_eventing-kafka-broker_main/1580175238883184640) | true | `/test channel-integration-tests-sasl-ssl`
unit-tests_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/unit-tests_eventing-kafka-broker_main/1580180304637726720) | true | `/test unit-tests`
upgrade-tests_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/upgrade-tests_eventing-kafka-broker_main/1580175238623137792) | true | `/test upgrade-tests`
channel-reconciler-tests-ssl_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-reconciler-tests-ssl_eventing-kafka-broker_main/1580175239302615040) | true | `/test channel-reconciler-tests-ssl`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
channel-integration-tests-ssl_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-integration-tests-ssl_eventing-kafka-broker_main/1580175238832852992) | true | `/test channel-integration-tests-ssl`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1580175238950293504) | true | `/test channel-integration-tests-sasl-plain`
channel-integration-tests-sasl-ssl_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-integration-tests-sasl-ssl_eventing-kafka-broker_main/1580175238883184640) | true | `/test channel-integration-tests-sasl-ssl`
unit-tests_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/unit-tests_eventing-kafka-broker_main/1580180304637726720) | true | `/test unit-tests`
upgrade-tests_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/upgrade-tests_eventing-kafka-broker_main/1580175238623137792) | true | `/test upgrade-tests`
channel-reconciler-tests-ssl_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-reconciler-tests-ssl_eventing-kafka-broker_main/1580175239302615040) | true | `/test channel-reconciler-tests-ssl`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1580175239940149248) | true | `/test channel-reconciler-tests-sasl-ssl`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
channel-integration-tests-ssl_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-integration-tests-ssl_eventing-kafka-broker_main/1580175238832852992) | true | `/test channel-integration-tests-ssl`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1580175238950293504) | true | `/test channel-integration-tests-sasl-plain`
channel-integration-tests-sasl-ssl_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-integration-tests-sasl-ssl_eventing-kafka-broker_main/1580175238883184640) | true | `/test channel-integration-tests-sasl-ssl`
unit-tests_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/unit-tests_eventing-kafka-broker_main/1580180304637726720) | true | `/test unit-tests`
upgrade-tests_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/upgrade-tests_eventing-kafka-broker_main/1580175238623137792) | true | `/test upgrade-tests`
channel-reconciler-tests-ssl_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-reconciler-tests-ssl_eventing-kafka-broker_main/1580175239302615040) | true | `/test channel-reconciler-tests-ssl`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1580175239940149248) | true | `/test channel-reconciler-tests-sasl-ssl`
channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main/1580175240791592960) | true | `/test channel-reconciler-tests-sasl-plain`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
channel-integration-tests-ssl_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-integration-tests-ssl_eventing-kafka-broker_main/1580175238832852992) | true | `/test channel-integration-tests-ssl`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1580175238950293504) | true | `/test channel-integration-tests-sasl-plain`
channel-integration-tests-sasl-ssl_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-integration-tests-sasl-ssl_eventing-kafka-broker_main/1580175238883184640) | true | `/test channel-integration-tests-sasl-ssl`
unit-tests_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/unit-tests_eventing-kafka-broker_main/1580180304637726720) | true | `/test unit-tests`
upgrade-tests_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/upgrade-tests_eventing-kafka-broker_main/1580175238623137792) | true | `/test upgrade-tests`
channel-reconciler-tests-ssl_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-reconciler-tests-ssl_eventing-kafka-broker_main/1580175239302615040) | true | `/test channel-reconciler-tests-ssl`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1580175239940149248) | true | `/test channel-reconciler-tests-sasl-ssl`
channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main/1580175240791592960) | true | `/test channel-reconciler-tests-sasl-plain`
integration-tests_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/integration-tests_eventing-kafka-broker_main/1580175238467948544) | true | `/test integration-tests`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
channel-integration-tests-ssl_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-integration-tests-ssl_eventing-kafka-broker_main/1580175238832852992) | true | `/test channel-integration-tests-ssl`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1580175238950293504) | true | `/test channel-integration-tests-sasl-plain`
channel-integration-tests-sasl-ssl_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-integration-tests-sasl-ssl_eventing-kafka-broker_main/1580175238883184640) | true | `/test channel-integration-tests-sasl-ssl`
unit-tests_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/unit-tests_eventing-kafka-broker_main/1580180304637726720) | true | `/test unit-tests`
upgrade-tests_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/upgrade-tests_eventing-kafka-broker_main/1580175238623137792) | true | `/test upgrade-tests`
channel-reconciler-tests-ssl_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-reconciler-tests-ssl_eventing-kafka-broker_main/1580175239302615040) | true | `/test channel-reconciler-tests-ssl`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1580175239940149248) | true | `/test channel-reconciler-tests-sasl-ssl`
channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main/1580175240791592960) | true | `/test channel-reconciler-tests-sasl-plain`
integration-tests_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/integration-tests_eventing-kafka-broker_main/1580175238467948544) | true | `/test integration-tests`
integration-tests-namespaced-broker_eventing-kafka-broker_main | a71e6eea31164dc007ff8acc60f2750e4d5acb5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2720/integration-tests-namespaced-broker_eventing-kafka-broker_main/1580175238556028928) | true | `/test integration-tests-namespaced-broker`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

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

Patch:
diff --git a/control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml b/control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml
index 666d47275..249b0ef23 100644
--- a/control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml
+++ b/control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml
@@ -83,10 +83,12 @@ rules:
     resources:
       - secrets
     verbs:
-      - list
       - get
-      - update
+      - list
       - watch
+      - update
+      - create
+      - delete
 
   # for namespaced brokers, we need to be able to manage additional resources within the user namespaces
   - apiGroups:
@@ -249,3 +251,19 @@ rules:
       - kafkasources/finalizers
     verbs:
       - update
+
+  - apiGroups:
+    - keda.sh
+    resources:
+    - scaledobjects
+    - scaledobjects/finalizers
+    - scaledobjects/status
+    - triggerauthentications
+    - triggerauthentications/status
+    verbs:
+      - get
+      - list
+      - watch
+      - update
+      - create
+      - delete
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_lifecycle.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_lifecycle.go
index 592a9503a..b0fbe2184 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_lifecycle.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_lifecycle.go
@@ -25,6 +25,7 @@ import (
 const (
 	ConditionConsumerGroupConsumers          apis.ConditionType = "Consumers"
 	ConditionConsumerGroupConsumersScheduled apis.ConditionType = "ConsumersScheduled"
+	ConditionAutoscaling                     apis.ConditionType = "Autoscaler"
 	// Labels
 	KafkaChannelNameLabel           = "kafkachannel-name"
 	ConsumerLabelSelector           = "kafka.eventing.knative.dev/metadata.uid"
@@ -35,6 +36,7 @@ var (
 	conditionSet = apis.NewLivingConditionSet(
 		ConditionConsumerGroupConsumers,
 		ConditionConsumerGroupConsumersScheduled,
+		ConditionAutoscaling,
 	)
 )
 
@@ -81,3 +83,17 @@ func (cg *ConsumerGroup) MarkInitializeOffsetFailed(reason string, err error) er
 func (cg *ConsumerGroup) MarkScheduleSucceeded() {
 	cg.GetConditionSet().Manage(cg.GetStatus()).MarkTrue(ConditionConsumerGroupConsumersScheduled)
 }
+
+func (cg *ConsumerGroup) MarkAutoscalerSucceeded() {
+	cg.GetConditionSet().Manage(cg.GetStatus()).MarkTrue(ConditionAutoscaling)
+}
+
+func (cg *ConsumerGroup) MarkAutoscalerDisabled() {
+	cg.GetConditionSet().Manage(cg.GetStatus()).MarkTrueWithReason(ConditionAutoscaling, "Autoscaler is disabled", "")
+}
+
+func (cg *ConsumerGroup) MarkAutoscalerFailed(reason string, err error) error {
+	err = fmt.Errorf("failed to set up autoscaler: %w", err)
+	cg.GetConditionSet().Manage(cg.GetStatus()).MarkFalse(ConditionAutoscaling, reason, err.Error())
+	return err
+}
diff --git a/control-plane/pkg/reconciler/consumer/consumer.go b/control-plane/pkg/reconciler/consumer/consumer.go
index 96ab3746b..f85d5b56c 100644
--- a/control-plane/pkg/reconciler/consumer/consumer.go
+++ b/control-plane/pkg/reconciler/consumer/consumer.go
@@ -193,7 +193,9 @@ func (r Reconciler) reconcileAuth(ctx context.Context, c *kafkainternals.Consume
 		if err != nil {
 			return fmt.Errorf("failed to resolve auth context: %w", err)
 		}
-		resource.Auth = &contract.Resource_MultiAuthSecret{MultiAuthSecret: authContext.MultiSecretReference}
+		resource.Auth = &contract.Resource_MultiAuthSecret{
+			MultiAuthSecret: authContext.MultiSecretReference,
+		}
 		return nil
 	}
 
@@ -202,13 +204,14 @@ func (r Reconciler) reconcileAuth(ctx context.Context, c *kafkainternals.Consume
 		if err != nil {
 			return fmt.Errorf("failed to get secret: %w", err)
 		}
-		resource.Auth = &contract.Resource_AuthSecret{
-			AuthSecret: &contract.Reference{
-				Uuid:      string(secret.UID),
-				Namespace: secret.Namespace,
-				Name:      secret.Name,
-				Version:   secret.ResourceVersion,
-			},
+
+		authContext, err := security.ResolveAuthContextFromLegacySecret(secret)
+		if err != nil {
+			return err
+		}
+
+		resource.Auth = &contract.Resource_MultiAuthSecret{
+			MultiAuthSecret: authContext.MultiSecretReference,
 		}
 		return nil
 	}
diff --git a/control-plane/pkg/reconciler/consumergroup/auth.go b/control-plane/pkg/reconciler/consumergroup/auth.go
index 7fabd0097..54cceccef 100644
--- a/control-plane/pkg/reconciler/consumergroup/auth.go
+++ b/control-plane/pkg/reconciler/consumergroup/auth.go
@@ -29,13 +29,19 @@ import (
 
 func (r Reconciler) newAuthConfigOption(ctx context.Context, cg *kafkainternals.ConsumerGroup) (kafka.ConfigOption, error) {
 	var secret *corev1.Secret
-	var err error
 
 	if hasAuthSpecAuthConfig(cg.Spec.Template.Spec.Auth) {
-		secret, err = security.Secret(ctx, &AuthSpecSecretLocator{cg}, security.DefaultSecretProviderFunc(r.SecretLister, r.KubeClient))
+		secret, err := security.Secret(ctx, &AuthSpecSecretLocator{cg}, security.DefaultSecretProviderFunc(r.SecretLister, r.KubeClient))
 		if err != nil {
 			return nil, err
 		}
+
+		authContext, err := security.ResolveAuthContextFromLegacySecret(secret)
+		if err != nil {
+			return nil, err
+		}
+		return security.NewSaramaSecurityOptionFromSecret(authContext.VirtualSecret), nil
+
 	} else if hasNetSpecAuthConfig(cg.Spec.Template.Spec.Auth) {
 		auth, err := security.ResolveAuthContextFromNetSpec(r.SecretLister, cg.GetNamespace(), *cg.Spec.Template.Spec.Auth.NetSpec)
 		if err != nil {
@@ -76,7 +82,9 @@ func hasAuthSpecAuthConfig(auth *kafkainternals.Auth) bool {
 }
 
 func hasNetSpecAuthConfig(auth *kafkainternals.Auth) bool {
-	return auth != nil && (auth.NetSpec.TLS.Enable || auth.NetSpec.SASL.Enable) &&
+	return auth != nil &&
+		auth.NetSpec != nil &&
+		(auth.NetSpec.TLS.Enable || auth.NetSpec.SASL.Enable) &&
 		(auth.NetSpec.TLS.Cert.SecretKeyRef != nil ||
 			auth.NetSpec.TLS.CACert.SecretKeyRef != nil ||
 			auth.NetSpec.TLS.Key.SecretKeyRef != nil ||
diff --git a/control-plane/pkg/reconciler/consumergroup/consumergroup.go b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
index e3f39e36f..eb3c8dc6b 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
@@ -25,6 +25,7 @@ import (
 
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -41,12 +42,17 @@ import (
 
 	"knative.dev/eventing/pkg/scheduler"
 
+	"knative.dev/eventing-autoscaler-keda/pkg/reconciler/keda"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/config"
 	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 	internalv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/clientset/versioned/typed/eventing/v1alpha1"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/reconciler/eventing/v1alpha1/consumergroup"
 	kafkainternalslisters "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/listers/eventing/v1alpha1"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
+
+	kedav1alpha1 "knative.dev/eventing-autoscaler-keda/third_party/pkg/apis/keda/v1alpha1"
+	kedaclientset "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/clientset/versioned"
+	kedafunc "knative.dev/eventing-kafka-broker/control-plane/pkg/keda"
 )
 
 var (
@@ -80,6 +86,7 @@ type Reconciler struct {
 	NewKafkaClusterAdminClient kafka.NewClusterAdminClientFunc
 
 	KafkaFeatureFlags *config.KafkaFeatureFlags
+	KedaClient        kedaclientset.Interface
 }
 
 func (r Reconciler) ReconcileKind(ctx context.Context, cg *kafkainternals.ConsumerGroup) reconciler.Event {
@@ -92,6 +99,16 @@ func (r Reconciler) ReconcileKind(ctx context.Context, cg *kafkainternals.Consum
 	}
 	cg.MarkScheduleSucceeded()
 
+	if r.isKEDAEnabled(ctx, cg.GetNamespace()) {
+		if err := r.reconcileKedaObjects(ctx, cg); err != nil {
+			return cg.MarkAutoscalerFailed("AutoscalerFailed", err)
+		}
+		cg.MarkAutoscalerSucceeded()
+	} else {
+		// If KEDA is not installed or autoscaler feature disabled, do nothing
+		cg.MarkAutoscalerDisabled()
+	}
+
 	if err := r.reconcileConsumers(ctx, cg); err != nil {
 		return err
 	}
@@ -104,15 +121,16 @@ func (r Reconciler) ReconcileKind(ctx context.Context, cg *kafkainternals.Consum
 		return cg.MarkReconcileConsumersFailedCondition(errCondition)
 	}
 
-	if cg.Status.SubscriberURI == nil {
-		_ = cg.MarkReconcileConsumersFailed("PropagateSubscriberURI", ErrNoSubscriberURI)
-		return nil
-	}
-	if cg.HasDeadLetterSink() && cg.Status.DeadLetterSinkURI == nil {
-		_ = cg.MarkReconcileConsumersFailed("PropagateDeadLetterSinkURI", ErrNoDeadLetterSinkURI)
-		return nil
+	if *cg.Spec.Replicas != 0 {
+		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
@@ -142,21 +160,21 @@ func (r Reconciler) FinalizeKind(ctx context.Context, cg *kafkainternals.Consume
 		}
 	}
 
-	options, err := r.newAuthConfigOption(ctx, cg)
+	saramaSecurityOption, err := r.newAuthConfigOption(ctx, cg)
 	if err != nil {
 		return fmt.Errorf("failed to create config options for Kafka cluster auth: %w", err)
 	}
 
-	saramaConfig, err := kafka.GetSaramaConfig(options)
+	kafkaClusterAdminSaramaConfig, err := kafka.GetSaramaConfig(saramaSecurityOption)
 	if err != nil {
-		return fmt.Errorf("failed to get cluster admin sarama config: %w", err)
+		return fmt.Errorf("failed to create Admin client config: %w", err)
 	}
+
 	bootstrapServers := kafka.BootstrapServersArray(cg.Spec.Template.Spec.Configs.Configs["bootstrap.servers"])
 
-	kafkaClusterAdminClient, err := r.NewKafkaClusterAdminClient(bootstrapServers, saramaConfig)
+	kafkaClusterAdminClient, err := r.NewKafkaClusterAdminClient(bootstrapServers, kafkaClusterAdminSaramaConfig)
 	if err != nil {
-		logging.FromContext(ctx).Errorw("unable to create a kafka client", zap.Error(err))
-		return err
+		return fmt.Errorf("cannot obtain Kafka cluster admin, %w", err)
 	}
 	defer kafkaClusterAdminClient.Close()
 
@@ -199,10 +217,7 @@ func (r Reconciler) reconcileConsumers(ctx context.Context, cg *kafkainternals.C
 	return nil
 }
 
-func (r Reconciler) reconcileConsumersInPlacement(
-	ctx context.Context,
-	cg *kafkainternals.ConsumerGroup,
-	pc ConsumersPerPlacement) error {
+func (r Reconciler) reconcileConsumersInPlacement(ctx context.Context, cg *kafkainternals.ConsumerGroup, pc ConsumersPerPlacement) error {
 
 	placement := *pc.Placement
 	consumers := pc.Consumers
@@ -391,39 +406,188 @@ func (r Reconciler) reconcileInitialOffset(ctx context.Context, cg *kafkainterna
 		return nil
 	}
 
-	options, err := r.newAuthConfigOption(ctx, cg)
+	saramaSecurityOption, err := r.newAuthConfigOption(ctx, cg)
 	if err != nil {
 		return fmt.Errorf("failed to create config options for Kafka cluster auth: %w", err)
 	}
 
-	saramaConfig, err := kafka.GetSaramaConfig(options)
+	kafkaClusterAdminSaramaConfig, err := kafka.GetSaramaConfig(saramaSecurityOption)
 	if err != nil {
 		return fmt.Errorf("failed to create Admin client config: %w", err)
 	}
+
+	kafkaClientSaramaConfig, err := kafka.GetSaramaConfig(saramaSecurityOption, kafka.DisableOffsetAutoCommitConfigOption)
+	if err != nil {
+		return fmt.Errorf("error getting client sarama config: %w", err)
+	}
+
 	bootstrapServers := kafka.BootstrapServersArray(cg.Spec.Template.Spec.Configs.Configs["bootstrap.servers"])
 
-	kafkaClient, err := r.NewKafkaClient(bootstrapServers, saramaConfig)
+	kafkaClusterAdminClient, err := r.NewKafkaClusterAdminClient(bootstrapServers, kafkaClusterAdminSaramaConfig)
 	if err != nil {
-		return fmt.Errorf("failed to create Kafka cluster client: %w", err)
+		return fmt.Errorf("cannot obtain Kafka cluster admin, %w", err)
 	}
-	defer kafkaClient.Close()
+	defer kafkaClusterAdminClient.Close()
 
-	kafkaAdminClient, err := sarama.NewClusterAdminFromClient(kafkaClient)
+	kafkaClient, err := r.NewKafkaClient(bootstrapServers, kafkaClientSaramaConfig)
 	if err != nil {
-		return fmt.Errorf("failed to create Kafka cluster admin client: %w", err)
+		return fmt.Errorf("failed to create Kafka cluster client: %w", err)
 	}
-	defer kafkaAdminClient.Close()
+	defer kafkaClient.Close()
 
 	groupId := cg.Spec.Template.Spec.Configs.Configs["group.id"]
 	topics := cg.Spec.Template.Spec.Topics
 
-	if _, err := r.InitOffsetsFunc(ctx, kafkaClient, kafkaAdminClient, topics, groupId); err != nil {
+	if _, err := r.InitOffsetsFunc(ctx, kafkaClient, kafkaClusterAdminClient, topics, groupId); err != nil {
 		return fmt.Errorf("failed to initialize offset: %w", err)
 	}
 
 	return nil
 }
 
+func (r Reconciler) reconcileKedaObjects(ctx context.Context, cg *kafkainternals.ConsumerGroup) error {
+	var triggerAuthentication *kedav1alpha1.TriggerAuthentication
+	var secret *corev1.Secret
+
+	if hasAuthSpecAuthConfig(cg.Spec.Template.Spec.Auth) || hasNetSpecAuthConfig(cg.Spec.Template.Spec.Auth) {
+		saslType, err := r.retrieveSaslTypeIfPresent(ctx, cg)
+		if err != nil {
+			return err
+		}
+
+		triggerAuthentication, secret, err = kedafunc.GenerateTriggerAuthentication(cg, saslType)
+		if err != nil {
+			return err
+		}
+	}
+
+	triggers, err := kedafunc.GenerateScaleTriggers(cg, triggerAuthentication)
+	if err != nil {
+		return err
+	}
+
+	scaledObject, err := keda.GenerateScaledObject(cg, cg.GetGroupVersionKind(), kedafunc.GenerateScaleTarget(cg), triggers)
+	if err != nil {
+		return err
+	}
+
+	if triggerAuthentication != nil && secret != nil {
+		if err = r.reconcileSecret(ctx, secret, cg); err != nil {
+			return err
+		}
+
+		if err = r.reconcileTriggerAuthentication(ctx, triggerAuthentication, cg); err != nil {
+			return err
+		}
+	}
+
+	if err = r.reconcileScaledObject(ctx, scaledObject, cg); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (r *Reconciler) retrieveSaslTypeIfPresent(ctx context.Context, cg *kafkainternals.ConsumerGroup) (*string, error) {
+	if hasNetSpecAuthConfig(cg.Spec.Template.Spec.Auth) && cg.Spec.Template.Spec.Auth.NetSpec.SASL.Enable {
+		if cg.Spec.Template.Spec.Auth.NetSpec.SASL.Type.SecretKeyRef != nil {
+			secretKeyRefName := cg.Spec.Template.Spec.Auth.NetSpec.SASL.Type.SecretKeyRef.Name
+			secretKeyRefKey := cg.Spec.Template.Spec.Auth.NetSpec.SASL.Type.SecretKeyRef.Key
+			secret, err := r.KubeClient.CoreV1().Secrets(cg.Namespace).Get(ctx, secretKeyRefName, metav1.GetOptions{})
+			if err != nil {
+				return nil, reconciler.NewEvent(corev1.EventTypeWarning, "SaslTypeSecretUnavailable", "Unable to get SASL type from secret: \"%s/%s\", %w", cg.Namespace, secretKeyRefName, err)
+			}
+			saslTypeValue := string(secret.Data[secretKeyRefKey])
+			return &saslTypeValue, nil
+		}
+	}
+	return nil, nil
+}
+
+func (r *Reconciler) reconcileScaledObject(ctx context.Context, expectedScaledObject *kedav1alpha1.ScaledObject, obj metav1.Object) error {
+	scaledObject, err := r.KedaClient.KedaV1alpha1().ScaledObjects(expectedScaledObject.Namespace).Get(ctx, expectedScaledObject.Name, metav1.GetOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to get scaledobject %s/%s: %w", expectedScaledObject.Namespace, expectedScaledObject.Name, err)
+	}
+	if apierrors.IsNotFound(err) {
+		_, err = r.KedaClient.KedaV1alpha1().ScaledObjects(expectedScaledObject.Namespace).Create(ctx, expectedScaledObject, metav1.CreateOptions{})
+		if err != nil {
+			return fmt.Errorf("failed to create scaledobject %s/%s: %w", expectedScaledObject.Namespace, expectedScaledObject.Name, err)
+		}
+		return nil
+	}
+	if !metav1.IsControlledBy(scaledObject, obj) {
+		return fmt.Errorf("scaledobject %s/%s is not owned by %s/%s", expectedScaledObject.Namespace, expectedScaledObject.Name, obj.GetNamespace(), obj.GetName())
+	}
+	if !equality.Semantic.DeepDerivative(expectedScaledObject.Spec, scaledObject.Spec) {
+		if _, err = r.KedaClient.KedaV1alpha1().ScaledObjects(expectedScaledObject.Namespace).Update(ctx, expectedScaledObject, metav1.UpdateOptions{}); err != nil {
+			return fmt.Errorf("failed to update scaled object %s/%s: %w", expectedScaledObject.Namespace, expectedScaledObject.Name, err)
+		}
+	}
+	return nil
+}
+
+func (r *Reconciler) reconcileTriggerAuthentication(ctx context.Context, expectedTriggerAuth *kedav1alpha1.TriggerAuthentication, obj metav1.Object) error {
+	triggerAuth, err := r.KedaClient.KedaV1alpha1().TriggerAuthentications(expectedTriggerAuth.Namespace).Get(ctx, expectedTriggerAuth.Name, metav1.GetOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to get triggerauthentication %s/%s: %w", expectedTriggerAuth.Namespace, expectedTriggerAuth.Name, err)
+	}
+	if apierrors.IsNotFound(err) {
+		_, err = r.KedaClient.KedaV1alpha1().TriggerAuthentications(expectedTriggerAuth.Namespace).Create(ctx, expectedTriggerAuth, metav1.CreateOptions{})
+		if err != nil {
+			return fmt.Errorf("failed to create triggerauthentication  object %s/%s: %w", expectedTriggerAuth.Namespace, expectedTriggerAuth.Name, err)
+		}
+		return nil
+	}
+	if !metav1.IsControlledBy(triggerAuth, obj) {
+		return fmt.Errorf("triggerauthentication object %s/%s is not owned by %s/%s", expectedTriggerAuth.Namespace, expectedTriggerAuth.Name, obj.GetNamespace(), obj.GetName())
+	}
+	if !equality.Semantic.DeepDerivative(expectedTriggerAuth.Spec, triggerAuth.Spec) {
+		if _, err = r.KedaClient.KedaV1alpha1().TriggerAuthentications(expectedTriggerAuth.Namespace).Update(ctx, expectedTriggerAuth, metav1.UpdateOptions{}); err != nil {
+			return fmt.Errorf("failed to update triggerauthentication object %s/%s: %w", expectedTriggerAuth.Namespace, expectedTriggerAuth.Name, err)
+		}
+	}
+	return nil
+}
+
+func (r *Reconciler) reconcileSecret(ctx context.Context, expectedSecret *corev1.Secret, obj metav1.Object) error {
+	secret, err := r.KubeClient.CoreV1().Secrets(expectedSecret.Namespace).Get(ctx, expectedSecret.Name, metav1.GetOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to get secret %s/%s: %w", expectedSecret.Namespace, expectedSecret.Name, err)
+	}
+	if apierrors.IsNotFound(err) {
+		_, err = r.KubeClient.CoreV1().Secrets(expectedSecret.Namespace).Create(ctx, expectedSecret, metav1.CreateOptions{})
+		if err != nil {
+			return fmt.Errorf("failed to create secret %s/%s: %w", expectedSecret.Namespace, expectedSecret.Name, err)
+		}
+		return nil
+	}
+	if !metav1.IsControlledBy(secret, obj) {
+		return fmt.Errorf("secret object %s/%s is not owned by %s/%s", expectedSecret.Namespace, expectedSecret.Name, obj.GetNamespace(), obj.GetName())
+	}
+	// StringData is not populated on read so for now always update the secret
+	if _, err = r.KubeClient.CoreV1().Secrets(expectedSecret.Namespace).Update(ctx, expectedSecret, metav1.UpdateOptions{}); err != nil {
+		return fmt.Errorf("failed to update secret object %s/%s: %w", expectedSecret.Namespace, expectedSecret.Name, err)
+	}
+	return nil
+}
+
+func (r *Reconciler) isKEDAEnabled(ctx context.Context, namespace string) bool {
+	// TODO: code below failing unit tests with err: "panic: interface conversion: testing.ActionImpl is not testing.GetAction: missing method GetName"
+	/*if err := discovery.ServerSupportsVersion(r.KubeClient.Discovery(), keda.KedaSchemeGroupVersion); err == nil {
+		return true
+	}*/
+
+	if r.KafkaFeatureFlags.IsControllerAutoscalerEnabled() {
+		if _, err := r.KedaClient.KedaV1alpha1().ScaledObjects(namespace).List(ctx, metav1.ListOptions{}); err != nil {
+			logging.FromContext(ctx).Debug("KEDA not installed, failed to list ScaledObjects")
+			return false
+		}
+		return true
+	}
+	return false
+}
+
 var (
 	_ consumergroup.Interface = &Reconciler{}
 )
diff --git a/control-plane/pkg/reconciler/consumergroup/controller.go b/control-plane/pkg/reconciler/consumergroup/controller.go
index f56cdac4d..d6a423a39 100644
--- a/control-plane/pkg/reconciler/consumergroup/controller.go
+++ b/control-plane/pkg/reconciler/consumergroup/controller.go
@@ -52,6 +52,8 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumer"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumergroup"
 	cgreconciler "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/reconciler/eventing/v1alpha1/consumergroup"
+
+	kedaclient "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/injection/client"
 )
 
 const (
@@ -114,6 +116,7 @@ func NewController(ctx context.Context, watcher configmap.Watcher) *controller.I
 		SystemNamespace:            system.Namespace(),
 		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
 		KafkaFeatureFlags:          config.DefaultFeaturesConfig(),
+		KedaClient:                 kedaclient.Get(ctx),
 	}
 
 	consumerInformer := consumer.Get(ctx)
@@ -137,6 +140,8 @@ func NewController(ctx context.Context, watcher configmap.Watcher) *controller.I
 
 	ResyncOnStatefulSetChange(ctx, globalResync)
 
+	//Todo: ScaledObject informer when KEDA is installed
+
 	return impl
 }
 
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java
index fc4aac196..e77a32993 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java
@@ -43,7 +43,7 @@ public final class WebClientCloudEventSender implements CloudEventSender {
 
   private static final Logger logger = LoggerFactory.getLogger(WebClientCloudEventSender.class);
 
-  private static final long DEFAULT_TIMEOUT_MS = 600_000L;
+  public static final long DEFAULT_TIMEOUT_MS = 600_000L;
 
   private final WebClient client;
   private final String target;
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImpl.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImpl.java
index 0bcc355c6..4ad3acff0 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImpl.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImpl.java
@@ -147,6 +147,14 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
       consumerConfigs.put(InvalidCloudEventInterceptor.SOURCE_NAMESPACE_CONFIG, resource.getReference().getNamespace());
     }
 
+    final var egressConfig =
+      egress.hasEgressConfig() ?
+        egress.getEgressConfig() :
+        resource.getEgressConfig();
+
+    final var maxProcessingTime = maxProcessingTimeMs(consumerConfigs, egressConfig);
+    consumerConfigs.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, maxProcessingTime);
+
     final var producerConfigs = new HashMap<>(this.producerConfigs);
     producerConfigs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, resource.getBootstrapServers());
 
@@ -160,14 +168,6 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
        final KafkaConsumer<Object, CloudEvent> consumer = createConsumer(vertx, consumerConfigs);
        final var metricsCloser = Metrics.register(consumer.unwrap());
 
-       final var egressConfig =
-         egress.hasEgressConfig() ?
-           egress.getEgressConfig() :
-           resource.getEgressConfig();
-
-       final var maxProcessingTime = maxProcessingTimeMs(consumerConfigs, egressConfig);
-       consumerConfigs.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, maxProcessingTime);
-
        final var egressSubscriberSender = createConsumerRecordSender(
          vertx,
          egress.getDestination(),
@@ -235,7 +235,7 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
   /**
    * For each handler call partitionRevoked and wait for the future to complete.
    *
-   * @param consumerGroup consumer group id
+   * @param consumerGroup            consumer group id
    * @param partitionRevokedHandlers partition revoked handlers
    * @return a single handler that dispatches the partition revoked event to the given handlers.
    */
@@ -369,14 +369,14 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
     return resource != null && !resource.getNamespace().isBlank() && !resource.getName().isBlank();
   }
 
-  private static long maxProcessingTimeMs(final HashMap<String, Object> consumerConfigs, final EgressConfig egressConfig) {
+  private static int maxProcessingTimeMs(final HashMap<String, Object> consumerConfigs, final EgressConfig egressConfig) {
     final var mpr = consumerConfigs.get(ConsumerConfig.MAX_POLL_RECORDS_CONFIG);
     final long maxPollRecords = mpr == null ? /* default max.poll.records */ 500L : Long.parseLong(mpr.toString());
     final var retryPolicy = WebClientCloudEventSender.computeRetryPolicy(egressConfig);
     final var retry = egressConfig.getRetry();
-    final var timeout = egressConfig.getTimeout();
+    final var timeout = egressConfig.getTimeout() > 0 ? egressConfig.getTimeout() : WebClientCloudEventSender.DEFAULT_TIMEOUT_MS;
 
-    long maxProcessingTimeForSingleRecord = 0L;
+    long maxProcessingTimeForSingleRecord = timeout;
     for (int i = 1; i <= retry; i++) {
       maxProcessingTimeForSingleRecord += timeout + retryPolicy.apply(i);
     }
@@ -386,6 +386,13 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
     // So far, the max processing time calculated is for one single record,
     // however, we poll records in batches based on the `max.poll.records`
     // configuration, so the total max processing time is:
-    return maxPollRecords * maxProcessingTimeForSingleRecord;
+    //
+    // Times 2 for dead letter sink retries.
+    final var total = 2 * maxPollRecords * maxProcessingTimeForSingleRecord;
+
+    if (total >= Integer.MAX_VALUE) {
+      return Integer.MAX_VALUE;
+    }
+    return (int) total;
   }
 }
diff --git a/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/factory.go b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/factory.go
new file mode 100644
index 000000000..f4d808ab3
--- /dev/null
+++ b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/factory.go
@@ -0,0 +1,180 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by informer-gen. DO NOT EDIT.
+
+package externalversions
+
+import (
+	reflect "reflect"
+	sync "sync"
+	time "time"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	runtime "k8s.io/apimachinery/pkg/runtime"
+	schema "k8s.io/apimachinery/pkg/runtime/schema"
+	cache "k8s.io/client-go/tools/cache"
+	versioned "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/clientset/versioned"
+	internalinterfaces "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/internalinterfaces"
+	keda "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/keda"
+)
+
+// SharedInformerOption defines the functional option type for SharedInformerFactory.
+type SharedInformerOption func(*sharedInformerFactory) *sharedInformerFactory
+
+type sharedInformerFactory struct {
+	client           versioned.Interface
+	namespace        string
+	tweakListOptions internalinterfaces.TweakListOptionsFunc
+	lock             sync.Mutex
+	defaultResync    time.Duration
+	customResync     map[reflect.Type]time.Duration
+
+	informers map[reflect.Type]cache.SharedIndexInformer
+	// startedInformers is used for tracking which informers have been started.
+	// This allows Start() to be called multiple times safely.
+	startedInformers map[reflect.Type]bool
+}
+
+// WithCustomResyncConfig sets a custom resync period for the specified informer types.
+func WithCustomResyncConfig(resyncConfig map[v1.Object]time.Duration) SharedInformerOption {
+	return func(factory *sharedInformerFactory) *sharedInformerFactory {
+		for k, v := range resyncConfig {
+			factory.customResync[reflect.TypeOf(k)] = v
+		}
+		return factory
+	}
+}
+
+// WithTweakListOptions sets a custom filter on all listers of the configured SharedInformerFactory.
+func WithTweakListOptions(tweakListOptions internalinterfaces.TweakListOptionsFunc) SharedInformerOption {
+	return func(factory *sharedInformerFactory) *sharedInformerFactory {
+		factory.tweakListOptions = tweakListOptions
+		return factory
+	}
+}
+
+// WithNamespace limits the SharedInformerFactory to the specified namespace.
+func WithNamespace(namespace string) SharedInformerOption {
+	return func(factory *sharedInformerFactory) *sharedInformerFactory {
+		factory.namespace = namespace
+		return factory
+	}
+}
+
+// NewSharedInformerFactory constructs a new instance of sharedInformerFactory for all namespaces.
+func NewSharedInformerFactory(client versioned.Interface, defaultResync time.Duration) SharedInformerFactory {
+	return NewSharedInformerFactoryWithOptions(client, defaultResync)
+}
+
+// NewFilteredSharedInformerFactory constructs a new instance of sharedInformerFactory.
+// Listers obtained via this SharedInformerFactory will be subject to the same filters
+// as specified here.
+// Deprecated: Please use NewSharedInformerFactoryWithOptions instead
+func NewFilteredSharedInformerFactory(client versioned.Interface, defaultResync time.Duration, namespace string, tweakListOptions internalinterfaces.TweakListOptionsFunc) SharedInformerFactory {
+	return NewSharedInformerFactoryWithOptions(client, defaultResync, WithNamespace(namespace), WithTweakListOptions(tweakListOptions))
+}
+
+// NewSharedInformerFactoryWithOptions constructs a new instance of a SharedInformerFactory with additional options.
+func NewSharedInformerFactoryWithOptions(client versioned.Interface, defaultResync time.Duration, options ...SharedInformerOption) SharedInformerFactory {
+	factory := &sharedInformerFactory{
+		client:           client,
+		namespace:        v1.NamespaceAll,
+		defaultResync:    defaultResync,
+		informers:        make(map[reflect.Type]cache.SharedIndexInformer),
+		startedInformers: make(map[reflect.Type]bool),
+		customResync:     make(map[reflect.Type]time.Duration),
+	}
+
+	// Apply all options
+	for _, opt := range options {
+		factory = opt(factory)
+	}
+
+	return factory
+}
+
+// Start initializes all requested informers.
+func (f *sharedInformerFactory) Start(stopCh <-chan struct{}) {
+	f.lock.Lock()
+	defer f.lock.Unlock()
+
+	for informerType, informer := range f.informers {
+		if !f.startedInformers[informerType] {
+			go informer.Run(stopCh)
+			f.startedInformers[informerType] = true
+		}
+	}
+}
+
+// WaitForCacheSync waits for all started informers' cache were synced.
+func (f *sharedInformerFactory) WaitForCacheSync(stopCh <-chan struct{}) map[reflect.Type]bool {
+	informers := func() map[reflect.Type]cache.SharedIndexInformer {
+		f.lock.Lock()
+		defer f.lock.Unlock()
+
+		informers := map[reflect.Type]cache.SharedIndexInformer{}
+		for informerType, informer := range f.informers {
+			if f.startedInformers[informerType] {
+				informers[informerType] = informer
+			}
+		}
+		return informers
+	}()
+
+	res := map[reflect.Type]bool{}
+	for informType, informer := range informers {
+		res[informType] = cache.WaitForCacheSync(stopCh, informer.HasSynced)
+	}
+	return res
+}
+
+// InternalInformerFor returns the SharedIndexInformer for obj using an internal
+// client.
+func (f *sharedInformerFactory) InformerFor(obj runtime.Object, newFunc internalinterfaces.NewInformerFunc) cache.SharedIndexInformer {
+	f.lock.Lock()
+	defer f.lock.Unlock()
+
+	informerType := reflect.TypeOf(obj)
+	informer, exists := f.informers[informerType]
+	if exists {
+		return informer
+	}
+
+	resyncPeriod, exists := f.customResync[informerType]
+	if !exists {
+		resyncPeriod = f.defaultResync
+	}
+
+	informer = newFunc(f.client, resyncPeriod)
+	f.informers[informerType] = informer
+
+	return informer
+}
+
+// SharedInformerFactory provides shared informers for resources in all known
+// API group versions.
+type SharedInformerFactory interface {
+	internalinterfaces.SharedInformerFactory
+	ForResource(resource schema.GroupVersionResource) (GenericInformer, error)
+	WaitForCacheSync(stopCh <-chan struct{}) map[reflect.Type]bool
+
+	Keda() keda.Interface
+}
+
+func (f *sharedInformerFactory) Keda() keda.Interface {
+	return keda.New(f, f.namespace, f.tweakListOptions)
+}
diff --git a/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/generic.go b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/generic.go
new file mode 100644
index 000000000..c5d88ad2a
--- /dev/null
+++ b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/generic.go
@@ -0,0 +1,68 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by informer-gen. DO NOT EDIT.
+
+package externalversions
+
+import (
+	"fmt"
+
+	schema "k8s.io/apimachinery/pkg/runtime/schema"
+	cache "k8s.io/client-go/tools/cache"
+	v1alpha1 "knative.dev/eventing-autoscaler-keda/third_party/pkg/apis/keda/v1alpha1"
+)
+
+// GenericInformer is type of SharedIndexInformer which will locate and delegate to other
+// sharedInformers based on type
+type GenericInformer interface {
+	Informer() cache.SharedIndexInformer
+	Lister() cache.GenericLister
+}
+
+type genericInformer struct {
+	informer cache.SharedIndexInformer
+	resource schema.GroupResource
+}
+
+// Informer returns the SharedIndexInformer.
+func (f *genericInformer) Informer() cache.SharedIndexInformer {
+	return f.informer
+}
+
+// Lister returns the GenericLister.
+func (f *genericInformer) Lister() cache.GenericLister {
+	return cache.NewGenericLister(f.Informer().GetIndexer(), f.resource)
+}
+
+// ForResource gives generic access to a shared informer of the matching type
+// TODO extend this to unknown resources with a client pool
+func (f *sharedInformerFactory) ForResource(resource schema.GroupVersionResource) (GenericInformer, error) {
+	switch resource {
+	// Group=keda, Version=v1alpha1
+	case v1alpha1.SchemeGroupVersion.WithResource("clustertriggerauthentications"):
+		return &genericInformer{resource: resource.GroupResource(), informer: f.Keda().V1alpha1().ClusterTriggerAuthentications().Informer()}, nil
+	case v1alpha1.SchemeGroupVersion.WithResource("scaledjobs"):
+		return &genericInformer{resource: resource.GroupResource(), informer: f.Keda().V1alpha1().ScaledJobs().Informer()}, nil
+	case v1alpha1.SchemeGroupVersion.WithResource("scaledobjects"):
+		return &genericInformer{resource: resource.GroupResource(), informer: f.Keda().V1alpha1().ScaledObjects().Informer()}, nil
+	case v1alpha1.SchemeGroupVersion.WithResource("triggerauthentications"):
+		return &genericInformer{resource: resource.GroupResource(), informer: f.Keda().V1alpha1().TriggerAuthentications().Informer()}, nil
+
+	}
+
+	return nil, fmt.Errorf("no informer found for %v", resource)
+}
diff --git a/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/internalinterfaces/factory_interfaces.go b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/internalinterfaces/factory_interfaces.go
new file mode 100644
index 000000000..6e147777c
--- /dev/null
+++ b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/internalinterfaces/factory_interfaces.go
@@ -0,0 +1,40 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by informer-gen. DO NOT EDIT.
+
+package internalinterfaces
+
+import (
+	time "time"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	runtime "k8s.io/apimachinery/pkg/runtime"
+	cache "k8s.io/client-go/tools/cache"
+	versioned "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/clientset/versioned"
+)
+
+// NewInformerFunc takes versioned.Interface and time.Duration to return a SharedIndexInformer.
+type NewInformerFunc func(versioned.Interface, time.Duration) cache.SharedIndexInformer
+
+// SharedInformerFactory a small interface to allow for adding an informer without an import cycle
+type SharedInformerFactory interface {
+	Start(stopCh <-chan struct{})
+	InformerFor(obj runtime.Object, newFunc NewInformerFunc) cache.SharedIndexInformer
+}
+
+// TweakListOptionsFunc is a function that transforms a v1.ListOptions.
+type TweakListOptionsFunc func(*v1.ListOptions)
diff --git a/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/keda/interface.go b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/keda/interface.go
new file mode 100644
index 000000000..e5f16420d
--- /dev/null
+++ b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/keda/interface.go
@@ -0,0 +1,46 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by informer-gen. DO NOT EDIT.
+
+package keda
+
+import (
+	internalinterfaces "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/internalinterfaces"
+	v1alpha1 "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/keda/v1alpha1"
+)
+
+// Interface provides access to each of this group's versions.
+type Interface interface {
+	// V1alpha1 provides access to shared informers for resources in V1alpha1.
+	V1alpha1() v1alpha1.Interface
+}
+
+type group struct {
+	factory          internalinterfaces.SharedInformerFactory
+	namespace        string
+	tweakListOptions internalinterfaces.TweakListOptionsFunc
+}
+
+// New returns a new Interface.
+func New(f internalinterfaces.SharedInformerFactory, namespace string, tweakListOptions internalinterfaces.TweakListOptionsFunc) Interface {
+	return &group{factory: f, namespace: namespace, tweakListOptions: tweakListOptions}
+}
+
+// V1alpha1 returns a new v1alpha1.Interface.
+func (g *group) V1alpha1() v1alpha1.Interface {
+	return v1alpha1.New(g.factory, g.namespace, g.tweakListOptions)
+}
diff --git a/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/keda/v1alpha1/clustertriggerauthentication.go b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/keda/v1alpha1/clustertriggerauthentication.go
new file mode 100644
index 000000000..25e0a7d43
--- /dev/null
+++ b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/keda/v1alpha1/clustertriggerauthentication.go
@@ -0,0 +1,89 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by informer-gen. DO NOT EDIT.
+
+package v1alpha1
+
+import (
+	"context"
+	time "time"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	runtime "k8s.io/apimachinery/pkg/runtime"
+	watch "k8s.io/apimachinery/pkg/watch"
+	cache "k8s.io/client-go/tools/cache"
+	kedav1alpha1 "knative.dev/eventing-autoscaler-keda/third_party/pkg/apis/keda/v1alpha1"
+	versioned "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/clientset/versioned"
+	internalinterfaces "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/internalinterfaces"
+	v1alpha1 "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/listers/keda/v1alpha1"
+)
+
+// ClusterTriggerAuthenticationInformer provides access to a shared informer and lister for
+// ClusterTriggerAuthentications.
+type ClusterTriggerAuthenticationInformer interface {
+	Informer() cache.SharedIndexInformer
+	Lister() v1alpha1.ClusterTriggerAuthenticationLister
+}
+
+type clusterTriggerAuthenticationInformer struct {
+	factory          internalinterfaces.SharedInformerFactory
+	tweakListOptions internalinterfaces.TweakListOptionsFunc
+}
+
+// NewClusterTriggerAuthenticationInformer constructs a new informer for ClusterTriggerAuthentication type.
+// Always prefer using an informer factory to get a shared informer instead of getting an independent
+// one. This reduces memory footprint and number of connections to the server.
+func NewClusterTriggerAuthenticationInformer(client versioned.Interface, resyncPeriod time.Duration, indexers cache.Indexers) cache.SharedIndexInformer {
+	return NewFilteredClusterTriggerAuthenticationInformer(client, resyncPeriod, indexers, nil)
+}
+
+// NewFilteredClusterTriggerAuthenticationInformer constructs a new informer for ClusterTriggerAuthentication type.
+// Always prefer using an informer factory to get a shared informer instead of getting an independent
+// one. This reduces memory footprint and number of connections to the server.
+func NewFilteredClusterTriggerAuthenticationInformer(client versioned.Interface, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer {
+	return cache.NewSharedIndexInformer(
+		&cache.ListWatch{
+			ListFunc: func(options v1.ListOptions) (runtime.Object, error) {
+				if tweakListOptions != nil {
+					tweakListOptions(&options)
+				}
+				return client.KedaV1alpha1().ClusterTriggerAuthentications().List(context.TODO(), options)
+			},
+			WatchFunc: func(options v1.ListOptions) (watch.Interface, error) {
+				if tweakListOptions != nil {
+					tweakListOptions(&options)
+				}
+				return client.KedaV1alpha1().ClusterTriggerAuthentications().Watch(context.TODO(), options)
+			},
+		},
+		&kedav1alpha1.ClusterTriggerAuthentication{},
+		resyncPeriod,
+		indexers,
+	)
+}
+
+func (f *clusterTriggerAuthenticationInformer) defaultInformer(client versioned.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {
+	return NewFilteredClusterTriggerAuthenticationInformer(client, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, f.tweakListOptions)
+}
+
+func (f *clusterTriggerAuthenticationInformer) Informer() cache.SharedIndexInformer {
+	return f.factory.InformerFor(&kedav1alpha1.ClusterTriggerAuthentication{}, f.defaultInformer)
+}
+
+func (f *clusterTriggerAuthenticationInformer) Lister() v1alpha1.ClusterTriggerAuthenticationLister {
+	return v1alpha1.NewClusterTriggerAuthenticationLister(f.Informer().GetIndexer())
+}
diff --git a/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/keda/v1alpha1/interface.go b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/keda/v1alpha1/interface.go
new file mode 100644
index 000000000..272fdcc30
--- /dev/null
+++ b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/keda/v1alpha1/interface.go
@@ -0,0 +1,66 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by informer-gen. DO NOT EDIT.
+
+package v1alpha1
+
+import (
+	internalinterfaces "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/internalinterfaces"
+)
+
+// Interface provides access to all the informers in this group version.
+type Interface interface {
+	// ClusterTriggerAuthentications returns a ClusterTriggerAuthenticationInformer.
+	ClusterTriggerAuthentications() ClusterTriggerAuthenticationInformer
+	// ScaledJobs returns a ScaledJobInformer.
+	ScaledJobs() ScaledJobInformer
+	// ScaledObjects returns a ScaledObjectInformer.
+	ScaledObjects() ScaledObjectInformer
+	// TriggerAuthentications returns a TriggerAuthenticationInformer.
+	TriggerAuthentications() TriggerAuthenticationInformer
+}
+
+type version struct {
+	factory          internalinterfaces.SharedInformerFactory
+	namespace        string
+	tweakListOptions internalinterfaces.TweakListOptionsFunc
+}
+
+// New returns a new Interface.
+func New(f internalinterfaces.SharedInformerFactory, namespace string, tweakListOptions internalinterfaces.TweakListOptionsFunc) Interface {
+	return &version{factory: f, namespace: namespace, tweakListOptions: tweakListOptions}
+}
+
+// ClusterTriggerAuthentications returns a ClusterTriggerAuthenticationInformer.
+func (v *version) ClusterTriggerAuthentications() ClusterTriggerAuthenticationInformer {
+	return &clusterTriggerAuthenticationInformer{factory: v.factory, tweakListOptions: v.tweakListOptions}
+}
+
+// ScaledJobs returns a ScaledJobInformer.
+func (v *version) ScaledJobs() ScaledJobInformer {
+	return &scaledJobInformer{factory: v.factory, namespace: v.namespace, tweakListOptions: v.tweakListOptions}
+}
+
+// ScaledObjects returns a ScaledObjectInformer.
+func (v *version) ScaledObjects() ScaledObjectInformer {
+	return &scaledObjectInformer{factory: v.factory, namespace: v.namespace, tweakListOptions: v.tweakListOptions}
+}
+
+// TriggerAuthentications returns a TriggerAuthenticationInformer.
+func (v *version) TriggerAuthentications() TriggerAuthenticationInformer {
+	return &triggerAuthenticationInformer{factory: v.factory, namespace: v.namespace, tweakListOptions: v.tweakListOptions}
+}
diff --git a/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/keda/v1alpha1/scaledjob.go b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/keda/v1alpha1/scaledjob.go
new file mode 100644
index 000000000..3024e25d3
--- /dev/null
+++ b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/keda/v1alpha1/scaledjob.go
@@ -0,0 +1,90 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by informer-gen. DO NOT EDIT.
+
+package v1alpha1
+
+import (
+	"context"
+	time "time"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	runtime "k8s.io/apimachinery/pkg/runtime"
+	watch "k8s.io/apimachinery/pkg/watch"
+	cache "k8s.io/client-go/tools/cache"
+	kedav1alpha1 "knative.dev/eventing-autoscaler-keda/third_party/pkg/apis/keda/v1alpha1"
+	versioned "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/clientset/versioned"
+	internalinterfaces "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/internalinterfaces"
+	v1alpha1 "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/listers/keda/v1alpha1"
+)
+
+// ScaledJobInformer provides access to a shared informer and lister for
+// ScaledJobs.
+type ScaledJobInformer interface {
+	Informer() cache.SharedIndexInformer
+	Lister() v1alpha1.ScaledJobLister
+}
+
+type scaledJobInformer struct {
+	factory          internalinterfaces.SharedInformerFactory
+	tweakListOptions internalinterfaces.TweakListOptionsFunc
+	namespace        string
+}
+
+// NewScaledJobInformer constructs a new informer for ScaledJob type.
+// Always prefer using an informer factory to get a shared informer instead of getting an independent
+// one. This reduces memory footprint and number of connections to the server.
+func NewScaledJobInformer(client versioned.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers) cache.SharedIndexInformer {
+	return NewFilteredScaledJobInformer(client, namespace, resyncPeriod, indexers, nil)
+}
+
+// NewFilteredScaledJobInformer constructs a new informer for ScaledJob type.
+// Always prefer using an informer factory to get a shared informer instead of getting an independent
+// one. This reduces memory footprint and number of connections to the server.
+func NewFilteredScaledJobInformer(client versioned.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer {
+	return cache.NewSharedIndexInformer(
+		&cache.ListWatch{
+			ListFunc: func(options v1.ListOptions) (runtime.Object, error) {
+				if tweakListOptions != nil {
+					tweakListOptions(&options)
+				}
+				return client.KedaV1alpha1().ScaledJobs(namespace).List(context.TODO(), options)
+			},
+			WatchFunc: func(options v1.ListOptions) (watch.Interface, error) {
+				if tweakListOptions != nil {
+					tweakListOptions(&options)
+				}
+				return client.KedaV1alpha1().ScaledJobs(namespace).Watch(context.TODO(), options)
+			},
+		},
+		&kedav1alpha1.ScaledJob{},
+		resyncPeriod,
+		indexers,
+	)
+}
+
+func (f *scaledJobInformer) defaultInformer(client versioned.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {
+	return NewFilteredScaledJobInformer(client, f.namespace, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, f.tweakListOptions)
+}
+
+func (f *scaledJobInformer) Informer() cache.SharedIndexInformer {
+	return f.factory.InformerFor(&kedav1alpha1.ScaledJob{}, f.defaultInformer)
+}
+
+func (f *scaledJobInformer) Lister() v1alpha1.ScaledJobLister {
+	return v1alpha1.NewScaledJobLister(f.Informer().GetIndexer())
+}
diff --git a/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/keda/v1alpha1/scaledobject.go b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/keda/v1alpha1/scaledobject.go
new file mode 100644
index 000000000..222fae56c
--- /dev/null
+++ b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/keda/v1alpha1/scaledobject.go
@@ -0,0 +1,90 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by informer-gen. DO NOT EDIT.
+
+package v1alpha1
+
+import (
+	"context"
+	time "time"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	runtime "k8s.io/apimachinery/pkg/runtime"
+	watch "k8s.io/apimachinery/pkg/watch"
+	cache "k8s.io/client-go/tools/cache"
+	kedav1alpha1 "knative.dev/eventing-autoscaler-keda/third_party/pkg/apis/keda/v1alpha1"
+	versioned "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/clientset/versioned"
+	internalinterfaces "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/internalinterfaces"
+	v1alpha1 "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/listers/keda/v1alpha1"
+)
+
+// ScaledObjectInformer provides access to a shared informer and lister for
+// ScaledObjects.
+type ScaledObjectInformer interface {
+	Informer() cache.SharedIndexInformer
+	Lister() v1alpha1.ScaledObjectLister
+}
+
+type scaledObjectInformer struct {
+	factory          internalinterfaces.SharedInformerFactory
+	tweakListOptions internalinterfaces.TweakListOptionsFunc
+	namespace        string
+}
+
+// NewScaledObjectInformer constructs a new informer for ScaledObject type.
+// Always prefer using an informer factory to get a shared informer instead of getting an independent
+// one. This reduces memory footprint and number of connections to the server.
+func NewScaledObjectInformer(client versioned.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers) cache.SharedIndexInformer {
+	return NewFilteredScaledObjectInformer(client, namespace, resyncPeriod, indexers, nil)
+}
+
+// NewFilteredScaledObjectInformer constructs a new informer for ScaledObject type.
+// Always prefer using an informer factory to get a shared informer instead of getting an independent
+// one. This reduces memory footprint and number of connections to the server.
+func NewFilteredScaledObjectInformer(client versioned.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer {
+	return cache.NewSharedIndexInformer(
+		&cache.ListWatch{
+			ListFunc: func(options v1.ListOptions) (runtime.Object, error) {
+				if tweakListOptions != nil {
+					tweakListOptions(&options)
+				}
+				return client.KedaV1alpha1().ScaledObjects(namespace).List(context.TODO(), options)
+			},
+			WatchFunc: func(options v1.ListOptions) (watch.Interface, error) {
+				if tweakListOptions != nil {
+					tweakListOptions(&options)
+				}
+				return client.KedaV1alpha1().ScaledObjects(namespace).Watch(context.TODO(), options)
+			},
+		},
+		&kedav1alpha1.ScaledObject{},
+		resyncPeriod,
+		indexers,
+	)
+}
+
+func (f *scaledObjectInformer) defaultInformer(client versioned.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {
+	return NewFilteredScaledObjectInformer(client, f.namespace, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, f.tweakListOptions)
+}
+
+func (f *scaledObjectInformer) Informer() cache.SharedIndexInformer {
+	return f.factory.InformerFor(&kedav1alpha1.ScaledObject{}, f.defaultInformer)
+}
+
+func (f *scaledObjectInformer) Lister() v1alpha1.ScaledObjectLister {
+	return v1alpha1.NewScaledObjectLister(f.Informer().GetIndexer())
+}
diff --git a/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/keda/v1alpha1/triggerauthentication.go b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/keda/v1alpha1/triggerauthentication.go
new file mode 100644
index 000000000..5fb825a5a
--- /dev/null
+++ b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/keda/v1alpha1/triggerauthentication.go
@@ -0,0 +1,90 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by informer-gen. DO NOT EDIT.
+
+package v1alpha1
+
+import (
+	"context"
+	time "time"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	runtime "k8s.io/apimachinery/pkg/runtime"
+	watch "k8s.io/apimachinery/pkg/watch"
+	cache "k8s.io/client-go/tools/cache"
+	kedav1alpha1 "knative.dev/eventing-autoscaler-keda/third_party/pkg/apis/keda/v1alpha1"
+	versioned "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/clientset/versioned"
+	internalinterfaces "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/internalinterfaces"
+	v1alpha1 "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/listers/keda/v1alpha1"
+)
+
+// TriggerAuthenticationInformer provides access to a shared informer and lister for
+// TriggerAuthentications.
+type TriggerAuthenticationInformer interface {
+	Informer() cache.SharedIndexInformer
+	Lister() v1alpha1.TriggerAuthenticationLister
+}
+
+type triggerAuthenticationInformer struct {
+	factory          internalinterfaces.SharedInformerFactory
+	tweakListOptions internalinterfaces.TweakListOptionsFunc
+	namespace        string
+}
+
+// NewTriggerAuthenticationInformer constructs a new informer for TriggerAuthentication type.
+// Always prefer using an informer factory to get a shared informer instead of getting an independent
+// one. This reduces memory footprint and number of connections to the server.
+func NewTriggerAuthenticationInformer(client versioned.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers) cache.SharedIndexInformer {
+	return NewFilteredTriggerAuthenticationInformer(client, namespace, resyncPeriod, indexers, nil)
+}
+
+// NewFilteredTriggerAuthenticationInformer constructs a new informer for TriggerAuthentication type.
+// Always prefer using an informer factory to get a shared informer instead of getting an independent
+// one. This reduces memory footprint and number of connections to the server.
+func NewFilteredTriggerAuthenticationInformer(client versioned.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer {
+	return cache.NewSharedIndexInformer(
+		&cache.ListWatch{
+			ListFunc: func(options v1.ListOptions) (runtime.Object, error) {
+				if tweakListOptions != nil {
+					tweakListOptions(&options)
+				}
+				return client.KedaV1alpha1().TriggerAuthentications(namespace).List(context.TODO(), options)
+			},
+			WatchFunc: func(options v1.ListOptions) (watch.Interface, error) {
+				if tweakListOptions != nil {
+					tweakListOptions(&options)
+				}
+				return client.KedaV1alpha1().TriggerAuthentications(namespace).Watch(context.TODO(), options)
+			},
+		},
+		&kedav1alpha1.TriggerAuthentication{},
+		resyncPeriod,
+		indexers,
+	)
+}
+
+func (f *triggerAuthenticationInformer) defaultInformer(client versioned.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {
+	return NewFilteredTriggerAuthenticationInformer(client, f.namespace, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, f.tweakListOptions)
+}
+
+func (f *triggerAuthenticationInformer) Informer() cache.SharedIndexInformer {
+	return f.factory.InformerFor(&kedav1alpha1.TriggerAuthentication{}, f.defaultInformer)
+}
+
+func (f *triggerAuthenticationInformer) Lister() v1alpha1.TriggerAuthenticationLister {
+	return v1alpha1.NewTriggerAuthenticationLister(f.Informer().GetIndexer())
+}
diff --git a/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/injection/informers/factory/factory.go b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/injection/informers/factory/factory.go
new file mode 100644
index 000000000..7952bc41f
--- /dev/null
+++ b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/injection/informers/factory/factory.go
@@ -0,0 +1,56 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package factory
+
+import (
+	context "context"
+
+	externalversions "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions"
+	client "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/injection/client"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+func init() {
+	injection.Default.RegisterInformerFactory(withInformerFactory)
+}
+
+// Key is used as the key for associating information with a context.Context.
+type Key struct{}
+
+func withInformerFactory(ctx context.Context) context.Context {
+	c := client.Get(ctx)
+	opts := make([]externalversions.SharedInformerOption, 0, 1)
+	if injection.HasNamespaceScope(ctx) {
+		opts = append(opts, externalversions.WithNamespace(injection.GetNamespaceScope(ctx)))
+	}
+	return context.WithValue(ctx, Key{},
+		externalversions.NewSharedInformerFactoryWithOptions(c, controller.GetResyncPeriod(ctx), opts...))
+}
+
+// Get extracts the InformerFactory from the context.
+func Get(ctx context.Context) externalversions.SharedInformerFactory {
+	untyped := ctx.Value(Key{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions.SharedInformerFactory from context.")
+	}
+	return untyped.(externalversions.SharedInformerFactory)
+}
diff --git a/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/injection/informers/factory/fake/fake.go b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/injection/informers/factory/fake/fake.go
new file mode 100644
index 000000000..91721496d
--- /dev/null
+++ b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/injection/informers/factory/fake/fake.go
@@ -0,0 +1,45 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	context "context"
+
+	externalversions "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions"
+	fake "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/injection/client/fake"
+	factory "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/injection/informers/factory"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+)
+
+var Get = factory.Get
+
+func init() {
+	injection.Fake.RegisterInformerFactory(withInformerFactory)
+}
+
+func withInformerFactory(ctx context.Context) context.Context {
+	c := fake.Get(ctx)
+	opts := make([]externalversions.SharedInformerOption, 0, 1)
+	if injection.HasNamespaceScope(ctx) {
+		opts = append(opts, externalversions.WithNamespace(injection.GetNamespaceScope(ctx)))
+	}
+	return context.WithValue(ctx, factory.Key{},
+		externalversions.NewSharedInformerFactoryWithOptions(c, controller.GetResyncPeriod(ctx), opts...))
+}
diff --git a/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/injection/informers/keda/v1alpha1/scaledobject/fake/fake.go b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/injection/informers/keda/v1alpha1/scaledobject/fake/fake.go
new file mode 100644
index 000000000..c131bc0f7
--- /dev/null
+++ b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/injection/informers/keda/v1alpha1/scaledobject/fake/fake.go
@@ -0,0 +1,40 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	context "context"
+
+	fake "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/injection/informers/factory/fake"
+	scaledobject "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/injection/informers/keda/v1alpha1/scaledobject"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+)
+
+var Get = scaledobject.Get
+
+func init() {
+	injection.Fake.RegisterInformer(withInformer)
+}
+
+func withInformer(ctx context.Context) (context.Context, controller.Informer) {
+	f := fake.Get(ctx)
+	inf := f.Keda().V1alpha1().ScaledObjects()
+	return context.WithValue(ctx, scaledobject.Key{}, inf), inf.Informer()
+}
diff --git a/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/injection/informers/keda/v1alpha1/scaledobject/scaledobject.go b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/injection/informers/keda/v1alpha1/scaledobject/scaledobject.go
new file mode 100644
index 000000000..8c145d6cd
--- /dev/null
+++ b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/injection/informers/keda/v1alpha1/scaledobject/scaledobject.go
@@ -0,0 +1,116 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package scaledobject
+
+import (
+	context "context"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	cache "k8s.io/client-go/tools/cache"
+	apiskedav1alpha1 "knative.dev/eventing-autoscaler-keda/third_party/pkg/apis/keda/v1alpha1"
+	versioned "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/clientset/versioned"
+	v1alpha1 "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/keda/v1alpha1"
+	client "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/injection/client"
+	factory "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/injection/informers/factory"
+	kedav1alpha1 "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/listers/keda/v1alpha1"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+func init() {
+	injection.Default.RegisterInformer(withInformer)
+	injection.Dynamic.RegisterDynamicInformer(withDynamicInformer)
+}
+
+// Key is used for associating the Informer inside the context.Context.
+type Key struct{}
+
+func withInformer(ctx context.Context) (context.Context, controller.Informer) {
+	f := factory.Get(ctx)
+	inf := f.Keda().V1alpha1().ScaledObjects()
+	return context.WithValue(ctx, Key{}, inf), inf.Informer()
+}
+
+func withDynamicInformer(ctx context.Context) context.Context {
+	inf := &wrapper{client: client.Get(ctx), resourceVersion: injection.GetResourceVersion(ctx)}
+	return context.WithValue(ctx, Key{}, inf)
+}
+
+// Get extracts the typed informer from the context.
+func Get(ctx context.Context) v1alpha1.ScaledObjectInformer {
+	untyped := ctx.Value(Key{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/keda/v1alpha1.ScaledObjectInformer from context.")
+	}
+	return untyped.(v1alpha1.ScaledObjectInformer)
+}
+
+type wrapper struct {
+	client versioned.Interface
+
+	namespace string
+
+	resourceVersion string
+}
+
+var _ v1alpha1.ScaledObjectInformer = (*wrapper)(nil)
+var _ kedav1alpha1.ScaledObjectLister = (*wrapper)(nil)
+
+func (w *wrapper) Informer() cache.SharedIndexInformer {
+	return cache.NewSharedIndexInformer(nil, &apiskedav1alpha1.ScaledObject{}, 0, nil)
+}
+
+func (w *wrapper) Lister() kedav1alpha1.ScaledObjectLister {
+	return w
+}
+
+func (w *wrapper) ScaledObjects(namespace string) kedav1alpha1.ScaledObjectNamespaceLister {
+	return &wrapper{client: w.client, namespace: namespace, resourceVersion: w.resourceVersion}
+}
+
+// SetResourceVersion allows consumers to adjust the minimum resourceVersion
+// used by the underlying client.  It is not accessible via the standard
+// lister interface, but can be accessed through a user-defined interface and
+// an implementation check e.g. rvs, ok := foo.(ResourceVersionSetter)
+func (w *wrapper) SetResourceVersion(resourceVersion string) {
+	w.resourceVersion = resourceVersion
+}
+
+func (w *wrapper) List(selector labels.Selector) (ret []*apiskedav1alpha1.ScaledObject, err error) {
+	lo, err := w.client.KedaV1alpha1().ScaledObjects(w.namespace).List(context.TODO(), v1.ListOptions{
+		LabelSelector:   selector.String(),
+		ResourceVersion: w.resourceVersion,
+	})
+	if err != nil {
+		return nil, err
+	}
+	for idx := range lo.Items {
+		ret = append(ret, &lo.Items[idx])
+	}
+	return ret, nil
+}
+
+func (w *wrapper) Get(name string) (*apiskedav1alpha1.ScaledObject, error) {
+	return w.client.KedaV1alpha1().ScaledObjects(w.namespace).Get(context.TODO(), name, v1.GetOptions{
+		ResourceVersion: w.resourceVersion,
+	})
+}
diff --git a/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/listers/keda/v1alpha1/clustertriggerauthentication.go b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/listers/keda/v1alpha1/clustertriggerauthentication.go
new file mode 100644
index 000000000..9065ff5c5
--- /dev/null
+++ b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/listers/keda/v1alpha1/clustertriggerauthentication.go
@@ -0,0 +1,68 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by lister-gen. DO NOT EDIT.
+
+package v1alpha1
+
+import (
+	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/tools/cache"
+	v1alpha1 "knative.dev/eventing-autoscaler-keda/third_party/pkg/apis/keda/v1alpha1"
+)
+
+// ClusterTriggerAuthenticationLister helps list ClusterTriggerAuthentications.
+// All objects returned here must be treated as read-only.
+type ClusterTriggerAuthenticationLister interface {
+	// List lists all ClusterTriggerAuthentications in the indexer.
+	// Objects returned here must be treated as read-only.
+	List(selector labels.Selector) (ret []*v1alpha1.ClusterTriggerAuthentication, err error)
+	// Get retrieves the ClusterTriggerAuthentication from the index for a given name.
+	// Objects returned here must be treated as read-only.
+	Get(name string) (*v1alpha1.ClusterTriggerAuthentication, error)
+	ClusterTriggerAuthenticationListerExpansion
+}
+
+// clusterTriggerAuthenticationLister implements the ClusterTriggerAuthenticationLister interface.
+type clusterTriggerAuthenticationLister struct {
+	indexer cache.Indexer
+}
+
+// NewClusterTriggerAuthenticationLister returns a new ClusterTriggerAuthenticationLister.
+func NewClusterTriggerAuthenticationLister(indexer cache.Indexer) ClusterTriggerAuthenticationLister {
+	return &clusterTriggerAuthenticationLister{indexer: indexer}
+}
+
+// List lists all ClusterTriggerAuthentications in the indexer.
+func (s *clusterTriggerAuthenticationLister) List(selector labels.Selector) (ret []*v1alpha1.ClusterTriggerAuthentication, err error) {
+	err = cache.ListAll(s.indexer, selector, func(m interface{}) {
+		ret = append(ret, m.(*v1alpha1.ClusterTriggerAuthentication))
+	})
+	return ret, err
+}
+
+// Get retrieves the ClusterTriggerAuthentication from the index for a given name.
+func (s *clusterTriggerAuthenticationLister) Get(name string) (*v1alpha1.ClusterTriggerAuthentication, error) {
+	obj, exists, err := s.indexer.GetByKey(name)
+	if err != nil {
+		return nil, err
+	}
+	if !exists {
+		return nil, errors.NewNotFound(v1alpha1.Resource("clustertriggerauthentication"), name)
+	}
+	return obj.(*v1alpha1.ClusterTriggerAuthentication), nil
+}
diff --git a/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/listers/keda/v1alpha1/expansion_generated.go b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/listers/keda/v1alpha1/expansion_generated.go
new file mode 100644
index 000000000..1b6d65f93
--- /dev/null
+++ b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/listers/keda/v1alpha1/expansion_generated.go
@@ -0,0 +1,47 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by lister-gen. DO NOT EDIT.
+
+package v1alpha1
+
+// ClusterTriggerAuthenticationListerExpansion allows custom methods to be added to
+// ClusterTriggerAuthenticationLister.
+type ClusterTriggerAuthenticationListerExpansion interface{}
+
+// ScaledJobListerExpansion allows custom methods to be added to
+// ScaledJobLister.
+type ScaledJobListerExpansion interface{}
+
+// ScaledJobNamespaceListerExpansion allows custom methods to be added to
+// ScaledJobNamespaceLister.
+type ScaledJobNamespaceListerExpansion interface{}
+
+// ScaledObjectListerExpansion allows custom methods to be added to
+// ScaledObjectLister.
+type ScaledObjectListerExpansion interface{}
+
+// ScaledObjectNamespaceListerExpansion allows custom methods to be added to
+// ScaledObjectNamespaceLister.
+type ScaledObjectNamespaceListerExpansion interface{}
+
+// TriggerAuthenticationListerExpansion allows custom methods to be added to
+// TriggerAuthenticationLister.
+type TriggerAuthenticationListerExpansion interface{}
+
+// TriggerAuthenticationNamespaceListerExpansion allows custom methods to be added to
+// TriggerAuthenticationNamespaceLister.
+type TriggerAuthenticationNamespaceListerExpansion interface{}
diff --git a/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/listers/keda/v1alpha1/scaledjob.go b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/listers/keda/v1alpha1/scaledjob.go
new file mode 100644
index 000000000..913e7cad7
--- /dev/null
+++ b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/listers/keda/v1alpha1/scaledjob.go
@@ -0,0 +1,99 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by lister-gen. DO NOT EDIT.
+
+package v1alpha1
+
+import (
+	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/tools/cache"
+	v1alpha1 "knative.dev/eventing-autoscaler-keda/third_party/pkg/apis/keda/v1alpha1"
+)
+
+// ScaledJobLister helps list ScaledJobs.
+// All objects returned here must be treated as read-only.
+type ScaledJobLister interface {
+	// List lists all ScaledJobs in the indexer.
+	// Objects returned here must be treated as read-only.
+	List(selector labels.Selector) (ret []*v1alpha1.ScaledJob, err error)
+	// ScaledJobs returns an object that can list and get ScaledJobs.
+	ScaledJobs(namespace string) ScaledJobNamespaceLister
+	ScaledJobListerExpansion
+}
+
+// scaledJobLister implements the ScaledJobLister interface.
+type scaledJobLister struct {
+	indexer cache.Indexer
+}
+
+// NewScaledJobLister returns a new ScaledJobLister.
+func NewScaledJobLister(indexer cache.Indexer) ScaledJobLister {
+	return &scaledJobLister{indexer: indexer}
+}
+
+// List lists all ScaledJobs in the indexer.
+func (s *scaledJobLister) List(selector labels.Selector) (ret []*v1alpha1.ScaledJob, err error) {
+	err = cache.ListAll(s.indexer, selector, func(m interface{}) {
+		ret = append(ret, m.(*v1alpha1.ScaledJob))
+	})
+	return ret, err
+}
+
+// ScaledJobs returns an object that can list and get ScaledJobs.
+func (s *scaledJobLister) ScaledJobs(namespace string) ScaledJobNamespaceLister {
+	return scaledJobNamespaceLister{indexer: s.indexer, namespace: namespace}
+}
+
+// ScaledJobNamespaceLister helps list and get ScaledJobs.
+// All objects returned here must be treated as read-only.
+type ScaledJobNamespaceLister interface {
+	// List lists all ScaledJobs in the indexer for a given namespace.
+	// Objects returned here must be treated as read-only.
+	List(selector labels.Selector) (ret []*v1alpha1.ScaledJob, err error)
+	// Get retrieves the ScaledJob from the indexer for a given namespace and name.
+	// Objects returned here must be treated as read-only.
+	Get(name string) (*v1alpha1.ScaledJob, error)
+	ScaledJobNamespaceListerExpansion
+}
+
+// scaledJobNamespaceLister implements the ScaledJobNamespaceLister
+// interface.
+type scaledJobNamespaceLister struct {
+	indexer   cache.Indexer
+	namespace string
+}
+
+// List lists all ScaledJobs in the indexer for a given namespace.
+func (s scaledJobNamespaceLister) List(selector labels.Selector) (ret []*v1alpha1.ScaledJob, err error) {
+	err = cache.ListAllByNamespace(s.indexer, s.namespace, selector, func(m interface{}) {
+		ret = append(ret, m.(*v1alpha1.ScaledJob))
+	})
+	return ret, err
+}
+
+// Get retrieves the ScaledJob from the indexer for a given namespace and name.
+func (s scaledJobNamespaceLister) Get(name string) (*v1alpha1.ScaledJob, error) {
+	obj, exists, err := s.indexer.GetByKey(s.namespace + "/" + name)
+	if err != nil {
+		return nil, err
+	}
+	if !exists {
+		return nil, errors.NewNotFound(v1alpha1.Resource("scaledjob"), name)
+	}
+	return obj.(*v1alpha1.ScaledJob), nil
+}
diff --git a/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/listers/keda/v1alpha1/scaledobject.go b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/listers/keda/v1alpha1/scaledobject.go
new file mode 100644
index 000000000..c3f74ace6
--- /dev/null
+++ b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/listers/keda/v1alpha1/scaledobject.go
@@ -0,0 +1,99 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by lister-gen. DO NOT EDIT.
+
+package v1alpha1
+
+import (
+	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/tools/cache"
+	v1alpha1 "knative.dev/eventing-autoscaler-keda/third_party/pkg/apis/keda/v1alpha1"
+)
+
+// ScaledObjectLister helps list ScaledObjects.
+// All objects returned here must be treated as read-only.
+type ScaledObjectLister interface {
+	// List lists all ScaledObjects in the indexer.
+	// Objects returned here must be treated as read-only.
+	List(selector labels.Selector) (ret []*v1alpha1.ScaledObject, err error)
+	// ScaledObjects returns an object that can list and get ScaledObjects.
+	ScaledObjects(namespace string) ScaledObjectNamespaceLister
+	ScaledObjectListerExpansion
+}
+
+// scaledObjectLister implements the ScaledObjectLister interface.
+type scaledObjectLister struct {
+	indexer cache.Indexer
+}
+
+// NewScaledObjectLister returns a new ScaledObjectLister.
+func NewScaledObjectLister(indexer cache.Indexer) ScaledObjectLister {
+	return &scaledObjectLister{indexer: indexer}
+}
+
+// List lists all ScaledObjects in the indexer.
+func (s *scaledObjectLister) List(selector labels.Selector) (ret []*v1alpha1.ScaledObject, err error) {
+	err = cache.ListAll(s.indexer, selector, func(m interface{}) {
+		ret = append(ret, m.(*v1alpha1.ScaledObject))
+	})
+	return ret, err
+}
+
+// ScaledObjects returns an object that can list and get ScaledObjects.
+func (s *scaledObjectLister) ScaledObjects(namespace string) ScaledObjectNamespaceLister {
+	return scaledObjectNamespaceLister{indexer: s.indexer, namespace: namespace}
+}
+
+// ScaledObjectNamespaceLister helps list and get ScaledObjects.
+// All objects returned here must be treated as read-only.
+type ScaledObjectNamespaceLister interface {
+	// List lists all ScaledObjects in the indexer for a given namespace.
+	// Objects returned here must be treated as read-only.
+	List(selector labels.Selector) (ret []*v1alpha1.ScaledObject, err error)
+	// Get retrieves the ScaledObject from the indexer for a given namespace and name.
+	// Objects returned here must be treated as read-only.
+	Get(name string) (*v1alpha1.ScaledObject, error)
+	ScaledObjectNamespaceListerExpansion
+}
+
+// scaledObjectNamespaceLister implements the ScaledObjectNamespaceLister
+// interface.
+type scaledObjectNamespaceLister struct {
+	indexer   cache.Indexer
+	namespace string
+}
+
+// List lists all ScaledObjects in the indexer for a given namespace.
+func (s scaledObjectNamespaceLister) List(selector labels.Selector) (ret []*v1alpha1.ScaledObject, err error) {
+	err = cache.ListAllByNamespace(s.indexer, s.namespace, selector, func(m interface{}) {
+		ret = append(ret, m.(*v1alpha1.ScaledObject))
+	})
+	return ret, err
+}
+
+// Get retrieves the ScaledObject from the indexer for a given namespace and name.
+func (s scaledObjectNamespaceLister) Get(name string) (*v1alpha1.ScaledObject, error) {
+	obj, exists, err := s.indexer.GetByKey(s.namespace + "/" + name)
+	if err != nil {
+		return nil, err
+	}
+	if !exists {
+		return nil, errors.NewNotFound(v1alpha1.Resource("scaledobject"), name)
+	}
+	return obj.(*v1alpha1.ScaledObject), nil
+}
diff --git a/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/listers/keda/v1alpha1/triggerauthentication.go b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/listers/keda/v1alpha1/triggerauthentication.go
new file mode 100644
index 000000000..1c8c16ddb
--- /dev/null
+++ b/vendor/knative.dev/eventing-autoscaler-keda/third_party/pkg/client/listers/keda/v1alpha1/triggerauthentication.go
@@ -0,0 +1,99 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by lister-gen. DO NOT EDIT.
+
+package v1alpha1
+
+import (
+	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/tools/cache"
+	v1alpha1 "knative.dev/eventing-autoscaler-keda/third_party/pkg/apis/keda/v1alpha1"
+)
+
+// TriggerAuthenticationLister helps list TriggerAuthentications.
+// All objects returned here must be treated as read-only.
+type TriggerAuthenticationLister interface {
+	// List lists all TriggerAuthentications in the indexer.
+	// Objects returned here must be treated as read-only.
+	List(selector labels.Selector) (ret []*v1alpha1.TriggerAuthentication, err error)
+	// TriggerAuthentications returns an object that can list and get TriggerAuthentications.
+	TriggerAuthentications(namespace string) TriggerAuthenticationNamespaceLister
+	TriggerAuthenticationListerExpansion
+}
+
+// triggerAuthenticationLister implements the TriggerAuthenticationLister interface.
+type triggerAuthenticationLister struct {
+	indexer cache.Indexer
+}
+
+// NewTriggerAuthenticationLister returns a new TriggerAuthenticationLister.
+func NewTriggerAuthenticationLister(indexer cache.Indexer) TriggerAuthenticationLister {
+	return &triggerAuthenticationLister{indexer: indexer}
+}
+
+// List lists all TriggerAuthentications in the indexer.
+func (s *triggerAuthenticationLister) List(selector labels.Selector) (ret []*v1alpha1.TriggerAuthentication, err error) {
+	err = cache.ListAll(s.indexer, selector, func(m interface{}) {
+		ret = append(ret, m.(*v1alpha1.TriggerAuthentication))
+	})
+	return ret, err
+}
+
+// TriggerAuthentications returns an object that can list and get TriggerAuthentications.
+func (s *triggerAuthenticationLister) TriggerAuthentications(namespace string) TriggerAuthenticationNamespaceLister {
+	return triggerAuthenticationNamespaceLister{indexer: s.indexer, namespace: namespace}
+}
+
+// TriggerAuthenticationNamespaceLister helps list and get TriggerAuthentications.
+// All objects returned here must be treated as read-only.
+type TriggerAuthenticationNamespaceLister interface {
+	// List lists all TriggerAuthentications in the indexer for a given namespace.
+	// Objects returned here must be treated as read-only.
+	List(selector labels.Selector) (ret []*v1alpha1.TriggerAuthentication, err error)
+	// Get retrieves the TriggerAuthentication from the indexer for a given namespace and name.
+	// Objects returned here must be treated as read-only.
+	Get(name string) (*v1alpha1.TriggerAuthentication, error)
+	TriggerAuthenticationNamespaceListerExpansion
+}
+
+// triggerAuthenticationNamespaceLister implements the TriggerAuthenticationNamespaceLister
+// interface.
+type triggerAuthenticationNamespaceLister struct {
+	indexer   cache.Indexer
+	namespace string
+}
+
+// List lists all TriggerAuthentications in the indexer for a given namespace.
+func (s triggerAuthenticationNamespaceLister) List(selector labels.Selector) (ret []*v1alpha1.TriggerAuthentication, err error) {
+	err = cache.ListAllByNamespace(s.indexer, s.namespace, selector, func(m interface{}) {
+		ret = append(ret, m.(*v1alpha1.TriggerAuthentication))
+	})
+	return ret, err
+}
+
+// Get retrieves the TriggerAuthentication from the indexer for a given namespace and name.
+func (s triggerAuthenticationNamespaceLister) Get(name string) (*v1alpha1.TriggerAuthentication, error) {
+	obj, exists, err := s.indexer.GetByKey(s.namespace + "/" + name)
+	if err != nil {
+		return nil, err
+	}
+	if !exists {
+		return nil, errors.NewNotFound(v1alpha1.Resource("triggerauthentication"), name)
+	}
+	return obj.(*v1alpha1.TriggerAuthentication), nil
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 252313c7a..31e0ae341 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1350,8 +1350,17 @@ knative.dev/eventing-autoscaler-keda/third_party/pkg/client/clientset/versioned/
 knative.dev/eventing-autoscaler-keda/third_party/pkg/client/clientset/versioned/scheme
 knative.dev/eventing-autoscaler-keda/third_party/pkg/client/clientset/versioned/typed/keda/v1alpha1
 knative.dev/eventing-autoscaler-keda/third_party/pkg/client/clientset/versioned/typed/keda/v1alpha1/fake
+knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions
+knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/internalinterfaces
+knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/keda
+knative.dev/eventing-autoscaler-keda/third_party/pkg/client/informers/externalversions/keda/v1alpha1
 knative.dev/eventing-autoscaler-keda/third_party/pkg/client/injection/client
 knative.dev/eventing-autoscaler-keda/third_party/pkg/client/injection/client/fake
+knative.dev/eventing-autoscaler-keda/third_party/pkg/client/injection/informers/factory
+knative.dev/eventing-autoscaler-keda/third_party/pkg/client/injection/informers/factory/fake
+knative.dev/eventing-autoscaler-keda/third_party/pkg/client/injection/informers/keda/v1alpha1/scaledobject
+knative.dev/eventing-autoscaler-keda/third_party/pkg/client/injection/informers/keda/v1alpha1/scaledobject/fake
+knative.dev/eventing-autoscaler-keda/third_party/pkg/client/listers/keda/v1alpha1
 # knative.dev/eventing-kafka v0.34.1-0.20220907020318-2f1fbd6dda1b
 ## explicit; go 1.16
 knative.dev/eventing-kafka/pkg/apis/bindings


Test patch:
diff --git a/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go b/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
index b425884dd..febd973ca 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
@@ -52,6 +52,9 @@ import (
 
 	configapis "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/config"
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/testing"
+
+	kedaclient "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/injection/client/fake"
+	cm "knative.dev/pkg/configmap/testing"
 )
 
 type SchedulerFunc func(vpod scheduler.VPod) ([]eventingduckv1alpha1.Placement, error)
@@ -74,7 +77,7 @@ var finalizerUpdatedEvent = Eventf(
 )
 
 func TestReconcileKind(t *testing.T) {
-
+	//TODO: Add tests with KEDA installed
 	tt := TableTest{
 		{
 			Name: "Consumers in multiple pods",
@@ -146,6 +149,7 @@ func TestReconcileKind(t *testing.T) {
 						}
 						_ = cg.MarkReconcileConsumersFailed("PropagateSubscriberURI", ErrNoSubscriberURI)
 						cg.MarkScheduleSucceeded()
+						cg.MarkAutoscalerDisabled() // KEDA not installed
 						return cg
 					}(),
 				},
@@ -348,6 +352,7 @@ func TestReconcileKind(t *testing.T) {
 							Message: "failed to bind resource to pod: EOF",
 						})
 						cg.MarkScheduleSucceeded()
+						cg.MarkAutoscalerDisabled() // KEDA not installed
 						return cg
 					}(),
 				},
@@ -458,6 +463,7 @@ func TestReconcileKind(t *testing.T) {
 						}
 						cg.MarkReconcileConsumersSucceeded()
 						cg.MarkScheduleSucceeded()
+						cg.MarkAutoscalerDisabled() // KEDA not installed
 						cg.Status.SubscriberURI = ConsumerSubscriberURI
 						return cg
 					}(),
@@ -587,7 +593,6 @@ func TestReconcileKind(t *testing.T) {
 				finalizerUpdatedEvent,
 				"Warning InternalError failed to initialize consumer group offset: failed to create config options for Kafka cluster auth: failed to read secret test-cg-ns/non-existing secret: secret \"non-existing secret\" not found",
 			},
-			WantCreates: []runtime.Object{},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
 					Object: func() runtime.Object {
@@ -858,6 +863,7 @@ func TestReconcileKind(t *testing.T) {
 						}
 						cg.MarkReconcileConsumersSucceeded()
 						cg.MarkScheduleSucceeded()
+						cg.MarkAutoscalerDisabled() // KEDA not installed
 						cg.Status.SubscriberURI = ConsumerSubscriberURI
 						return cg
 					}(),
@@ -944,6 +950,7 @@ func TestReconcileKind(t *testing.T) {
 						}
 						_ = cg.MarkReconcileConsumersFailed("PropagateSubscriberURI", ErrNoSubscriberURI)
 						cg.MarkScheduleSucceeded()
+						cg.MarkAutoscalerDisabled() // KEDA not installed
 						return cg
 					}(),
 				},
@@ -1027,6 +1034,7 @@ func TestReconcileKind(t *testing.T) {
 						}
 						_ = cg.MarkReconcileConsumersFailed("PropagateSubscriberURI", ErrNoSubscriberURI)
 						cg.MarkScheduleSucceeded()
+						cg.MarkAutoscalerDisabled() // KEDA not installed
 						return cg
 					}(),
 				},
@@ -1111,6 +1119,7 @@ func TestReconcileKind(t *testing.T) {
 						}
 						cg.MarkReconcileConsumersSucceeded()
 						cg.MarkScheduleSucceeded()
+						cg.MarkAutoscalerDisabled() // KEDA not installed
 						cg.Status.SubscriberURI = ConsumerSubscriberURI
 						return cg
 					}(),
@@ -1236,6 +1245,7 @@ func TestReconcileKind(t *testing.T) {
 						}
 						cg.MarkReconcileConsumersSucceeded()
 						cg.MarkScheduleSucceeded()
+						cg.MarkAutoscalerDisabled() // KEDA not installed
 						cg.Status.SubscriberURI = ConsumerSubscriberURI
 						cg.Status.DeadLetterSinkURI = ConsumerDeadLetterSinkURI
 						cg.Status.Replicas = pointer.Int32(1)
@@ -1344,6 +1354,7 @@ func TestReconcileKind(t *testing.T) {
 						}
 						cg.MarkReconcileConsumersSucceeded()
 						cg.MarkScheduleSucceeded()
+						cg.MarkAutoscalerDisabled() // KEDA not installed
 						cg.Status.SubscriberURI = ConsumerSubscriberURI
 						cg.Status.Replicas = pointer.Int32(1)
 						return cg
@@ -1450,6 +1461,7 @@ func TestReconcileKind(t *testing.T) {
 						}
 						_ = cg.MarkReconcileConsumersFailed("PropagateSubscriberURI", ErrNoSubscriberURI)
 						cg.MarkScheduleSucceeded()
+						cg.MarkAutoscalerDisabled() // KEDA not installed
 						cg.Status.Replicas = pointer.Int32(0)
 						return cg
 					}(),
@@ -1516,6 +1528,11 @@ func TestReconcileKind(t *testing.T) {
 
 	tt.Test(t, NewFactory(nil, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
 
+		ctx, _ = kedaclient.With(ctx)
+		store := configapis.NewStore(ctx)
+		_, exampleConfig := cm.ConfigMapsFromTestFile(t, configapis.FlagsConfigName)
+		store.OnConfigChanged(exampleConfig)
+
 		r := Reconciler{
 			SchedulerFunc: func(s string) scheduler.Scheduler {
 				return row.OtherTestData[testSchedulerKey].(scheduler.Scheduler)
@@ -1524,17 +1541,157 @@ func TestReconcileKind(t *testing.T) {
 			InternalsClient: fakekafkainternalsclient.Get(ctx).InternalV1alpha1(),
 			SecretLister:    listers.GetSecretLister(),
 			KubeClient:      kubeclient.Get(ctx),
+			KedaClient:      kedaclient.Get(ctx),
 			NameGenerator:   &CounterGenerator{},
 			NewKafkaClient: func(addrs []string, config *sarama.Config) (sarama.Client, error) {
 				return &kafkatesting.MockKafkaClient{}, nil
 			},
+			NewKafkaClusterAdminClient: func(_ []string, _ *sarama.Config) (sarama.ClusterAdmin, error) {
+				return &kafkatesting.MockKafkaClusterAdmin{
+					T: t,
+				}, nil
+			},
 			InitOffsetsFunc: func(ctx context.Context, kafkaClient sarama.Client, kafkaAdminClient sarama.ClusterAdmin, topics []string, consumerGroup string) (int32, error) {
 				return 1, nil
 			},
-			SystemNamespace:   systemNamespace,
-			KafkaFeatureFlags: configapis.DefaultFeaturesConfig(),
+			SystemNamespace: systemNamespace,
+		}
+
+		r.KafkaFeatureFlags = configapis.FromContext(store.ToContext(ctx))
+
+		return consumergroup.NewReconciler(
+			ctx,
+			logging.FromContext(ctx),
+			fakekafkainternalsclient.Get(ctx),
+			listers.GetConsumerGroupLister(),
+			controller.GetEventRecorder(ctx),
+			r,
+		)
+	}))
+
+}
+
+func TestReconcileKindNoAutoscaler(t *testing.T) {
+
+	tt := TableTest{
+		{
+			Name: "Consumers in multiple pods with autoscaler disabled",
+			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,
+						}),
+					)),
+					ConsumerReady(),
+				),
+				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(1),
+							ConsumerForTrigger(),
+						)
+						cg.Status.Placements = []eventingduckv1alpha1.Placement{
+							{PodName: "p1", VReplicas: 1},
+							{PodName: "p2", VReplicas: 1},
+						}
+						cg.MarkReconcileConsumersSucceeded()
+						cg.MarkScheduleSucceeded()
+						cg.MarkAutoscalerDisabled() // autoscaler feature disabled
+						cg.Status.SubscriberURI = ConsumerSubscriberURI
+						return cg
+					}(),
+				},
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+		},
+	}
+
+	tt.Test(t, NewFactory(nil, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
+
+		ctx, _ = kedaclient.With(ctx)
+
+		r := Reconciler{
+			SchedulerFunc: func(s string) scheduler.Scheduler {
+				return row.OtherTestData[testSchedulerKey].(scheduler.Scheduler)
+			},
+			ConsumerLister:  listers.GetConsumerLister(),
+			InternalsClient: fakekafkainternalsclient.Get(ctx).InternalV1alpha1(),
+			SecretLister:    listers.GetSecretLister(),
+			KubeClient:      kubeclient.Get(ctx),
+			KedaClient:      kedaclient.Get(ctx),
+			NameGenerator:   &CounterGenerator{},
+			NewKafkaClient: func(addrs []string, config *sarama.Config) (sarama.Client, error) {
+				return &kafkatesting.MockKafkaClient{}, nil
+			},
+			NewKafkaClusterAdminClient: func(_ []string, _ *sarama.Config) (sarama.ClusterAdmin, error) {
+				return &kafkatesting.MockKafkaClusterAdmin{
+					T: t,
+				}, nil
+			},
+			InitOffsetsFunc: func(ctx context.Context, kafkaClient sarama.Client, kafkaAdminClient sarama.ClusterAdmin, topics []string, consumerGroup string) (int32, error) {
+				return 1, nil
+			},
+			SystemNamespace: systemNamespace,
 		}
 
+		r.KafkaFeatureFlags = configapis.DefaultFeaturesConfig()
+
 		return consumergroup.NewReconciler(
 			ctx,
 			logging.FromContext(ctx),
@@ -1749,6 +1906,9 @@ func TestFinalizeKind(t *testing.T) {
 			ConsumerLister:  listers.GetConsumerLister(),
 			InternalsClient: fakekafkainternalsclient.Get(ctx).InternalV1alpha1(),
 			SecretLister:    listers.GetSecretLister(),
+			NewKafkaClient: func(addrs []string, config *sarama.Config) (sarama.Client, error) {
+				return &kafkatesting.MockKafkaClient{}, nil
+			},
 			NewKafkaClusterAdminClient: func(_ []string, _ *sarama.Config) (sarama.ClusterAdmin, error) {
 				return &kafkatesting.MockKafkaClusterAdmin{
 					T: t,
diff --git a/control-plane/pkg/reconciler/consumergroup/controller_test.go b/control-plane/pkg/reconciler/consumergroup/controller_test.go
index 786b7fcb3..d6769f46d 100644
--- a/control-plane/pkg/reconciler/consumergroup/controller_test.go
+++ b/control-plane/pkg/reconciler/consumergroup/controller_test.go
@@ -32,9 +32,12 @@ import (
 	"knative.dev/pkg/configmap"
 	reconcilertesting "knative.dev/pkg/reconciler/testing"
 
+	_ "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/injection/informers/keda/v1alpha1/scaledobject/fake"
 	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 	_ "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumer/fake"
 	_ "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumergroup/fake"
+
+	kedaclient "knative.dev/eventing-autoscaler-keda/third_party/pkg/client/injection/client/fake"
 )
 
 const (
@@ -48,6 +51,7 @@ const (
 
 func TestNewController(t *testing.T) {
 	ctx, _ := reconcilertesting.SetupFakeContext(t)
+	ctx, _ = kedaclient.With(ctx)
 
 	t.Setenv("SYSTEM_NAMESPACE", systemNamespace)
 
diff --git a/control-plane/pkg/reconciler/consumergroup/testdata/config-kafka-features.yaml b/control-plane/pkg/reconciler/consumergroup/testdata/config-kafka-features.yaml
new file mode 100644
index 000000000..d0057f72b
--- /dev/null
+++ b/control-plane/pkg/reconciler/consumergroup/testdata/config-kafka-features.yaml
@@ -0,0 +1,8 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: config-kafka-features
+  namespace: knative-eventing
+data:
+  _example: |
+    controller.autoscaler: "enabled"
diff --git a/control-plane/pkg/reconciler/testing/objects_common.go b/control-plane/pkg/reconciler/testing/objects_common.go
index f66c2247c..32b2e0019 100644
--- a/control-plane/pkg/reconciler/testing/objects_common.go
+++ b/control-plane/pkg/reconciler/testing/objects_common.go
@@ -315,6 +315,23 @@ func NewSASLSSLSecret(ns, name string) *corev1.Secret {
 	}
 }
 
+func NewKedaSecret(ns, name string) *corev1.Secret {
+
+	return &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: ns,
+			Name:      name,
+			OwnerReferences: []metav1.OwnerReference{
+				ConsumerGroupAsOwnerRef(),
+			},
+		},
+		StringData: map[string]string{
+			"sasl": "plaintext",
+			"tls":  "enable",
+		},
+	}
+}
+
 func loadCerts() (ca, userKey, userCert []byte) {
 	ca, err := ioutil.ReadFile("testdata/ca.crt")
 	if err != nil {