Repository: knative-sandbox/eventing-kafka-broker
Pull request #2171 (2022-05-10 10:45:52)

Set vReplicas in consumer reconciler 
Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

Part of https://github.com/knative-sandbox/eventing-kafka-broker/issues/2125

- [Add vReplicas to contract](https://github.com/knative-sandbox/eventing-kafka-broker/commit/2ba6b70de2a4bfeb6e1537ed52758454f2867b55)
- [Set vReplicas in consumer reconciler](https://github.com/knative-sandbox/eventing-kafka-broker/commit/2e8f97e5cd5bb78dd57aba361d1d227d9aba4e57)

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

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

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

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

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

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

```diff
@@             Coverage Diff              @@
##               main    #2171      +/-   ##
============================================
- Coverage     66.55%   61.24%   -5.31%     
============================================
  Files           142       70      -72     
  Lines          9176     6861    -2315     
  Branches        202        0     -202     
============================================
- Hits           6107     4202    -1905     
+ Misses         2654     2360     -294     
+ Partials        415      299     -116     
```

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

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

| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2171?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/2171/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvY29udHJhY3QvY29udHJhY3QucGIuZ28=) | `7.86% <0.00%> (-0.04%)` | :arrow_down: |
| [control-plane/pkg/reconciler/consumer/consumer.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2171/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lci9jb25zdW1lci5nbw==) | `51.44% <100.00%> (+0.17%)` | :arrow_up: |
| [...ver/impl/handler/ControlPlaneProbeRequestUtil.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2171/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL2hhbmRsZXIvQ29udHJvbFBsYW5lUHJvYmVSZXF1ZXN0VXRpbC5qYXZh) | | |
| [.../dispatcher/impl/RecordDispatcherMutatorChain.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2171/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9SZWNvcmREaXNwYXRjaGVyTXV0YXRvckNoYWluLmphdmE=) | | |
| [...er/dispatcher/impl/consumer/InvalidCloudEvent.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2171/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9JbnZhbGlkQ2xvdWRFdmVudC5qYXZh) | | |
| [.../broker/receiver/main/ReceiverVerticleFactory.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2171/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9tYWluL1JlY2VpdmVyVmVydGljbGVGYWN0b3J5LmphdmE=) | | |
| [...fka/broker/receiver/impl/handler/ProbeHandler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2171/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL2hhbmRsZXIvUHJvYmVIYW5kbGVyLmphdmE=) | | |
| [...ispatcher/impl/http/WebClientCloudEventSender.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2171/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9odHRwL1dlYkNsaWVudENsb3VkRXZlbnRTZW5kZXIuamF2YQ==) | | |
| [...tive/eventing/kafka/broker/core/utils/BaseEnv.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2171/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdXRpbHMvQmFzZUVudi5qYXZh) | | |
| [...atcher/impl/filter/subscriptionsapi/NotFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2171/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvc3Vic2NyaXB0aW9uc2FwaS9Ob3RGaWx0ZXIuamF2YQ==) | | |
| ... and [65 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2171/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/2171?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/2171?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [1f1b765...2e8f97e](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2171?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).


Patch:
diff --git a/OWNERS_ALIASES b/OWNERS_ALIASES
index d98fc841e..c9d0ba642 100644
--- a/OWNERS_ALIASES
+++ b/OWNERS_ALIASES
@@ -5,6 +5,7 @@ aliases:
   api-core-wg-leads:
   - dprotaso
   async-component-approvers:
+  - AngeloDanducci
   - bvennam
   - julz
   - maximilien
@@ -202,6 +203,7 @@ aliases:
   - knative-test-reporter-robot
   kperf-approvers:
   - maximilien
+  - psschwei
   - zhanggbj
   net-certmanager-approvers:
   - ZhiminXiang
@@ -259,6 +261,7 @@ aliases:
   - dprotaso
   - julz
   - nak3
+  - psschwei
   steering-committee:
   - csantanapr
   - itsmurugappan
diff --git a/control-plane/pkg/contract/contract.pb.go b/control-plane/pkg/contract/contract.pb.go
index 06cd7c68c..8c959e7cb 100644
--- a/control-plane/pkg/contract/contract.pb.go
+++ b/control-plane/pkg/contract/contract.pb.go
@@ -7,11 +7,10 @@
 package contract
 
 import (
-	reflect "reflect"
-	sync "sync"
-
 	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
 	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
+	reflect "reflect"
+	sync "sync"
 )
 
 const (
@@ -1023,6 +1022,8 @@ type Egress struct {
 	Reference *Reference `protobuf:"bytes,11,opt,name=reference,proto3" json:"reference,omitempty"`
 	// CNCF CloudEvents SubscriptionsAPI compliant filters
 	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"`
 }
 
 func (x *Egress) Reset() {
@@ -1148,6 +1149,13 @@ func (x *Egress) GetDialectedFilter() []*DialectedFilter {
 	return nil
 }
 
+func (x *Egress) GetVReplicas() int32 {
+	if x != nil {
+		return x.VReplicas
+	}
+	return 0
+}
+
 type isEgress_ReplyStrategy interface {
 	isEgress_ReplyStrategy()
 }
@@ -1883,7 +1891,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,
-	0x91, 0x04, 0x0a, 0x06, 0x45, 0x67, 0x72, 0x65, 0x73, 0x73, 0x12, 0x24, 0x0a, 0x0d, 0x63, 0x6f,
+	0xaf, 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,
@@ -1915,117 +1923,119 @@ var file_contract_proto_rawDesc = []byte{
 	0x74, 0x65, 0x64, 0x46, 0x69, 0x6c, 0x74, 0x65, 0x72, 0x18, 0x0c, 0x20, 0x03, 0x28, 0x0b, 0x32,
 	0x10, 0x2e, 0x44, 0x69, 0x61, 0x6c, 0x65, 0x63, 0x74, 0x65, 0x64, 0x46, 0x69, 0x6c, 0x74, 0x65,
 	0x72, 0x52, 0x0f, 0x64, 0x69, 0x61, 0x6c, 0x65, 0x63, 0x74, 0x65, 0x64, 0x46, 0x69, 0x6c, 0x74,
-	0x65, 0x72, 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, 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,
+	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,
+	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 (
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index 8c59fb072..50665fc07 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -222,11 +222,6 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 	}
 	statusConditionManager.Addressable(address)
 
-	// TODO(pierDipi) remove after some releases (released in 1.4)
-	if err := r.removeFinalizerCM(ctx, finalizerCM(broker), brokerConfig); err != nil {
-		return err
-	}
-
 	return nil
 }
 
@@ -342,11 +337,6 @@ func (r *Reconciler) finalizeKind(ctx context.Context, broker *eventing.Broker)
 		return err
 	}
 
-	// TODO(pierDipi) remove after some releases (released in 1.4)
-	if err := r.removeFinalizerCM(ctx, finalizerCM(broker), brokerConfig); err != nil {
-		return err
-	}
-
 	return nil
 }
 
@@ -478,25 +468,3 @@ func (r *Reconciler) reconcilerBrokerResource(ctx context.Context, topic string,
 
 	return resource, nil
 }
-
-func (r *Reconciler) removeFinalizerCM(ctx context.Context, finalizer string, cm *corev1.ConfigMap) error {
-	newFinalizers := make([]string, 0, len(cm.Finalizers))
-	for _, f := range cm.Finalizers {
-		if f != finalizer {
-			newFinalizers = append(newFinalizers, f)
-		}
-	}
-	if len(newFinalizers) != len(cm.Finalizers) {
-		cm := cm.DeepCopy() // Do not modify informer copy.
-		cm.Finalizers = newFinalizers
-		_, err := r.KubeClient.CoreV1().ConfigMaps(cm.GetNamespace()).Update(ctx, cm, metav1.UpdateOptions{})
-		if err != nil && !apierrors.IsNotFound(err) {
-			return fmt.Errorf("failed to remove finalizer %s to ConfigMap %s/%s: %w", finalizer, cm.GetNamespace(), cm.GetName(), err)
-		}
-	}
-	return nil
-}
-
-func finalizerCM(object metav1.Object) string {
-	return fmt.Sprintf("%s/%s-%s", "kafka.brokers.eventing.knative.dev", object.GetNamespace(), object.GetName())
-}
diff --git a/control-plane/pkg/reconciler/consumer/consumer.go b/control-plane/pkg/reconciler/consumer/consumer.go
index 02892d234..d4f2669e7 100644
--- a/control-plane/pkg/reconciler/consumer/consumer.go
+++ b/control-plane/pkg/reconciler/consumer/consumer.go
@@ -112,6 +112,7 @@ func (r Reconciler) reconcileContractResource(ctx context.Context, c *kafkainter
 	}
 
 	egress.Reference = userFacingResourceRef
+	egress.VReplicas = *c.Spec.VReplicas
 
 	resource := &contract.Resource{
 		Uid:                 string(c.UID),
diff --git a/control-plane/pkg/reconciler/source/v2/sourcev2.go b/control-plane/pkg/reconciler/source/v2/sourcev2.go
index 2cede37c4..54e3f342d 100644
--- a/control-plane/pkg/reconciler/source/v2/sourcev2.go
+++ b/control-plane/pkg/reconciler/source/v2/sourcev2.go
@@ -24,6 +24,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/utils/pointer"
 	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
@@ -58,6 +59,12 @@ type Reconciler struct {
 
 func (r *Reconciler) ReconcileKind(ctx context.Context, ks *sources.KafkaSource) reconciler.Event {
 
+	selector, err := GetLabelsAsSelector(ks.Name)
+	if err != nil {
+		return fmt.Errorf("getting labels as selector: %v", err)
+	}
+	ks.Status.Selector = selector.String()
+
 	cg, err := r.reconcileConsumerGroup(ctx, ks)
 	if err != nil {
 		ks.GetConditionSet().Manage(&ks.Status).MarkFalse(KafkaConditionConsumerGroup, "failed to reconcile consumer group", err.Error())
@@ -69,6 +76,21 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, ks *sources.KafkaSource)
 	return nil
 }
 
+func GetLabelsAsSelector(name string) (labels.Selector, error) {
+	labels := GetLabels(name)
+	var labelSelector metav1.LabelSelector
+	labelSelector.MatchLabels = labels
+	selector, err := metav1.LabelSelectorAsSelector(&labelSelector)
+	return selector, err
+}
+
+func GetLabels(name string) map[string]string {
+	return map[string]string{
+		"eventing.knative.dev/source":     "kafka-source-controller",
+		"eventing.knative.dev/sourceName": name,
+	}
+}
+
 func (r Reconciler) reconcileConsumerGroup(ctx context.Context, ks *sources.KafkaSource) (*internalscg.ConsumerGroup, error) {
 
 	backoffPolicy := eventingduck.BackoffPolicyExponential
diff --git a/data-plane/config/broker/500-dispatcher.yaml b/data-plane/config/broker/500-dispatcher.yaml
index 7de01d67f..e0104374f 100644
--- a/data-plane/config/broker/500-dispatcher.yaml
+++ b/data-plane/config/broker/500-dispatcher.yaml
@@ -116,6 +116,7 @@ spec:
           command:
             - "java"
           args:
+            - "-XX:+CrashOnOutOfMemoryError"
             - "-Dlogback.configurationFile=/etc/logging/config.xml"
             - "-jar"
             - "/app/app.jar"
diff --git a/data-plane/config/broker/500-receiver.yaml b/data-plane/config/broker/500-receiver.yaml
index 50be74302..92edc3a24 100644
--- a/data-plane/config/broker/500-receiver.yaml
+++ b/data-plane/config/broker/500-receiver.yaml
@@ -117,6 +117,7 @@ spec:
           command:
             - "java"
           args:
+            - "-XX:+CrashOnOutOfMemoryError"
             - "-Dlogback.configurationFile=/etc/logging/config.xml"
             - "-jar"
             - "/app/app.jar"
diff --git a/data-plane/config/brokerv2/500-dispatcher.yaml b/data-plane/config/brokerv2/500-dispatcher.yaml
index e581a5b7f..2b5d05237 100644
--- a/data-plane/config/brokerv2/500-dispatcher.yaml
+++ b/data-plane/config/brokerv2/500-dispatcher.yaml
@@ -119,6 +119,7 @@ spec:
           command:
             - "java"
           args:
+            - "-XX:+CrashOnOutOfMemoryError"
             - "-Dlogback.configurationFile=/etc/logging/config.xml"
             - "-jar"
             - "/app/app.jar"
diff --git a/data-plane/config/channel/500-dispatcher.yaml b/data-plane/config/channel/500-dispatcher.yaml
index a3741a2af..b0bf91df4 100644
--- a/data-plane/config/channel/500-dispatcher.yaml
+++ b/data-plane/config/channel/500-dispatcher.yaml
@@ -116,6 +116,7 @@ spec:
           command:
             - "java"
           args:
+            - "-XX:+CrashOnOutOfMemoryError"
             - "-Dlogback.configurationFile=/etc/logging/config.xml"
             - "-jar"
             - "/app/app.jar"
diff --git a/data-plane/config/channel/500-receiver.yaml b/data-plane/config/channel/500-receiver.yaml
index 6afd14ece..945b1da3a 100644
--- a/data-plane/config/channel/500-receiver.yaml
+++ b/data-plane/config/channel/500-receiver.yaml
@@ -117,6 +117,7 @@ spec:
           command:
             - "java"
           args:
+            - "-XX:+CrashOnOutOfMemoryError"
             - "-Dlogback.configurationFile=/etc/logging/config.xml"
             - "-jar"
             - "/app/app.jar"
diff --git a/data-plane/config/channelv2/500-dispatcher.yaml b/data-plane/config/channelv2/500-dispatcher.yaml
index 2e0bf12ff..3f05df806 100644
--- a/data-plane/config/channelv2/500-dispatcher.yaml
+++ b/data-plane/config/channelv2/500-dispatcher.yaml
@@ -119,6 +119,7 @@ spec:
           command:
             - "java"
           args:
+            - "-XX:+CrashOnOutOfMemoryError"
             - "-Dlogback.configurationFile=/etc/logging/config.xml"
             - "-jar"
             - "/app/app.jar"
diff --git a/data-plane/config/sink/500-receiver.yaml b/data-plane/config/sink/500-receiver.yaml
index a0cbf2124..6341322a7 100644
--- a/data-plane/config/sink/500-receiver.yaml
+++ b/data-plane/config/sink/500-receiver.yaml
@@ -117,6 +117,7 @@ spec:
           command:
             - "java"
           args:
+            - "-XX:+CrashOnOutOfMemoryError"
             - "-Dlogback.configurationFile=/etc/logging/config.xml"
             - "-jar"
             - "/app/app.jar"
diff --git a/data-plane/config/source/500-dispatcher.yaml b/data-plane/config/source/500-dispatcher.yaml
index 13b5402e6..7544526f4 100644
--- a/data-plane/config/source/500-dispatcher.yaml
+++ b/data-plane/config/source/500-dispatcher.yaml
@@ -116,6 +116,7 @@ spec:
           command:
             - "java"
           args:
+            - "-XX:+CrashOnOutOfMemoryError"
             - "-Dlogback.configurationFile=/etc/logging/config.xml"
             - "-jar"
             - "/app/app.jar"
diff --git a/data-plane/config/sourcev2/500-dispatcher.yaml b/data-plane/config/sourcev2/500-dispatcher.yaml
index 52a4c19f4..0556b70a2 100644
--- a/data-plane/config/sourcev2/500-dispatcher.yaml
+++ b/data-plane/config/sourcev2/500-dispatcher.yaml
@@ -119,6 +119,7 @@ spec:
           command:
             - "java"
           args:
+            - "-XX:+CrashOnOutOfMemoryError"
             - "-Dlogback.configurationFile=/etc/logging/config.xml"
             - "-jar"
             - "/app/app.jar"
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 330eed3a5..4aca10aaf 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
@@ -10498,6 +10498,16 @@ public final class DataPlaneContract {
     dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilterOrBuilder getDialectedFilterOrBuilder(
         int index);
 
+    /**
+     * <pre>
+     * Number of virtual replicas.
+     * </pre>
+     *
+     * <code>int32 vReplicas = 13;</code>
+     * @return The vReplicas.
+     */
+    int getVReplicas();
+
     public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Egress.ReplyStrategyCase getReplyStrategyCase();
   }
   /**
@@ -10664,6 +10674,11 @@ public final class DataPlaneContract {
                   input.readMessage(dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DialectedFilter.parser(), extensionRegistry));
               break;
             }
+            case 104: {
+
+              vReplicas_ = input.readInt32();
+              break;
+            }
             default: {
               if (!parseUnknownField(
                   input, unknownFields, extensionRegistry, tag)) {
@@ -11274,6 +11289,21 @@ public final class DataPlaneContract {
       return dialectedFilter_.get(index);
     }
 
+    public static final int VREPLICAS_FIELD_NUMBER = 13;
+    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_;
+    }
+
     private byte memoizedIsInitialized = -1;
     @java.lang.Override
     public final boolean isInitialized() {
@@ -11324,6 +11354,9 @@ public final class DataPlaneContract {
       for (int i = 0; i < dialectedFilter_.size(); i++) {
         output.writeMessage(12, dialectedFilter_.get(i));
       }
+      if (vReplicas_ != 0) {
+        output.writeInt32(13, vReplicas_);
+      }
       unknownFields.writeTo(output);
     }
 
@@ -11377,6 +11410,10 @@ public final class DataPlaneContract {
         size += com.google.protobuf.CodedOutputStream
           .computeMessageSize(12, dialectedFilter_.get(i));
       }
+      if (vReplicas_ != 0) {
+        size += com.google.protobuf.CodedOutputStream
+          .computeInt32Size(13, vReplicas_);
+      }
       size += unknownFields.getSerializedSize();
       memoizedSize = size;
       return size;
@@ -11417,6 +11454,8 @@ public final class DataPlaneContract {
       }
       if (!getDialectedFilterList()
           .equals(other.getDialectedFilterList())) return false;
+      if (getVReplicas()
+          != other.getVReplicas()) return false;
       if (!getReplyStrategyCase().equals(other.getReplyStrategyCase())) return false;
       switch (replyStrategyCase_) {
         case 3:
@@ -11471,6 +11510,8 @@ public final class DataPlaneContract {
         hash = (37 * hash) + DIALECTEDFILTER_FIELD_NUMBER;
         hash = (53 * hash) + getDialectedFilterList().hashCode();
       }
+      hash = (37 * hash) + VREPLICAS_FIELD_NUMBER;
+      hash = (53 * hash) + getVReplicas();
       switch (replyStrategyCase_) {
         case 3:
           hash = (37 * hash) + REPLYURL_FIELD_NUMBER;
@@ -11655,6 +11696,8 @@ public final class DataPlaneContract {
         } else {
           dialectedFilterBuilder_.clear();
         }
+        vReplicas_ = 0;
+
         replyStrategyCase_ = 0;
         replyStrategy_ = null;
         return this;
@@ -11730,6 +11773,7 @@ public final class DataPlaneContract {
         } else {
           result.dialectedFilter_ = dialectedFilterBuilder_.build();
         }
+        result.vReplicas_ = vReplicas_;
         result.replyStrategyCase_ = replyStrategyCase_;
         onBuilt();
         return result;
@@ -11832,6 +11876,9 @@ public final class DataPlaneContract {
             }
           }
         }
+        if (other.getVReplicas() != 0) {
+          setVReplicas(other.getVReplicas());
+        }
         switch (other.getReplyStrategyCase()) {
           case REPLYURL: {
             replyStrategyCase_ = 3;
@@ -13629,6 +13676,49 @@ public final class DataPlaneContract {
         }
         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;
+      }
       @java.lang.Override
       public final Builder setUnknownFields(
           final com.google.protobuf.UnknownFieldSet unknownFields) {
@@ -24584,7 +24674,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\"\370\002\n\006Egress\022\025\n\rconsumerGroup\030\001 \001(" +
+      "t\030\005 \001(\004\"\213\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" +
@@ -24593,43 +24683,43 @@ 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" +
-      "tedFilterB\017\n\rreplyStrategy\"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\004uui" +
-      "d\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\017SecretReference\022\035\n\t" +
-      "reference\030\001 \001(\0132\n.Reference\022.\n\022keyFieldR" +
-      "eferences\030\002 \003(\0132\022.KeyFieldReference\"C\n\021K" +
-      "eyFieldReference\022\021\n\tsecretKey\030\002 \001(\t\022\033\n\005f" +
-      "ield\030\003 \001(\0162\014.SecretField\"Y\n\024MultiSecretR" +
-      "eference\022\033\n\010protocol\030\001 \001(\0162\t.Protocol\022$\n" +
-      "\nreferences\030\002 \003(\0132\020.SecretReference\"\202\001\n\023" +
-      "CloudEventOverrides\0228\n\nextensions\030\001 \003(\0132" +
-      "$.CloudEventOverrides.ExtensionsEntry\0321\n" +
-      "\017ExtensionsEntry\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\006to" +
-      "pics\030\002 \003(\t\022\030\n\020bootstrapServers\030\003 \001(\t\022\031\n\007" +
-      "ingress\030\004 \001(\0132\010.Ingress\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.ReferenceH\000\0220\n\017multi" +
-      "AuthSecret\030\t \001(\0132\025.MultiSecretReferenceH" +
-      "\000\0221\n\023cloudEventOverrides\030\n \001(\0132\024.CloudEv" +
-      "entOverrides\022\035\n\treference\030\013 \001(\0132\n.Refere" +
-      "nceB\006\n\004Auth\"<\n\010Contract\022\022\n\ngeneration\030\001 " +
-      "\001(\004\022\034\n\tresources\030\002 \003(\0132\t.Resource*,\n\rBac" +
-      "koffPolicy\022\017\n\013Exponential\020\000\022\n\n\006Linear\020\001*" +
-      "+\n\rDeliveryOrder\022\r\n\tUNORDERED\020\000\022\013\n\007ORDER" +
-      "ED\020\001*=\n\007KeyType\022\n\n\006String\020\000\022\013\n\007Integer\020\001" +
-      "\022\n\n\006Double\020\002\022\r\n\tByteArray\020\003*)\n\013ContentMo" +
-      "de\022\n\n\006BINARY\020\000\022\016\n\nSTRUCTURED\020\001*a\n\013Secret" +
-      "Field\022\022\n\016SASL_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\010P" +
-      "ASSWORD\020\005*D\n\010Protocol\022\r\n\tPLAINTEXT\020\000\022\022\n\016" +
-      "SASL_PLAINTEXT\020\001\022\007\n\003SSL\020\002\022\014\n\010SASL_SSL\020\003B" +
-      "[\n*dev.knative.eventing.kafka.broker.con" +
-      "tractB\021DataPlaneContractZ\032control-plane/" +
-      "pkg/contractb\006proto3"
+      "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"
     };
     descriptor = com.google.protobuf.Descriptors.FileDescriptor
       .internalBuildGeneratedFileFrom(descriptorData,
@@ -24730,7 +24820,7 @@ 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", "ReplyStrategy", });
+        new java.lang.String[] { "ConsumerGroup", "Destination", "ReplyUrl", "ReplyToOriginalTopic", "DiscardReply", "Filter", "Uid", "EgressConfig", "DeliveryOrder", "KeyType", "Reference", "DialectedFilter", "VReplicas", "ReplyStrategy", });
     internal_static_Ingress_descriptor =
       getDescriptor().getMessageTypes().get(12);
     internal_static_Ingress_fieldAccessorTable = new
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/AsyncCloseable.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/AsyncCloseable.java
index bd89e836e..370a77789 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/AsyncCloseable.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/AsyncCloseable.java
@@ -79,38 +79,4 @@ public interface AsyncCloseable extends Closeable {
         v -> compose(closeableIterator)
       );
   }
-
-  /**
-   * Wrap the provided blocking {@link AutoCloseable} into an {@link AsyncCloseable}.
-   * This is going to use the current context when the close is invoked.
-   *
-   * @param closeable the closeable to wrap
-   * @return the wrapped closeable
-   */
-  static AsyncCloseable wrapAutoCloseable(AutoCloseable closeable) {
-    return () -> {
-      Context context = Vertx.currentContext();
-
-      if (context == null) {
-        // I'm not on the event loop, I can just execute this as is!
-        try {
-          closeable.close();
-          return Future.succeededFuture();
-        } catch (Exception e) {
-          return Future.failedFuture(e);
-        }
-      }
-
-      // I'm on the event loop, I need to execute blocking.
-      return context.executeBlocking(promise -> {
-        try {
-          closeable.close();
-          promise.complete();
-        } catch (Exception e) {
-          promise.fail(e);
-        }
-      });
-    };
-  }
-
 }
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/metrics/Metrics.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/metrics/Metrics.java
index 5a63f663f..5c51105dd 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/metrics/Metrics.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/metrics/Metrics.java
@@ -15,6 +15,7 @@
  */
 package dev.knative.eventing.kafka.broker.core.metrics;
 
+import dev.knative.eventing.kafka.broker.core.AsyncCloseable;
 import dev.knative.eventing.kafka.broker.core.utils.BaseEnv;
 import io.micrometer.core.instrument.Counter;
 import io.micrometer.core.instrument.DistributionSummary;
@@ -23,6 +24,8 @@ import io.micrometer.core.instrument.binder.BaseUnits;
 import io.micrometer.core.instrument.binder.kafka.KafkaClientMetrics;
 import io.micrometer.prometheus.PrometheusConfig;
 import io.micrometer.prometheus.PrometheusMeterRegistry;
+import io.vertx.core.Future;
+import io.vertx.core.Promise;
 import io.vertx.core.http.HttpServerOptions;
 import io.vertx.core.metrics.MetricsOptions;
 import io.vertx.core.net.PemKeyCertOptions;
@@ -37,6 +40,12 @@ import org.apache.kafka.clients.producer.Producer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.function.Supplier;
+
 public class Metrics {
 
   private static final Logger logger = LoggerFactory.getLogger(Metrics.class);
@@ -46,6 +55,27 @@ public class Metrics {
 
   public static final String METRICS_REGISTRY_NAME = "metrics";
 
+  public static final boolean DISABLE_KAFKA_CLIENTS_METRICS =
+    Boolean.parseBoolean(System.getenv("DISABLE_KAFKA_CLIENTS_METRICS"));
+
+  // There are different thread polls usable, mainly, each with its own drawbacks for our use case:
+  //   - cached thread pools
+  //   - fixed thread pools
+  //
+  // A cached thread might grow unbounded and since creating, updating and deleting resources
+  // trigger the usage of this executor, a bad actor might start continuously creating, updating
+  // and deleting resources which will cause resource exhaustion.
+  //
+  // A fixed thread poll doesn't give the best possible latency for every resource, but it's
+  // bounded, so we keep the resource usage under control.
+  // We might want to provide configs to make it bigger than a single thread but a single thread
+  // to start with is good enough for now.
+  public static final ExecutorService meterBinderExecutor = Executors.newSingleThreadExecutor();
+
+  static {
+    Runtime.getRuntime().addShutdownHook(new Thread(meterBinderExecutor::shutdown));
+  }
+
   // Micrometer employs a naming convention that separates lowercase words with a '.' (dot) character.
   // Different monitoring systems have different recommendations regarding naming convention, and some naming
   // conventions may be incompatible for one system and not another.
@@ -77,7 +107,7 @@ public class Metrics {
    */
   public static final String EVENT_PROCESSING_LATENCY = "event_processing_latencies";
 
-    /**
+  /**
    * @link https://knative.dev/docs/eventing/observability/metrics/eventing-metrics/
    * @see Metrics#discardedEventCount(io.micrometer.core.instrument.Tags)
    */
@@ -160,14 +190,8 @@ public class Metrics {
    * @param <V>      Record value type.
    * @return A meter binder to close once the consumer is closed.
    */
-  public static <K, V> AutoCloseable register(final Consumer<K, V> consumer) {
-    final var registry = getRegistry();
-    if (registry != null) {
-      final var clientMetrics = new KafkaClientMetrics(consumer);
-      clientMetrics.bindTo(registry);
-      return clientMetrics;
-    }
-    return () -> {};
+  public static <K, V> AsyncCloseable register(final Consumer<K, V> consumer) {
+    return register(() -> new KafkaClientMetrics(consumer));
   }
 
   /**
@@ -178,14 +202,39 @@ public class Metrics {
    * @param <V>      Record value type.
    * @return A meter binder to close once the producer is closed.
    */
-  public static <K, V> AutoCloseable register(final Producer<K, V> producer) {
+  public static <K, V> AsyncCloseable register(final Producer<K, V> producer) {
+    return register(() -> new KafkaClientMetrics(producer));
+  }
+
+  /**
+   * Register the given metrics provider to the global meter registry.
+   *
+   * @param metricsProvider metrics to bind to the global registry.
+   * @return A meter binder to close once the consumer or producer is closed.
+   */
+  private static AsyncCloseable register(final Supplier<KafkaClientMetrics> metricsProvider) {
     final var registry = getRegistry();
-    if (registry != null) {
-      final var clientMetrics = new KafkaClientMetrics(producer);
-      clientMetrics.bindTo(registry);
-      return clientMetrics;
+    if (registry != null && !DISABLE_KAFKA_CLIENTS_METRICS) {
+      final var clientMetrics = metricsProvider.get();
+      try {
+        // The binding and close process is blocking, so execute them asynchronously.
+        meterBinderExecutor.execute(() -> clientMetrics.bindTo(registry));
+
+        return () -> {
+          final Promise<Void> p = Promise.promise();
+          meterBinderExecutor.execute(() -> {
+            clientMetrics.close();
+            p.complete();
+          });
+          return p.future();
+        };
+
+      } catch (final RejectedExecutionException ex) {
+        // if this task cannot be accepted for execution when the executor has been shutdown.
+        logger.warn("Failed to bind metrics for Kafka client", ex);
+      }
     }
-    return () -> {};
+    return Future::succeededFuture;
   }
 
   public static PemKeyCertOptions permKeyCertOptions() {
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/ResponseToKafkaTopicHandler.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/ResponseToKafkaTopicHandler.java
index 86fec03cb..0c9e6d1a4 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/ResponseToKafkaTopicHandler.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/ResponseToKafkaTopicHandler.java
@@ -38,7 +38,7 @@ public final class ResponseToKafkaTopicHandler extends BaseResponseHandler imple
 
   private final String topic;
   private final KafkaProducer<String, CloudEvent> producer;
-  private final AutoCloseable producerMeterBinder;
+  private final AsyncCloseable producerMeterBinder;
 
   private int inFlightEvents = 0;
   private boolean closed = false;
@@ -103,10 +103,9 @@ public final class ResponseToKafkaTopicHandler extends BaseResponseHandler imple
       closePromise.tryComplete(null);
     }
 
-    final Function<Void, Future<Void>> closeF = v -> CompositeFuture.all(
-      this.producer.close(),
-      AsyncCloseable.wrapAutoCloseable(producerMeterBinder).close()
-    ).mapEmpty();
+    final Function<Void, Future<Void>> closeF = v -> AsyncCloseable
+      .compose(producerMeterBinder, this.producer::close)
+      .close();
 
     return closePromise.future()
       .compose(
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 c27ef8402..e3fb8d753 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
@@ -128,9 +128,14 @@ public final class OffsetManager implements RecordDispatcherListener {
   }
 
   private void commit(final KafkaConsumerRecord<?, ?> record) {
-    this.offsetTrackers
-      .get(new TopicPartition(record.topic(), record.partition()))
-      .recordNewOffset(record.offset());
+    // We need to handle the case when the offset tracker was removed from our Map since when partitions are revoked we
+    // remove the associated offset tracker, however, we may get a response from the sink for a previously owned
+    // partition after a partition has been revoked.
+    // Note: it's not possible to commit offsets of partitions that this a particular consumer instance doesn't own.
+    final var ot = this.offsetTrackers.get(new TopicPartition(record.topic(), record.partition()));
+    if (ot != null) {
+      ot.recordNewOffset(record.offset());
+    }
   }
 
   private synchronized Future<Void> commit(final TopicPartition topicPartition, final OffsetTracker tracker) {
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 223e23053..b82665705 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
@@ -17,7 +17,6 @@ package dev.knative.eventing.kafka.broker.dispatcher.main;
 
 import dev.knative.eventing.kafka.broker.contract.DataPlaneContract;
 import dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressConfig;
-import dev.knative.eventing.kafka.broker.core.AsyncCloseable;
 import dev.knative.eventing.kafka.broker.core.metrics.Metrics;
 import dev.knative.eventing.kafka.broker.core.security.AuthProvider;
 import dev.knative.eventing.kafka.broker.core.security.KafkaClientsAuth;
@@ -48,7 +47,6 @@ import dev.knative.eventing.kafka.broker.dispatcher.impl.filter.subscriptionsapi
 import dev.knative.eventing.kafka.broker.dispatcher.impl.filter.subscriptionsapi.SuffixFilter;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.http.WebClientCloudEventSender;
 import io.cloudevents.CloudEvent;
-import io.micrometer.core.instrument.Counter;
 import io.micrometer.core.instrument.MeterRegistry;
 import io.vertx.circuitbreaker.CircuitBreaker;
 import io.vertx.circuitbreaker.CircuitBreakerOptions;
@@ -62,6 +60,10 @@ import io.vertx.kafka.client.common.KafkaClientOptions;
 import io.vertx.kafka.client.common.tracing.ConsumerTracer;
 import io.vertx.kafka.client.consumer.KafkaConsumer;
 import io.vertx.kafka.client.producer.KafkaProducer;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.AbstractMap.SimpleImmutableEntry;
 import java.util.HashMap;
@@ -76,11 +78,6 @@ import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
-import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.clients.producer.ProducerConfig;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import static dev.knative.eventing.kafka.broker.core.utils.Logging.keyValue;
 
 public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
@@ -159,7 +156,7 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
        KafkaClientsAuth.attachCredentials(producerConfigs, credentials);
 
        final KafkaConsumer<Object, CloudEvent> consumer = createConsumer(vertx, consumerConfigs);
-       AutoCloseable metricsCloser = Metrics.register(consumer.unwrap());
+       final var metricsCloser = Metrics.register(consumer.unwrap());
 
        final var egressConfig =
          egress.hasEgressConfig() ?
@@ -208,7 +205,7 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
        // Set all the built objects in the consumer verticle
        consumerVerticle.setRecordDispatcher(recordDispatcher);
        consumerVerticle.setConsumer(consumer);
-       consumerVerticle.setCloser(AsyncCloseable.wrapAutoCloseable(metricsCloser));
+       consumerVerticle.setCloser(metricsCloser);
      })
        .mapEmpty();
 
diff --git a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStore.java b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStore.java
index 5e5146c2b..88832b4c5 100644
--- a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStore.java
+++ b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStore.java
@@ -137,14 +137,14 @@ public class IngressProducerReconcilableStore implements IngressReconcilerListen
         producerProps
       );
 
-      if(isRootPath(ingress.getPath()) && Strings.isNullOrEmpty(ingress.getHost())){
+      if (isRootPath(ingress.getPath()) && Strings.isNullOrEmpty(ingress.getHost())) {
         throw new IllegalArgumentException("Ingress path and host is blank. One of them should be defined. Resource UID: " + resource.getUid());
       }
 
-      if(!isRootPath(ingress.getPath())){
+      if (!isRootPath(ingress.getPath())) {
         this.pathMapper.put(ingress.getPath(), ingressInfo);
       }
-      if (!Strings.isNullOrEmpty(ingress.getHost())){
+      if (!Strings.isNullOrEmpty(ingress.getHost())) {
         this.hostMapper.put(ingress.getHost(), ingressInfo);
       }
       this.ingressInfos.put(resource.getUid(), ingressInfo);
@@ -183,20 +183,20 @@ public class IngressProducerReconcilableStore implements IngressReconcilerListen
       return rc.getValue().close()
         .onSuccess(r -> {
           // Remove ingress info from the maps
-          if(!isRootPath(ingressInfo.getPath())){
+          if (!isRootPath(ingressInfo.getPath())) {
             this.pathMapper.remove(ingressInfo.getPath());
           }
-          if(!Strings.isNullOrEmpty(ingressInfo.getHost())){
+          if (!Strings.isNullOrEmpty(ingressInfo.getHost())) {
             this.hostMapper.remove(ingressInfo.getHost());
           }
           this.ingressInfos.remove(resource.getUid());
         });
     }
     // Remove ingress info from the maps
-    if(!isRootPath(ingressInfo.getPath())){
+    if (!isRootPath(ingressInfo.getPath())) {
       this.pathMapper.remove(ingressInfo.getPath());
     }
-    if(!Strings.isNullOrEmpty(ingressInfo.getHost())){
+    if (!Strings.isNullOrEmpty(ingressInfo.getHost())) {
       this.hostMapper.remove(ingressInfo.getHost());
     }
     this.ingressInfos.remove(resource.getUid());
@@ -217,7 +217,7 @@ public class IngressProducerReconcilableStore implements IngressReconcilerListen
     private static final Logger logger = LoggerFactory.getLogger(ProducerHolder.class);
 
     private final KafkaProducer<String, CloudEvent> producer;
-    private final AutoCloseable producerMeterBinder;
+    private final AsyncCloseable producerMeterBinder;
 
     ProducerHolder(final KafkaProducer<String, CloudEvent> producer) {
       this.producer = producer;
@@ -241,10 +241,9 @@ public class IngressProducerReconcilableStore implements IngressReconcilerListen
     }
 
     private Future<Void> closeNow() {
-      return AsyncCloseable.compose(
-        producer::close,
-        AsyncCloseable.wrapAutoCloseable(this.producerMeterBinder)
-      ).close();
+      return AsyncCloseable
+        .compose(this.producerMeterBinder, producer::close)
+        .close();
     }
   }
 
diff --git a/go.mod b/go.mod
index abe0d88e3..fbee388dd 100644
--- a/go.mod
+++ b/go.mod
@@ -23,10 +23,10 @@ require (
 	k8s.io/apiserver v0.23.4
 	k8s.io/client-go v0.23.5
 	k8s.io/utils v0.0.0-20220210201930-3a6ce19ff2f9
-	knative.dev/eventing v0.31.1-0.20220509111632-59e25b88478f
-	knative.dev/eventing-kafka v0.31.1-0.20220509113232-9885a361e09d
-	knative.dev/hack v0.0.0-20220506160929-a8076b0510ed
-	knative.dev/pkg v0.0.0-20220505013700-a8b7d99374a3
-	knative.dev/reconciler-test v0.0.0-20220505183729-e295cedfd97b
+	knative.dev/eventing v0.31.1-0.20220512085854-005d6b6127da
+	knative.dev/eventing-kafka v0.31.1-0.20220512052054-8e0c94a88836
+	knative.dev/hack v0.0.0-20220512014059-f4972b4daff9
+	knative.dev/pkg v0.0.0-20220512013937-2d8305b2e59a
+	knative.dev/reconciler-test v0.0.0-20220512014341-24984453a0c8
 	sigs.k8s.io/yaml v1.3.0
 )
diff --git a/go.sum b/go.sum
index e67b2ec9e..a064a778f 100644
--- a/go.sum
+++ b/go.sum
@@ -1392,30 +1392,27 @@ k8s.io/utils v0.0.0-20210802155522-efc7438f0176/go.mod h1:jPW/WVKK9YHAvNhRxK0md/
 k8s.io/utils v0.0.0-20211116205334-6203023598ed/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 k8s.io/utils v0.0.0-20220210201930-3a6ce19ff2f9 h1:HNSDgDCrr/6Ly3WEGKZftiE7IY19Vz2GdbOCyI4qqhc=
 k8s.io/utils v0.0.0-20220210201930-3a6ce19ff2f9/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
-knative.dev/control-protocol v0.0.0-20220428151655-bec9426db2df/go.mod h1:RDxYU9MRrP2+upLtVVjqD/8aq3CFZiBsHqqar8iIiac=
-knative.dev/eventing v0.31.1-0.20220428204853-01f56122bf2a/go.mod h1:jdXjZKlLMwEj2KFKatzbqKWyOJAjUtn3072V3SkKIGI=
-knative.dev/eventing v0.31.1-0.20220509111632-59e25b88478f h1:CwMRHnr1hASYqMyvMqhWRHeKSoHXAK4BI+HAGMhN36Q=
-knative.dev/eventing v0.31.1-0.20220509111632-59e25b88478f/go.mod h1:pZVuKFUqxDEcfzqimrzJvftUXx3W0lGwkPyMMq7L2O0=
-knative.dev/eventing-kafka v0.31.1-0.20220509113232-9885a361e09d h1:ICERZgWG9WMp7KrJR/2W35hsP0sDVuJ1KTrDCJXBk24=
-knative.dev/eventing-kafka v0.31.1-0.20220509113232-9885a361e09d/go.mod h1:PM2C2rSVgTDIDdZPhhBZdyXj8tpaYceY5mrKCZgSWAw=
-knative.dev/hack v0.0.0-20220411131823-6ffd8417de7c/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
+knative.dev/control-protocol v0.0.0-20220512014117-d1ec88ff4436/go.mod h1:98BfMAI1Nj+1JnmWRFADhYecJswQpIH/9NVR/mpECTs=
+knative.dev/eventing v0.31.1-0.20220512013937-02a6fa0e9ee3/go.mod h1:aqiJ8eEXtXBz3vERz/2lbkRjA1jbG6yXujaEGHqKLgs=
+knative.dev/eventing v0.31.1-0.20220512085854-005d6b6127da h1:cs6blaapfYXxTmT44/Oez/RRe4Eo5VIA7jSJGXHcGRg=
+knative.dev/eventing v0.31.1-0.20220512085854-005d6b6127da/go.mod h1:aqiJ8eEXtXBz3vERz/2lbkRjA1jbG6yXujaEGHqKLgs=
+knative.dev/eventing-kafka v0.31.1-0.20220512052054-8e0c94a88836 h1:SD5K1AlUVvy8fb9f7IahVH8O87gJg0vLa/CIgWnNRPE=
+knative.dev/eventing-kafka v0.31.1-0.20220512052054-8e0c94a88836/go.mod h1:j1y7gx2uV0J2WD7RC9MiddTnF1/pLLYK/jlHwALcjfc=
 knative.dev/hack v0.0.0-20220427014036-5f473869d377/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack v0.0.0-20220503220458-46c77f157e20/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack v0.0.0-20220505013956-ccf82ad03f32/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/hack v0.0.0-20220506160929-a8076b0510ed h1:6mFHY1IFMRnO9O1z9q6xL9mAy6Gpm8SmRQ2KbahXKz0=
 knative.dev/hack v0.0.0-20220506160929-a8076b0510ed/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/hack/schema v0.0.0-20220427014036-5f473869d377/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
-knative.dev/hack/schema v0.0.0-20220503220458-46c77f157e20/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
-knative.dev/pkg v0.0.0-20220427013826-1f681e126af6/go.mod h1:j2MeD8s+JoCu1vegX80GbRXV/xd20Jm1NznxBYtVXiM=
+knative.dev/hack v0.0.0-20220512014059-f4972b4daff9 h1:/KjTyToOXNMD6SyBzDMnXgMBhsAePkJHY5uhpK+rYOc=
+knative.dev/hack v0.0.0-20220512014059-f4972b4daff9/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
+knative.dev/hack/schema v0.0.0-20220506160929-a8076b0510ed/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
 knative.dev/pkg v0.0.0-20220427171752-2d552be030f6/go.mod h1:NXK3p/UMCbFybBM9xQGii3TuMN/WKHByXcYsTwZ6Y6U=
-knative.dev/pkg v0.0.0-20220502014956-6ec9c1a62fa8/go.mod h1:NXK3p/UMCbFybBM9xQGii3TuMN/WKHByXcYsTwZ6Y6U=
-knative.dev/pkg v0.0.0-20220504135159-0719b405eca5/go.mod h1:mRzBIY8eoQurpADZ4+3LzNuucKOBhsSvhX6lXqMVpIc=
-knative.dev/pkg v0.0.0-20220505013700-a8b7d99374a3 h1:odf5hudTbupqdYDsxeqofZtMsZuU0FqPZqWI6xETYIA=
 knative.dev/pkg v0.0.0-20220505013700-a8b7d99374a3/go.mod h1:mRzBIY8eoQurpADZ4+3LzNuucKOBhsSvhX6lXqMVpIc=
-knative.dev/reconciler-test v0.0.0-20220427134652-f14902d1c7a5/go.mod h1:IZTJfC1sgr70/39NIne/MmaNbDWRHwT6cWvJB3AeHIU=
-knative.dev/reconciler-test v0.0.0-20220428151655-74b18404decf/go.mod h1:hSnnO5QjvMwH8E83UlL81jNFA9HWZqG3qEChPz85GDY=
-knative.dev/reconciler-test v0.0.0-20220505183729-e295cedfd97b h1:9sDSbOr5QZzg56Mo54DWRBrHddjRj8kLSf7pagNBLR4=
+knative.dev/pkg v0.0.0-20220512013937-2d8305b2e59a h1:lVm5ZtTrQZg3yid38zu13QgXN02clVzPkMTEyB9FRUo=
+knative.dev/pkg v0.0.0-20220512013937-2d8305b2e59a/go.mod h1:mRzBIY8eoQurpADZ4+3LzNuucKOBhsSvhX6lXqMVpIc=
+knative.dev/reconciler-test v0.0.0-20220505014032-dbe7d3721ba3/go.mod h1:hSnnO5QjvMwH8E83UlL81jNFA9HWZqG3qEChPz85GDY=
 knative.dev/reconciler-test v0.0.0-20220505183729-e295cedfd97b/go.mod h1:blANTpykbOpFykGxucL/8RkU8hH4WmZRWaxifP8heF4=
+knative.dev/reconciler-test v0.0.0-20220512014341-24984453a0c8 h1:blFrJrEGUg80Cq7inJotD3DKrtLgcLbYV5AiQjopV58=
+knative.dev/reconciler-test v0.0.0-20220512014341-24984453a0c8/go.mod h1:blANTpykbOpFykGxucL/8RkU8hH4WmZRWaxifP8heF4=
 modernc.org/cc v1.0.0/go.mod h1:1Sk4//wdnYJiUIxnW8ddKpaOJCF37yAdqYnkxUpaYxw=
 modernc.org/golex v1.0.0/go.mod h1:b/QX9oBD/LhixY6NDh+IdGv17hgB+51fET1i2kPSmvk=
 modernc.org/mathutil v1.0.0/go.mod h1:wU0vUrJsVWBZ4P6e7xtFJEhFSNsfRLJ8H458uRjg03k=
diff --git a/proto/contract.proto b/proto/contract.proto
index b1f69010d..8e1dd0aac 100644
--- a/proto/contract.proto
+++ b/proto/contract.proto
@@ -148,6 +148,9 @@ message Egress {
 
   // CNCF CloudEvents SubscriptionsAPI compliant filters
   repeated DialectedFilter dialectedFilter = 12;
+
+  // Number of virtual replicas.
+  int32 vReplicas = 13;
 }
 
 // CloudEvent content mode
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 1d1906521..2f25a65bd 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1150,7 +1150,7 @@ k8s.io/utils/internal/third_party/forked/golang/net
 k8s.io/utils/net
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.31.1-0.20220509111632-59e25b88478f
+# knative.dev/eventing v0.31.1-0.20220512085854-005d6b6127da
 ## explicit
 knative.dev/eventing/pkg/apis/config
 knative.dev/eventing/pkg/apis/duck
@@ -1298,7 +1298,7 @@ knative.dev/eventing/test/upgrade/prober/wathola/fetcher
 knative.dev/eventing/test/upgrade/prober/wathola/forwarder
 knative.dev/eventing/test/upgrade/prober/wathola/receiver
 knative.dev/eventing/test/upgrade/prober/wathola/sender
-# knative.dev/eventing-kafka v0.31.1-0.20220509113232-9885a361e09d
+# knative.dev/eventing-kafka v0.31.1-0.20220512052054-8e0c94a88836
 ## explicit
 knative.dev/eventing-kafka/pkg/apis/bindings
 knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1
@@ -1363,11 +1363,11 @@ knative.dev/eventing-kafka/test/rekt/resources/resetoffset
 knative.dev/eventing-kafka/test/upgrade
 knative.dev/eventing-kafka/test/upgrade/continual
 knative.dev/eventing-kafka/test/upgrade/installation
-# knative.dev/hack v0.0.0-20220506160929-a8076b0510ed
+# knative.dev/hack v0.0.0-20220512014059-f4972b4daff9
 ## explicit
 knative.dev/hack
 knative.dev/hack/shell
-# knative.dev/pkg v0.0.0-20220505013700-a8b7d99374a3
+# knative.dev/pkg v0.0.0-20220512013937-2d8305b2e59a
 ## explicit
 knative.dev/pkg/apiextensions/storageversion
 knative.dev/pkg/apiextensions/storageversion/cmd/migrate
@@ -1470,7 +1470,7 @@ knative.dev/pkg/webhook/json
 knative.dev/pkg/webhook/resourcesemantics
 knative.dev/pkg/webhook/resourcesemantics/defaulting
 knative.dev/pkg/webhook/resourcesemantics/validation
-# knative.dev/reconciler-test v0.0.0-20220505183729-e295cedfd97b
+# knative.dev/reconciler-test v0.0.0-20220512014341-24984453a0c8
 ## explicit
 knative.dev/reconciler-test/cmd/eventshub
 knative.dev/reconciler-test/pkg/environment


Test patch:
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index 7acf64112..a6dcec5e2 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -22,7 +22,6 @@ import (
 	"net/url"
 	"testing"
 
-	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/utils/pointer"
 	"knative.dev/pkg/network"
 
@@ -1772,23 +1771,6 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 	useTable(t, table, &env)
 }
 
-func ConfigMapFinalizerUpdateRemove(name string, opts ...CMOption) clientgotesting.UpdateActionImpl {
-	return clientgotesting.NewUpdateAction(
-		schema.GroupVersionResource{
-			Group:    "*",
-			Version:  "v1",
-			Resource: "ConfigMap",
-		},
-		ConfigMapNamespace,
-		BrokerConfig(bootstrapServers, 20, 5,
-			append(
-				[]CMOption{BrokerConfigFinalizerRemove(name)},
-				opts...,
-			)...,
-		),
-	)
-}
-
 func TestBrokerFinalizer(t *testing.T) {
 	t.Parallel()
 
@@ -1808,9 +1790,7 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 			Name: "Reconciled normal - no DLS",
 			Objects: []runtime.Object{
 				NewDeletedBroker(),
-				BrokerConfig(bootstrapServers, 20, 5,
-					BrokerConfigFinalizer(ConfigMapFinalizerName),
-				),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -1828,7 +1808,6 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 					Resources:  []*contract.Resource{},
 					Generation: 2,
 				}),
-				ConfigMapFinalizerUpdateRemove(ConfigMapFinalizerName),
 			},
 			OtherTestData: map[string]interface{}{
 				testProber: probertesting.MockProber(prober.StatusNotReady),
@@ -1866,9 +1845,7 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 			Name: "Reconciled failed - probe not ready",
 			Objects: []runtime.Object{
 				NewDeletedBroker(),
-				BrokerConfig(bootstrapServers, 20, 5,
-					BrokerConfigFinalizer(ConfigMapFinalizerName),
-				),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -1898,9 +1875,7 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 				NewDeletedBroker(
 					WithDelivery(),
 				),
-				BrokerConfig(bootstrapServers, 20, 5,
-					BrokerConfigFinalizer(ConfigMapFinalizerName),
-				),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -1918,7 +1893,6 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 				ConfigMapUpdate(&env, &contract.Contract{
 					Generation: 2,
 				}),
-				ConfigMapFinalizerUpdateRemove(ConfigMapFinalizerName),
 			},
 			OtherTestData: map[string]interface{}{
 				testProber: probertesting.MockProber(prober.StatusNotReady),
@@ -1928,9 +1902,7 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 			Name: "Failed to delete topic",
 			Objects: []runtime.Object{
 				NewDeletedBroker(),
-				BrokerConfig(bootstrapServers, 20, 5,
-					BrokerConfigFinalizer(ConfigMapFinalizerName),
-				),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -1968,15 +1940,11 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 				NewDeletedBroker(
 					WithDelivery(),
 				),
-				BrokerConfig(bootstrapServers, 20, 5,
-					BrokerConfigFinalizer(ConfigMapFinalizerName),
-				),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewService(),
 			},
-			Key: testKey,
-			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapFinalizerUpdateRemove(ConfigMapFinalizerName),
-			},
+			Key:         testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{},
 			WantCreates: []runtime.Object{
 				NewConfigMapWithBinaryData(&env, nil),
 			},
@@ -1989,9 +1957,7 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 			Name: "Reconciled normal - preserve config map previous state",
 			Objects: []runtime.Object{
 				NewDeletedBroker(),
-				BrokerConfig(bootstrapServers, 20, 5,
-					BrokerConfigFinalizer(ConfigMapFinalizerName),
-				),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -2020,7 +1986,6 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 					},
 					Generation: 6,
 				}),
-				ConfigMapFinalizerUpdateRemove(ConfigMapFinalizerName),
 			},
 			OtherTestData: map[string]interface{}{
 				testProber: probertesting.MockProber(prober.StatusNotReady),
@@ -2030,9 +1995,7 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 			Name: "Reconciled normal - topic doesn't exist",
 			Objects: []runtime.Object{
 				NewDeletedBroker(),
-				BrokerConfig(bootstrapServers, 20, 5,
-					BrokerConfigFinalizer(ConfigMapFinalizerName),
-				),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -2061,7 +2024,6 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 					},
 					Generation: 6,
 				}),
-				ConfigMapFinalizerUpdateRemove(ConfigMapFinalizerName),
 			},
 			OtherTestData: map[string]interface{}{
 				wantErrorOnDeleteTopic: sarama.ErrUnknownTopicOrPartition,
@@ -2072,9 +2034,7 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 			Name: "Reconciled normal - no broker found in config map",
 			Objects: []runtime.Object{
 				NewDeletedBroker(),
-				BrokerConfig(bootstrapServers, 20, 5,
-					BrokerConfigFinalizer(ConfigMapFinalizerName),
-				),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -2086,10 +2046,8 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 					Generation: 5,
 				}, &env),
 			},
-			Key: testKey,
-			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapFinalizerUpdateRemove(ConfigMapFinalizerName),
-			},
+			Key:         testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{},
 			OtherTestData: map[string]interface{}{
 				testProber: probertesting.MockProber(prober.StatusNotReady),
 			},
diff --git a/control-plane/pkg/reconciler/consumer/consumer_test.go b/control-plane/pkg/reconciler/consumer/consumer_test.go
index fe58c2f20..e81d34e7e 100644
--- a/control-plane/pkg/reconciler/consumer/consumer_test.go
+++ b/control-plane/pkg/reconciler/consumer/consumer_test.go
@@ -115,6 +115,7 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           ConsumerUUID,
 								DeliveryOrder: contract.DeliveryOrder_UNORDERED,
 								KeyType:       0,
+								VReplicas:     1,
 								Reference: &contract.Reference{
 									Uuid:      SourceUUID,
 									Namespace: ConsumerNamespace,
@@ -166,6 +167,116 @@ func TestReconcileKind(t *testing.T) {
 				},
 			},
 		},
+		{
+			Name: "Reconciled normal - multiple replicas",
+			Objects: []runtime.Object{
+				NewService(),
+				NewConsumerGroup(ConsumerGroupOwnerRef(SourceAsOwnerReference())),
+				NewDispatcherPod("p1", PodRunning()),
+				NewConsumer(1,
+					ConsumerUID(ConsumerUUID),
+					ConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(SourceTopics...),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(SourceBootstrapServers),
+							ConsumerGroupIdConfig(SourceConsumerGroup),
+						),
+						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerVReplicas(2),
+						ConsumerPlacement(kafkainternals.PodBind{PodName: "p1", PodNamespace: SystemNamespace}),
+					)),
+					ConsumerOwnerRef(ConsumerGroupAsOwnerRef()),
+				),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantCreates: []runtime.Object{
+				NewConfigMapWithBinaryData(&config.Env{
+					DataPlaneConfigFormat:       base.Json,
+					DataPlaneConfigMapNamespace: SystemNamespace,
+					DataPlaneConfigMapName:      "p1",
+				}, []byte(""),
+					DispatcherPodAsOwnerReference("p1"),
+				),
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&config.Env{
+					DataPlaneConfigFormat:       base.Json,
+					DataPlaneConfigMapNamespace: SystemNamespace,
+					DataPlaneConfigMapName:      "p1",
+				}, &contract.Contract{
+					Generation: 1,
+					Resources: []*contract.Resource{
+						{
+							Uid:              ConsumerUUID,
+							Topics:           SourceTopics,
+							BootstrapServers: SourceBootstrapServers,
+							Egresses: []*contract.Egress{{
+								ConsumerGroup: SourceConsumerGroup,
+								Destination:   ServiceURL,
+								ReplyStrategy: nil,
+								Filter:        nil,
+								Uid:           ConsumerUUID,
+								DeliveryOrder: contract.DeliveryOrder_UNORDERED,
+								KeyType:       0,
+								VReplicas:     2,
+								Reference: &contract.Reference{
+									Uuid:      SourceUUID,
+									Namespace: ConsumerNamespace,
+									Name:      SourceName,
+								},
+							}},
+							Auth:                nil,
+							CloudEventOverrides: nil,
+							Reference: &contract.Reference{
+								Uuid:      SourceUUID,
+								Namespace: ConsumerNamespace,
+								Name:      SourceName,
+							},
+						},
+					},
+				},
+					DispatcherPodAsOwnerReference("p1"),
+				),
+				{Object: NewDispatcherPod("p1",
+					PodRunning(),
+					PodAnnotations(map[string]string{
+						base.VolumeGenerationAnnotationKey: "1",
+					}),
+				)},
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: func() runtime.Object {
+						c := NewConsumer(1,
+							ConsumerUID(ConsumerUUID),
+							ConsumerSpec(NewConsumerSpec(
+								ConsumerTopics(SourceTopics...),
+								ConsumerConfigs(
+									ConsumerBootstrapServersConfig(SourceBootstrapServers),
+									ConsumerGroupIdConfig(SourceConsumerGroup),
+								),
+								ConsumerSubscriber(NewSourceSinkReference()),
+								ConsumerVReplicas(2),
+								ConsumerPlacement(kafkainternals.PodBind{PodName: "p1", PodNamespace: SystemNamespace}),
+							)),
+							ConsumerOwnerRef(ConsumerGroupAsOwnerRef()),
+						)
+						c.GetConditionSet().Manage(c.GetStatus()).InitializeConditions()
+						c.MarkReconcileContractSucceeded()
+						c.MarkBindSucceeded()
+						c.Status.SubscriberURI, _ = apis.ParseURL(ServiceURL)
+						return c
+					}(),
+				},
+			},
+		},
 		{
 			Name: "Reconciled normal - consumer delivery",
 			Objects: []runtime.Object{
@@ -230,6 +341,7 @@ func TestReconcileKind(t *testing.T) {
 								ReplyStrategy: nil,
 								Filter:        nil,
 								Uid:           ConsumerUUID,
+								VReplicas:     1,
 								EgressConfig: &contract.EgressConfig{
 									DeadLetter:    ConsumerDeadLetterSinkURI.String(),
 									Retry:         10,
diff --git a/control-plane/pkg/reconciler/source/v2/sourcev2_test.go b/control-plane/pkg/reconciler/source/v2/sourcev2_test.go
index 1e02418ee..407aeca03 100644
--- a/control-plane/pkg/reconciler/source/v2/sourcev2_test.go
+++ b/control-plane/pkg/reconciler/source/v2/sourcev2_test.go
@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"testing"
 
+	"github.com/google/go-cmp/cmp"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -48,6 +49,36 @@ import (
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/testing"
 )
 
+func TestGetLabels(t *testing.T) {
+
+	testLabels := GetLabels("testSourceName")
+
+	wantLabels := map[string]string{
+		"eventing.knative.dev/source":     "kafka-source-controller",
+		"eventing.knative.dev/sourceName": "testSourceName",
+	}
+
+	eq := cmp.Equal(testLabels, wantLabels)
+	if !eq {
+		t.Fatalf("%v is not equal to %v", testLabels, wantLabels)
+	}
+}
+
+func TestGetLabelsAsSelector(t *testing.T) {
+
+	testLabels, err := GetLabelsAsSelector("testSourceName")
+	if err != nil {
+		t.Fatalf("Unable to get labels as selector")
+	}
+	testLabelsString := testLabels.String()
+
+	wantLabels := "eventing.knative.dev/source=kafka-source-controller,eventing.knative.dev/sourceName=testSourceName"
+
+	if testLabelsString != wantLabels {
+		t.Fatalf("%v is not equal to %v", testLabels, wantLabels)
+	}
+}
+
 func TestReconcileKind(t *testing.T) {
 
 	testKey := fmt.Sprintf("%s/%s", SourceNamespace, SourceName)
@@ -95,6 +126,7 @@ func TestReconcileKind(t *testing.T) {
 					Object: NewSource(
 						StatusSourceConsumerGroupUnknown(),
 						StatusSourceSinkResolved(""),
+						StatusSourceSelector(),
 					),
 				},
 			},
@@ -222,6 +254,7 @@ func TestReconcileKind(t *testing.T) {
 						StatusSourceConsumerGroupUnknown(),
 						StatusSourceSinkResolved(""),
 						SourceNetSaslTls(true),
+						StatusSourceSelector(),
 					),
 				},
 			},
@@ -333,6 +366,7 @@ func TestReconcileKind(t *testing.T) {
 						StatusSourceConsumerGroupUnknown(),
 						StatusSourceSinkResolved(""),
 						SourceNetSaslTls(false),
+						StatusSourceSelector(),
 					),
 				},
 			},
@@ -386,6 +420,7 @@ func TestReconcileKind(t *testing.T) {
 						}),
 						StatusSourceConsumerGroupUnknown(),
 						StatusSourceSinkResolved(""),
+						StatusSourceSelector(),
 					),
 				},
 			},
