Repository: knative-sandbox/eventing-kafka-broker
Pull request #2235 (2022-05-30 09:54:27)

Introduce a rate limiter feature flag
Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

Part of #2125

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Introduce a rate limiter feature flag

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- :gift: Add new feature
- :bug: Fix bug
- :broom: Update or clean up current behavior
- :wastebasket: Remove feature or internal logic

Otherwise delete the rest of this template.
-->

The feature flag setup is similar to the one in eventing and serving.

**Docs**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


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

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

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
knative-prow[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 | f78893516b07cd34800fc8722b246cd595f19e45 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/unit-tests_eventing-kafka-broker_main/1531212440308551680) | true | `/test unit-tests_eventing-kafka-broker_main`

[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 -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2235?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2235](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2235?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (f788935) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/dbb7929242fde897a0ee1d266b5b884fbc5fa2e4?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (dbb7929) will **increase** coverage by `15.87%`.
> The diff coverage is `n/a`.

```diff
@@              Coverage Diff              @@
##               main    #2235       +/-   ##
=============================================
+ Coverage     66.53%   82.40%   +15.87%     
- Complexity      687      688        +1     
=============================================
  Files           142       72       -70     
  Lines          9206     2313     -6893     
  Branches        201      201               
=============================================
- Hits           6125     1906     -4219     
+ Misses         2663      293     -2370     
+ Partials        418      114      -304     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `82.40% <ø> (+0.04%)` | :arrow_up: |

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/2235?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [control-plane/pkg/contract/contract.pb.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2235/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvY29udHJhY3QvY29udHJhY3QucGIuZ28=) | | |
| [control-plane/pkg/reconciler/consumer/consumer.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2235/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lci9jb25zdW1lci5nbw==) | | |
| [...ontrol-plane/pkg/reconciler/consumer/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2235/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lci9jb250cm9sbGVyLmdv) | | |
| [control-plane/pkg/reconciler/channel/channel.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2235/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jaGFubmVsL2NoYW5uZWwuZ28=) | | |
| [control-plane/pkg/reconciler/sink/auth\_provider.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2235/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zaW5rL2F1dGhfcHJvdmlkZXIuZ28=) | | |
| [...als/kafka/eventing/v1alpha1/consumer\_validation.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2235/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9pbnRlcm5hbHMva2Fma2EvZXZlbnRpbmcvdjFhbHBoYTEvY29uc3VtZXJfdmFsaWRhdGlvbi5nbw==) | | |
| [...ntrol-plane/pkg/reconciler/source/auth\_provider.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2235/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zb3VyY2UvYXV0aF9wcm92aWRlci5nbw==) | | |
| [control-plane/pkg/contract/subscriptionsapi.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2235/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvY29udHJhY3Qvc3Vic2NyaXB0aW9uc2FwaS5nbw==) | | |
| [...ntrol-plane/pkg/apis/eventing/v1alpha1/register.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2235/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9yZWdpc3Rlci5nbw==) | | |
| [...trol-plane/pkg/prober/probertesting/mock\_prober.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2235/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcHJvYmVyL3Byb2JlcnRlc3RpbmcvbW9ja19wcm9iZXIuZ28=) | | |
| ... and [61 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2235/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2235?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2235?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [dbb7929...f788935](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2235?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).

pierDipi: /cc @aavarghese @matzew 
pierDipi: /cc @steven0711dong 
matzew: /test unit-tests_eventing-kafka-broker_main
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 | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/unit-tests_eventing-kafka-broker_main/1531658708923191296) | true | `/test unit-tests_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1531658720776294400) | true | `/test channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main`

[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 | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/unit-tests_eventing-kafka-broker_main/1531658708923191296) | true | `/test unit-tests_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1531658720776294400) | true | `/test channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main`
channel-integration-tests-sasl-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-sasl-ssl_eventing-kafka-broker_main/1531658717425045504) | true | `/test channel-integration-tests-sasl-ssl_eventing-kafka-broker_main`

[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: /retest
/retest-required
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 | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/unit-tests_eventing-kafka-broker_main/1531658708923191296) | true | `/test unit-tests_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1531658720776294400) | true | `/test channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main`
channel-integration-tests-sasl-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-sasl-ssl_eventing-kafka-broker_main/1531658717425045504) | true | `/test channel-integration-tests-sasl-ssl_eventing-kafka-broker_main`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1531658719098572800) | true | `/test channel-integration-tests-sasl-plain_eventing-kafka-broker_main`

[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 | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/unit-tests_eventing-kafka-broker_main/1531658708923191296) | true | `/test unit-tests_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1531658720776294400) | true | `/test channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main`
channel-integration-tests-sasl-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-sasl-ssl_eventing-kafka-broker_main/1531658717425045504) | true | `/test channel-integration-tests-sasl-ssl_eventing-kafka-broker_main`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1531658719098572800) | true | `/test channel-integration-tests-sasl-plain_eventing-kafka-broker_main`
channel-integration-tests-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-ssl_eventing-kafka-broker_main/1531658716204503040) | true | `/test channel-integration-tests-ssl_eventing-kafka-broker_main`

[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 | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/unit-tests_eventing-kafka-broker_main/1531658708923191296) | true | `/test unit-tests_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1531658720776294400) | true | `/test channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main`
channel-integration-tests-sasl-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-sasl-ssl_eventing-kafka-broker_main/1531658717425045504) | true | `/test channel-integration-tests-sasl-ssl_eventing-kafka-broker_main`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1531658719098572800) | true | `/test channel-integration-tests-sasl-plain_eventing-kafka-broker_main`
channel-integration-tests-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-ssl_eventing-kafka-broker_main/1531658716204503040) | true | `/test channel-integration-tests-ssl_eventing-kafka-broker_main`
reconciler-tests_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/reconciler-tests_eventing-kafka-broker_main/1531658713234935808) | true | `/test reconciler-tests_eventing-kafka-broker_main`

[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 | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/unit-tests_eventing-kafka-broker_main/1531658708923191296) | true | `/test unit-tests_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1531658720776294400) | true | `/test channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main`
channel-integration-tests-sasl-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-sasl-ssl_eventing-kafka-broker_main/1531658717425045504) | true | `/test channel-integration-tests-sasl-ssl_eventing-kafka-broker_main`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1531658719098572800) | true | `/test channel-integration-tests-sasl-plain_eventing-kafka-broker_main`
channel-integration-tests-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-ssl_eventing-kafka-broker_main/1531658716204503040) | true | `/test channel-integration-tests-ssl_eventing-kafka-broker_main`
reconciler-tests_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/reconciler-tests_eventing-kafka-broker_main/1531658713234935808) | true | `/test reconciler-tests_eventing-kafka-broker_main`
integration-tests_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/integration-tests_eventing-kafka-broker_main/1531658709875298304) | true | `/test integration-tests_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main/1531658722080722944) | true | `/test channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main`

[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 | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/unit-tests_eventing-kafka-broker_main/1531658708923191296) | true | `/test unit-tests_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1531658720776294400) | true | `/test channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main`
channel-integration-tests-sasl-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-sasl-ssl_eventing-kafka-broker_main/1531658717425045504) | true | `/test channel-integration-tests-sasl-ssl_eventing-kafka-broker_main`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1531658719098572800) | true | `/test channel-integration-tests-sasl-plain_eventing-kafka-broker_main`
channel-integration-tests-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-ssl_eventing-kafka-broker_main/1531658716204503040) | true | `/test channel-integration-tests-ssl_eventing-kafka-broker_main`
reconciler-tests_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/reconciler-tests_eventing-kafka-broker_main/1531658713234935808) | true | `/test reconciler-tests_eventing-kafka-broker_main`
integration-tests_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/integration-tests_eventing-kafka-broker_main/1531658709875298304) | true | `/test integration-tests_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main/1531658722080722944) | true | `/test channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main`
channel-reconciler-tests-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-ssl_eventing-kafka-broker_main/1531658720411389952) | true | `/test channel-reconciler-tests-ssl_eventing-kafka-broker_main`

[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-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1531658720776294400) | true | `/test channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1531658719098572800) | true | `/test channel-integration-tests-sasl-plain_eventing-kafka-broker_main`
channel-integration-tests-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-ssl_eventing-kafka-broker_main/1531658716204503040) | true | `/test channel-integration-tests-ssl_eventing-kafka-broker_main`
reconciler-tests_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/reconciler-tests_eventing-kafka-broker_main/1531658713234935808) | true | `/test reconciler-tests_eventing-kafka-broker_main`
integration-tests_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/integration-tests_eventing-kafka-broker_main/1531658709875298304) | true | `/test integration-tests_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main/1531658722080722944) | true | `/test channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main`
channel-reconciler-tests-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-ssl_eventing-kafka-broker_main/1531658720411389952) | true | `/test channel-reconciler-tests-ssl_eventing-kafka-broker_main`
channel-integration-tests-sasl-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-sasl-ssl_eventing-kafka-broker_main/1531663563834265600) | true | `/test channel-integration-tests-sasl-ssl_eventing-kafka-broker_main`

[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-sasl-plain_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1531658719098572800) | true | `/test channel-integration-tests-sasl-plain_eventing-kafka-broker_main`
channel-integration-tests-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-ssl_eventing-kafka-broker_main/1531658716204503040) | true | `/test channel-integration-tests-ssl_eventing-kafka-broker_main`
reconciler-tests_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/reconciler-tests_eventing-kafka-broker_main/1531658713234935808) | true | `/test reconciler-tests_eventing-kafka-broker_main`
integration-tests_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/integration-tests_eventing-kafka-broker_main/1531658709875298304) | true | `/test integration-tests_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main/1531658722080722944) | true | `/test channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main`
channel-reconciler-tests-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-ssl_eventing-kafka-broker_main/1531658720411389952) | true | `/test channel-reconciler-tests-ssl_eventing-kafka-broker_main`
channel-integration-tests-sasl-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-sasl-ssl_eventing-kafka-broker_main/1531663563834265600) | true | `/test channel-integration-tests-sasl-ssl_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1531663563872014336) | true | `/test channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main`

[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: /retest
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-sasl-plain_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1531658719098572800) | true | `/test channel-integration-tests-sasl-plain_eventing-kafka-broker_main`
channel-integration-tests-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-ssl_eventing-kafka-broker_main/1531658716204503040) | true | `/test channel-integration-tests-ssl_eventing-kafka-broker_main`
integration-tests_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/integration-tests_eventing-kafka-broker_main/1531658709875298304) | true | `/test integration-tests_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main/1531658722080722944) | true | `/test channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main`
channel-reconciler-tests-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-ssl_eventing-kafka-broker_main/1531658720411389952) | true | `/test channel-reconciler-tests-ssl_eventing-kafka-broker_main`
channel-integration-tests-sasl-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-sasl-ssl_eventing-kafka-broker_main/1531663563834265600) | true | `/test channel-integration-tests-sasl-ssl_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1531663563872014336) | true | `/test channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main`
reconciler-tests_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/reconciler-tests_eventing-kafka-broker_main/1531670008562716672) | true | `/test reconciler-tests_eventing-kafka-broker_main`

[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-sasl-plain_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1531658719098572800) | true | `/test channel-integration-tests-sasl-plain_eventing-kafka-broker_main`
channel-integration-tests-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-ssl_eventing-kafka-broker_main/1531658716204503040) | true | `/test channel-integration-tests-ssl_eventing-kafka-broker_main`
integration-tests_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/integration-tests_eventing-kafka-broker_main/1531658709875298304) | true | `/test integration-tests_eventing-kafka-broker_main`
channel-reconciler-tests-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-ssl_eventing-kafka-broker_main/1531658720411389952) | true | `/test channel-reconciler-tests-ssl_eventing-kafka-broker_main`
channel-integration-tests-sasl-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-sasl-ssl_eventing-kafka-broker_main/1531663563834265600) | true | `/test channel-integration-tests-sasl-ssl_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1531663563872014336) | true | `/test channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main`
reconciler-tests_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/reconciler-tests_eventing-kafka-broker_main/1531670008562716672) | true | `/test reconciler-tests_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main/1531670008814374912) | true | `/test channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main`

[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-sasl-plain_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1531658719098572800) | true | `/test channel-integration-tests-sasl-plain_eventing-kafka-broker_main`
channel-integration-tests-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-ssl_eventing-kafka-broker_main/1531658716204503040) | true | `/test channel-integration-tests-ssl_eventing-kafka-broker_main`
integration-tests_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/integration-tests_eventing-kafka-broker_main/1531658709875298304) | true | `/test integration-tests_eventing-kafka-broker_main`
channel-integration-tests-sasl-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-sasl-ssl_eventing-kafka-broker_main/1531663563834265600) | true | `/test channel-integration-tests-sasl-ssl_eventing-kafka-broker_main`
reconciler-tests_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/reconciler-tests_eventing-kafka-broker_main/1531670008562716672) | true | `/test reconciler-tests_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main/1531670008814374912) | true | `/test channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1531670008785014784) | true | `/test channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main`
channel-reconciler-tests-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-ssl_eventing-kafka-broker_main/1531670008730488832) | true | `/test channel-reconciler-tests-ssl_eventing-kafka-broker_main`

[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-sasl-plain_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1531658719098572800) | true | `/test channel-integration-tests-sasl-plain_eventing-kafka-broker_main`
integration-tests_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/integration-tests_eventing-kafka-broker_main/1531658709875298304) | true | `/test integration-tests_eventing-kafka-broker_main`
channel-integration-tests-sasl-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-sasl-ssl_eventing-kafka-broker_main/1531663563834265600) | true | `/test channel-integration-tests-sasl-ssl_eventing-kafka-broker_main`
reconciler-tests_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/reconciler-tests_eventing-kafka-broker_main/1531670008562716672) | true | `/test reconciler-tests_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main/1531670008814374912) | true | `/test channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1531670008785014784) | true | `/test channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main`
channel-reconciler-tests-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-ssl_eventing-kafka-broker_main/1531670008730488832) | true | `/test channel-reconciler-tests-ssl_eventing-kafka-broker_main`
channel-integration-tests-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-ssl_eventing-kafka-broker_main/1531670008600465408) | true | `/test channel-integration-tests-ssl_eventing-kafka-broker_main`