@@ -462,6 +497,7 @@ func TestReconcileKind(t *testing.T) {
 						WithSourceSink(NewSourceSink2Reference()),
 						StatusSourceConsumerGroup(),
 						StatusSourceSinkResolved(""),
+						StatusSourceSelector(),
 					),
 				},
 			},
@@ -518,6 +554,7 @@ func TestReconcileKind(t *testing.T) {
 					Object: NewSource(
 						StatusSourceConsumerGroup(),
 						StatusSourceSinkResolved(""),
+						StatusSourceSelector(),
 					),
 				},
 			},
@@ -573,6 +610,7 @@ func TestReconcileKind(t *testing.T) {
 					Object: NewSource(
 						StatusSourceConsumerGroupUnknown(),
 						StatusSourceSinkResolved(""),
+						StatusSourceSelector(),
 					),
 				},
 			},
@@ -616,6 +654,7 @@ func TestReconcileKind(t *testing.T) {
 					Object: NewSource(
 						StatusSourceConsumerGroup(),
 						StatusSourceSinkResolved(""),
+						StatusSourceSelector(),
 					),
 				},
 			},
@@ -659,6 +698,7 @@ func TestReconcileKind(t *testing.T) {
 					Object: NewSource(
 						StatusSourceConsumerGroupUnknown(),
 						StatusSourceSinkResolved(""),
+						StatusSourceSelector(),
 					),
 				},
 			},
@@ -703,6 +743,7 @@ func TestReconcileKind(t *testing.T) {
 					Object: NewSource(
 						StatusSourceConsumerGroupFailed("failed", "failed"),
 						StatusSourceSinkResolved(""),
+						StatusSourceSelector(),
 					),
 				},
 			},
@@ -749,6 +790,7 @@ func TestReconcileKind(t *testing.T) {
 						StatusSourceConsumerGroup(),
 						StatusSourceSinkResolved(""),
 						StatusSourceConsumerGroupReplicas(1),
+						StatusSourceSelector(),
 					),
 				},
 			},
@@ -774,6 +816,13 @@ func TestReconcileKind(t *testing.T) {
 	}))
 }
 
+func StatusSourceSelector() KRShapedOption {
+	return func(obj duckv1.KRShaped) {
+		ks := obj.(*sources.KafkaSource)
+		ks.Status.Selector = "eventing.knative.dev/source=kafka-source-controller,eventing.knative.dev/sourceName=" + ks.Name
+	}
+}
+
 func StatusSourceConsumerGroup() KRShapedOption {
 	return func(obj duckv1.KRShaped) {
 		ks := obj.(*sources.KafkaSource)
diff --git a/control-plane/pkg/reconciler/testing/objects_broker.go b/control-plane/pkg/reconciler/testing/objects_broker.go
index 8e57cb46a..2ea14d1f9 100644
--- a/control-plane/pkg/reconciler/testing/objects_broker.go
+++ b/control-plane/pkg/reconciler/testing/objects_broker.go
@@ -24,7 +24,6 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
-	"k8s.io/apimachinery/pkg/util/sets"
 	clientgotesting "k8s.io/client-go/testing"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
@@ -50,8 +49,6 @@ const (
 
 	TriggerName      = "test-trigger"
 	TriggerNamespace = "test-namespace"
-
-	ConfigMapFinalizerName = "kafka.brokers.eventing.knative.dev/" + BrokerNamespace + "-" + BrokerName
 )
 
 func BrokerTopic() string {
@@ -155,18 +152,6 @@ func BrokerConfig(bootstrapServers string, numPartitions, replicationFactor int,
 	return cm
 }
 
-func BrokerConfigFinalizer(finalizerName string) CMOption {
-	return func(cm *corev1.ConfigMap) {
-		cm.Finalizers = append(cm.Finalizers, finalizerName)
-	}
-}
-
-func BrokerConfigFinalizerRemove(finalizerName string) CMOption {
-	return func(cm *corev1.ConfigMap) {
-		cm.Finalizers = sets.NewString(cm.Finalizers...).Delete(finalizerName).List()
-	}
-}
-
 func BrokerAuthConfig(name string) CMOption {
 	return func(cm *corev1.ConfigMap) {
 		if cm.Data == nil {
diff --git a/test/e2e_new/resources/kafkasink/kafkasink_test.go b/test/e2e_new/resources/kafkasink/kafkasink_test.go
index 54427418a..2cf75fabf 100644
--- a/test/e2e_new/resources/kafkasink/kafkasink_test.go
+++ b/test/e2e_new/resources/kafkasink/kafkasink_test.go
@@ -17,6 +17,7 @@
 package kafkasink_test
 
 import (
+	"embed"
 	"os"
 
 	eventingv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
@@ -25,6 +26,9 @@ import (
 	"knative.dev/reconciler-test/pkg/manifest"
 )
 
+//go:embed *.yaml
+var yaml embed.FS
+
 // The following examples validate the processing of the With* helper methods
 // applied to config and go template parser.
 
@@ -37,7 +41,7 @@ func Example_zero() {
 		"bootstrapServers": []string{"my-bootstrap-server:8082"},
 	}
 
-	files, err := manifest.ExecuteLocalYAML(images, cfg)
+	files, err := manifest.ExecuteYAML(yaml, images, cfg)
 	if err != nil {
 		panic(err)
 	}
@@ -68,7 +72,7 @@ func Example_full() {
 	kafkasink.WithNumPartitions(10)(cfg)
 	kafkasink.WithAuthSecretName("abc")(cfg)
 
-	files, err := manifest.ExecuteLocalYAML(images, cfg)
+	files, err := manifest.ExecuteYAML(yaml, images, cfg)
 	if err != nil {
 		panic(err)
 	}
diff --git a/vendor/knative.dev/eventing-kafka/test/e2e-common.sh b/vendor/knative.dev/eventing-kafka/test/e2e-common.sh
index ccf2e87a2..6f0a93aa7 100644
--- a/vendor/knative.dev/eventing-kafka/test/e2e-common.sh
+++ b/vendor/knative.dev/eventing-kafka/test/e2e-common.sh
@@ -73,11 +73,11 @@ readonly EVENTING_KAFKA_SECRET_TEMPLATE="300-kafka-secret.yaml"
 readonly EVENTING_KAFKA_CONFIG_TEMPLATE="300-eventing-kafka-configmap.yaml"
 
 # Strimzi installation config template used for starting up Kafka clusters.
-readonly STRIMZI_INSTALLATION_CONFIG_TEMPLATE="test/config/100-strimzi-cluster-operator-0.27.0.yaml"
+readonly STRIMZI_INSTALLATION_CONFIG_TEMPLATE="test/config/100-strimzi-cluster-operator-0.28.0.yaml"
 # Strimzi installation config.
 readonly STRIMZI_INSTALLATION_CONFIG="$(mktemp "${ARTIFACTS}/strimzi-XXXXX.yaml")"
 # Kafka cluster CR config file.
-readonly KAFKA_INSTALLATION_CONFIG="test/config/100-kafka-ephemeral-triple-3.0.0.yaml"
+readonly KAFKA_INSTALLATION_CONFIG="test/config/100-kafka-ephemeral-triple-3.1.0.yaml"
 # Kafka TLS ConfigMap.
 readonly KAFKA_TLS_CONFIG="test/config/config-kafka-tls.yaml"
 # Kafka SASL ConfigMap.
diff --git a/vendor/knative.dev/eventing/test/lib/resources/eventing.go b/vendor/knative.dev/eventing/test/lib/resources/eventing.go
index d4720d2b8..05af1b366 100644
--- a/vendor/knative.dev/eventing/test/lib/resources/eventing.go
+++ b/vendor/knative.dev/eventing/test/lib/resources/eventing.go
@@ -171,6 +171,19 @@ func WithBrokerClassForBroker(brokerClass string) BrokerOption {
 	}
 }
 
+// WithCustomAnnotationForBroker returns a function that adds a custom
+// annotation to the given v1 Broker.
+func WithCustomAnnotationForBroker(annotationKey, annotationValue string) BrokerOption {
+	return func(b *eventingv1.Broker) {
+		annotations := b.GetAnnotations()
+		if annotations == nil {
+			annotations = make(map[string]string, 1)
+		}
+		annotations[annotationKey] = annotationValue
+		b.SetAnnotations(annotations)
+	}
+}
+
 // WithDeliveryForBroker returns a function that adds a Delivery for the given
 // v1 Broker.
 func WithDeliveryForBroker(delivery *eventingduckv1.DeliverySpec) BrokerOption {