[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.
pierDipi: One more missing namespace https://github.com/knative-sandbox/eventing-kafka-broker/pull/2235/commits/11b1376acd2add4ca382da026ed0623535acde14
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-sasl-plain_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1531658719098572800) | true | `/test channel-integration-tests-sasl-plain_eventing-kafka-broker_main`
integration-tests_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/integration-tests_eventing-kafka-broker_main/1531658709875298304) | true | `/test integration-tests_eventing-kafka-broker_main`
channel-integration-tests-sasl-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-sasl-ssl_eventing-kafka-broker_main/1531663563834265600) | true | `/test channel-integration-tests-sasl-ssl_eventing-kafka-broker_main`
reconciler-tests_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/reconciler-tests_eventing-kafka-broker_main/1531670008562716672) | true | `/test reconciler-tests_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main/1531670008814374912) | true | `/test channel-reconciler-tests-sasl-plain_eventing-kafka-broker_main`
channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main/1531670008785014784) | true | `/test channel-reconciler-tests-sasl-ssl_eventing-kafka-broker_main`
channel-reconciler-tests-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-reconciler-tests-ssl_eventing-kafka-broker_main/1531670008730488832) | true | `/test channel-reconciler-tests-ssl_eventing-kafka-broker_main`
channel-integration-tests-ssl_eventing-kafka-broker_main | 5ac29718059dc498a7ff722b46ef007752a5b702 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/channel-integration-tests-ssl_eventing-kafka-broker_main/1531670008600465408) | true | `/test channel-integration-tests-ssl_eventing-kafka-broker_main`
unit-tests_eventing-kafka-broker_main | 11b1376acd2add4ca382da026ed0623535acde14 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/unit-tests_eventing-kafka-broker_main/1531675466916696064) | true | `/test unit-tests_eventing-kafka-broker_main`

[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 -->
pierDipi: /test unit-tests_eventing-kafka-broker_main 
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/2235#pullrequestreview-990796342" title="LGTM">aavarghese</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2235#pullrequestreview-990686354" title="Approved">matzew</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2235#" 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
--- | --- | --- | --- | ---
reconciler-tests_eventing-kafka-broker_main | 11b1376acd2add4ca382da026ed0623535acde14 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/reconciler-tests_eventing-kafka-broker_main/1531675467050913792) | true | `/test reconciler-tests_eventing-kafka-broker_main`

[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 reconciler-tests_eventing-kafka-broker_main

On Tue 31. May 2022 at 19:47, knative-prow[bot] ***@***.***>
wrote:

> @pierDipi <https://github.com/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
> reconciler-tests_eventing-kafka-broker_main 11b1376
> <https://github.com/knative-sandbox/eventing-kafka-broker/commit/11b1376acd2add4ca382da026ed0623535acde14>
> link
> <https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2235/reconciler-tests_eventing-kafka-broker_main/1531675467050913792>
> true /test reconciler-tests_eventing-kafka-broker_main
>
> Your PR dashboard <https://prow.knative.dev/pr>.
>
> 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>.
>
> —
> Reply to this email directly, view it on GitHub
> <https://github.com/knative-sandbox/eventing-kafka-broker/pull/2235#issuecomment-1142431443>,
> or unsubscribe
> <https://github.com/notifications/unsubscribe-auth/AABGPTUNZVNF5OA75G6P7IDVMZGCFANCNFSM5XJ766QQ>
> .
> You are receiving this because you were assigned.Message ID:
> ***@***.***>
>
-- 
Sent from Gmail Mobile


Patch:
diff --git a/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-features.yaml b/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-features.yaml
new file mode 100644
index 000000000..abec8b93c
--- /dev/null
+++ b/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-features.yaml
@@ -0,0 +1,28 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: config-kafka-features
+  namespace: knative-eventing
+  annotations:
+    knative.dev/example-checksum: "e914c9f1"
+data:
+  _example: |-
+    ################################
+    #                              #
+    #    EXAMPLE CONFIGURATION     #
+    #                              #
+    ################################
+    # This block is not actually functional configuration,
+    # but serves to illustrate the available configuration
+    # options and document them in a way that is accessible
+    # to users that `kubectl edit` this config map.
+    #
+    # These sample configuration options may be copied out of
+    # this example block and unindented to be in the data block
+    # to actually change the configuration.
+    #
+    # Controls whether the dispatcher should use the rate limiter based on the number of virtual replicas.
+    # 1. Enabled: The rate limiter is applied.
+    # 2. Disabled: The rate limiter is not applied.
+    dispatcher.rate-limiter: "disabled"
+  dispatcher.rate-limiter: "disabled"
diff --git a/control-plane/config/eventing-kafka-source/200-controller/100-config-kafka-feautures.yaml b/control-plane/config/eventing-kafka-source/200-controller/100-config-kafka-feautures.yaml
new file mode 120000
index 000000000..ebb42bad7
--- /dev/null
+++ b/control-plane/config/eventing-kafka-source/200-controller/100-config-kafka-feautures.yaml
@@ -0,0 +1,1 @@
+../../eventing-kafka-broker/200-controller/100-config-kafka-features.yaml
\ No newline at end of file
diff --git a/control-plane/pkg/apis/config/features.go b/control-plane/pkg/apis/config/features.go
new file mode 100644
index 000000000..cde9e5d3d
--- /dev/null
+++ b/control-plane/pkg/apis/config/features.go
@@ -0,0 +1,136 @@
+/*
+ * Copyright 2022 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package config
+
+import (
+	"context"
+	"strings"
+	"sync"
+
+	corev1 "k8s.io/api/core/v1"
+	"knative.dev/eventing/pkg/apis/feature"
+	"knative.dev/pkg/configmap"
+	"knative.dev/pkg/logging"
+)
+
+const (
+	FlagsConfigName = "config-kafka-features"
+)
+
+type features struct {
+	DispatcherRateLimiter feature.Flag
+}
+
+type KafkaFeatureFlags struct {
+	features features
+	m        sync.RWMutex
+}
+
+func DefaultFeaturesConfig() *KafkaFeatureFlags {
+	return &KafkaFeatureFlags{
+		features: features{
+			DispatcherRateLimiter: feature.Disabled,
+		},
+	}
+}
+
+// newFeaturesConfigFromMap creates a Features from the supplied Map
+func newFeaturesConfigFromMap(cm *corev1.ConfigMap) (*KafkaFeatureFlags, error) {
+	nc := DefaultFeaturesConfig()
+	err := configmap.Parse(cm.Data,
+		asFlag("dispatcher.rate-limiter", &nc.features.DispatcherRateLimiter),
+	)
+	return nc, err
+}
+
+func (f *KafkaFeatureFlags) Reset(ff *KafkaFeatureFlags) {
+	f.m.Lock()
+	defer f.m.Unlock()
+	f.features = ff.features
+}
+
+func (f *KafkaFeatureFlags) IsDispatcherRateLimiterEnabled() bool {
+	return f.features.DispatcherRateLimiter == feature.Enabled
+}
+
+// Store is a typed wrapper around configmap.Untyped store to handle our configmaps.
+// +k8s:deepcopy-gen=false
+type Store struct {
+	*configmap.UntypedStore
+}
+
+// NewStore creates a new store of Configs and optionally calls functions when ConfigMaps are updated.
+func NewStore(ctx context.Context, onAfterStore ...func(name string, value *KafkaFeatureFlags)) *Store {
+	store := &Store{
+		UntypedStore: configmap.NewUntypedStore(
+			"config-kafka-features",
+			logging.FromContext(ctx).Named("config-kafka-features"),
+			configmap.Constructors{
+				FlagsConfigName: newFeaturesConfigFromMap,
+			},
+			func(name string, value interface{}) {
+				for _, f := range onAfterStore {
+					f(name, value.(*KafkaFeatureFlags))
+				}
+			},
+		),
+	}
+
+	return store
+}
+
+// ToContext attaches the current Config state to the provided context.
+func (s *Store) ToContext(ctx context.Context) context.Context {
+	return ToContext(ctx, s.Load())
+}
+
+type cfgKey struct{}
+
+func ToContext(ctx context.Context, features *KafkaFeatureFlags) context.Context {
+	return context.WithValue(ctx, cfgKey{}, features)
+}
+
+func FromContext(ctx context.Context) *KafkaFeatureFlags {
+	if v := ctx.Value(cfgKey{}); v != nil {
+		return v.(*KafkaFeatureFlags)
+	}
+	return DefaultFeaturesConfig()
+}
+
+// Load creates a Config from the current config state of the Store.
+func (s *Store) Load() *KafkaFeatureFlags {
+	loaded := s.UntypedLoad(FlagsConfigName)
+	if loaded == nil {
+		return DefaultFeaturesConfig()
+	}
+	return loaded.(*KafkaFeatureFlags)
+}
+
+// asFlag parses the value at key as a Flag into the target, if it exists.
+func asFlag(key string, target *feature.Flag) configmap.ParseFunc {
+	return func(data map[string]string) error {
+		if raw, ok := data[key]; ok {
+			for _, flag := range []feature.Flag{feature.Enabled, feature.Allowed, feature.Disabled} {
+				if strings.EqualFold(raw, string(flag)) {
+					*target = flag
+					return nil
+				}
+			}
+		}
+		return nil
+	}
+}
diff --git a/control-plane/pkg/contract/contract.pb.go b/control-plane/pkg/contract/contract.pb.go
index 8c959e7cb..0924b2fdb 100644
--- a/control-plane/pkg/contract/contract.pb.go
+++ b/control-plane/pkg/contract/contract.pb.go
@@ -1024,6 +1024,8 @@ type Egress struct {
 	DialectedFilter []*DialectedFilter `protobuf:"bytes,12,rep,name=dialectedFilter,proto3" json:"dialectedFilter,omitempty"`
 	// Number of virtual replicas.
 	VReplicas int32 `protobuf:"varint,13,opt,name=vReplicas,proto3" json:"vReplicas,omitempty"`
+	// Egress feature flags.
+	FeatureFlags *EgressFeatureFlags `protobuf:"bytes,14,opt,name=featureFlags,proto3" json:"featureFlags,omitempty"`
 }
 
 func (x *Egress) Reset() {
@@ -1156,6 +1158,13 @@ func (x *Egress) GetVReplicas() int32 {
 	return 0
 }
 
+func (x *Egress) GetFeatureFlags() *EgressFeatureFlags {
+	if x != nil {
+		return x.FeatureFlags
+	}
+	return nil
+}
+
 type isEgress_ReplyStrategy interface {
 	isEgress_ReplyStrategy()
 }
@@ -1181,6 +1190,54 @@ func (*Egress_ReplyToOriginalTopic) isEgress_ReplyStrategy() {}
 
 func (*Egress_DiscardReply) isEgress_ReplyStrategy() {}
 
+type EgressFeatureFlags struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	// Enable rateLimiter
+	EnableRateLimiter bool `protobuf:"varint,1,opt,name=enableRateLimiter,proto3" json:"enableRateLimiter,omitempty"`
+}
+
+func (x *EgressFeatureFlags) Reset() {
+	*x = EgressFeatureFlags{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_contract_proto_msgTypes[12]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *EgressFeatureFlags) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*EgressFeatureFlags) ProtoMessage() {}
+
+func (x *EgressFeatureFlags) ProtoReflect() protoreflect.Message {
+	mi := &file_contract_proto_msgTypes[12]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use EgressFeatureFlags.ProtoReflect.Descriptor instead.
+func (*EgressFeatureFlags) Descriptor() ([]byte, []int) {
+	return file_contract_proto_rawDescGZIP(), []int{12}
+}
+
+func (x *EgressFeatureFlags) GetEnableRateLimiter() bool {
+	if x != nil {
+		return x.EnableRateLimiter
+	}
+	return false
+}
+
 // Ingress is the definition for HTTP ingress that is receiving the events
 // into the Knative Kafka component.
 //
@@ -1205,7 +1262,7 @@ type Ingress struct {
 func (x *Ingress) Reset() {
 	*x = Ingress{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_contract_proto_msgTypes[12]
+		mi := &file_contract_proto_msgTypes[13]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -1218,7 +1275,7 @@ func (x *Ingress) String() string {
 func (*Ingress) ProtoMessage() {}
 
 func (x *Ingress) ProtoReflect() protoreflect.Message {
-	mi := &file_contract_proto_msgTypes[12]
+	mi := &file_contract_proto_msgTypes[13]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -1231,7 +1288,7 @@ func (x *Ingress) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use Ingress.ProtoReflect.Descriptor instead.
 func (*Ingress) Descriptor() ([]byte, []int) {
-	return file_contract_proto_rawDescGZIP(), []int{12}
+	return file_contract_proto_rawDescGZIP(), []int{13}
 }
 
 func (x *Ingress) GetContentMode() ContentMode {
@@ -1274,7 +1331,7 @@ type Reference struct {
 func (x *Reference) Reset() {
 	*x = Reference{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_contract_proto_msgTypes[13]
+		mi := &file_contract_proto_msgTypes[14]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -1287,7 +1344,7 @@ func (x *Reference) String() string {
 func (*Reference) ProtoMessage() {}
 
 func (x *Reference) ProtoReflect() protoreflect.Message {
-	mi := &file_contract_proto_msgTypes[13]
+	mi := &file_contract_proto_msgTypes[14]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -1300,7 +1357,7 @@ func (x *Reference) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use Reference.ProtoReflect.Descriptor instead.
 func (*Reference) Descriptor() ([]byte, []int) {
-	return file_contract_proto_rawDescGZIP(), []int{13}
+	return file_contract_proto_rawDescGZIP(), []int{14}
 }
 
 func (x *Reference) GetUuid() string {
@@ -1345,7 +1402,7 @@ type SecretReference struct {
 func (x *SecretReference) Reset() {
 	*x = SecretReference{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_contract_proto_msgTypes[14]
+		mi := &file_contract_proto_msgTypes[15]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -1358,7 +1415,7 @@ func (x *SecretReference) String() string {
 func (*SecretReference) ProtoMessage() {}
 
 func (x *SecretReference) ProtoReflect() protoreflect.Message {
-	mi := &file_contract_proto_msgTypes[14]
+	mi := &file_contract_proto_msgTypes[15]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -1371,7 +1428,7 @@ func (x *SecretReference) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use SecretReference.ProtoReflect.Descriptor instead.
 func (*SecretReference) Descriptor() ([]byte, []int) {
-	return file_contract_proto_rawDescGZIP(), []int{14}
+	return file_contract_proto_rawDescGZIP(), []int{15}
 }
 
 func (x *SecretReference) GetReference() *Reference {
@@ -1402,7 +1459,7 @@ type KeyFieldReference struct {
 func (x *KeyFieldReference) Reset() {
 	*x = KeyFieldReference{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_contract_proto_msgTypes[15]
+		mi := &file_contract_proto_msgTypes[16]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -1415,7 +1472,7 @@ func (x *KeyFieldReference) String() string {
 func (*KeyFieldReference) ProtoMessage() {}
 
 func (x *KeyFieldReference) ProtoReflect() protoreflect.Message {
-	mi := &file_contract_proto_msgTypes[15]
+	mi := &file_contract_proto_msgTypes[16]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -1428,7 +1485,7 @@ func (x *KeyFieldReference) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use KeyFieldReference.ProtoReflect.Descriptor instead.
 func (*KeyFieldReference) Descriptor() ([]byte, []int) {
-	return file_contract_proto_rawDescGZIP(), []int{15}
+	return file_contract_proto_rawDescGZIP(), []int{16}
 }
 
 func (x *KeyFieldReference) GetSecretKey() string {
@@ -1459,7 +1516,7 @@ type MultiSecretReference struct {
 func (x *MultiSecretReference) Reset() {
 	*x = MultiSecretReference{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_contract_proto_msgTypes[16]
+		mi := &file_contract_proto_msgTypes[17]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -1472,7 +1529,7 @@ func (x *MultiSecretReference) String() string {
 func (*MultiSecretReference) ProtoMessage() {}
 
 func (x *MultiSecretReference) ProtoReflect() protoreflect.Message {
-	mi := &file_contract_proto_msgTypes[16]
+	mi := &file_contract_proto_msgTypes[17]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -1485,7 +1542,7 @@ func (x *MultiSecretReference) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use MultiSecretReference.ProtoReflect.Descriptor instead.
 func (*MultiSecretReference) Descriptor() ([]byte, []int) {
-	return file_contract_proto_rawDescGZIP(), []int{16}
+	return file_contract_proto_rawDescGZIP(), []int{17}
 }
 
 func (x *MultiSecretReference) GetProtocol() Protocol {
@@ -1514,7 +1571,7 @@ type CloudEventOverrides struct {
 func (x *CloudEventOverrides) Reset() {
 	*x = CloudEventOverrides{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_contract_proto_msgTypes[17]
+		mi := &file_contract_proto_msgTypes[18]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -1527,7 +1584,7 @@ func (x *CloudEventOverrides) String() string {
 func (*CloudEventOverrides) ProtoMessage() {}
 
 func (x *CloudEventOverrides) ProtoReflect() protoreflect.Message {
-	mi := &file_contract_proto_msgTypes[17]
+	mi := &file_contract_proto_msgTypes[18]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -1540,7 +1597,7 @@ func (x *CloudEventOverrides) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use CloudEventOverrides.ProtoReflect.Descriptor instead.
 func (*CloudEventOverrides) Descriptor() ([]byte, []int) {
-	return file_contract_proto_rawDescGZIP(), []int{17}
+	return file_contract_proto_rawDescGZIP(), []int{18}
 }
 
 func (x *CloudEventOverrides) GetExtensions() map[string]string {
@@ -1589,7 +1646,7 @@ type Resource struct {
 func (x *Resource) Reset() {
 	*x = Resource{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_contract_proto_msgTypes[18]
+		mi := &file_contract_proto_msgTypes[19]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -1602,7 +1659,7 @@ func (x *Resource) String() string {
 func (*Resource) ProtoMessage() {}
 
 func (x *Resource) ProtoReflect() protoreflect.Message {
-	mi := &file_contract_proto_msgTypes[18]
+	mi := &file_contract_proto_msgTypes[19]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -1615,7 +1672,7 @@ func (x *Resource) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use Resource.ProtoReflect.Descriptor instead.
 func (*Resource) Descriptor() ([]byte, []int) {
-	return file_contract_proto_rawDescGZIP(), []int{18}
+	return file_contract_proto_rawDescGZIP(), []int{19}
 }
 
 func (x *Resource) GetUid() string {
@@ -1772,7 +1829,7 @@ type Contract struct {
 func (x *Contract) Reset() {
 	*x = Contract{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_contract_proto_msgTypes[19]
+		mi := &file_contract_proto_msgTypes[20]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -1785,7 +1842,7 @@ func (x *Contract) String() string {
 func (*Contract) ProtoMessage() {}
 
 func (x *Contract) ProtoReflect() protoreflect.Message {
-	mi := &file_contract_proto_msgTypes[19]
+	mi := &file_contract_proto_msgTypes[20]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -1798,7 +1855,7 @@ func (x *Contract) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use Contract.ProtoReflect.Descriptor instead.
 func (*Contract) Descriptor() ([]byte, []int) {
-	return file_contract_proto_rawDescGZIP(), []int{19}
+	return file_contract_proto_rawDescGZIP(), []int{20}
 }
 
 func (x *Contract) GetGeneration() uint64 {
@@ -1891,7 +1948,7 @@ var file_contract_proto_rawDesc = []byte{
 	0x79, 0x18, 0x04, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0c, 0x62, 0x61, 0x63, 0x6b, 0x6f, 0x66, 0x66,
 	0x44, 0x65, 0x6c, 0x61, 0x79, 0x12, 0x18, 0x0a, 0x07, 0x74, 0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74,
 	0x18, 0x05, 0x20, 0x01, 0x28, 0x04, 0x52, 0x07, 0x74, 0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74, 0x22,
-	0xaf, 0x04, 0x0a, 0x06, 0x45, 0x67, 0x72, 0x65, 0x73, 0x73, 0x12, 0x24, 0x0a, 0x0d, 0x63, 0x6f,
+	0xe8, 0x04, 0x0a, 0x06, 0x45, 0x67, 0x72, 0x65, 0x73, 0x73, 0x12, 0x24, 0x0a, 0x0d, 0x63, 0x6f,
 	0x6e, 0x73, 0x75, 0x6d, 0x65, 0x72, 0x47, 0x72, 0x6f, 0x75, 0x70, 0x18, 0x01, 0x20, 0x01, 0x28,
 	0x09, 0x52, 0x0d, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6d, 0x65, 0x72, 0x47, 0x72, 0x6f, 0x75, 0x70,
 	0x12, 0x20, 0x0a, 0x0b, 0x64, 0x65, 0x73, 0x74, 0x69, 0x6e, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x18,
@@ -1925,117 +1982,124 @@ var file_contract_proto_rawDesc = []byte{
 	0x72, 0x52, 0x0f, 0x64, 0x69, 0x61, 0x6c, 0x65, 0x63, 0x74, 0x65, 0x64, 0x46, 0x69, 0x6c, 0x74,
 	0x65, 0x72, 0x12, 0x1c, 0x0a, 0x09, 0x76, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x73, 0x18,
 	0x0d, 0x20, 0x01, 0x28, 0x05, 0x52, 0x09, 0x76, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x73,
-	0x42, 0x0f, 0x0a, 0x0d, 0x72, 0x65, 0x70, 0x6c, 0x79, 0x53, 0x74, 0x72, 0x61, 0x74, 0x65, 0x67,
-	0x79, 0x22, 0x61, 0x0a, 0x07, 0x49, 0x6e, 0x67, 0x72, 0x65, 0x73, 0x73, 0x12, 0x2e, 0x0a, 0x0b,
-	0x63, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x4d, 0x6f, 0x64, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28,
-	0x0e, 0x32, 0x0c, 0x2e, 0x43, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x4d, 0x6f, 0x64, 0x65, 0x52,
-	0x0b, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x4d, 0x6f, 0x64, 0x65, 0x12, 0x12, 0x0a, 0x04,
-	0x70, 0x61, 0x74, 0x68, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x70, 0x61, 0x74, 0x68,
-	0x12, 0x12, 0x0a, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04,
-	0x68, 0x6f, 0x73, 0x74, 0x22, 0x6b, 0x0a, 0x09, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63,
-	0x65, 0x12, 0x12, 0x0a, 0x04, 0x75, 0x75, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52,
-	0x04, 0x75, 0x75, 0x69, 0x64, 0x12, 0x1c, 0x0a, 0x09, 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x70, 0x61,
-	0x63, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x70,
-	0x61, 0x63, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28,
-	0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x18, 0x0a, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69,
-	0x6f, 0x6e, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f,
-	0x6e, 0x22, 0x7f, 0x0a, 0x0f, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x52, 0x65, 0x66, 0x65, 0x72,
-	0x65, 0x6e, 0x63, 0x65, 0x12, 0x28, 0x0a, 0x09, 0x72, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63,
-	0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0a, 0x2e, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65,
-	0x6e, 0x63, 0x65, 0x52, 0x09, 0x72, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x12, 0x42,
-	0x0a, 0x12, 0x6b, 0x65, 0x79, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65,
-	0x6e, 0x63, 0x65, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x4b, 0x65, 0x79,
-	0x46, 0x69, 0x65, 0x6c, 0x64, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x52, 0x12,
-	0x6b, 0x65, 0x79, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63,
-	0x65, 0x73, 0x22, 0x55, 0x0a, 0x11, 0x4b, 0x65, 0x79, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x52, 0x65,
-	0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x12, 0x1c, 0x0a, 0x09, 0x73, 0x65, 0x63, 0x72, 0x65,
-	0x74, 0x4b, 0x65, 0x79, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x73, 0x65, 0x63, 0x72,
-	0x65, 0x74, 0x4b, 0x65, 0x79, 0x12, 0x22, 0x0a, 0x05, 0x66, 0x69, 0x65, 0x6c, 0x64, 0x18, 0x03,
-	0x20, 0x01, 0x28, 0x0e, 0x32, 0x0c, 0x2e, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x46, 0x69, 0x65,
-	0x6c, 0x64, 0x52, 0x05, 0x66, 0x69, 0x65, 0x6c, 0x64, 0x22, 0x6f, 0x0a, 0x14, 0x4d, 0x75, 0x6c,
-	0x74, 0x69, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63,
-	0x65, 0x12, 0x25, 0x0a, 0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x18, 0x01, 0x20,
-	0x01, 0x28, 0x0e, 0x32, 0x09, 0x2e, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x52, 0x08,
-	0x70, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x12, 0x30, 0x0a, 0x0a, 0x72, 0x65, 0x66, 0x65,
-	0x72, 0x65, 0x6e, 0x63, 0x65, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x10, 0x2e, 0x53,
-	0x65, 0x63, 0x72, 0x65, 0x74, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x52, 0x0a,
-	0x72, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x73, 0x22, 0x9a, 0x01, 0x0a, 0x13, 0x43,
-	0x6c, 0x6f, 0x75, 0x64, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x4f, 0x76, 0x65, 0x72, 0x72, 0x69, 0x64,
-	0x65, 0x73, 0x12, 0x44, 0x0a, 0x0a, 0x65, 0x78, 0x74, 0x65, 0x6e, 0x73, 0x69, 0x6f, 0x6e, 0x73,
-	0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x24, 0x2e, 0x43, 0x6c, 0x6f, 0x75, 0x64, 0x45, 0x76,
-	0x65, 0x6e, 0x74, 0x4f, 0x76, 0x65, 0x72, 0x72, 0x69, 0x64, 0x65, 0x73, 0x2e, 0x45, 0x78, 0x74,
-	0x65, 0x6e, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x0a, 0x65, 0x78,
-	0x74, 0x65, 0x6e, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x1a, 0x3d, 0x0a, 0x0f, 0x45, 0x78, 0x74, 0x65,
-	0x6e, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x12, 0x10, 0x0a, 0x03, 0x6b,
-	0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a,
-	0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61,
-	0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x22, 0xf1, 0x03, 0x0a, 0x08, 0x52, 0x65, 0x73, 0x6f,
-	0x75, 0x72, 0x63, 0x65, 0x12, 0x10, 0x0a, 0x03, 0x75, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28,
-	0x09, 0x52, 0x03, 0x75, 0x69, 0x64, 0x12, 0x16, 0x0a, 0x06, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x73,
-	0x18, 0x02, 0x20, 0x03, 0x28, 0x09, 0x52, 0x06, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x73, 0x12, 0x2a,
-	0x0a, 0x10, 0x62, 0x6f, 0x6f, 0x74, 0x73, 0x74, 0x72, 0x61, 0x70, 0x53, 0x65, 0x72, 0x76, 0x65,
-	0x72, 0x73, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x10, 0x62, 0x6f, 0x6f, 0x74, 0x73, 0x74,
-	0x72, 0x61, 0x70, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x12, 0x22, 0x0a, 0x07, 0x69, 0x6e,
-	0x67, 0x72, 0x65, 0x73, 0x73, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x08, 0x2e, 0x49, 0x6e,
-	0x67, 0x72, 0x65, 0x73, 0x73, 0x52, 0x07, 0x69, 0x6e, 0x67, 0x72, 0x65, 0x73, 0x73, 0x12, 0x31,
-	0x0a, 0x0c, 0x65, 0x67, 0x72, 0x65, 0x73, 0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x18, 0x05,
-	0x20, 0x01, 0x28, 0x0b, 0x32, 0x0d, 0x2e, 0x45, 0x67, 0x72, 0x65, 0x73, 0x73, 0x43, 0x6f, 0x6e,
-	0x66, 0x69, 0x67, 0x52, 0x0c, 0x65, 0x67, 0x72, 0x65, 0x73, 0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69,
-	0x67, 0x12, 0x23, 0x0a, 0x08, 0x65, 0x67, 0x72, 0x65, 0x73, 0x73, 0x65, 0x73, 0x18, 0x06, 0x20,
-	0x03, 0x28, 0x0b, 0x32, 0x07, 0x2e, 0x45, 0x67, 0x72, 0x65, 0x73, 0x73, 0x52, 0x08, 0x65, 0x67,
-	0x72, 0x65, 0x73, 0x73, 0x65, 0x73, 0x12, 0x28, 0x0a, 0x0a, 0x61, 0x62, 0x73, 0x65, 0x6e, 0x74,
-	0x41, 0x75, 0x74, 0x68, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x06, 0x2e, 0x45, 0x6d, 0x70,
-	0x74, 0x79, 0x48, 0x00, 0x52, 0x0a, 0x61, 0x62, 0x73, 0x65, 0x6e, 0x74, 0x41, 0x75, 0x74, 0x68,
-	0x12, 0x2c, 0x0a, 0x0a, 0x61, 0x75, 0x74, 0x68, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x18, 0x08,
+	0x12, 0x37, 0x0a, 0x0c, 0x66, 0x65, 0x61, 0x74, 0x75, 0x72, 0x65, 0x46, 0x6c, 0x61, 0x67, 0x73,
+	0x18, 0x0e, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x45, 0x67, 0x72, 0x65, 0x73, 0x73, 0x46,
+	0x65, 0x61, 0x74, 0x75, 0x72, 0x65, 0x46, 0x6c, 0x61, 0x67, 0x73, 0x52, 0x0c, 0x66, 0x65, 0x61,
+	0x74, 0x75, 0x72, 0x65, 0x46, 0x6c, 0x61, 0x67, 0x73, 0x42, 0x0f, 0x0a, 0x0d, 0x72, 0x65, 0x70,
+	0x6c, 0x79, 0x53, 0x74, 0x72, 0x61, 0x74, 0x65, 0x67, 0x79, 0x22, 0x42, 0x0a, 0x12, 0x45, 0x67,
+	0x72, 0x65, 0x73, 0x73, 0x46, 0x65, 0x61, 0x74, 0x75, 0x72, 0x65, 0x46, 0x6c, 0x61, 0x67, 0x73,
+	0x12, 0x2c, 0x0a, 0x11, 0x65, 0x6e, 0x61, 0x62, 0x6c, 0x65, 0x52, 0x61, 0x74, 0x65, 0x4c, 0x69,
+	0x6d, 0x69, 0x74, 0x65, 0x72, 0x18, 0x01, 0x20, 0x01, 0x28, 0x08, 0x52, 0x11, 0x65, 0x6e, 0x61,
+	0x62, 0x6c, 0x65, 0x52, 0x61, 0x74, 0x65, 0x4c, 0x69, 0x6d, 0x69, 0x74, 0x65, 0x72, 0x22, 0x61,
+	0x0a, 0x07, 0x49, 0x6e, 0x67, 0x72, 0x65, 0x73, 0x73, 0x12, 0x2e, 0x0a, 0x0b, 0x63, 0x6f, 0x6e,
+	0x74, 0x65, 0x6e, 0x74, 0x4d, 0x6f, 0x64, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x0c,
+	0x2e, 0x43, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x4d, 0x6f, 0x64, 0x65, 0x52, 0x0b, 0x63, 0x6f,
+	0x6e, 0x74, 0x65, 0x6e, 0x74, 0x4d, 0x6f, 0x64, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x70, 0x61, 0x74,
+	0x68, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x70, 0x61, 0x74, 0x68, 0x12, 0x12, 0x0a,
+	0x04, 0x68, 0x6f, 0x73, 0x74, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x68, 0x6f, 0x73,
+	0x74, 0x22, 0x6b, 0x0a, 0x09, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x12, 0x12,
+	0x0a, 0x04, 0x75, 0x75, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x75, 0x75,
+	0x69, 0x64, 0x12, 0x1c, 0x0a, 0x09, 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x70, 0x61, 0x63, 0x65, 0x18,
+	0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x70, 0x61, 0x63, 0x65,
+	0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04,
+	0x6e, 0x61, 0x6d, 0x65, 0x12, 0x18, 0x0a, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x18,
+	0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x22, 0x7f,
+	0x0a, 0x0f, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63,
+	0x65, 0x12, 0x28, 0x0a, 0x09, 0x72, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x18, 0x01,
 	0x20, 0x01, 0x28, 0x0b, 0x32, 0x0a, 0x2e, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65,
-	0x48, 0x00, 0x52, 0x0a, 0x61, 0x75, 0x74, 0x68, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x12, 0x41,
-	0x0a, 0x0f, 0x6d, 0x75, 0x6c, 0x74, 0x69, 0x41, 0x75, 0x74, 0x68, 0x53, 0x65, 0x63, 0x72, 0x65,
-	0x74, 0x18, 0x09, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x15, 0x2e, 0x4d, 0x75, 0x6c, 0x74, 0x69, 0x53,
-	0x65, 0x63, 0x72, 0x65, 0x74, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x48, 0x00,
-	0x52, 0x0f, 0x6d, 0x75, 0x6c, 0x74, 0x69, 0x41, 0x75, 0x74, 0x68, 0x53, 0x65, 0x63, 0x72, 0x65,
-	0x74, 0x12, 0x46, 0x0a, 0x13, 0x63, 0x6c, 0x6f, 0x75, 0x64, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x4f,
-	0x76, 0x65, 0x72, 0x72, 0x69, 0x64, 0x65, 0x73, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x14,
-	0x2e, 0x43, 0x6c, 0x6f, 0x75, 0x64, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x4f, 0x76, 0x65, 0x72, 0x72,
-	0x69, 0x64, 0x65, 0x73, 0x52, 0x13, 0x63, 0x6c, 0x6f, 0x75, 0x64, 0x45, 0x76, 0x65, 0x6e, 0x74,
-	0x4f, 0x76, 0x65, 0x72, 0x72, 0x69, 0x64, 0x65, 0x73, 0x12, 0x28, 0x0a, 0x09, 0x72, 0x65, 0x66,
-	0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x18, 0x0b, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0a, 0x2e, 0x52,
-	0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x52, 0x09, 0x72, 0x65, 0x66, 0x65, 0x72, 0x65,
-	0x6e, 0x63, 0x65, 0x42, 0x06, 0x0a, 0x04, 0x41, 0x75, 0x74, 0x68, 0x22, 0x53, 0x0a, 0x08, 0x43,
-	0x6f, 0x6e, 0x74, 0x72, 0x61, 0x63, 0x74, 0x12, 0x1e, 0x0a, 0x0a, 0x67, 0x65, 0x6e, 0x65, 0x72,
-	0x61, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0a, 0x67, 0x65, 0x6e,
-	0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x27, 0x0a, 0x09, 0x72, 0x65, 0x73, 0x6f, 0x75,
-	0x72, 0x63, 0x65, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x09, 0x2e, 0x52, 0x65, 0x73,
-	0x6f, 0x75, 0x72, 0x63, 0x65, 0x52, 0x09, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x73,
-	0x2a, 0x2c, 0x0a, 0x0d, 0x42, 0x61, 0x63, 0x6b, 0x6f, 0x66, 0x66, 0x50, 0x6f, 0x6c, 0x69, 0x63,
-	0x79, 0x12, 0x0f, 0x0a, 0x0b, 0x45, 0x78, 0x70, 0x6f, 0x6e, 0x65, 0x6e, 0x74, 0x69, 0x61, 0x6c,
-	0x10, 0x00, 0x12, 0x0a, 0x0a, 0x06, 0x4c, 0x69, 0x6e, 0x65, 0x61, 0x72, 0x10, 0x01, 0x2a, 0x2b,
-	0x0a, 0x0d, 0x44, 0x65, 0x6c, 0x69, 0x76, 0x65, 0x72, 0x79, 0x4f, 0x72, 0x64, 0x65, 0x72, 0x12,
-	0x0d, 0x0a, 0x09, 0x55, 0x4e, 0x4f, 0x52, 0x44, 0x45, 0x52, 0x45, 0x44, 0x10, 0x00, 0x12, 0x0b,
-	0x0a, 0x07, 0x4f, 0x52, 0x44, 0x45, 0x52, 0x45, 0x44, 0x10, 0x01, 0x2a, 0x3d, 0x0a, 0x07, 0x4b,
-	0x65, 0x79, 0x54, 0x79, 0x70, 0x65, 0x12, 0x0a, 0x0a, 0x06, 0x53, 0x74, 0x72, 0x69, 0x6e, 0x67,
-	0x10, 0x00, 0x12, 0x0b, 0x0a, 0x07, 0x49, 0x6e, 0x74, 0x65, 0x67, 0x65, 0x72, 0x10, 0x01, 0x12,
-	0x0a, 0x0a, 0x06, 0x44, 0x6f, 0x75, 0x62, 0x6c, 0x65, 0x10, 0x02, 0x12, 0x0d, 0x0a, 0x09, 0x42,
-	0x79, 0x74, 0x65, 0x41, 0x72, 0x72, 0x61, 0x79, 0x10, 0x03, 0x2a, 0x29, 0x0a, 0x0b, 0x43, 0x6f,
-	0x6e, 0x74, 0x65, 0x6e, 0x74, 0x4d, 0x6f, 0x64, 0x65, 0x12, 0x0a, 0x0a, 0x06, 0x42, 0x49, 0x4e,
-	0x41, 0x52, 0x59, 0x10, 0x00, 0x12, 0x0e, 0x0a, 0x0a, 0x53, 0x54, 0x52, 0x55, 0x43, 0x54, 0x55,
-	0x52, 0x45, 0x44, 0x10, 0x01, 0x2a, 0x61, 0x0a, 0x0b, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x46,
-	0x69, 0x65, 0x6c, 0x64, 0x12, 0x12, 0x0a, 0x0e, 0x53, 0x41, 0x53, 0x4c, 0x5f, 0x4d, 0x45, 0x43,
-	0x48, 0x41, 0x4e, 0x49, 0x53, 0x4d, 0x10, 0x00, 0x12, 0x0a, 0x0a, 0x06, 0x43, 0x41, 0x5f, 0x43,
-	0x52, 0x54, 0x10, 0x01, 0x12, 0x0c, 0x0a, 0x08, 0x55, 0x53, 0x45, 0x52, 0x5f, 0x43, 0x52, 0x54,
-	0x10, 0x02, 0x12, 0x0c, 0x0a, 0x08, 0x55, 0x53, 0x45, 0x52, 0x5f, 0x4b, 0x45, 0x59, 0x10, 0x03,
-	0x12, 0x08, 0x0a, 0x04, 0x55, 0x53, 0x45, 0x52, 0x10, 0x04, 0x12, 0x0c, 0x0a, 0x08, 0x50, 0x41,
-	0x53, 0x53, 0x57, 0x4f, 0x52, 0x44, 0x10, 0x05, 0x2a, 0x44, 0x0a, 0x08, 0x50, 0x72, 0x6f, 0x74,
-	0x6f, 0x63, 0x6f, 0x6c, 0x12, 0x0d, 0x0a, 0x09, 0x50, 0x4c, 0x41, 0x49, 0x4e, 0x54, 0x45, 0x58,
-	0x54, 0x10, 0x00, 0x12, 0x12, 0x0a, 0x0e, 0x53, 0x41, 0x53, 0x4c, 0x5f, 0x50, 0x4c, 0x41, 0x49,
-	0x4e, 0x54, 0x45, 0x58, 0x54, 0x10, 0x01, 0x12, 0x07, 0x0a, 0x03, 0x53, 0x53, 0x4c, 0x10, 0x02,
-	0x12, 0x0c, 0x0a, 0x08, 0x53, 0x41, 0x53, 0x4c, 0x5f, 0x53, 0x53, 0x4c, 0x10, 0x03, 0x42, 0x5b,
-	0x0a, 0x2a, 0x64, 0x65, 0x76, 0x2e, 0x6b, 0x6e, 0x61, 0x74, 0x69, 0x76, 0x65, 0x2e, 0x65, 0x76,
-	0x65, 0x6e, 0x74, 0x69, 0x6e, 0x67, 0x2e, 0x6b, 0x61, 0x66, 0x6b, 0x61, 0x2e, 0x62, 0x72, 0x6f,
-	0x6b, 0x65, 0x72, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x61, 0x63, 0x74, 0x42, 0x11, 0x44, 0x61,
-	0x74, 0x61, 0x50, 0x6c, 0x61, 0x6e, 0x65, 0x43, 0x6f, 0x6e, 0x74, 0x72, 0x61, 0x63, 0x74, 0x5a,
-	0x1a, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x2d, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x2f, 0x70,
-	0x6b, 0x67, 0x2f, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x61, 0x63, 0x74, 0x62, 0x06, 0x70, 0x72, 0x6f,
-	0x74, 0x6f, 0x33,
+	0x52, 0x09, 0x72, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x12, 0x42, 0x0a, 0x12, 0x6b,
+	0x65, 0x79, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65,
+	0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x4b, 0x65, 0x79, 0x46, 0x69, 0x65,
+	0x6c, 0x64, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x52, 0x12, 0x6b, 0x65, 0x79,
+	0x46, 0x69, 0x65, 0x6c, 0x64, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x73, 0x22,
+	0x55, 0x0a, 0x11, 0x4b, 0x65, 0x79, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x52, 0x65, 0x66, 0x65, 0x72,
+	0x65, 0x6e, 0x63, 0x65, 0x12, 0x1c, 0x0a, 0x09, 0x73, 0x65, 0x63, 0x72, 0x65, 0x74, 0x4b, 0x65,
+	0x79, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x73, 0x65, 0x63, 0x72, 0x65, 0x74, 0x4b,
+	0x65, 0x79, 0x12, 0x22, 0x0a, 0x05, 0x66, 0x69, 0x65, 0x6c, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28,
+	0x0e, 0x32, 0x0c, 0x2e, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x52,
+	0x05, 0x66, 0x69, 0x65, 0x6c, 0x64, 0x22, 0x6f, 0x0a, 0x14, 0x4d, 0x75, 0x6c, 0x74, 0x69, 0x53,
+	0x65, 0x63, 0x72, 0x65, 0x74, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x12, 0x25,
+	0x0a, 0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e,
+	0x32, 0x09, 0x2e, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x52, 0x08, 0x70, 0x72, 0x6f,
+	0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x12, 0x30, 0x0a, 0x0a, 0x72, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e,
+	0x63, 0x65, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x10, 0x2e, 0x53, 0x65, 0x63, 0x72,
+	0x65, 0x74, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x52, 0x0a, 0x72, 0x65, 0x66,
+	0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x73, 0x22, 0x9a, 0x01, 0x0a, 0x13, 0x43, 0x6c, 0x6f, 0x75,
+	0x64, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x4f, 0x76, 0x65, 0x72, 0x72, 0x69, 0x64, 0x65, 0x73, 0x12,
+	0x44, 0x0a, 0x0a, 0x65, 0x78, 0x74, 0x65, 0x6e, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x18, 0x01, 0x20,
+	0x03, 0x28, 0x0b, 0x32, 0x24, 0x2e, 0x43, 0x6c, 0x6f, 0x75, 0x64, 0x45, 0x76, 0x65, 0x6e, 0x74,
+	0x4f, 0x76, 0x65, 0x72, 0x72, 0x69, 0x64, 0x65, 0x73, 0x2e, 0x45, 0x78, 0x74, 0x65, 0x6e, 0x73,
+	0x69, 0x6f, 0x6e, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x0a, 0x65, 0x78, 0x74, 0x65, 0x6e,
+	0x73, 0x69, 0x6f, 0x6e, 0x73, 0x1a, 0x3d, 0x0a, 0x0f, 0x45, 0x78, 0x74, 0x65, 0x6e, 0x73, 0x69,
+	0x6f, 0x6e, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18,
+	0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61,
+	0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65,
+	0x3a, 0x02, 0x38, 0x01, 0x22, 0xf1, 0x03, 0x0a, 0x08, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63,
+	0x65, 0x12, 0x10, 0x0a, 0x03, 0x75, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03,
+	0x75, 0x69, 0x64, 0x12, 0x16, 0x0a, 0x06, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x73, 0x18, 0x02, 0x20,
+	0x03, 0x28, 0x09, 0x52, 0x06, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x73, 0x12, 0x2a, 0x0a, 0x10, 0x62,
+	0x6f, 0x6f, 0x74, 0x73, 0x74, 0x72, 0x61, 0x70, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x18,
+	0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x10, 0x62, 0x6f, 0x6f, 0x74, 0x73, 0x74, 0x72, 0x61, 0x70,
+	0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x12, 0x22, 0x0a, 0x07, 0x69, 0x6e, 0x67, 0x72, 0x65,
+	0x73, 0x73, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x08, 0x2e, 0x49, 0x6e, 0x67, 0x72, 0x65,
+	0x73, 0x73, 0x52, 0x07, 0x69, 0x6e, 0x67, 0x72, 0x65, 0x73, 0x73, 0x12, 0x31, 0x0a, 0x0c, 0x65,
+	0x67, 0x72, 0x65, 0x73, 0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x18, 0x05, 0x20, 0x01, 0x28,
+	0x0b, 0x32, 0x0d, 0x2e, 0x45, 0x67, 0x72, 0x65, 0x73, 0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67,
+	0x52, 0x0c, 0x65, 0x67, 0x72, 0x65, 0x73, 0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x23,
+	0x0a, 0x08, 0x65, 0x67, 0x72, 0x65, 0x73, 0x73, 0x65, 0x73, 0x18, 0x06, 0x20, 0x03, 0x28, 0x0b,
+	0x32, 0x07, 0x2e, 0x45, 0x67, 0x72, 0x65, 0x73, 0x73, 0x52, 0x08, 0x65, 0x67, 0x72, 0x65, 0x73,
+	0x73, 0x65, 0x73, 0x12, 0x28, 0x0a, 0x0a, 0x61, 0x62, 0x73, 0x65, 0x6e, 0x74, 0x41, 0x75, 0x74,
+	0x68, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x06, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x48,
+	0x00, 0x52, 0x0a, 0x61, 0x62, 0x73, 0x65, 0x6e, 0x74, 0x41, 0x75, 0x74, 0x68, 0x12, 0x2c, 0x0a,
+	0x0a, 0x61, 0x75, 0x74, 0x68, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x18, 0x08, 0x20, 0x01, 0x28,
+	0x0b, 0x32, 0x0a, 0x2e, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x48, 0x00, 0x52,
+	0x0a, 0x61, 0x75, 0x74, 0x68, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x12, 0x41, 0x0a, 0x0f, 0x6d,
+	0x75, 0x6c, 0x74, 0x69, 0x41, 0x75, 0x74, 0x68, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x18, 0x09,
+	0x20, 0x01, 0x28, 0x0b, 0x32, 0x15, 0x2e, 0x4d, 0x75, 0x6c, 0x74, 0x69, 0x53, 0x65, 0x63, 0x72,
+	0x65, 0x74, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x48, 0x00, 0x52, 0x0f, 0x6d,
+	0x75, 0x6c, 0x74, 0x69, 0x41, 0x75, 0x74, 0x68, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x12, 0x46,
+	0x0a, 0x13, 0x63, 0x6c, 0x6f, 0x75, 0x64, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x4f, 0x76, 0x65, 0x72,
+	0x72, 0x69, 0x64, 0x65, 0x73, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x14, 0x2e, 0x43, 0x6c,
+	0x6f, 0x75, 0x64, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x4f, 0x76, 0x65, 0x72, 0x72, 0x69, 0x64, 0x65,
+	0x73, 0x52, 0x13, 0x63, 0x6c, 0x6f, 0x75, 0x64, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x4f, 0x76, 0x65,
+	0x72, 0x72, 0x69, 0x64, 0x65, 0x73, 0x12, 0x28, 0x0a, 0x09, 0x72, 0x65, 0x66, 0x65, 0x72, 0x65,
+	0x6e, 0x63, 0x65, 0x18, 0x0b, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0a, 0x2e, 0x52, 0x65, 0x66, 0x65,
+	0x72, 0x65, 0x6e, 0x63, 0x65, 0x52, 0x09, 0x72, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65,
+	0x42, 0x06, 0x0a, 0x04, 0x41, 0x75, 0x74, 0x68, 0x22, 0x53, 0x0a, 0x08, 0x43, 0x6f, 0x6e, 0x74,
+	0x72, 0x61, 0x63, 0x74, 0x12, 0x1e, 0x0a, 0x0a, 0x67, 0x65, 0x6e, 0x65, 0x72, 0x61, 0x74, 0x69,
+	0x6f, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0a, 0x67, 0x65, 0x6e, 0x65, 0x72, 0x61,
+	0x74, 0x69, 0x6f, 0x6e, 0x12, 0x27, 0x0a, 0x09, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
+	0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x09, 0x2e, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72,
+	0x63, 0x65, 0x52, 0x09, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x73, 0x2a, 0x2c, 0x0a,
+	0x0d, 0x42, 0x61, 0x63, 0x6b, 0x6f, 0x66, 0x66, 0x50, 0x6f, 0x6c, 0x69, 0x63, 0x79, 0x12, 0x0f,
+	0x0a, 0x0b, 0x45, 0x78, 0x70, 0x6f, 0x6e, 0x65, 0x6e, 0x74, 0x69, 0x61, 0x6c, 0x10, 0x00, 0x12,
+	0x0a, 0x0a, 0x06, 0x4c, 0x69, 0x6e, 0x65, 0x61, 0x72, 0x10, 0x01, 0x2a, 0x2b, 0x0a, 0x0d, 0x44,
+	0x65, 0x6c, 0x69, 0x76, 0x65, 0x72, 0x79, 0x4f, 0x72, 0x64, 0x65, 0x72, 0x12, 0x0d, 0x0a, 0x09,
+	0x55, 0x4e, 0x4f, 0x52, 0x44, 0x45, 0x52, 0x45, 0x44, 0x10, 0x00, 0x12, 0x0b, 0x0a, 0x07, 0x4f,
+	0x52, 0x44, 0x45, 0x52, 0x45, 0x44, 0x10, 0x01, 0x2a, 0x3d, 0x0a, 0x07, 0x4b, 0x65, 0x79, 0x54,
+	0x79, 0x70, 0x65, 0x12, 0x0a, 0x0a, 0x06, 0x53, 0x74, 0x72, 0x69, 0x6e, 0x67, 0x10, 0x00, 0x12,
+	0x0b, 0x0a, 0x07, 0x49, 0x6e, 0x74, 0x65, 0x67, 0x65, 0x72, 0x10, 0x01, 0x12, 0x0a, 0x0a, 0x06,
+	0x44, 0x6f, 0x75, 0x62, 0x6c, 0x65, 0x10, 0x02, 0x12, 0x0d, 0x0a, 0x09, 0x42, 0x79, 0x74, 0x65,
+	0x41, 0x72, 0x72, 0x61, 0x79, 0x10, 0x03, 0x2a, 0x29, 0x0a, 0x0b, 0x43, 0x6f, 0x6e, 0x74, 0x65,
+	0x6e, 0x74, 0x4d, 0x6f, 0x64, 0x65, 0x12, 0x0a, 0x0a, 0x06, 0x42, 0x49, 0x4e, 0x41, 0x52, 0x59,
+	0x10, 0x00, 0x12, 0x0e, 0x0a, 0x0a, 0x53, 0x54, 0x52, 0x55, 0x43, 0x54, 0x55, 0x52, 0x45, 0x44,
+	0x10, 0x01, 0x2a, 0x61, 0x0a, 0x0b, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x46, 0x69, 0x65, 0x6c,
+	0x64, 0x12, 0x12, 0x0a, 0x0e, 0x53, 0x41, 0x53, 0x4c, 0x5f, 0x4d, 0x45, 0x43, 0x48, 0x41, 0x4e,
+	0x49, 0x53, 0x4d, 0x10, 0x00, 0x12, 0x0a, 0x0a, 0x06, 0x43, 0x41, 0x5f, 0x43, 0x52, 0x54, 0x10,
+	0x01, 0x12, 0x0c, 0x0a, 0x08, 0x55, 0x53, 0x45, 0x52, 0x5f, 0x43, 0x52, 0x54, 0x10, 0x02, 0x12,
+	0x0c, 0x0a, 0x08, 0x55, 0x53, 0x45, 0x52, 0x5f, 0x4b, 0x45, 0x59, 0x10, 0x03, 0x12, 0x08, 0x0a,
+	0x04, 0x55, 0x53, 0x45, 0x52, 0x10, 0x04, 0x12, 0x0c, 0x0a, 0x08, 0x50, 0x41, 0x53, 0x53, 0x57,
+	0x4f, 0x52, 0x44, 0x10, 0x05, 0x2a, 0x44, 0x0a, 0x08, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f,
+	0x6c, 0x12, 0x0d, 0x0a, 0x09, 0x50, 0x4c, 0x41, 0x49, 0x4e, 0x54, 0x45, 0x58, 0x54, 0x10, 0x00,
+	0x12, 0x12, 0x0a, 0x0e, 0x53, 0x41, 0x53, 0x4c, 0x5f, 0x50, 0x4c, 0x41, 0x49, 0x4e, 0x54, 0x45,
+	0x58, 0x54, 0x10, 0x01, 0x12, 0x07, 0x0a, 0x03, 0x53, 0x53, 0x4c, 0x10, 0x02, 0x12, 0x0c, 0x0a,
+	0x08, 0x53, 0x41, 0x53, 0x4c, 0x5f, 0x53, 0x53, 0x4c, 0x10, 0x03, 0x42, 0x5b, 0x0a, 0x2a, 0x64,
+	0x65, 0x76, 0x2e, 0x6b, 0x6e, 0x61, 0x74, 0x69, 0x76, 0x65, 0x2e, 0x65, 0x76, 0x65, 0x6e, 0x74,
+	0x69, 0x6e, 0x67, 0x2e, 0x6b, 0x61, 0x66, 0x6b, 0x61, 0x2e, 0x62, 0x72, 0x6f, 0x6b, 0x65, 0x72,
+	0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x61, 0x63, 0x74, 0x42, 0x11, 0x44, 0x61, 0x74, 0x61, 0x50,
+	0x6c, 0x61, 0x6e, 0x65, 0x43, 0x6f, 0x6e, 0x74, 0x72, 0x61, 0x63, 0x74, 0x5a, 0x1a, 0x63, 0x6f,
+	0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x2d, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x2f, 0x70, 0x6b, 0x67, 0x2f,
+	0x63, 0x6f, 0x6e, 0x74, 0x72, 0x61, 0x63, 0x74, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
@@ -2051,7 +2115,7 @@ func file_contract_proto_rawDescGZIP() []byte {
 }
 
 var file_contract_proto_enumTypes = make([]protoimpl.EnumInfo, 6)
-var file_contract_proto_msgTypes = make([]protoimpl.MessageInfo, 25)
+var file_contract_proto_msgTypes = make([]protoimpl.MessageInfo, 26)
 var file_contract_proto_goTypes = []interface{}{
 	(BackoffPolicy)(0),           // 0: BackoffPolicy
 	(DeliveryOrder)(0),           // 1: DeliveryOrder
@@ -2071,24 +2135,25 @@ var file_contract_proto_goTypes = []interface{}{
 	(*Filter)(nil),               // 15: Filter
 	(*EgressConfig)(nil),         // 16: EgressConfig
 	(*Egress)(nil),               // 17: Egress
-	(*Ingress)(nil),              // 18: Ingress
-	(*Reference)(nil),            // 19: Reference
-	(*SecretReference)(nil),      // 20: SecretReference
-	(*KeyFieldReference)(nil),    // 21: KeyFieldReference
-	(*MultiSecretReference)(nil), // 22: MultiSecretReference
-	(*CloudEventOverrides)(nil),  // 23: CloudEventOverrides
-	(*Resource)(nil),             // 24: Resource
-	(*Contract)(nil),             // 25: Contract
-	nil,                          // 26: Exact.AttributesEntry
-	nil,                          // 27: Prefix.AttributesEntry
-	nil,                          // 28: Suffix.AttributesEntry
-	nil,                          // 29: Filter.AttributesEntry
-	nil,                          // 30: CloudEventOverrides.ExtensionsEntry
+	(*EgressFeatureFlags)(nil),   // 18: EgressFeatureFlags
+	(*Ingress)(nil),              // 19: Ingress
+	(*Reference)(nil),            // 20: Reference
+	(*SecretReference)(nil),      // 21: SecretReference
+	(*KeyFieldReference)(nil),    // 22: KeyFieldReference
+	(*MultiSecretReference)(nil), // 23: MultiSecretReference
+	(*CloudEventOverrides)(nil),  // 24: CloudEventOverrides
+	(*Resource)(nil),             // 25: Resource
+	(*Contract)(nil),             // 26: Contract
+	nil,                          // 27: Exact.AttributesEntry
+	nil,                          // 28: Prefix.AttributesEntry
+	nil,                          // 29: Suffix.AttributesEntry
+	nil,                          // 30: Filter.AttributesEntry
+	nil,                          // 31: CloudEventOverrides.ExtensionsEntry
 }
 var file_contract_proto_depIdxs = []int32{
-	26, // 0: Exact.attributes:type_name -> Exact.AttributesEntry
-	27, // 1: Prefix.attributes:type_name -> Prefix.AttributesEntry
-	28, // 2: Suffix.attributes:type_name -> Suffix.AttributesEntry
+	27, // 0: Exact.attributes:type_name -> Exact.AttributesEntry
+	28, // 1: Prefix.attributes:type_name -> Prefix.AttributesEntry
+	29, // 2: Suffix.attributes:type_name -> Suffix.AttributesEntry
 	14, // 3: All.filters:type_name -> DialectedFilter
 	14, // 4: Any.filters:type_name -> DialectedFilter
 	14, // 5: Not.filter:type_name -> DialectedFilter
@@ -2099,7 +2164,7 @@ var file_contract_proto_depIdxs = []int32{
 	11, // 10: DialectedFilter.any:type_name -> Any
 	12, // 11: DialectedFilter.not:type_name -> Not
 	13, // 12: DialectedFilter.cesql:type_name -> CESQL
-	29, // 13: Filter.attributes:type_name -> Filter.AttributesEntry
+	30, // 13: Filter.attributes:type_name -> Filter.AttributesEntry
 	0,  // 14: EgressConfig.backoffPolicy:type_name -> BackoffPolicy
 	6,  // 15: Egress.replyToOriginalTopic:type_name -> Empty
 	6,  // 16: Egress.discardReply:type_name -> Empty
@@ -2107,29 +2172,30 @@ var file_contract_proto_depIdxs = []int32{
 	16, // 18: Egress.egressConfig:type_name -> EgressConfig
 	1,  // 19: Egress.deliveryOrder:type_name -> DeliveryOrder
 	2,  // 20: Egress.keyType:type_name -> KeyType
-	19, // 21: Egress.reference:type_name -> Reference
+	20, // 21: Egress.reference:type_name -> Reference
 	14, // 22: Egress.dialectedFilter:type_name -> DialectedFilter
-	3,  // 23: Ingress.contentMode:type_name -> ContentMode
-	19, // 24: SecretReference.reference:type_name -> Reference
-	21, // 25: SecretReference.keyFieldReferences:type_name -> KeyFieldReference
-	4,  // 26: KeyFieldReference.field:type_name -> SecretField
-	5,  // 27: MultiSecretReference.protocol:type_name -> Protocol
-	20, // 28: MultiSecretReference.references:type_name -> SecretReference
-	30, // 29: CloudEventOverrides.extensions:type_name -> CloudEventOverrides.ExtensionsEntry
-	18, // 30: Resource.ingress:type_name -> Ingress
-	16, // 31: Resource.egressConfig:type_name -> EgressConfig
-	17, // 32: Resource.egresses:type_name -> Egress
-	6,  // 33: Resource.absentAuth:type_name -> Empty
-	19, // 34: Resource.authSecret:type_name -> Reference
-	22, // 35: Resource.multiAuthSecret:type_name -> MultiSecretReference
-	23, // 36: Resource.cloudEventOverrides:type_name -> CloudEventOverrides
-	19, // 37: Resource.reference:type_name -> Reference
-	24, // 38: Contract.resources:type_name -> Resource
-	39, // [39:39] is the sub-list for method output_type
-	39, // [39:39] is the sub-list for method input_type
-	39, // [39:39] is the sub-list for extension type_name
-	39, // [39:39] is the sub-list for extension extendee
-	0,  // [0:39] is the sub-list for field type_name
+	18, // 23: Egress.featureFlags:type_name -> EgressFeatureFlags
+	3,  // 24: Ingress.contentMode:type_name -> ContentMode
+	20, // 25: SecretReference.reference:type_name -> Reference
+	22, // 26: SecretReference.keyFieldReferences:type_name -> KeyFieldReference
+	4,  // 27: KeyFieldReference.field:type_name -> SecretField
+	5,  // 28: MultiSecretReference.protocol:type_name -> Protocol
+	21, // 29: MultiSecretReference.references:type_name -> SecretReference
+	31, // 30: CloudEventOverrides.extensions:type_name -> CloudEventOverrides.ExtensionsEntry
+	19, // 31: Resource.ingress:type_name -> Ingress
+	16, // 32: Resource.egressConfig:type_name -> EgressConfig
+	17, // 33: Resource.egresses:type_name -> Egress
+	6,  // 34: Resource.absentAuth:type_name -> Empty
+	20, // 35: Resource.authSecret:type_name -> Reference
+	23, // 36: Resource.multiAuthSecret:type_name -> MultiSecretReference
+	24, // 37: Resource.cloudEventOverrides:type_name -> CloudEventOverrides
+	20, // 38: Resource.reference:type_name -> Reference
+	25, // 39: Contract.resources:type_name -> Resource
+	40, // [40:40] is the sub-list for method output_type
+	40, // [40:40] is the sub-list for method input_type
+	40, // [40:40] is the sub-list for extension type_name
+	40, // [40:40] is the sub-list for extension extendee
+	0,  // [0:40] is the sub-list for field type_name
 }
 
 func init() { file_contract_proto_init() }
@@ -2283,7 +2349,7 @@ func file_contract_proto_init() {
 			}
 		}
 		file_contract_proto_msgTypes[12].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*Ingress); i {
+			switch v := v.(*EgressFeatureFlags); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -2295,7 +2361,7 @@ func file_contract_proto_init() {
 			}
 		}
 		file_contract_proto_msgTypes[13].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*Reference); i {
+			switch v := v.(*Ingress); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -2307,7 +2373,7 @@ func file_contract_proto_init() {
 			}
 		}
 		file_contract_proto_msgTypes[14].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*SecretReference); i {
+			switch v := v.(*Reference); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -2319,7 +2385,7 @@ func file_contract_proto_init() {
 			}
 		}
 		file_contract_proto_msgTypes[15].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*KeyFieldReference); i {
+			switch v := v.(*SecretReference); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -2331,7 +2397,7 @@ func file_contract_proto_init() {
 			}
 		}
 		file_contract_proto_msgTypes[16].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*MultiSecretReference); i {
+			switch v := v.(*KeyFieldReference); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -2343,7 +2409,7 @@ func file_contract_proto_init() {
 			}
 		}
 		file_contract_proto_msgTypes[17].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*CloudEventOverrides); i {
+			switch v := v.(*MultiSecretReference); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -2355,7 +2421,7 @@ func file_contract_proto_init() {
 			}
 		}
 		file_contract_proto_msgTypes[18].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*Resource); i {
+			switch v := v.(*CloudEventOverrides); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -2367,6 +2433,18 @@ func file_contract_proto_init() {
 			}
 		}
 		file_contract_proto_msgTypes[19].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*Resource); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_contract_proto_msgTypes[20].Exporter = func(v interface{}, i int) interface{} {
 			switch v := v.(*Contract); i {
 			case 0:
 				return &v.state
@@ -2393,7 +2471,7 @@ func file_contract_proto_init() {
 		(*Egress_ReplyToOriginalTopic)(nil),
 		(*Egress_DiscardReply)(nil),
 	}
-	file_contract_proto_msgTypes[18].OneofWrappers = []interface{}{
+	file_contract_proto_msgTypes[19].OneofWrappers = []interface{}{
 		(*Resource_AbsentAuth)(nil),
 		(*Resource_AuthSecret)(nil),
 		(*Resource_MultiAuthSecret)(nil),
@@ -2404,7 +2482,7 @@ func file_contract_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: file_contract_proto_rawDesc,
 			NumEnums:      6,
-			NumMessages:   25,
+			NumMessages:   26,
 			NumExtensions: 0,
 			NumServices:   0,
 		},
diff --git a/control-plane/pkg/reconciler/consumer/consumer.go b/control-plane/pkg/reconciler/consumer/consumer.go
index d4f2669e7..5f805891e 100644
--- a/control-plane/pkg/reconciler/consumer/consumer.go
+++ b/control-plane/pkg/reconciler/consumer/consumer.go
@@ -33,6 +33,7 @@ import (
 	"knative.dev/pkg/resolver"
 	"knative.dev/pkg/tracker"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/reconciler/eventing/v1alpha1/consumer"
@@ -51,6 +52,7 @@ type Reconciler struct {
 	SecretLister        corelisters.SecretLister
 	PodLister           corelisters.PodLister
 	KubeClient          kubernetes.Interface
+	KafkaFeatureFlags   *config.KafkaFeatureFlags
 }
 
 var (
@@ -159,6 +161,10 @@ func (r Reconciler) reconcileContractEgress(ctx context.Context, c *kafkainterna
 		DeliveryOrder: reconcileDeliveryOrder(c),
 
 		KeyType: 0, // TODO handle key type
+
+		FeatureFlags: &contract.EgressFeatureFlags{
+			EnableRateLimiter: r.KafkaFeatureFlags.IsDispatcherRateLimiterEnabled(),
+		},
 	}
 
 	if err := r.reconcileReplyStrategy(ctx, c, egress); err != nil {
diff --git a/control-plane/pkg/reconciler/consumer/controller.go b/control-plane/pkg/reconciler/consumer/controller.go
index ba10fc064..97ac54f20 100644
--- a/control-plane/pkg/reconciler/consumer/controller.go
+++ b/control-plane/pkg/reconciler/consumer/controller.go
@@ -27,6 +27,7 @@ import (
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/resolver"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/config"
 	"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"
 	creconciler "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/reconciler/eventing/v1alpha1/consumer"
@@ -40,22 +41,30 @@ type ControllerConfig struct {
 
 func NewController(ctx context.Context) *controller.Impl {
 
-	config := &ControllerConfig{}
-	if err := envconfig.Process("CONSUMER", config); err != nil {
+	controllerConfig := &ControllerConfig{}
+	if err := envconfig.Process("CONSUMER", controllerConfig); err != nil {
 		panic(fmt.Errorf("failed to process env variables for consumer controller, prefix CONSUMER: %v", err))
 	}
 
 	consumerInformer := consumer.Get(ctx)
 
 	r := &Reconciler{
-		SerDe:               formatSerDeFromString(config.DataPlaneConfigFormat),
+		SerDe:               formatSerDeFromString(controllerConfig.DataPlaneConfigFormat),
 		ConsumerGroupLister: consumergroup.Get(ctx).Lister(),
 		SecretLister:        secretinformer.Get(ctx).Lister(),
 		PodLister:           podinformer.Get(ctx).Lister(),
 		KubeClient:          kubeclient.Get(ctx),
+		KafkaFeatureFlags:   config.DefaultFeaturesConfig(),
 	}
 
-	impl := creconciler.NewImpl(ctx, r)
+	impl := creconciler.NewImpl(ctx, r, func(impl *controller.Impl) controller.Options {
+		return controller.Options{
+			ConfigStore: config.NewStore(ctx, func(name string, value *config.KafkaFeatureFlags) {
+				r.KafkaFeatureFlags.Reset(value)
+				impl.GlobalResync(consumerInformer.Informer())
+			}),
+		}
+	})
 
 	r.Resolver = resolver.NewURIResolverFromTracker(ctx, impl.Tracker)
 
diff --git a/data-plane/contract/src/main/java/dev/knative/eventing/kafka/broker/contract/DataPlaneContract.java b/data-plane/contract/src/main/java/dev/knative/eventing/kafka/broker/contract/DataPlaneContract.java
index 4aca10aaf..3cd87db91 100644
--- a/data-plane/contract/src/main/java/dev/knative/eventing/kafka/broker/contract/DataPlaneContract.java
+++ b/data-plane/contract/src/main/java/dev/knative/eventing/kafka/broker/contract/DataPlaneContract.java
@@ -10508,6 +10508,33 @@ public final class DataPlaneContract {
      */
     int getVReplicas();
 
+    /**
+     * <pre>
+     * Egress feature flags.
+     * </pre>
+     *
+     * <code>.EgressFeatureFlags featureFlags = 14;</code>
+     * @return Whether the featureFlags field is set.
+     */
+    boolean hasFeatureFlags();
+    /**
+     * <pre>
+     * Egress feature flags.
+     * </pre>
+     *
+     * <code>.EgressFeatureFlags featureFlags = 14;</code>
+     * @return The featureFlags.
+     */
+    dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags getFeatureFlags();
+    /**
+     * <pre>
+     * Egress feature flags.
+     * </pre>
+     *
+     * <code>.EgressFeatureFlags featureFlags = 14;</code>
+     */
+    dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlagsOrBuilder getFeatureFlagsOrBuilder();
+
     public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Egress.ReplyStrategyCase getReplyStrategyCase();
   }
   /**
@@ -10679,6 +10706,19 @@ public final class DataPlaneContract {
               vReplicas_ = input.readInt32();
               break;
             }
+            case 114: {
+              dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags.Builder subBuilder = null;
+              if (featureFlags_ != null) {
+                subBuilder = featureFlags_.toBuilder();
+              }
+              featureFlags_ = input.readMessage(dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags.parser(), extensionRegistry);
+              if (subBuilder != null) {
+                subBuilder.mergeFrom(featureFlags_);
+                featureFlags_ = subBuilder.buildPartial();
+              }
+
+              break;
+            }
             default: {
               if (!parseUnknownField(
                   input, unknownFields, extensionRegistry, tag)) {
@@ -11304,6 +11344,44 @@ public final class DataPlaneContract {
       return vReplicas_;
     }
 
+    public static final int FEATUREFLAGS_FIELD_NUMBER = 14;
+    private dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags featureFlags_;
+    /**
+     * <pre>
+     * Egress feature flags.
+     * </pre>
+     *
+     * <code>.EgressFeatureFlags featureFlags = 14;</code>
+     * @return Whether the featureFlags field is set.
+     */
+    @java.lang.Override
+    public boolean hasFeatureFlags() {
+      return featureFlags_ != null;
+    }
+    /**
+     * <pre>
+     * Egress feature flags.
+     * </pre>
+     *
+     * <code>.EgressFeatureFlags featureFlags = 14;</code>
+     * @return The featureFlags.
+     */
+    @java.lang.Override
+    public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags getFeatureFlags() {
+      return featureFlags_ == null ? dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags.getDefaultInstance() : featureFlags_;
+    }
+    /**
+     * <pre>
+     * Egress feature flags.
+     * </pre>
+     *
+     * <code>.EgressFeatureFlags featureFlags = 14;</code>
+     */
+    @java.lang.Override
+    public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlagsOrBuilder getFeatureFlagsOrBuilder() {
+      return getFeatureFlags();
+    }
+
     private byte memoizedIsInitialized = -1;
     @java.lang.Override
     public final boolean isInitialized() {
@@ -11357,6 +11435,9 @@ public final class DataPlaneContract {
       if (vReplicas_ != 0) {
         output.writeInt32(13, vReplicas_);
       }
+      if (featureFlags_ != null) {
+        output.writeMessage(14, getFeatureFlags());
+      }
       unknownFields.writeTo(output);
     }
 
@@ -11414,6 +11495,10 @@ public final class DataPlaneContract {
         size += com.google.protobuf.CodedOutputStream
           .computeInt32Size(13, vReplicas_);
       }
+      if (featureFlags_ != null) {
+        size += com.google.protobuf.CodedOutputStream
+          .computeMessageSize(14, getFeatureFlags());
+      }
       size += unknownFields.getSerializedSize();
       memoizedSize = size;
       return size;
@@ -11456,6 +11541,11 @@ public final class DataPlaneContract {
           .equals(other.getDialectedFilterList())) return false;
       if (getVReplicas()
           != other.getVReplicas()) return false;
+      if (hasFeatureFlags() != other.hasFeatureFlags()) return false;
+      if (hasFeatureFlags()) {
+        if (!getFeatureFlags()
+            .equals(other.getFeatureFlags())) return false;
+      }
       if (!getReplyStrategyCase().equals(other.getReplyStrategyCase())) return false;
       switch (replyStrategyCase_) {
         case 3:
@@ -11512,6 +11602,10 @@ public final class DataPlaneContract {
       }
       hash = (37 * hash) + VREPLICAS_FIELD_NUMBER;
       hash = (53 * hash) + getVReplicas();
+      if (hasFeatureFlags()) {
+        hash = (37 * hash) + FEATUREFLAGS_FIELD_NUMBER;
+        hash = (53 * hash) + getFeatureFlags().hashCode();
+      }
       switch (replyStrategyCase_) {
         case 3:
           hash = (37 * hash) + REPLYURL_FIELD_NUMBER;
@@ -11698,6 +11792,12 @@ public final class DataPlaneContract {
         }
         vReplicas_ = 0;
 
+        if (featureFlagsBuilder_ == null) {
+          featureFlags_ = null;
+        } else {
+          featureFlags_ = null;
+          featureFlagsBuilder_ = null;
+        }
         replyStrategyCase_ = 0;
         replyStrategy_ = null;
         return this;
@@ -11774,6 +11874,11 @@ public final class DataPlaneContract {
           result.dialectedFilter_ = dialectedFilterBuilder_.build();
         }
         result.vReplicas_ = vReplicas_;
+        if (featureFlagsBuilder_ == null) {
+          result.featureFlags_ = featureFlags_;
+        } else {
+          result.featureFlags_ = featureFlagsBuilder_.build();
+        }
         result.replyStrategyCase_ = replyStrategyCase_;
         onBuilt();
         return result;
@@ -11879,6 +11984,9 @@ public final class DataPlaneContract {
         if (other.getVReplicas() != 0) {
           setVReplicas(other.getVReplicas());
         }
+        if (other.hasFeatureFlags()) {
+          mergeFeatureFlags(other.getFeatureFlags());
+        }
         switch (other.getReplyStrategyCase()) {
           case REPLYURL: {
             replyStrategyCase_ = 3;
@@ -13533,189 +13641,853 @@ public final class DataPlaneContract {
         } else {
           dialectedFilterBuilder_.addMessage(index, builderForValue.build());
         }
-        return this;
+        return this;
+      }
+      /**
+       * <pre>
+       * CNCF CloudEvents SubscriptionsAPI compliant filters
+       * </pre>
+       *
+       * <code>repeated .DialectedFilter dialectedFilter = 12;</code>
+       */
+      public Builder addAllDialectedFilter(
+          java.lang.Iterable<? extends dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilter> values) {
+        if (dialectedFilterBuilder_ == null) {
+          ensureDialectedFilterIsMutable();
+          com.google.protobuf.AbstractMessageLite.Builder.addAll(
+              values, dialectedFilter_);
+          onChanged();
+        } else {
+          dialectedFilterBuilder_.addAllMessages(values);
+        }
+        return this;
+      }
+      /**
+       * <pre>
+       * CNCF CloudEvents SubscriptionsAPI compliant filters
+       * </pre>
+       *
+       * <code>repeated .DialectedFilter dialectedFilter = 12;</code>
+       */
+      public Builder clearDialectedFilter() {
+        if (dialectedFilterBuilder_ == null) {
+          dialectedFilter_ = java.util.Collections.emptyList();
+          bitField0_ = (bitField0_ & ~0x00000001);
+          onChanged();
+        } else {
+          dialectedFilterBuilder_.clear();
+        }
+        return this;
+      }
+      /**
+       * <pre>
+       * CNCF CloudEvents SubscriptionsAPI compliant filters
+       * </pre>
+       *
+       * <code>repeated .DialectedFilter dialectedFilter = 12;</code>
+       */
+      public Builder removeDialectedFilter(int index) {
+        if (dialectedFilterBuilder_ == null) {
+          ensureDialectedFilterIsMutable();
+          dialectedFilter_.remove(index);
+          onChanged();
+        } else {
+          dialectedFilterBuilder_.remove(index);
+        }
+        return this;
+      }
+      /**
+       * <pre>
+       * CNCF CloudEvents SubscriptionsAPI compliant filters
+       * </pre>
+       *
+       * <code>repeated .DialectedFilter dialectedFilter = 12;</code>
+       */
+      public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilter.Builder getDialectedFilterBuilder(
+          int index) {
+        return getDialectedFilterFieldBuilder().getBuilder(index);
+      }
+      /**
+       * <pre>
+       * CNCF CloudEvents SubscriptionsAPI compliant filters
+       * </pre>
+       *
+       * <code>repeated .DialectedFilter dialectedFilter = 12;</code>
+       */
+      public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilterOrBuilder getDialectedFilterOrBuilder(
+          int index) {
+        if (dialectedFilterBuilder_ == null) {
+          return dialectedFilter_.get(index);  } else {
+          return dialectedFilterBuilder_.getMessageOrBuilder(index);
+        }
+      }
+      /**
+       * <pre>
+       * CNCF CloudEvents SubscriptionsAPI compliant filters
+       * </pre>
+       *
+       * <code>repeated .DialectedFilter dialectedFilter = 12;</code>
+       */
+      public java.util.List<? extends dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilterOrBuilder> 
+           getDialectedFilterOrBuilderList() {
+        if (dialectedFilterBuilder_ != null) {
+          return dialectedFilterBuilder_.getMessageOrBuilderList();
+        } else {
+          return java.util.Collections.unmodifiableList(dialectedFilter_);
+        }
+      }
+      /**
+       * <pre>
+       * CNCF CloudEvents SubscriptionsAPI compliant filters
+       * </pre>
+       *
+       * <code>repeated .DialectedFilter dialectedFilter = 12;</code>
+       */
+      public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilter.Builder addDialectedFilterBuilder() {
+        return getDialectedFilterFieldBuilder().addBuilder(
+            dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilter.getDefaultInstance());
+      }
+      /**
+       * <pre>
+       * CNCF CloudEvents SubscriptionsAPI compliant filters
+       * </pre>
+       *
+       * <code>repeated .DialectedFilter dialectedFilter = 12;</code>
+       */
+      public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilter.Builder addDialectedFilterBuilder(
+          int index) {
+        return getDialectedFilterFieldBuilder().addBuilder(
+            index, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilter.getDefaultInstance());
+      }
+      /**
+       * <pre>
+       * CNCF CloudEvents SubscriptionsAPI compliant filters
+       * </pre>
+       *
+       * <code>repeated .DialectedFilter dialectedFilter = 12;</code>
+       */
+      public java.util.List<dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilter.Builder> 
+           getDialectedFilterBuilderList() {
+        return getDialectedFilterFieldBuilder().getBuilderList();
+      }
+      private com.google.protobuf.RepeatedFieldBuilderV3<
+          dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilter, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilter.Builder, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilterOrBuilder> 
+          getDialectedFilterFieldBuilder() {
+        if (dialectedFilterBuilder_ == null) {
+          dialectedFilterBuilder_ = new com.google.protobuf.RepeatedFieldBuilderV3<
+              dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilter, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilter.Builder, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilterOrBuilder>(
+                  dialectedFilter_,
+                  ((bitField0_ & 0x00000001) != 0),
+                  getParentForChildren(),
+                  isClean());
+          dialectedFilter_ = null;
+        }
+        return dialectedFilterBuilder_;
+      }
+
+      private int vReplicas_ ;
+      /**
+       * <pre>
+       * Number of virtual replicas.
+       * </pre>
+       *
+       * <code>int32 vReplicas = 13;</code>
+       * @return The vReplicas.
+       */
+      @java.lang.Override
+      public int getVReplicas() {
+        return vReplicas_;
+      }
+      /**
+       * <pre>
+       * Number of virtual replicas.
+       * </pre>
+       *
+       * <code>int32 vReplicas = 13;</code>
+       * @param value The vReplicas to set.
+       * @return This builder for chaining.
+       */
+      public Builder setVReplicas(int value) {
+        
+        vReplicas_ = value;
+        onChanged();
+        return this;
+      }
+      /**
+       * <pre>
+       * Number of virtual replicas.
+       * </pre>
+       *
+       * <code>int32 vReplicas = 13;</code>
+       * @return This builder for chaining.
+       */
+      public Builder clearVReplicas() {
+        
+        vReplicas_ = 0;
+        onChanged();
+        return this;
+      }
+
+      private dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags featureFlags_;
+      private com.google.protobuf.SingleFieldBuilderV3<
+          dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags.Builder, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlagsOrBuilder> featureFlagsBuilder_;
+      /**
+       * <pre>
+       * Egress feature flags.
+       * </pre>
+       *
+       * <code>.EgressFeatureFlags featureFlags = 14;</code>
+       * @return Whether the featureFlags field is set.
+       */
+      public boolean hasFeatureFlags() {
+        return featureFlagsBuilder_ != null || featureFlags_ != null;
+      }
+      /**
+       * <pre>
+       * Egress feature flags.
+       * </pre>
+       *
+       * <code>.EgressFeatureFlags featureFlags = 14;</code>
+       * @return The featureFlags.
+       */
+      public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags getFeatureFlags() {
+        if (featureFlagsBuilder_ == null) {
+          return featureFlags_ == null ? dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags.getDefaultInstance() : featureFlags_;
+        } else {
+          return featureFlagsBuilder_.getMessage();
+        }
+      }
+      /**
+       * <pre>
+       * Egress feature flags.
+       * </pre>
+       *
+       * <code>.EgressFeatureFlags featureFlags = 14;</code>
+       */
+      public Builder setFeatureFlags(dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags value) {
+        if (featureFlagsBuilder_ == null) {
+          if (value == null) {
+            throw new NullPointerException();
+          }
+          featureFlags_ = value;
+          onChanged();
+        } else {
+          featureFlagsBuilder_.setMessage(value);
+        }
+
+        return this;
+      }
+      /**
+       * <pre>
+       * Egress feature flags.
+       * </pre>
+       *
+       * <code>.EgressFeatureFlags featureFlags = 14;</code>
+       */
+      public Builder setFeatureFlags(
+          dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags.Builder builderForValue) {
+        if (featureFlagsBuilder_ == null) {
+          featureFlags_ = builderForValue.build();
+          onChanged();
+        } else {
+          featureFlagsBuilder_.setMessage(builderForValue.build());
+        }
+
+        return this;
+      }
+      /**
+       * <pre>
+       * Egress feature flags.
+       * </pre>
+       *
+       * <code>.EgressFeatureFlags featureFlags = 14;</code>
+       */
+      public Builder mergeFeatureFlags(dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags value) {
+        if (featureFlagsBuilder_ == null) {
+          if (featureFlags_ != null) {
+            featureFlags_ =
+              dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags.newBuilder(featureFlags_).mergeFrom(value).buildPartial();
+          } else {
+            featureFlags_ = value;
+          }
+          onChanged();
+        } else {
+          featureFlagsBuilder_.mergeFrom(value);
+        }
+
+        return this;
+      }
+      /**
+       * <pre>
+       * Egress feature flags.
+       * </pre>
+       *
+       * <code>.EgressFeatureFlags featureFlags = 14;</code>
+       */
+      public Builder clearFeatureFlags() {
+        if (featureFlagsBuilder_ == null) {
+          featureFlags_ = null;
+          onChanged();
+        } else {
+          featureFlags_ = null;
+          featureFlagsBuilder_ = null;
+        }
+
+        return this;
+      }
+      /**
+       * <pre>
+       * Egress feature flags.
+       * </pre>
+       *
+       * <code>.EgressFeatureFlags featureFlags = 14;</code>
+       */
+      public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags.Builder getFeatureFlagsBuilder() {
+        
+        onChanged();
+        return getFeatureFlagsFieldBuilder().getBuilder();
+      }
+      /**
+       * <pre>
+       * Egress feature flags.
+       * </pre>
+       *
+       * <code>.EgressFeatureFlags featureFlags = 14;</code>
+       */
+      public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlagsOrBuilder getFeatureFlagsOrBuilder() {
+        if (featureFlagsBuilder_ != null) {
+          return featureFlagsBuilder_.getMessageOrBuilder();
+        } else {
+          return featureFlags_ == null ?
+              dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags.getDefaultInstance() : featureFlags_;
+        }
+      }
+      /**
+       * <pre>
+       * Egress feature flags.
+       * </pre>
+       *
+       * <code>.EgressFeatureFlags featureFlags = 14;</code>
+       */
+      private com.google.protobuf.SingleFieldBuilderV3<
+          dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags.Builder, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlagsOrBuilder> 
+          getFeatureFlagsFieldBuilder() {
+        if (featureFlagsBuilder_ == null) {
+          featureFlagsBuilder_ = new com.google.protobuf.SingleFieldBuilderV3<
+              dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags.Builder, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlagsOrBuilder>(
+                  getFeatureFlags(),
+                  getParentForChildren(),
+                  isClean());
+          featureFlags_ = null;
+        }
+        return featureFlagsBuilder_;
+      }
+      @java.lang.Override
+      public final Builder setUnknownFields(
+          final com.google.protobuf.UnknownFieldSet unknownFields) {
+        return super.setUnknownFields(unknownFields);
+      }
+
+      @java.lang.Override
+      public final Builder mergeUnknownFields(
+          final com.google.protobuf.UnknownFieldSet unknownFields) {
+        return super.mergeUnknownFields(unknownFields);
+      }
+
+
+      // @@protoc_insertion_point(builder_scope:Egress)
+    }
+
+    // @@protoc_insertion_point(class_scope:Egress)
+    private static final dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Egress DEFAULT_INSTANCE;
+    static {
+      DEFAULT_INSTANCE = new dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Egress();
+    }
+
+    public static dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Egress getDefaultInstance() {
+      return DEFAULT_INSTANCE;
+    }
+
+    private static final com.google.protobuf.Parser<Egress>
+        PARSER = new com.google.protobuf.AbstractParser<Egress>() {
+      @java.lang.Override
+      public Egress parsePartialFrom(
+          com.google.protobuf.CodedInputStream input,
+          com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+          throws com.google.protobuf.InvalidProtocolBufferException {
+        return new Egress(input, extensionRegistry);
+      }
+    };
+
+    public static com.google.protobuf.Parser<Egress> parser() {
+      return PARSER;
+    }
+
+    @java.lang.Override
+    public com.google.protobuf.Parser<Egress> getParserForType() {
+      return PARSER;
+    }
+
+    @java.lang.Override
+    public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Egress getDefaultInstanceForType() {
+      return DEFAULT_INSTANCE;
+    }
+
+  }
+
+  public interface EgressFeatureFlagsOrBuilder extends
+      // @@protoc_insertion_point(interface_extends:EgressFeatureFlags)
+      com.google.protobuf.MessageOrBuilder {
+
+    /**
+     * <pre>
+     * Enable rateLimiter
+     * </pre>
+     *
+     * <code>bool enableRateLimiter = 1;</code>
+     * @return The enableRateLimiter.
+     */
+    boolean getEnableRateLimiter();
+  }
+  /**
+   * Protobuf type {@code EgressFeatureFlags}
+   */
+  public static final class EgressFeatureFlags extends
+      com.google.protobuf.GeneratedMessageV3 implements
+      // @@protoc_insertion_point(message_implements:EgressFeatureFlags)
+      EgressFeatureFlagsOrBuilder {
+  private static final long serialVersionUID = 0L;
+    // Use EgressFeatureFlags.newBuilder() to construct.
+    private EgressFeatureFlags(com.google.protobuf.GeneratedMessageV3.Builder<?> builder) {
+      super(builder);
+    }
+    private EgressFeatureFlags() {
+    }
+
+    @java.lang.Override
+    @SuppressWarnings({"unused"})
+    protected java.lang.Object newInstance(
+        UnusedPrivateParameter unused) {
+      return new EgressFeatureFlags();
+    }
+
+    @java.lang.Override
+    public final com.google.protobuf.UnknownFieldSet
+    getUnknownFields() {
+      return this.unknownFields;
+    }
+    private EgressFeatureFlags(
+        com.google.protobuf.CodedInputStream input,
+        com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+        throws com.google.protobuf.InvalidProtocolBufferException {
+      this();
+      if (extensionRegistry == null) {
+        throw new java.lang.NullPointerException();
+      }
+      com.google.protobuf.UnknownFieldSet.Builder unknownFields =
+          com.google.protobuf.UnknownFieldSet.newBuilder();
+      try {
+        boolean done = false;
+        while (!done) {
+          int tag = input.readTag();
+          switch (tag) {
+            case 0:
+              done = true;
+              break;
+            case 8: {
+
+              enableRateLimiter_ = input.readBool();
+              break;
+            }
+            default: {
+              if (!parseUnknownField(
+                  input, unknownFields, extensionRegistry, tag)) {
+                done = true;
+              }
+              break;
+            }
+          }
+        }
+      } catch (com.google.protobuf.InvalidProtocolBufferException e) {
+        throw e.setUnfinishedMessage(this);
+      } catch (java.io.IOException e) {
+        throw new com.google.protobuf.InvalidProtocolBufferException(
+            e).setUnfinishedMessage(this);
+      } finally {
+        this.unknownFields = unknownFields.build();
+        makeExtensionsImmutable();
+      }
+    }
+    public static final com.google.protobuf.Descriptors.Descriptor
+        getDescriptor() {
+      return dev.knative.eventing.kafka.broker.contract.DataPlaneContract.internal_static_EgressFeatureFlags_descriptor;
+    }
+
+    @java.lang.Override
+    protected com.google.protobuf.GeneratedMessageV3.FieldAccessorTable
+        internalGetFieldAccessorTable() {
+      return dev.knative.eventing.kafka.broker.contract.DataPlaneContract.internal_static_EgressFeatureFlags_fieldAccessorTable
+          .ensureFieldAccessorsInitialized(
+              dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags.class, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags.Builder.class);
+    }
+
+    public static final int ENABLERATELIMITER_FIELD_NUMBER = 1;
+    private boolean enableRateLimiter_;
+    /**
+     * <pre>
+     * Enable rateLimiter
+     * </pre>
+     *
+     * <code>bool enableRateLimiter = 1;</code>
+     * @return The enableRateLimiter.
+     */
+    @java.lang.Override
+    public boolean getEnableRateLimiter() {
+      return enableRateLimiter_;
+    }
+
+    private byte memoizedIsInitialized = -1;
+    @java.lang.Override
+    public final boolean isInitialized() {
+      byte isInitialized = memoizedIsInitialized;
+      if (isInitialized == 1) return true;
+      if (isInitialized == 0) return false;
+
+      memoizedIsInitialized = 1;
+      return true;
+    }
+
+    @java.lang.Override
+    public void writeTo(com.google.protobuf.CodedOutputStream output)
+                        throws java.io.IOException {
+      if (enableRateLimiter_ != false) {
+        output.writeBool(1, enableRateLimiter_);
+      }
+      unknownFields.writeTo(output);
+    }
+
+    @java.lang.Override
+    public int getSerializedSize() {
+      int size = memoizedSize;
+      if (size != -1) return size;
+
+      size = 0;
+      if (enableRateLimiter_ != false) {
+        size += com.google.protobuf.CodedOutputStream
+          .computeBoolSize(1, enableRateLimiter_);
+      }
+      size += unknownFields.getSerializedSize();
+      memoizedSize = size;
+      return size;
+    }
+
+    @java.lang.Override
+    public boolean equals(final java.lang.Object obj) {
+      if (obj == this) {
+       return true;
+      }
+      if (!(obj instanceof dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags)) {
+        return super.equals(obj);
+      }
+      dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags other = (dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags) obj;
+
+      if (getEnableRateLimiter()
+          != other.getEnableRateLimiter()) return false;
+      if (!unknownFields.equals(other.unknownFields)) return false;
+      return true;
+    }
+
+    @java.lang.Override
+    public int hashCode() {
+      if (memoizedHashCode != 0) {
+        return memoizedHashCode;
+      }
+      int hash = 41;
+      hash = (19 * hash) + getDescriptor().hashCode();
+      hash = (37 * hash) + ENABLERATELIMITER_FIELD_NUMBER;
+      hash = (53 * hash) + com.google.protobuf.Internal.hashBoolean(
+          getEnableRateLimiter());
+      hash = (29 * hash) + unknownFields.hashCode();
+      memoizedHashCode = hash;
+      return hash;
+    }
+
+    public static dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags parseFrom(
+        java.nio.ByteBuffer data)
+        throws com.google.protobuf.InvalidProtocolBufferException {
+      return PARSER.parseFrom(data);
+    }
+    public static dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags parseFrom(
+        java.nio.ByteBuffer data,
+        com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+        throws com.google.protobuf.InvalidProtocolBufferException {
+      return PARSER.parseFrom(data, extensionRegistry);
+    }
+    public static dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags parseFrom(
+        com.google.protobuf.ByteString data)
+        throws com.google.protobuf.InvalidProtocolBufferException {
+      return PARSER.parseFrom(data);
+    }
+    public static dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags parseFrom(
+        com.google.protobuf.ByteString data,
+        com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+        throws com.google.protobuf.InvalidProtocolBufferException {
+      return PARSER.parseFrom(data, extensionRegistry);
+    }
+    public static dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags parseFrom(byte[] data)
+        throws com.google.protobuf.InvalidProtocolBufferException {
+      return PARSER.parseFrom(data);
+    }
+    public static dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags parseFrom(
+        byte[] data,
+        com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+        throws com.google.protobuf.InvalidProtocolBufferException {
+      return PARSER.parseFrom(data, extensionRegistry);
+    }
+    public static dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags parseFrom(java.io.InputStream input)
+        throws java.io.IOException {
+      return com.google.protobuf.GeneratedMessageV3
+          .parseWithIOException(PARSER, input);
+    }
+    public static dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags parseFrom(
+        java.io.InputStream input,
+        com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+        throws java.io.IOException {
+      return com.google.protobuf.GeneratedMessageV3
+          .parseWithIOException(PARSER, input, extensionRegistry);
+    }
+    public static dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags parseDelimitedFrom(java.io.InputStream input)
+        throws java.io.IOException {
+      return com.google.protobuf.GeneratedMessageV3
+          .parseDelimitedWithIOException(PARSER, input);
+    }
+    public static dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags parseDelimitedFrom(
+        java.io.InputStream input,
+        com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+        throws java.io.IOException {
+      return com.google.protobuf.GeneratedMessageV3
+          .parseDelimitedWithIOException(PARSER, input, extensionRegistry);
+    }
+    public static dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags parseFrom(
+        com.google.protobuf.CodedInputStream input)
+        throws java.io.IOException {
+      return com.google.protobuf.GeneratedMessageV3
+          .parseWithIOException(PARSER, input);
+    }
+    public static dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags parseFrom(
+        com.google.protobuf.CodedInputStream input,
+        com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+        throws java.io.IOException {
+      return com.google.protobuf.GeneratedMessageV3
+          .parseWithIOException(PARSER, input, extensionRegistry);
+    }
+
+    @java.lang.Override
+    public Builder newBuilderForType() { return newBuilder(); }
+    public static Builder newBuilder() {
+      return DEFAULT_INSTANCE.toBuilder();
+    }
+    public static Builder newBuilder(dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags prototype) {
+      return DEFAULT_INSTANCE.toBuilder().mergeFrom(prototype);
+    }
+    @java.lang.Override
+    public Builder toBuilder() {
+      return this == DEFAULT_INSTANCE
+          ? new Builder() : new Builder().mergeFrom(this);
+    }
+
+    @java.lang.Override
+    protected Builder newBuilderForType(
+        com.google.protobuf.GeneratedMessageV3.BuilderParent parent) {
+      Builder builder = new Builder(parent);
+      return builder;
+    }
+    /**
+     * Protobuf type {@code EgressFeatureFlags}
+     */
+    public static final class Builder extends
+        com.google.protobuf.GeneratedMessageV3.Builder<Builder> implements
+        // @@protoc_insertion_point(builder_implements:EgressFeatureFlags)
+        dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlagsOrBuilder {
+      public static final com.google.protobuf.Descriptors.Descriptor
+          getDescriptor() {
+        return dev.knative.eventing.kafka.broker.contract.DataPlaneContract.internal_static_EgressFeatureFlags_descriptor;
+      }
+
+      @java.lang.Override
+      protected com.google.protobuf.GeneratedMessageV3.FieldAccessorTable
+          internalGetFieldAccessorTable() {
+        return dev.knative.eventing.kafka.broker.contract.DataPlaneContract.internal_static_EgressFeatureFlags_fieldAccessorTable
+            .ensureFieldAccessorsInitialized(
+                dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags.class, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags.Builder.class);
+      }
+
+      // Construct using dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags.newBuilder()
+      private Builder() {
+        maybeForceBuilderInitialization();
+      }
+
+      private Builder(
+          com.google.protobuf.GeneratedMessageV3.BuilderParent parent) {
+        super(parent);
+        maybeForceBuilderInitialization();
+      }
+      private void maybeForceBuilderInitialization() {
+        if (com.google.protobuf.GeneratedMessageV3
+                .alwaysUseFieldBuilders) {
+        }
       }
-      /**
-       * <pre>
-       * CNCF CloudEvents SubscriptionsAPI compliant filters
-       * </pre>
-       *
-       * <code>repeated .DialectedFilter dialectedFilter = 12;</code>
-       */
-      public Builder addAllDialectedFilter(
-          java.lang.Iterable<? extends dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilter> values) {
-        if (dialectedFilterBuilder_ == null) {
-          ensureDialectedFilterIsMutable();
-          com.google.protobuf.AbstractMessageLite.Builder.addAll(
-              values, dialectedFilter_);
-          onChanged();
-        } else {
-          dialectedFilterBuilder_.addAllMessages(values);
-        }
+      @java.lang.Override
+      public Builder clear() {
+        super.clear();
+        enableRateLimiter_ = false;
+
         return this;
       }
-      /**
-       * <pre>
-       * CNCF CloudEvents SubscriptionsAPI compliant filters
-       * </pre>
-       *
-       * <code>repeated .DialectedFilter dialectedFilter = 12;</code>
-       */
-      public Builder clearDialectedFilter() {
-        if (dialectedFilterBuilder_ == null) {
-          dialectedFilter_ = java.util.Collections.emptyList();
-          bitField0_ = (bitField0_ & ~0x00000001);
-          onChanged();
-        } else {
-          dialectedFilterBuilder_.clear();
-        }
-        return this;
+
+      @java.lang.Override
+      public com.google.protobuf.Descriptors.Descriptor
+          getDescriptorForType() {
+        return dev.knative.eventing.kafka.broker.contract.DataPlaneContract.internal_static_EgressFeatureFlags_descriptor;
       }
-      /**
-       * <pre>
-       * CNCF CloudEvents SubscriptionsAPI compliant filters
-       * </pre>
-       *
-       * <code>repeated .DialectedFilter dialectedFilter = 12;</code>
-       */
-      public Builder removeDialectedFilter(int index) {
-        if (dialectedFilterBuilder_ == null) {
-          ensureDialectedFilterIsMutable();
-          dialectedFilter_.remove(index);
-          onChanged();
-        } else {
-          dialectedFilterBuilder_.remove(index);
+
+      @java.lang.Override
+      public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags getDefaultInstanceForType() {
+        return dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags.getDefaultInstance();
+      }
+
+      @java.lang.Override
+      public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags build() {
+        dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags result = buildPartial();
+        if (!result.isInitialized()) {
+          throw newUninitializedMessageException(result);
         }
-        return this;
+        return result;
       }
-      /**
-       * <pre>
-       * CNCF CloudEvents SubscriptionsAPI compliant filters
-       * </pre>
-       *
-       * <code>repeated .DialectedFilter dialectedFilter = 12;</code>
-       */
-      public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilter.Builder getDialectedFilterBuilder(
-          int index) {
-        return getDialectedFilterFieldBuilder().getBuilder(index);
+
+      @java.lang.Override
+      public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags buildPartial() {
+        dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags result = new dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags(this);
+        result.enableRateLimiter_ = enableRateLimiter_;
+        onBuilt();
+        return result;
       }
-      /**
-       * <pre>
-       * CNCF CloudEvents SubscriptionsAPI compliant filters
-       * </pre>
-       *
-       * <code>repeated .DialectedFilter dialectedFilter = 12;</code>
-       */
-      public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilterOrBuilder getDialectedFilterOrBuilder(
-          int index) {
-        if (dialectedFilterBuilder_ == null) {
-          return dialectedFilter_.get(index);  } else {
-          return dialectedFilterBuilder_.getMessageOrBuilder(index);
-        }
+
+      @java.lang.Override
+      public Builder clone() {
+        return super.clone();
       }
-      /**
-       * <pre>
-       * CNCF CloudEvents SubscriptionsAPI compliant filters
-       * </pre>
-       *
-       * <code>repeated .DialectedFilter dialectedFilter = 12;</code>
-       */
-      public java.util.List<? extends dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilterOrBuilder> 
-           getDialectedFilterOrBuilderList() {
-        if (dialectedFilterBuilder_ != null) {
-          return dialectedFilterBuilder_.getMessageOrBuilderList();
+      @java.lang.Override
+      public Builder setField(
+          com.google.protobuf.Descriptors.FieldDescriptor field,
+          java.lang.Object value) {
+        return super.setField(field, value);
+      }
+      @java.lang.Override
+      public Builder clearField(
+          com.google.protobuf.Descriptors.FieldDescriptor field) {
+        return super.clearField(field);
+      }
+      @java.lang.Override
+      public Builder clearOneof(
+          com.google.protobuf.Descriptors.OneofDescriptor oneof) {
+        return super.clearOneof(oneof);
+      }
+      @java.lang.Override
+      public Builder setRepeatedField(
+          com.google.protobuf.Descriptors.FieldDescriptor field,
+          int index, java.lang.Object value) {
+        return super.setRepeatedField(field, index, value);
+      }
+      @java.lang.Override
+      public Builder addRepeatedField(
+          com.google.protobuf.Descriptors.FieldDescriptor field,
+          java.lang.Object value) {
+        return super.addRepeatedField(field, value);
+      }
+      @java.lang.Override
+      public Builder mergeFrom(com.google.protobuf.Message other) {
+        if (other instanceof dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags) {
+          return mergeFrom((dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags)other);
         } else {
-          return java.util.Collections.unmodifiableList(dialectedFilter_);
+          super.mergeFrom(other);
+          return this;
         }
       }
-      /**
-       * <pre>
-       * CNCF CloudEvents SubscriptionsAPI compliant filters
-       * </pre>
-       *
-       * <code>repeated .DialectedFilter dialectedFilter = 12;</code>
-       */
-      public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilter.Builder addDialectedFilterBuilder() {
-        return getDialectedFilterFieldBuilder().addBuilder(
-            dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilter.getDefaultInstance());
-      }
-      /**
-       * <pre>
-       * CNCF CloudEvents SubscriptionsAPI compliant filters
-       * </pre>
-       *
-       * <code>repeated .DialectedFilter dialectedFilter = 12;</code>
-       */
-      public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilter.Builder addDialectedFilterBuilder(
-          int index) {
-        return getDialectedFilterFieldBuilder().addBuilder(
-            index, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilter.getDefaultInstance());
+
+      public Builder mergeFrom(dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags other) {
+        if (other == dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags.getDefaultInstance()) return this;
+        if (other.getEnableRateLimiter() != false) {
+          setEnableRateLimiter(other.getEnableRateLimiter());
+        }
+        this.mergeUnknownFields(other.unknownFields);
+        onChanged();
+        return this;
       }
-      /**
-       * <pre>
-       * CNCF CloudEvents SubscriptionsAPI compliant filters
-       * </pre>
-       *
-       * <code>repeated .DialectedFilter dialectedFilter = 12;</code>
-       */
-      public java.util.List<dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilter.Builder> 
-           getDialectedFilterBuilderList() {
-        return getDialectedFilterFieldBuilder().getBuilderList();
+
+      @java.lang.Override
+      public final boolean isInitialized() {
+        return true;
       }
-      private com.google.protobuf.RepeatedFieldBuilderV3<
-          dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilter, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilter.Builder, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilterOrBuilder> 
-          getDialectedFilterFieldBuilder() {
-        if (dialectedFilterBuilder_ == null) {
-          dialectedFilterBuilder_ = new com.google.protobuf.RepeatedFieldBuilderV3<
-              dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilter, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilter.Builder, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilterOrBuilder>(
-                  dialectedFilter_,
-                  ((bitField0_ & 0x00000001) != 0),
-                  getParentForChildren(),
-                  isClean());
-          dialectedFilter_ = null;
+
+      @java.lang.Override
+      public Builder mergeFrom(
+          com.google.protobuf.CodedInputStream input,
+          com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+          throws java.io.IOException {
+        dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags parsedMessage = null;
+        try {
+          parsedMessage = PARSER.parsePartialFrom(input, extensionRegistry);
+        } catch (com.google.protobuf.InvalidProtocolBufferException e) {
+          parsedMessage = (dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags) e.getUnfinishedMessage();
+          throw e.unwrapIOException();
+        } finally {
+          if (parsedMessage != null) {
+            mergeFrom(parsedMessage);
+          }
         }
-        return dialectedFilterBuilder_;
+        return this;
       }
 
-      private int vReplicas_ ;
+      private boolean enableRateLimiter_ ;
       /**
        * <pre>
-       * Number of virtual replicas.
+       * Enable rateLimiter
        * </pre>
        *
-       * <code>int32 vReplicas = 13;</code>
-       * @return The vReplicas.
+       * <code>bool enableRateLimiter = 1;</code>
+       * @return The enableRateLimiter.
        */
       @java.lang.Override
-      public int getVReplicas() {
-        return vReplicas_;
+      public boolean getEnableRateLimiter() {
+        return enableRateLimiter_;
       }
       /**
        * <pre>
-       * Number of virtual replicas.
+       * Enable rateLimiter
        * </pre>
        *
-       * <code>int32 vReplicas = 13;</code>
-       * @param value The vReplicas to set.
+       * <code>bool enableRateLimiter = 1;</code>
+       * @param value The enableRateLimiter to set.
        * @return This builder for chaining.
        */
-      public Builder setVReplicas(int value) {
+      public Builder setEnableRateLimiter(boolean value) {
         
-        vReplicas_ = value;
+        enableRateLimiter_ = value;
         onChanged();
         return this;
       }
       /**
        * <pre>
-       * Number of virtual replicas.
+       * Enable rateLimiter
        * </pre>
        *
-       * <code>int32 vReplicas = 13;</code>
+       * <code>bool enableRateLimiter = 1;</code>
        * @return This builder for chaining.
        */
-      public Builder clearVReplicas() {
+      public Builder clearEnableRateLimiter() {
         
-        vReplicas_ = 0;
+        enableRateLimiter_ = false;
         onChanged();
         return this;
       }
@@ -13732,41 +14504,41 @@ public final class DataPlaneContract {
       }
 
 
-      // @@protoc_insertion_point(builder_scope:Egress)
+      // @@protoc_insertion_point(builder_scope:EgressFeatureFlags)
     }
 
-    // @@protoc_insertion_point(class_scope:Egress)
-    private static final dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Egress DEFAULT_INSTANCE;
+    // @@protoc_insertion_point(class_scope:EgressFeatureFlags)
+    private static final dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags DEFAULT_INSTANCE;
     static {
-      DEFAULT_INSTANCE = new dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Egress();
+      DEFAULT_INSTANCE = new dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags();
     }
 
-    public static dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Egress getDefaultInstance() {
+    public static dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags getDefaultInstance() {
       return DEFAULT_INSTANCE;
     }
 
-    private static final com.google.protobuf.Parser<Egress>
-        PARSER = new com.google.protobuf.AbstractParser<Egress>() {
+    private static final com.google.protobuf.Parser<EgressFeatureFlags>
+        PARSER = new com.google.protobuf.AbstractParser<EgressFeatureFlags>() {
       @java.lang.Override
-      public Egress parsePartialFrom(
+      public EgressFeatureFlags parsePartialFrom(
           com.google.protobuf.CodedInputStream input,
           com.google.protobuf.ExtensionRegistryLite extensionRegistry)
           throws com.google.protobuf.InvalidProtocolBufferException {
-        return new Egress(input, extensionRegistry);
+        return new EgressFeatureFlags(input, extensionRegistry);
       }
     };
 
-    public static com.google.protobuf.Parser<Egress> parser() {
+    public static com.google.protobuf.Parser<EgressFeatureFlags> parser() {
       return PARSER;
     }
 
     @java.lang.Override
-    public com.google.protobuf.Parser<Egress> getParserForType() {
+    public com.google.protobuf.Parser<EgressFeatureFlags> getParserForType() {
       return PARSER;
     }
 
     @java.lang.Override
-    public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Egress getDefaultInstanceForType() {
+    public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressFeatureFlags getDefaultInstanceForType() {
       return DEFAULT_INSTANCE;
     }
 
@@ -24597,6 +25369,11 @@ public final class DataPlaneContract {
   private static final 
     com.google.protobuf.GeneratedMessageV3.FieldAccessorTable
       internal_static_Egress_fieldAccessorTable;
+  private static final com.google.protobuf.Descriptors.Descriptor
+    internal_static_EgressFeatureFlags_descriptor;
+  private static final 
+    com.google.protobuf.GeneratedMessageV3.FieldAccessorTable
+      internal_static_EgressFeatureFlags_fieldAccessorTable;
   private static final com.google.protobuf.Descriptors.Descriptor
     internal_static_Ingress_descriptor;
   private static final 
@@ -24674,7 +25451,7 @@ public final class DataPlaneContract {
       "\n\014EgressConfig\022\022\n\ndeadLetter\030\001 \001(\t\022\r\n\005re" +
       "try\030\002 \001(\r\022%\n\rbackoffPolicy\030\003 \001(\0162\016.Backo" +
       "ffPolicy\022\024\n\014backoffDelay\030\004 \001(\004\022\017\n\007timeou" +
-      "t\030\005 \001(\004\"\213\003\n\006Egress\022\025\n\rconsumerGroup\030\001 \001(" +
+      "t\030\005 \001(\004\"\266\003\n\006Egress\022\025\n\rconsumerGroup\030\001 \001(" +
       "\t\022\023\n\013destination\030\002 \001(\t\022\022\n\010replyUrl\030\003 \001(\t" +
       "H\000\022&\n\024replyToOriginalTopic\030\004 \001(\0132\006.Empty" +
       "H\000\022\036\n\014discardReply\030\t \001(\0132\006.EmptyH\000\022\027\n\006fi" +
@@ -24683,43 +25460,46 @@ public final class DataPlaneContract {
       "yOrder\030\010 \001(\0162\016.DeliveryOrder\022\031\n\007keyType\030" +
       "\n \001(\0162\010.KeyType\022\035\n\treference\030\013 \001(\0132\n.Ref" +
       "erence\022)\n\017dialectedFilter\030\014 \003(\0132\020.Dialec" +
-      "tedFilter\022\021\n\tvReplicas\030\r \001(\005B\017\n\rreplyStr" +
-      "ategy\"H\n\007Ingress\022!\n\013contentMode\030\001 \001(\0162\014." +
-      "ContentMode\022\014\n\004path\030\002 \001(\t\022\014\n\004host\030\003 \001(\t\"" +
-      "K\n\tReference\022\014\n\004uuid\030\001 \001(\t\022\021\n\tnamespace\030" +
-      "\002 \001(\t\022\014\n\004name\030\003 \001(\t\022\017\n\007version\030\004 \001(\t\"`\n\017" +
-      "SecretReference\022\035\n\treference\030\001 \001(\0132\n.Ref" +
-      "erence\022.\n\022keyFieldReferences\030\002 \003(\0132\022.Key" +
-      "FieldReference\"C\n\021KeyFieldReference\022\021\n\ts" +
-      "ecretKey\030\002 \001(\t\022\033\n\005field\030\003 \001(\0162\014.SecretFi" +
-      "eld\"Y\n\024MultiSecretReference\022\033\n\010protocol\030" +
-      "\001 \001(\0162\t.Protocol\022$\n\nreferences\030\002 \003(\0132\020.S" +
-      "ecretReference\"\202\001\n\023CloudEventOverrides\0228" +
-      "\n\nextensions\030\001 \003(\0132$.CloudEventOverrides" +
-      ".ExtensionsEntry\0321\n\017ExtensionsEntry\022\013\n\003k" +
-      "ey\030\001 \001(\t\022\r\n\005value\030\002 \001(\t:\0028\001\"\350\002\n\010Resource" +
-      "\022\013\n\003uid\030\001 \001(\t\022\016\n\006topics\030\002 \003(\t\022\030\n\020bootstr" +
-      "apServers\030\003 \001(\t\022\031\n\007ingress\030\004 \001(\0132\010.Ingre" +
-      "ss\022#\n\014egressConfig\030\005 \001(\0132\r.EgressConfig\022" +
-      "\031\n\010egresses\030\006 \003(\0132\007.Egress\022\034\n\nabsentAuth" +
-      "\030\007 \001(\0132\006.EmptyH\000\022 \n\nauthSecret\030\010 \001(\0132\n.R" +
-      "eferenceH\000\0220\n\017multiAuthSecret\030\t \001(\0132\025.Mu" +
-      "ltiSecretReferenceH\000\0221\n\023cloudEventOverri" +
-      "des\030\n \001(\0132\024.CloudEventOverrides\022\035\n\trefer" +
-      "ence\030\013 \001(\0132\n.ReferenceB\006\n\004Auth\"<\n\010Contra" +
-      "ct\022\022\n\ngeneration\030\001 \001(\004\022\034\n\tresources\030\002 \003(" +
-      "\0132\t.Resource*,\n\rBackoffPolicy\022\017\n\013Exponen" +
-      "tial\020\000\022\n\n\006Linear\020\001*+\n\rDeliveryOrder\022\r\n\tU" +
-      "NORDERED\020\000\022\013\n\007ORDERED\020\001*=\n\007KeyType\022\n\n\006St" +
-      "ring\020\000\022\013\n\007Integer\020\001\022\n\n\006Double\020\002\022\r\n\tByteA" +
-      "rray\020\003*)\n\013ContentMode\022\n\n\006BINARY\020\000\022\016\n\nSTR" +
-      "UCTURED\020\001*a\n\013SecretField\022\022\n\016SASL_MECHANI" +
-      "SM\020\000\022\n\n\006CA_CRT\020\001\022\014\n\010USER_CRT\020\002\022\014\n\010USER_K" +
-      "EY\020\003\022\010\n\004USER\020\004\022\014\n\010PASSWORD\020\005*D\n\010Protocol" +
-      "\022\r\n\tPLAINTEXT\020\000\022\022\n\016SASL_PLAINTEXT\020\001\022\007\n\003S" +
-      "SL\020\002\022\014\n\010SASL_SSL\020\003B[\n*dev.knative.eventi" +
-      "ng.kafka.broker.contractB\021DataPlaneContr" +
-      "actZ\032control-plane/pkg/contractb\006proto3"
+      "tedFilter\022\021\n\tvReplicas\030\r \001(\005\022)\n\014featureF" +
+      "lags\030\016 \001(\0132\023.EgressFeatureFlagsB\017\n\rreply" +
+      "Strategy\"/\n\022EgressFeatureFlags\022\031\n\021enable" +
+      "RateLimiter\030\001 \001(\010\"H\n\007Ingress\022!\n\013contentM" +
+      "ode\030\001 \001(\0162\014.ContentMode\022\014\n\004path\030\002 \001(\t\022\014\n" +
+      "\004host\030\003 \001(\t\"K\n\tReference\022\014\n\004uuid\030\001 \001(\t\022\021" +
+      "\n\tnamespace\030\002 \001(\t\022\014\n\004name\030\003 \001(\t\022\017\n\007versi" +
+      "on\030\004 \001(\t\"`\n\017SecretReference\022\035\n\treference" +
+      "\030\001 \001(\0132\n.Reference\022.\n\022keyFieldReferences" +
+      "\030\002 \003(\0132\022.KeyFieldReference\"C\n\021KeyFieldRe" +
+      "ference\022\021\n\tsecretKey\030\002 \001(\t\022\033\n\005field\030\003 \001(" +
+      "\0162\014.SecretField\"Y\n\024MultiSecretReference\022" +
+      "\033\n\010protocol\030\001 \001(\0162\t.Protocol\022$\n\nreferenc" +
+      "es\030\002 \003(\0132\020.SecretReference\"\202\001\n\023CloudEven" +
+      "tOverrides\0228\n\nextensions\030\001 \003(\0132$.CloudEv" +
+      "entOverrides.ExtensionsEntry\0321\n\017Extensio" +
+      "nsEntry\022\013\n\003key\030\001 \001(\t\022\r\n\005value\030\002 \001(\t:\0028\001\"" +
+      "\350\002\n\010Resource\022\013\n\003uid\030\001 \001(\t\022\016\n\006topics\030\002 \003(" +
+      "\t\022\030\n\020bootstrapServers\030\003 \001(\t\022\031\n\007ingress\030\004" +
+      " \001(\0132\010.Ingress\022#\n\014egressConfig\030\005 \001(\0132\r.E" +
+      "gressConfig\022\031\n\010egresses\030\006 \003(\0132\007.Egress\022\034" +
+      "\n\nabsentAuth\030\007 \001(\0132\006.EmptyH\000\022 \n\nauthSecr" +
+      "et\030\010 \001(\0132\n.ReferenceH\000\0220\n\017multiAuthSecre" +
+      "t\030\t \001(\0132\025.MultiSecretReferenceH\000\0221\n\023clou" +
+      "dEventOverrides\030\n \001(\0132\024.CloudEventOverri" +
+      "des\022\035\n\treference\030\013 \001(\0132\n.ReferenceB\006\n\004Au" +
+      "th\"<\n\010Contract\022\022\n\ngeneration\030\001 \001(\004\022\034\n\tre" +
+      "sources\030\002 \003(\0132\t.Resource*,\n\rBackoffPolic" +
+      "y\022\017\n\013Exponential\020\000\022\n\n\006Linear\020\001*+\n\rDelive" +
+      "ryOrder\022\r\n\tUNORDERED\020\000\022\013\n\007ORDERED\020\001*=\n\007K" +
+      "eyType\022\n\n\006String\020\000\022\013\n\007Integer\020\001\022\n\n\006Doubl" +
+      "e\020\002\022\r\n\tByteArray\020\003*)\n\013ContentMode\022\n\n\006BIN" +
+      "ARY\020\000\022\016\n\nSTRUCTURED\020\001*a\n\013SecretField\022\022\n\016" +
+      "SASL_MECHANISM\020\000\022\n\n\006CA_CRT\020\001\022\014\n\010USER_CRT" +
+      "\020\002\022\014\n\010USER_KEY\020\003\022\010\n\004USER\020\004\022\014\n\010PASSWORD\020\005" +
+      "*D\n\010Protocol\022\r\n\tPLAINTEXT\020\000\022\022\n\016SASL_PLAI" +
+      "NTEXT\020\001\022\007\n\003SSL\020\002\022\014\n\010SASL_SSL\020\003B[\n*dev.kn" +
+      "ative.eventing.kafka.broker.contractB\021Da" +
+      "taPlaneContractZ\032control-plane/pkg/contr" +
+      "actb\006proto3"
     };
     descriptor = com.google.protobuf.Descriptors.FileDescriptor
       .internalBuildGeneratedFileFrom(descriptorData,
@@ -24820,39 +25600,45 @@ public final class DataPlaneContract {
     internal_static_Egress_fieldAccessorTable = new
       com.google.protobuf.GeneratedMessageV3.FieldAccessorTable(
         internal_static_Egress_descriptor,
-        new java.lang.String[] { "ConsumerGroup", "Destination", "ReplyUrl", "ReplyToOriginalTopic", "DiscardReply", "Filter", "Uid", "EgressConfig", "DeliveryOrder", "KeyType", "Reference", "DialectedFilter", "VReplicas", "ReplyStrategy", });
-    internal_static_Ingress_descriptor =
+        new java.lang.String[] { "ConsumerGroup", "Destination", "ReplyUrl", "ReplyToOriginalTopic", "DiscardReply", "Filter", "Uid", "EgressConfig", "DeliveryOrder", "KeyType", "Reference", "DialectedFilter", "VReplicas", "FeatureFlags", "ReplyStrategy", });
+    internal_static_EgressFeatureFlags_descriptor =
       getDescriptor().getMessageTypes().get(12);
+    internal_static_EgressFeatureFlags_fieldAccessorTable = new
+      com.google.protobuf.GeneratedMessageV3.FieldAccessorTable(
+        internal_static_EgressFeatureFlags_descriptor,
+        new java.lang.String[] { "EnableRateLimiter", });
+    internal_static_Ingress_descriptor =
+      getDescriptor().getMessageTypes().get(13);
     internal_static_Ingress_fieldAccessorTable = new
       com.google.protobuf.GeneratedMessageV3.FieldAccessorTable(
         internal_static_Ingress_descriptor,
         new java.lang.String[] { "ContentMode", "Path", "Host", });
     internal_static_Reference_descriptor =
-      getDescriptor().getMessageTypes().get(13);
+      getDescriptor().getMessageTypes().get(14);
     internal_static_Reference_fieldAccessorTable = new
       com.google.protobuf.GeneratedMessageV3.FieldAccessorTable(
         internal_static_Reference_descriptor,
         new java.lang.String[] { "Uuid", "Namespace", "Name", "Version", });
     internal_static_SecretReference_descriptor =
-      getDescriptor().getMessageTypes().get(14);
+      getDescriptor().getMessageTypes().get(15);
     internal_static_SecretReference_fieldAccessorTable = new
       com.google.protobuf.GeneratedMessageV3.FieldAccessorTable(
         internal_static_SecretReference_descriptor,
         new java.lang.String[] { "Reference", "KeyFieldReferences", });
     internal_static_KeyFieldReference_descriptor =
-      getDescriptor().getMessageTypes().get(15);
+      getDescriptor().getMessageTypes().get(16);
     internal_static_KeyFieldReference_fieldAccessorTable = new
       com.google.protobuf.GeneratedMessageV3.FieldAccessorTable(
         internal_static_KeyFieldReference_descriptor,
         new java.lang.String[] { "SecretKey", "Field", });
     internal_static_MultiSecretReference_descriptor =
-      getDescriptor().getMessageTypes().get(16);
+      getDescriptor().getMessageTypes().get(17);
     internal_static_MultiSecretReference_fieldAccessorTable = new
       com.google.protobuf.GeneratedMessageV3.FieldAccessorTable(
         internal_static_MultiSecretReference_descriptor,
         new java.lang.String[] { "Protocol", "References", });
     internal_static_CloudEventOverrides_descriptor =
-      getDescriptor().getMessageTypes().get(17);
+      getDescriptor().getMessageTypes().get(18);
     internal_static_CloudEventOverrides_fieldAccessorTable = new
       com.google.protobuf.GeneratedMessageV3.FieldAccessorTable(
         internal_static_CloudEventOverrides_descriptor,
@@ -24864,13 +25650,13 @@ public final class DataPlaneContract {
         internal_static_CloudEventOverrides_ExtensionsEntry_descriptor,
         new java.lang.String[] { "Key", "Value", });
     internal_static_Resource_descriptor =
-      getDescriptor().getMessageTypes().get(18);
+      getDescriptor().getMessageTypes().get(19);
     internal_static_Resource_fieldAccessorTable = new
       com.google.protobuf.GeneratedMessageV3.FieldAccessorTable(
         internal_static_Resource_descriptor,
         new java.lang.String[] { "Uid", "Topics", "BootstrapServers", "Ingress", "EgressConfig", "Egresses", "AbsentAuth", "AuthSecret", "MultiAuthSecret", "CloudEventOverrides", "Reference", "Auth", });
     internal_static_Contract_descriptor =
-      getDescriptor().getMessageTypes().get(19);
+      getDescriptor().getMessageTypes().get(20);
     internal_static_Contract_fieldAccessorTable = new
       com.google.protobuf.GeneratedMessageV3.FieldAccessorTable(
         internal_static_Contract_descriptor,
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/BaseConsumerVerticle.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/BaseConsumerVerticle.java
index 36f86df12..27668d56c 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/BaseConsumerVerticle.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/BaseConsumerVerticle.java
@@ -100,4 +100,6 @@ public abstract class BaseConsumerVerticle extends AbstractVerticle {
       this.context.exceptionHandler().handle(cause);
     }
   }
+
+  public abstract PartitionRevokedHandler getPartitionsRevokedHandler();
 }
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OffsetManager.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OffsetManager.java
index e3fb8d753..0d941242f 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OffsetManager.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OffsetManager.java
@@ -51,6 +51,7 @@ public final class OffsetManager implements RecordDispatcherListener {
   private final Consumer<Integer> onCommit;
   private final long timerId;
   private final Vertx vertx;
+  private final PartitionRevokedHandler partitionRevokedHandler;
 
   /**
    * All args constructor.
@@ -71,14 +72,26 @@ public final class OffsetManager implements RecordDispatcherListener {
     this.timerId = vertx.setPeriodic(commitIntervalMs, l -> commitAll());
     this.vertx = vertx;
 
-    this.consumer.partitionsRevokedHandler(partitions -> {
-      logPartitions("revoked", partitions);
-      partitions.forEach(offsetTrackers::remove);
-    });
+    partitionRevokedHandler = partitions -> {
+      try {
+        // Async commit offsets.
+        final var commitFuture = commit(partitions);
+        // Remove revoked partitions.
+        partitions.forEach(offsetTrackers::remove);
+        return commitFuture;
+      } catch (final Exception ex) {
+        logger.warn("Failed to commit revoked partitions offsets {}", keyValue("partitions", partitions), ex);
+        return Future.failedFuture(ex);
+      } finally {
+        // Remove revoked partitions in any case.
+        partitions.forEach(offsetTrackers::remove);
+        logPartitions("revoked", partitions);
+      }
+    };
+  }
 
-    this.consumer.partitionsAssignedHandler(partitions ->
-      partitions.forEach(tp -> logPartitions("assigned", partitions))
-    );
+  public PartitionRevokedHandler getPartitionRevokedHandler() {
+    return partitionRevokedHandler;
   }
 
   /**
@@ -179,6 +192,22 @@ public final class OffsetManager implements RecordDispatcherListener {
     ).mapEmpty();
   }
 
+  /**
+   * Commit all tracked offsets by colling commit on every offsetTracker entry.
+   *
+   * @return succeeded or failed future.
+   */
+  private Future<Void> commit(final Set<TopicPartition> partitions) {
+    return CompositeFuture.all(
+      this.offsetTrackers.entrySet()
+        .stream()
+        .filter(kv -> partitions.contains(kv.getKey()))
+        .map(e -> commit(e.getKey(), e.getValue()))
+        .filter(Objects::nonNull)
+        .collect(Collectors.toList())
+    ).mapEmpty();
+  }
+
   @Override
   public Future<Void> close() {
     vertx.cancelTimer(timerId);
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OrderedConsumerVerticle.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OrderedConsumerVerticle.java
index c79300332..bc9e4ed99 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OrderedConsumerVerticle.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OrderedConsumerVerticle.java
@@ -40,6 +40,7 @@ public class OrderedConsumerVerticle extends BaseConsumerVerticle {
   private static final Duration POLLING_TIMEOUT = Duration.ofMillis(1000L);
 
   private final Map<TopicPartition, OrderedAsyncExecutor> recordDispatcherExecutors;
+  private final PartitionRevokedHandler partitionRevokedHandler;
 
   private boolean closed;
   private long pollTimer;
@@ -50,6 +51,18 @@ public class OrderedConsumerVerticle extends BaseConsumerVerticle {
     this.recordDispatcherExecutors = new HashMap<>();
     this.closed = false;
     this.isPollInFlight = false;
+
+    partitionRevokedHandler = partitions -> {
+      // Stop executors associated with revoked partitions.
+      for (final TopicPartition partition : partitions) {
+        final var executor = recordDispatcherExecutors.remove(partition);
+        if (executor != null) {
+          logger.info("Stopping executor {}", keyValue("topicPartition", partition));
+          executor.stop();
+        }
+      }
+      return Future.succeededFuture();
+    };
   }
 
   @Override
@@ -94,6 +107,11 @@ public class OrderedConsumerVerticle extends BaseConsumerVerticle {
     super.stop(stopPromise);
   }
 
+  @Override
+  public PartitionRevokedHandler getPartitionsRevokedHandler() {
+    return partitionRevokedHandler;
+  }
+
   private void recordsHandler(KafkaConsumerRecords<Object, CloudEvent> records) {
     if (this.closed) {
       return;
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/PartitionRevokedHandler.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/PartitionRevokedHandler.java
new file mode 100644
index 000000000..bc628e1a9
--- /dev/null
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/PartitionRevokedHandler.java
@@ -0,0 +1,34 @@
+/*
+ * Copyright © 2018 Knative Authors (knative-dev@googlegroups.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package dev.knative.eventing.kafka.broker.dispatcher.impl.consumer;
+
+import io.vertx.core.Future;
+import io.vertx.kafka.client.common.TopicPartition;
+
+import java.util.Set;
+
+/**
+ * {@link PartitionRevokedHandler} is the handler called when some partitions are revoked to a
+ * {@link org.apache.kafka.clients.consumer.Consumer}
+ */
+public interface PartitionRevokedHandler {
+
+  /**
+   * @param partitions revoked partitions
+   * @return a successful or a failed future
+   */
+  Future<Void> partitionRevoked(final Set<TopicPartition> partitions);
+}
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/UnorderedConsumerVerticle.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/UnorderedConsumerVerticle.java
index f03027e34..f9f92858f 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/UnorderedConsumerVerticle.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/UnorderedConsumerVerticle.java
@@ -17,6 +17,7 @@ package dev.knative.eventing.kafka.broker.dispatcher.impl.consumer;
 
 import dev.knative.eventing.kafka.broker.dispatcher.DeliveryOrder;
 import io.cloudevents.CloudEvent;
+import io.vertx.core.Future;
 import io.vertx.core.Promise;
 import io.vertx.kafka.client.consumer.KafkaConsumerRecords;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -135,4 +136,9 @@ public final class UnorderedConsumerVerticle extends BaseConsumerVerticle {
     // Stop the consumer
     super.stop(stopPromise);
   }
+
+  @Override
+  public PartitionRevokedHandler getPartitionsRevokedHandler() {
+    return partitions -> Future.succeededFuture();
+  }
 }
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 b82665705..acf212f38 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
@@ -37,6 +37,7 @@ import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.InvalidCloudEv
 import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.KeyDeserializer;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.OffsetManager;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.OrderedConsumerVerticle;
+import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.PartitionRevokedHandler;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.UnorderedConsumerVerticle;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.filter.subscriptionsapi.AllFilter;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.filter.subscriptionsapi.AnyFilter;
@@ -51,12 +52,15 @@ import io.micrometer.core.instrument.MeterRegistry;
 import io.vertx.circuitbreaker.CircuitBreaker;
 import io.vertx.circuitbreaker.CircuitBreakerOptions;
 import io.vertx.core.AbstractVerticle;
+import io.vertx.core.Future;
+import io.vertx.core.Handler;
 import io.vertx.core.Vertx;
 import io.vertx.core.impl.VertxInternal;
 import io.vertx.core.tracing.TracingPolicy;
 import io.vertx.ext.web.client.WebClient;
 import io.vertx.ext.web.client.WebClientOptions;
 import io.vertx.kafka.client.common.KafkaClientOptions;
+import io.vertx.kafka.client.common.TopicPartition;
 import io.vertx.kafka.client.common.tracing.ConsumerTracer;
 import io.vertx.kafka.client.consumer.KafkaConsumer;
 import io.vertx.kafka.client.producer.KafkaProducer;
@@ -66,6 +70,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.AbstractMap.SimpleImmutableEntry;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -74,6 +79,7 @@ import java.util.Map.Entry;
 import java.util.Objects;
 import java.util.Properties;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
@@ -182,26 +188,35 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
          () -> new ResponseToHttpEndpointHandler(createConsumerRecordSender(vertx, egress.getReplyUrl(), egressConfig)));
        final var commitIntervalMs = Integer.parseInt(String.valueOf(consumerConfigs.get(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG)));
 
-       final var recordDispatcher = new RecordDispatcherMutatorChain(
-         new RecordDispatcherImpl(
-           new ResourceContext(resource, egress),
-           filter,
-           egressSubscriberSender,
-           egressDeadLetterSender,
-           responseHandler,
-           new OffsetManager(vertx, consumer, (v) -> {}, commitIntervalMs),
-           ConsumerTracer.create(
-             ((VertxInternal) vertx).tracer(),
-             new KafkaClientOptions()
-               .setConfig(consumerConfigs)
-               // Make sure the policy is propagate for the manually instantiated consumer tracer
-               .setTracingPolicy(TracingPolicy.PROPAGATE)
-           ),
-           Metrics.getRegistry()
+       final var offsetManager = new OffsetManager(vertx, consumer, (v) -> {}, commitIntervalMs);
+       final var recordDispatcherImpl = new RecordDispatcherImpl(
+         new ResourceContext(resource, egress),
+         filter,
+         egressSubscriberSender,
+         egressDeadLetterSender,
+         responseHandler,
+         offsetManager,
+         ConsumerTracer.create(
+           ((VertxInternal) vertx).tracer(),
+           new KafkaClientOptions()
+             .setConfig(consumerConfigs)
+             // Make sure the policy is propagate for the manually instantiated consumer tracer
+             .setTracingPolicy(TracingPolicy.PROPAGATE)
          ),
+         Metrics.getRegistry()
+       );
+
+       final var recordDispatcher = new RecordDispatcherMutatorChain(
+         recordDispatcherImpl,
          new CloudEventOverridesMutator(resource.getCloudEventOverrides())
        );
 
+       final var partitionRevokedHandlers = List.of(
+         consumerVerticle.getPartitionsRevokedHandler(),
+         offsetManager.getPartitionRevokedHandler()
+       );
+       consumer.partitionsRevokedHandler(handlePartitionsRevoked(egress.getConsumerGroup(), partitionRevokedHandlers));
+
        // Set all the built objects in the consumer verticle
        consumerVerticle.setRecordDispatcher(recordDispatcher);
        consumerVerticle.setConsumer(consumer);
@@ -217,6 +232,39 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
     );
   }
 
+  /**
+   * For each handler call partitionRevoked and wait for the future to complete.
+   *
+   * @param consumerGroup consumer group id
+   * @param partitionRevokedHandlers partition revoked handlers
+   * @return a single handler that dispatches the partition revoked event to the given handlers.
+   */
+  private Handler<Set<TopicPartition>> handlePartitionsRevoked(final String consumerGroup,
+                                                               final List<PartitionRevokedHandler> partitionRevokedHandlers) {
+    return partitions -> {
+
+      logger.info("Received revoke partitions for consumer group {} {}",
+        keyValue("group.id", consumerGroup),
+        keyValue("topicPartitions", partitions)
+      );
+
+      final var futures = new ArrayList<Future<Void>>(partitionRevokedHandlers.size());
+      for (PartitionRevokedHandler partitionRevokedHandler : partitionRevokedHandlers) {
+        futures.add(partitionRevokedHandler.partitionRevoked(partitions));
+      }
+
+      for (final var future : futures) {
+        try {
+          future
+            .toCompletionStage()
+            .toCompletableFuture()
+            .get(1, TimeUnit.SECONDS);
+        } catch (final Exception ignored) {
+        }
+      }
+    };
+  }
+
   private Filter getFilter(DataPlaneContract.Egress egress) {
     // Dialected filters should override the attributes filter
     if (egress.getDialectedFilterCount() > 0) {
diff --git a/hack/update-checksums.sh b/hack/update-checksums.sh
index df7c14e6c..ad2c6310b 100755
--- a/hack/update-checksums.sh
+++ b/hack/update-checksums.sh
@@ -29,3 +29,4 @@ source $(dirname $0)/../vendor/knative.dev/hack/library.sh
 go run "${REPO_ROOT_DIR}/vendor/knative.dev/pkg/configmap/hash-gen" "${REPO_ROOT_DIR}"/control-plane/config/eventing-kafka-broker/200-controller/100-config-tracing.yaml
 go run "${REPO_ROOT_DIR}/vendor/knative.dev/pkg/configmap/hash-gen" "${REPO_ROOT_DIR}"/control-plane/config/eventing-kafka-broker/200-controller/100-config-logging.yaml
 go run "${REPO_ROOT_DIR}/vendor/knative.dev/pkg/configmap/hash-gen" "${REPO_ROOT_DIR}"/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-leader-election.yaml
+go run "${REPO_ROOT_DIR}/vendor/knative.dev/pkg/configmap/hash-gen" "${REPO_ROOT_DIR}"/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-features.yaml
diff --git a/proto/contract.proto b/proto/contract.proto
index 8e1dd0aac..c7c253dfe 100644
--- a/proto/contract.proto
+++ b/proto/contract.proto
@@ -151,6 +151,14 @@ message Egress {
 
   // Number of virtual replicas.
   int32 vReplicas = 13;
+
+  // Egress feature flags.
+  EgressFeatureFlags featureFlags = 14;
+}
+
+message EgressFeatureFlags {
+  // Enable rateLimiter
+  bool enableRateLimiter = 1;
 }
 
 // CloudEvent content mode


Test patch:
diff --git a/control-plane/pkg/reconciler/consumer/consumer_test.go b/control-plane/pkg/reconciler/consumer/consumer_test.go
index e81d34e7e..e87a4e709 100644
--- a/control-plane/pkg/reconciler/consumer/consumer_test.go
+++ b/control-plane/pkg/reconciler/consumer/consumer_test.go
@@ -34,6 +34,7 @@ import (
 	"knative.dev/pkg/resolver"
 	"knative.dev/pkg/tracker"
 
+	configapis "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/config"
 	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 	fakekafkainternalsclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client/fake"
@@ -50,6 +51,10 @@ var (
 		"FinalizerUpdate",
 		fmt.Sprintf(`Updated %q finalizers`, ConsumerName),
 	)
+
+	defaultContractFeatureFlags = &contract.EgressFeatureFlags{
+		EnableRateLimiter: false,
+	}
 )
 
 func TestReconcileKind(t *testing.T) {
@@ -121,6 +126,7 @@ func TestReconcileKind(t *testing.T) {
 									Namespace: ConsumerNamespace,
 									Name:      SourceName,
 								},
+								FeatureFlags: defaultContractFeatureFlags,
 							}},
 							Auth:                nil,
 							CloudEventOverrides: nil,
@@ -231,6 +237,7 @@ func TestReconcileKind(t *testing.T) {
 									Namespace: ConsumerNamespace,
 									Name:      SourceName,
 								},
+								FeatureFlags: defaultContractFeatureFlags,
 							}},
 							Auth:                nil,
 							CloudEventOverrides: nil,
@@ -356,6 +363,7 @@ func TestReconcileKind(t *testing.T) {
 									Namespace: ConsumerNamespace,
 									Name:      SourceName,
 								},
+								FeatureFlags: defaultContractFeatureFlags,
 							}},
 							Auth:                nil,
 							CloudEventOverrides: nil,
@@ -476,6 +484,7 @@ func TestReconcileKind(t *testing.T) {
 			SecretLister:        listers.GetSecretLister(),
 			PodLister:           listers.GetPodLister(),
 			KubeClient:          kubeclient.Get(ctx),
+			KafkaFeatureFlags:   configapis.DefaultFeaturesConfig(),
 		}
 
 		return creconciler.NewReconciler(
diff --git a/test/config/100-config-kafka-features.yaml b/test/config/100-config-kafka-features.yaml
new file mode 100644
index 000000000..443e5838f
--- /dev/null
+++ b/test/config/100-config-kafka-features.yaml
@@ -0,0 +1,7 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: config-kafka-features
+  namespace: knative-eventing
+data:
+  dispatcher.rate-limiter: "enabled"
diff --git a/test/e2e-common.sh b/test/e2e-common.sh
index 24186eae5..b12ea5012 100644
--- a/test/e2e-common.sh
+++ b/test/e2e-common.sh
@@ -151,8 +151,9 @@ function install_latest_release() {
   kubectl apply -f "${PREVIOUS_RELEASE_URL}/${EVENTING_KAFKA_SOURCE_ARTIFACT}" || return $?
   kubectl apply -f "${PREVIOUS_RELEASE_URL}/${EVENTING_KAFKA_CHANNEL_ARTIFACT}" || return $?
 
-  # Restore test config-tracing.
+  # Restore test config.
   kubectl replace -f ./test/config/100-config-tracing.yaml
+  kubectl replace -f ./test/config/100-config-kafka-features.yaml
 }
 
 function install_head() {
@@ -165,8 +166,9 @@ function install_head() {
   kubectl apply -f "${EVENTING_KAFKA_CHANNEL_ARTIFACT}" || return $?
   kubectl apply -f "${EVENTING_KAFKA_POST_INSTALL_ARTIFACT}" || return $?
 
-  # Restore test config-tracing.
+  # Restore test config.
   kubectl replace -f ./test/config/100-config-tracing.yaml
+  kubectl replace -f ./test/config/100-config-kafka-features.yaml
 }
 
 function install_latest_release_source() {
@@ -176,8 +178,9 @@ function install_latest_release_source() {
 
   kubectl apply -f "${PREVIOUS_RELEASE_URL}/${EVENTING_KAFKA_SOURCE_BUNDLE_ARTIFACT}" || return $?
 
-  # Restore test config-tracing.
+  # Restore test config.
   kubectl replace -f ./test/config/100-config-tracing.yaml
+  kubectl replace -f ./test/config/100-config-kafka-features.yaml
 }
 
 function install_head_source() {
@@ -186,8 +189,9 @@ function install_head_source() {
   kubectl apply -f "${EVENTING_KAFKA_SOURCE_BUNDLE_ARTIFACT}" || return $?
   kubectl apply -f "${EVENTING_KAFKA_POST_INSTALL_ARTIFACT}" || return $?
 
-  # Restore test config-tracing.
+  # Restore test config.
   kubectl replace -f ./test/config/100-config-tracing.yaml
+  kubectl replace -f ./test/config/100-config-kafka-features.yaml
 }
 
 function test_setup() {