Repository: knative-sandbox/eventing-kafka-broker
Pull request #1789 (2022-01-24 08:26:13)

Promote some logs to info in data plane reconciler
These logs are useful to watch control plane activities in
the data plane and auditing the behavior and they should be
IMO set to `info` since they are not a lot or in the hot path.

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

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

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

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

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

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

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

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-reconciler-tests | ee032b6506d42d8d9fb80ddfe1915a5559657005 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1789/pull-knative-sandbox-eventing-kafka-broker-reconciler-tests/1485529363586748416) | true | `/test pull-knative-sandbox-eventing-kafka-broker-reconciler-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1789). [Your PR dashboard](https://gubernator.knative.dev/pr/pierDipi).

<details>

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

Patch:
diff --git a/control-plane/config/200-controller/200-controller-cluster-role.yaml b/control-plane/config/200-controller/200-controller-cluster-role.yaml
index fd54da4f3..dfb85a5d7 100644
--- a/control-plane/config/200-controller/200-controller-cluster-role.yaml
+++ b/control-plane/config/200-controller/200-controller-cluster-role.yaml
@@ -108,6 +108,15 @@ rules:
       - watch
       - update
       - patch
+  - apiGroups:
+      - messaging.knative.dev
+    resources:
+      - subscriptions
+      - subscriptions/status
+    verbs:
+      - get
+      - list
+      - watch
   - apiGroups:
       - messaging.knative.dev
     resources:
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go
index 0f769d11b..4693af4db 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go
@@ -167,3 +167,11 @@ func (cg *ConsumerGroup) IsReady() bool {
 	return cg.Generation == cg.Status.ObservedGeneration &&
 		cg.GetConditionSet().Manage(cg.GetStatus()).IsHappy()
 }
+
+func (cg *ConsumerGroup) IsNotScheduled() bool {
+	// We want to return true when:
+	// - the condition isn't present, or
+	// - the condition isn't ready (aka status=false)
+	cond := cg.Status.GetCondition(ConditionConsumerGroupConsumersScheduled)
+	return cond.IsFalse() || cond.IsUnknown()
+}
diff --git a/control-plane/pkg/contract/contract.pb.go b/control-plane/pkg/contract/contract.pb.go
index 3ae8af054..49076b003 100644
--- a/control-plane/pkg/contract/contract.pb.go
+++ b/control-plane/pkg/contract/contract.pb.go
@@ -534,6 +534,12 @@ type Egress struct {
 	DeliveryOrder DeliveryOrder `protobuf:"varint,8,opt,name=deliveryOrder,proto3,enum=DeliveryOrder" json:"deliveryOrder,omitempty"`
 	// Kafka record key type.
 	KeyType KeyType `protobuf:"varint,10,opt,name=keyType,proto3,enum=KeyType" json:"keyType,omitempty"`
+	// Resource reference.
+	//
+	// This reference is used to reference the associated resource for data plane
+	// activities such as:
+	// - tagging metrics
+	Reference *Reference `protobuf:"bytes,11,opt,name=reference,proto3" json:"reference,omitempty"`
 }
 
 func (x *Egress) Reset() {
@@ -645,6 +651,13 @@ func (x *Egress) GetKeyType() KeyType {
 	return KeyType_String
 }
 
+func (x *Egress) GetReference() *Reference {
+	if x != nil {
+		return x.Reference
+	}
+	return nil
+}
+
 type isEgress_ReplyStrategy interface {
 	isEgress_ReplyStrategy()
 }
@@ -1348,7 +1361,7 @@ var file_contract_proto_rawDesc = []byte{
 	0x6b, 0x6f, 0x66, 0x66, 0x44, 0x65, 0x6c, 0x61, 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, 0xab, 0x03, 0x0a, 0x06, 0x45, 0x67, 0x72, 0x65,
+	0x74, 0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74, 0x22, 0xd5, 0x03, 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,
@@ -1374,118 +1387,121 @@ var file_contract_proto_rawDesc = []byte{
 	0x72, 0x52, 0x0d, 0x64, 0x65, 0x6c, 0x69, 0x76, 0x65, 0x72, 0x79, 0x4f, 0x72, 0x64, 0x65, 0x72,
 	0x12, 0x22, 0x0a, 0x07, 0x6b, 0x65, 0x79, 0x54, 0x79, 0x70, 0x65, 0x18, 0x0a, 0x20, 0x01, 0x28,
 	0x0e, 0x32, 0x08, 0x2e, 0x4b, 0x65, 0x79, 0x54, 0x79, 0x70, 0x65, 0x52, 0x07, 0x6b, 0x65, 0x79,
-	0x54, 0x79, 0x70, 0x65, 0x42, 0x0f, 0x0a, 0x0d, 0x72, 0x65, 0x70, 0x6c, 0x79, 0x53, 0x74, 0x72,
-	0x61, 0x74, 0x65, 0x67, 0x79, 0x22, 0x74, 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, 0x14, 0x0a, 0x04, 0x70, 0x61, 0x74, 0x68, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x48, 0x00,
-	0x52, 0x04, 0x70, 0x61, 0x74, 0x68, 0x12, 0x14, 0x0a, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x18, 0x03,
-	0x20, 0x01, 0x28, 0x09, 0x48, 0x00, 0x52, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x42, 0x0d, 0x0a, 0x0b,
-	0x69, 0x6e, 0x67, 0x72, 0x65, 0x73, 0x73, 0x54, 0x79, 0x70, 0x65, 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,
+	0x54, 0x79, 0x70, 0x65, 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, 0x0f,
+	0x0a, 0x0d, 0x72, 0x65, 0x70, 0x6c, 0x79, 0x53, 0x74, 0x72, 0x61, 0x74, 0x65, 0x67, 0x79, 0x22,
+	0x74, 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, 0x14, 0x0a, 0x04, 0x70, 0x61,
+	0x74, 0x68, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x48, 0x00, 0x52, 0x04, 0x70, 0x61, 0x74, 0x68,
+	0x12, 0x14, 0x0a, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x48, 0x00,
+	0x52, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x42, 0x0d, 0x0a, 0x0b, 0x69, 0x6e, 0x67, 0x72, 0x65, 0x73,
+	0x73, 0x54, 0x79, 0x70, 0x65, 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, 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,
+	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 (
@@ -1533,27 +1549,28 @@ var file_contract_proto_depIdxs = []int32{
 	8,  // 5: Egress.egressConfig:type_name -> EgressConfig
 	1,  // 6: Egress.deliveryOrder:type_name -> DeliveryOrder
 	2,  // 7: Egress.keyType:type_name -> KeyType
-	3,  // 8: Ingress.contentMode:type_name -> ContentMode
-	11, // 9: SecretReference.reference:type_name -> Reference
-	13, // 10: SecretReference.keyFieldReferences:type_name -> KeyFieldReference
-	4,  // 11: KeyFieldReference.field:type_name -> SecretField
-	5,  // 12: MultiSecretReference.protocol:type_name -> Protocol
-	12, // 13: MultiSecretReference.references:type_name -> SecretReference
-	19, // 14: CloudEventOverrides.extensions:type_name -> CloudEventOverrides.ExtensionsEntry
-	10, // 15: Resource.ingress:type_name -> Ingress
-	8,  // 16: Resource.egressConfig:type_name -> EgressConfig
-	9,  // 17: Resource.egresses:type_name -> Egress
-	6,  // 18: Resource.absentAuth:type_name -> Empty
-	11, // 19: Resource.authSecret:type_name -> Reference
-	14, // 20: Resource.multiAuthSecret:type_name -> MultiSecretReference
-	15, // 21: Resource.cloudEventOverrides:type_name -> CloudEventOverrides
-	11, // 22: Resource.reference:type_name -> Reference
-	16, // 23: Contract.resources:type_name -> Resource
-	24, // [24:24] is the sub-list for method output_type
-	24, // [24:24] is the sub-list for method input_type
-	24, // [24:24] is the sub-list for extension type_name
-	24, // [24:24] is the sub-list for extension extendee
-	0,  // [0:24] is the sub-list for field type_name
+	11, // 8: Egress.reference:type_name -> Reference
+	3,  // 9: Ingress.contentMode:type_name -> ContentMode
+	11, // 10: SecretReference.reference:type_name -> Reference
+	13, // 11: SecretReference.keyFieldReferences:type_name -> KeyFieldReference
+	4,  // 12: KeyFieldReference.field:type_name -> SecretField
+	5,  // 13: MultiSecretReference.protocol:type_name -> Protocol
+	12, // 14: MultiSecretReference.references:type_name -> SecretReference
+	19, // 15: CloudEventOverrides.extensions:type_name -> CloudEventOverrides.ExtensionsEntry
+	10, // 16: Resource.ingress:type_name -> Ingress
+	8,  // 17: Resource.egressConfig:type_name -> EgressConfig
+	9,  // 18: Resource.egresses:type_name -> Egress
+	6,  // 19: Resource.absentAuth:type_name -> Empty
+	11, // 20: Resource.authSecret:type_name -> Reference
+	14, // 21: Resource.multiAuthSecret:type_name -> MultiSecretReference
+	15, // 22: Resource.cloudEventOverrides:type_name -> CloudEventOverrides
+	11, // 23: Resource.reference:type_name -> Reference
+	16, // 24: Contract.resources:type_name -> Resource
+	25, // [25:25] is the sub-list for method output_type
+	25, // [25:25] is the sub-list for method input_type
+	25, // [25:25] is the sub-list for extension type_name
+	25, // [25:25] is the sub-list for extension extendee
+	0,  // [0:25] is the sub-list for field type_name
 }
 
 func init() { file_contract_proto_init() }
diff --git a/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index 620b29361..c3a6be022 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -18,13 +18,15 @@ package channel
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"strconv"
 
 	"github.com/Shopify/sarama"
-
 	"go.uber.org/multierr"
 	"go.uber.org/zap"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -37,6 +39,7 @@ import (
 	"knative.dev/pkg/system"
 
 	v1 "knative.dev/eventing/pkg/apis/duck/v1"
+	messaginglisters "knative.dev/eventing/pkg/client/listers/messaging/v1"
 
 	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	"knative.dev/eventing-kafka/pkg/common/constants"
@@ -77,7 +80,8 @@ type Reconciler struct {
 	// reconciliation loop.
 	InitOffsetsFunc kafka.InitOffsetsFunc
 
-	ConfigMapLister corelisters.ConfigMapLister
+	ConfigMapLister    corelisters.ConfigMapLister
+	SubscriptionLister messaginglisters.SubscriptionLister
 
 	Prober prober.Prober
 
@@ -455,7 +459,7 @@ func (r *Reconciler) reconcileSubscriber(ctx context.Context, kafkaClient sarama
 	subscriberIndex := coreconfig.FindEgress(channelContractResource.Egresses, subscriberSpec.UID)
 	subscriberConfig, err := r.getSubscriberConfig(ctx, channel, subscriberSpec)
 	if err != nil {
-		return fmt.Errorf("failed to resolve subscriber config: %v", zap.Error(err))
+		return fmt.Errorf("failed to resolve subscriber config: %w", err)
 	}
 
 	coreconfig.AddOrUpdateEgressConfigForResource(channelContractResource, subscriberConfig, subscriberIndex)
@@ -476,11 +480,21 @@ func (r *Reconciler) reconcileInitialOffset(ctx context.Context, channel *messag
 }
 
 func (r *Reconciler) getSubscriberConfig(ctx context.Context, channel *messagingv1beta1.KafkaChannel, subscriber *v1.SubscriberSpec) (*contract.Egress, error) {
+	subscriptionName, err := r.getSubscriptionName(channel, subscriber)
+	if err != nil {
+		return nil, fmt.Errorf("failed to extract subscription name for subscriber %v: %w", subscriber, err)
+	}
+
 	egress := &contract.Egress{
 		Destination:   subscriber.SubscriberURI.String(),
 		ConsumerGroup: consumerGroup(channel, subscriber),
 		DeliveryOrder: DefaultDeliveryOrder,
 		Uid:           string(subscriber.UID),
+		Reference: &contract.Reference{
+			Uuid:      string(subscriber.UID),
+			Namespace: channel.GetNamespace(),
+			Name:      subscriptionName,
+		},
 		ReplyStrategy: &contract.Egress_DiscardReply{},
 	}
 
@@ -581,6 +595,21 @@ func (r *Reconciler) getChannelContractResource(ctx context.Context, topic strin
 	return resource, nil
 }
 
+func (r *Reconciler) getSubscriptionName(channel *messagingv1beta1.KafkaChannel, subscriber *v1.SubscriberSpec) (string, error) {
+	subscriptions, err := r.SubscriptionLister.Subscriptions(channel.GetNamespace()).List(labels.Everything())
+	if err != nil {
+		return "", fmt.Errorf("failed to list subscriptions in namespace %s: %w", channel.GetNamespace(), err)
+	}
+
+	for _, s := range subscriptions {
+		if s.UID == subscriber.UID {
+			return s.Name, nil
+		}
+	}
+
+	return "", errors.New(string(metav1.StatusReasonNotFound))
+}
+
 // consumerGroup returns a consumerGroup name for the given channel and subscription
 func consumerGroup(channel *messagingv1beta1.KafkaChannel, sub *v1.SubscriberSpec) string {
 	return fmt.Sprintf("kafka.%s.%s.%s", channel.Namespace, channel.Name, string(sub.UID))
diff --git a/control-plane/pkg/reconciler/channel/controller.go b/control-plane/pkg/reconciler/channel/controller.go
index 8ccfa800b..db8296757 100644
--- a/control-plane/pkg/reconciler/channel/controller.go
+++ b/control-plane/pkg/reconciler/channel/controller.go
@@ -25,6 +25,8 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/tools/cache"
 
+	subscriptioninformer "knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription"
+
 	messagingv1beta "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	kafkachannelinformer "knative.dev/eventing-kafka/pkg/client/injection/informers/messaging/v1beta1/kafkachannel"
 	kafkachannelreconciler "knative.dev/eventing-kafka/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel"
@@ -65,6 +67,7 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *conf
 			DispatcherLabel:             base.ChannelDispatcherLabel,
 			ReceiverLabel:               base.ChannelReceiverLabel,
 		},
+		SubscriptionLister:         subscriptioninformer.Get(ctx).Lister(),
 		NewKafkaClient:             sarama.NewClient,
 		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
 		InitOffsetsFunc:            offset.InitOffsets,
diff --git a/control-plane/pkg/reconciler/consumergroup/evictor.go b/control-plane/pkg/reconciler/consumergroup/evictor.go
new file mode 100644
index 000000000..28b2149d6
--- /dev/null
+++ b/control-plane/pkg/reconciler/consumergroup/evictor.go
@@ -0,0 +1,129 @@
+/*
+ * Copyright 2022 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package consumergroup
+
+import (
+	"context"
+	"fmt"
+
+	"go.uber.org/zap"
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	eventingduckv1alpha1 "knative.dev/eventing/pkg/apis/duck/v1alpha1"
+	"knative.dev/eventing/pkg/scheduler"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	"knative.dev/pkg/logging"
+
+	internalv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/clientset/versioned/typed/eventing/v1alpha1"
+	kafkainternalsclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client"
+)
+
+type evictor struct {
+	ctx context.Context
+
+	kubeClient      kubernetes.Interface
+	InternalsClient internalv1alpha1.InternalV1alpha1Interface
+
+	logger *zap.Logger
+}
+
+// newEvictor creates a new evictor.
+//
+// fields are additional logger fields to be attached to the evictor logger.
+func newEvictor(ctx context.Context, fields ...zap.Field) evictor {
+	return evictor{
+		ctx:             ctx,
+		kubeClient:      kubeclient.Get(ctx),
+		InternalsClient: kafkainternalsclient.Get(ctx).InternalV1alpha1(),
+		logger: logging.FromContext(ctx).
+			Desugar().
+			With(zap.String("component", "evictor")).
+			With(fields...),
+	}
+}
+
+func (e *evictor) evict(pod *corev1.Pod, vpod scheduler.VPod, from *eventingduckv1alpha1.Placement) error {
+	key := vpod.GetKey()
+
+	logger := e.logger.
+		With(zap.String("consumergroup", key.String())).
+		With(zap.String("pod", fmt.Sprintf("%s/%s", pod.GetNamespace(), pod.GetName())))
+
+	if err := e.disablePodScheduling(logger, pod.DeepCopy() /* Do not modify informer copy. */); err != nil {
+		return fmt.Errorf("failed to mark pod unschedulable: %w", err)
+	}
+
+	cg, err := e.InternalsClient.
+		ConsumerGroups(key.Namespace).
+		Get(e.ctx, key.Name, metav1.GetOptions{})
+	if apierrors.IsNotFound(err) {
+		return nil
+	}
+	if err != nil {
+		return fmt.Errorf("failed to get consumer group %s/%s: %w", key.Namespace, key.Name, err)
+	}
+
+	// Do not evict when the consumer group isn't scheduled yet.
+	if cg.IsNotScheduled() {
+		return nil
+	}
+
+	cg.Status.Placements = removePlacement(cg.GetPlacements(), from)
+
+	_, err = e.InternalsClient.
+		ConsumerGroups(cg.GetNamespace()).
+		Update(e.ctx, cg, metav1.UpdateOptions{})
+	if apierrors.IsNotFound(err) {
+		return nil
+	}
+	if err != nil {
+		return fmt.Errorf("failed to update consumer group %s/%s: %w", cg.GetNamespace(), cg.GetName(), err)
+	}
+
+	return nil
+}
+
+func (e *evictor) disablePodScheduling(logger *zap.Logger, pod *corev1.Pod) error {
+	if pod.Annotations == nil {
+		pod.Annotations = make(map[string]string, 1)
+	}
+	// scheduling disabled.
+	pod.Annotations[scheduler.PodAnnotationKey] = "true"
+
+	_, err := e.kubeClient.CoreV1().
+		Pods(pod.GetNamespace()).
+		Update(e.ctx, pod, metav1.UpdateOptions{})
+	if err != nil {
+		return fmt.Errorf("failed to update pod %s/%s: %w", pod.GetNamespace(), pod.GetName(), err)
+	}
+
+	logger.Info("Marked pod as unschedulable")
+
+	return nil
+}
+
+func removePlacement(before []eventingduckv1alpha1.Placement, toRemove *eventingduckv1alpha1.Placement) []eventingduckv1alpha1.Placement {
+	after := make([]eventingduckv1alpha1.Placement, 0, len(before))
+	for _, p := range before {
+		if p.PodName != toRemove.PodName {
+			after = append(after, p)
+		}
+	}
+	return after
+}
diff --git a/control-plane/pkg/reconciler/source/source.go b/control-plane/pkg/reconciler/source/source.go
index e3980959c..935fb5f16 100644
--- a/control-plane/pkg/reconciler/source/source.go
+++ b/control-plane/pkg/reconciler/source/source.go
@@ -275,6 +275,11 @@ func (r *Reconciler) reconcileKafkaSourceResource(ctx context.Context, ks *sourc
 		Uid:           string(ks.GetUID()),
 		EgressConfig:  egressConfig,
 		DeliveryOrder: DefaultDeliveryOrder,
+		Reference: &contract.Reference{
+			Uuid:      string(ks.GetUID()),
+			Namespace: ks.GetNamespace(),
+			Name:      ks.GetName(),
+		},
 	}
 	// Set key type hint (if any).
 	if keyType, ok := ks.Labels[sources.KafkaKeyTypeLabel]; ok {
@@ -286,10 +291,7 @@ func (r *Reconciler) reconcileKafkaSourceResource(ctx context.Context, ks *sourc
 		BootstrapServers: strings.Join(ks.Spec.BootstrapServers, ","),
 		Egresses:         []*contract.Egress{egress},
 		Auth:             &contract.Resource_AbsentAuth{},
-		Reference: &contract.Reference{
-			Namespace: ks.GetNamespace(),
-			Name:      ks.GetName(),
-		},
+		Reference:        egress.Reference,
 	}
 	if ks.Spec.CloudEventOverrides != nil {
 		resource.CloudEventOverrides = &contract.CloudEventOverrides{
diff --git a/control-plane/pkg/reconciler/source/v2/controllerv2.go b/control-plane/pkg/reconciler/source/v2/controllerv2.go
index dbd161ca1..b9b072de1 100644
--- a/control-plane/pkg/reconciler/source/v2/controllerv2.go
+++ b/control-plane/pkg/reconciler/source/v2/controllerv2.go
@@ -33,14 +33,14 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/consumergroup"
 )
 
-func NewControllerV2(ctx context.Context, configs *config.Env) *controller.Impl {
+func NewController(ctx context.Context, configs *config.Env) *controller.Impl {
 
 	kafkaInformer := kafkainformer.Get(ctx)
 	consumerGroupInformer := consumergroupinformer.Get(ctx)
 
 	sources.RegisterAlternateKafkaConditionSet(conditionSet)
 
-	r := &ReconcilerV2{
+	r := &Reconciler{
 		ConsumerGroupLister: consumerGroupInformer.Lister(),
 		InternalsClient:     consumergroupclient.Get(ctx),
 	}
diff --git a/control-plane/pkg/reconciler/source/v2/sourcev2.go b/control-plane/pkg/reconciler/source/v2/sourcev2.go
index fc79cb028..e68eb8950 100644
--- a/control-plane/pkg/reconciler/source/v2/sourcev2.go
+++ b/control-plane/pkg/reconciler/source/v2/sourcev2.go
@@ -45,15 +45,16 @@ const (
 var (
 	conditionSet = apis.NewLivingConditionSet(
 		KafkaConditionConsumerGroup,
+		sources.KafkaConditionSinkProvided,
 	)
 )
 
-type ReconcilerV2 struct {
+type Reconciler struct {
 	ConsumerGroupLister internalslst.ConsumerGroupLister
 	InternalsClient     internalsclient.Interface
 }
 
-func (r *ReconcilerV2) ReconcileKind(ctx context.Context, ks *sources.KafkaSource) reconciler.Event {
+func (r *Reconciler) ReconcileKind(ctx context.Context, ks *sources.KafkaSource) reconciler.Event {
 
 	cg, err := r.reconcileConsumerGroup(ctx, ks)
 	if err != nil {
@@ -64,11 +65,15 @@ func (r *ReconcilerV2) ReconcileKind(ctx context.Context, ks *sources.KafkaSourc
 		ks.GetConditionSet().Manage(&ks.Status).MarkTrue(KafkaConditionConsumerGroup)
 	} else {
 		topLevelCondition := cg.GetConditionSet().Manage(cg.GetStatus()).GetTopLevelCondition()
-		ks.GetConditionSet().Manage(&ks.Status).MarkFalse(
-			KafkaConditionConsumerGroup,
-			topLevelCondition.Reason,
-			topLevelCondition.Message,
-		)
+		if topLevelCondition == nil {
+			ks.GetConditionSet().Manage(&ks.Status).MarkUnknown(KafkaConditionConsumerGroup, "failed to reconcile consumer group", "consumer group not ready")
+		} else {
+			ks.GetConditionSet().Manage(&ks.Status).MarkFalse(
+				KafkaConditionConsumerGroup,
+				topLevelCondition.Reason,
+				topLevelCondition.Message,
+			)
+		}
 	}
 
 	ks.Status.MarkSink(cg.Status.SubscriberURI)
@@ -80,7 +85,7 @@ func (r *ReconcilerV2) ReconcileKind(ctx context.Context, ks *sources.KafkaSourc
 	return nil
 }
 
-func (r ReconcilerV2) reconcileConsumerGroup(ctx context.Context, ks *sources.KafkaSource) (*internalscg.ConsumerGroup, error) {
+func (r Reconciler) reconcileConsumerGroup(ctx context.Context, ks *sources.KafkaSource) (*internalscg.ConsumerGroup, error) {
 
 	newcg := &internalscg.ConsumerGroup{
 		ObjectMeta: metav1.ObjectMeta{
@@ -105,15 +110,18 @@ func (r ReconcilerV2) reconcileConsumerGroup(ctx context.Context, ks *sources.Ka
 						Ordering: DefaultDeliveryOrder,
 					},
 					Subscriber: ks.Spec.Sink,
-					CloudEventOverrides: &duckv1.CloudEventOverrides{
-						Extensions: ks.Spec.CloudEventOverrides.Extensions,
-					},
 				},
 			},
 			Replicas: ks.Spec.Consumers,
 		},
 	}
 
+	if ks.Spec.CloudEventOverrides != nil {
+		newcg.Spec.Template.Spec.CloudEventOverrides = &duckv1.CloudEventOverrides{
+			Extensions: ks.Spec.CloudEventOverrides.Extensions,
+		}
+	}
+
 	cg, err := r.ConsumerGroupLister.ConsumerGroups(ks.GetNamespace()).Get(string(ks.UID)) //Get by consumer group id
 	if err != nil && !apierrors.IsNotFound(err) {
 		return nil, err
diff --git a/control-plane/pkg/reconciler/trigger/trigger.go b/control-plane/pkg/reconciler/trigger/trigger.go
index e1614b381..a4a8b8b55 100644
--- a/control-plane/pkg/reconciler/trigger/trigger.go
+++ b/control-plane/pkg/reconciler/trigger/trigger.go
@@ -276,6 +276,11 @@ func (r *Reconciler) reconcileTriggerEgress(ctx context.Context, broker *eventin
 		Destination:   destination.String(),
 		ConsumerGroup: string(trigger.UID),
 		Uid:           string(trigger.UID),
+		Reference: &contract.Reference{
+			Uuid:      string(trigger.GetUID()),
+			Namespace: trigger.GetNamespace(),
+			Name:      trigger.GetName(),
+		},
 	}
 
 	if trigger.Spec.Filter != nil && trigger.Spec.Filter.Attributes != nil {
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 1d81cacea..fc4b899fe 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
@@ -3325,6 +3325,42 @@ public final class DataPlaneContract {
      */
     dev.knative.eventing.kafka.broker.contract.DataPlaneContract.KeyType getKeyType();
 
+    /**
+     * <pre>
+     * Resource reference.
+     * This reference is used to reference the associated resource for data plane
+     * activities such as:
+     * - tagging metrics
+     * </pre>
+     *
+     * <code>.Reference reference = 11;</code>
+     * @return Whether the reference field is set.
+     */
+    boolean hasReference();
+    /**
+     * <pre>
+     * Resource reference.
+     * This reference is used to reference the associated resource for data plane
+     * activities such as:
+     * - tagging metrics
+     * </pre>
+     *
+     * <code>.Reference reference = 11;</code>
+     * @return The reference.
+     */
+    dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference getReference();
+    /**
+     * <pre>
+     * Resource reference.
+     * This reference is used to reference the associated resource for data plane
+     * activities such as:
+     * - tagging metrics
+     * </pre>
+     *
+     * <code>.Reference reference = 11;</code>
+     */
+    dev.knative.eventing.kafka.broker.contract.DataPlaneContract.ReferenceOrBuilder getReferenceOrBuilder();
+
     public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Egress.ReplyStrategyCase getReplyStrategyCase();
   }
   /**
@@ -3467,6 +3503,19 @@ public final class DataPlaneContract {
               keyType_ = rawValue;
               break;
             }
+            case 90: {
+              dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference.Builder subBuilder = null;
+              if (reference_ != null) {
+                subBuilder = reference_.toBuilder();
+              }
+              reference_ = input.readMessage(dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference.parser(), extensionRegistry);
+              if (subBuilder != null) {
+                subBuilder.mergeFrom(reference_);
+                reference_ = subBuilder.buildPartial();
+              }
+
+              break;
+            }
             default: {
               if (!parseUnknownField(
                   input, unknownFields, extensionRegistry, tag)) {
@@ -3955,6 +4004,53 @@ public final class DataPlaneContract {
       return result == null ? dev.knative.eventing.kafka.broker.contract.DataPlaneContract.KeyType.UNRECOGNIZED : result;
     }
 
+    public static final int REFERENCE_FIELD_NUMBER = 11;
+    private dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference reference_;
+    /**
+     * <pre>
+     * Resource reference.
+     * This reference is used to reference the associated resource for data plane
+     * activities such as:
+     * - tagging metrics
+     * </pre>
+     *
+     * <code>.Reference reference = 11;</code>
+     * @return Whether the reference field is set.
+     */
+    @java.lang.Override
+    public boolean hasReference() {
+      return reference_ != null;
+    }
+    /**
+     * <pre>
+     * Resource reference.
+     * This reference is used to reference the associated resource for data plane
+     * activities such as:
+     * - tagging metrics
+     * </pre>
+     *
+     * <code>.Reference reference = 11;</code>
+     * @return The reference.
+     */
+    @java.lang.Override
+    public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference getReference() {
+      return reference_ == null ? dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference.getDefaultInstance() : reference_;
+    }
+    /**
+     * <pre>
+     * Resource reference.
+     * This reference is used to reference the associated resource for data plane
+     * activities such as:
+     * - tagging metrics
+     * </pre>
+     *
+     * <code>.Reference reference = 11;</code>
+     */
+    @java.lang.Override
+    public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.ReferenceOrBuilder getReferenceOrBuilder() {
+      return getReference();
+    }
+
     private byte memoizedIsInitialized = -1;
     @java.lang.Override
     public final boolean isInitialized() {
@@ -3999,6 +4095,9 @@ public final class DataPlaneContract {
       if (keyType_ != dev.knative.eventing.kafka.broker.contract.DataPlaneContract.KeyType.String.getNumber()) {
         output.writeEnum(10, keyType_);
       }
+      if (reference_ != null) {
+        output.writeMessage(11, getReference());
+      }
       unknownFields.writeTo(output);
     }
 
@@ -4044,6 +4143,10 @@ public final class DataPlaneContract {
         size += com.google.protobuf.CodedOutputStream
           .computeEnumSize(10, keyType_);
       }
+      if (reference_ != null) {
+        size += com.google.protobuf.CodedOutputStream
+          .computeMessageSize(11, getReference());
+      }
       size += unknownFields.getSerializedSize();
       memoizedSize = size;
       return size;
@@ -4077,6 +4180,11 @@ public final class DataPlaneContract {
       }
       if (deliveryOrder_ != other.deliveryOrder_) return false;
       if (keyType_ != other.keyType_) return false;
+      if (hasReference() != other.hasReference()) return false;
+      if (hasReference()) {
+        if (!getReference()
+            .equals(other.getReference())) return false;
+      }
       if (!getReplyStrategyCase().equals(other.getReplyStrategyCase())) return false;
       switch (replyStrategyCase_) {
         case 3:
@@ -4123,6 +4231,10 @@ public final class DataPlaneContract {
       hash = (53 * hash) + deliveryOrder_;
       hash = (37 * hash) + KEYTYPE_FIELD_NUMBER;
       hash = (53 * hash) + keyType_;
+      if (hasReference()) {
+        hash = (37 * hash) + REFERENCE_FIELD_NUMBER;
+        hash = (53 * hash) + getReference().hashCode();
+      }
       switch (replyStrategyCase_) {
         case 3:
           hash = (37 * hash) + REPLYURL_FIELD_NUMBER;
@@ -4294,6 +4406,12 @@ public final class DataPlaneContract {
 
         keyType_ = 0;
 
+        if (referenceBuilder_ == null) {
+          reference_ = null;
+        } else {
+          reference_ = null;
+          referenceBuilder_ = null;
+        }
         replyStrategyCase_ = 0;
         replyStrategy_ = null;
         return this;
@@ -4354,6 +4472,11 @@ public final class DataPlaneContract {
         }
         result.deliveryOrder_ = deliveryOrder_;
         result.keyType_ = keyType_;
+        if (referenceBuilder_ == null) {
+          result.reference_ = reference_;
+        } else {
+          result.reference_ = referenceBuilder_.build();
+        }
         result.replyStrategyCase_ = replyStrategyCase_;
         onBuilt();
         return result;
@@ -4427,6 +4550,9 @@ public final class DataPlaneContract {
         if (other.keyType_ != 0) {
           setKeyTypeValue(other.getKeyTypeValue());
         }
+        if (other.hasReference()) {
+          mergeReference(other.getReference());
+        }
         switch (other.getReplyStrategyCase()) {
           case REPLYURL: {
             replyStrategyCase_ = 3;
@@ -5693,6 +5819,188 @@ public final class DataPlaneContract {
         onChanged();
         return this;
       }
+
+      private dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference reference_;
+      private com.google.protobuf.SingleFieldBuilderV3<
+          dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference.Builder, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.ReferenceOrBuilder> referenceBuilder_;
+      /**
+       * <pre>
+       * Resource reference.
+       * This reference is used to reference the associated resource for data plane
+       * activities such as:
+       * - tagging metrics
+       * </pre>
+       *
+       * <code>.Reference reference = 11;</code>
+       * @return Whether the reference field is set.
+       */
+      public boolean hasReference() {
+        return referenceBuilder_ != null || reference_ != null;
+      }
+      /**
+       * <pre>
+       * Resource reference.
+       * This reference is used to reference the associated resource for data plane
+       * activities such as:
+       * - tagging metrics
+       * </pre>
+       *
+       * <code>.Reference reference = 11;</code>
+       * @return The reference.
+       */
+      public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference getReference() {
+        if (referenceBuilder_ == null) {
+          return reference_ == null ? dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference.getDefaultInstance() : reference_;
+        } else {
+          return referenceBuilder_.getMessage();
+        }
+      }
+      /**
+       * <pre>
+       * Resource reference.
+       * This reference is used to reference the associated resource for data plane
+       * activities such as:
+       * - tagging metrics
+       * </pre>
+       *
+       * <code>.Reference reference = 11;</code>
+       */
+      public Builder setReference(dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference value) {
+        if (referenceBuilder_ == null) {
+          if (value == null) {
+            throw new NullPointerException();
+          }
+          reference_ = value;
+          onChanged();
+        } else {
+          referenceBuilder_.setMessage(value);
+        }
+
+        return this;
+      }
+      /**
+       * <pre>
+       * Resource reference.
+       * This reference is used to reference the associated resource for data plane
+       * activities such as:
+       * - tagging metrics
+       * </pre>
+       *
+       * <code>.Reference reference = 11;</code>
+       */
+      public Builder setReference(
+          dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference.Builder builderForValue) {
+        if (referenceBuilder_ == null) {
+          reference_ = builderForValue.build();
+          onChanged();
+        } else {
+          referenceBuilder_.setMessage(builderForValue.build());
+        }
+
+        return this;
+      }
+      /**
+       * <pre>
+       * Resource reference.
+       * This reference is used to reference the associated resource for data plane
+       * activities such as:
+       * - tagging metrics
+       * </pre>
+       *
+       * <code>.Reference reference = 11;</code>
+       */
+      public Builder mergeReference(dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference value) {
+        if (referenceBuilder_ == null) {
+          if (reference_ != null) {
+            reference_ =
+              dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference.newBuilder(reference_).mergeFrom(value).buildPartial();
+          } else {
+            reference_ = value;
+          }
+          onChanged();
+        } else {
+          referenceBuilder_.mergeFrom(value);
+        }
+
+        return this;
+      }
+      /**
+       * <pre>
+       * Resource reference.
+       * This reference is used to reference the associated resource for data plane
+       * activities such as:
+       * - tagging metrics
+       * </pre>
+       *
+       * <code>.Reference reference = 11;</code>
+       */
+      public Builder clearReference() {
+        if (referenceBuilder_ == null) {
+          reference_ = null;
+          onChanged();
+        } else {
+          reference_ = null;
+          referenceBuilder_ = null;
+        }
+
+        return this;
+      }
+      /**
+       * <pre>
+       * Resource reference.
+       * This reference is used to reference the associated resource for data plane
+       * activities such as:
+       * - tagging metrics
+       * </pre>
+       *
+       * <code>.Reference reference = 11;</code>
+       */
+      public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference.Builder getReferenceBuilder() {
+        
+        onChanged();
+        return getReferenceFieldBuilder().getBuilder();
+      }
+      /**
+       * <pre>
+       * Resource reference.
+       * This reference is used to reference the associated resource for data plane
+       * activities such as:
+       * - tagging metrics
+       * </pre>
+       *
+       * <code>.Reference reference = 11;</code>
+       */
+      public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.ReferenceOrBuilder getReferenceOrBuilder() {
+        if (referenceBuilder_ != null) {
+          return referenceBuilder_.getMessageOrBuilder();
+        } else {
+          return reference_ == null ?
+              dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference.getDefaultInstance() : reference_;
+        }
+      }
+      /**
+       * <pre>
+       * Resource reference.
+       * This reference is used to reference the associated resource for data plane
+       * activities such as:
+       * - tagging metrics
+       * </pre>
+       *
+       * <code>.Reference reference = 11;</code>
+       */
+      private com.google.protobuf.SingleFieldBuilderV3<
+          dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference.Builder, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.ReferenceOrBuilder> 
+          getReferenceFieldBuilder() {
+        if (referenceBuilder_ == null) {
+          referenceBuilder_ = new com.google.protobuf.SingleFieldBuilderV3<
+              dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Reference.Builder, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.ReferenceOrBuilder>(
+                  getReference(),
+                  getParentForChildren(),
+                  isClean());
+          reference_ = null;
+        }
+        return referenceBuilder_;
+      }
       @java.lang.Override
       public final Builder setUnknownFields(
           final com.google.protobuf.UnknownFieldSet unknownFields) {
@@ -16745,51 +17053,52 @@ public final class DataPlaneContract {
       "e\030\002 \001(\t:\0028\001\"\177\n\014EgressConfig\022\022\n\ndeadLette" +
       "r\030\001 \001(\t\022\r\n\005retry\030\002 \001(\r\022%\n\rbackoffPolicy\030" +
       "\003 \001(\0162\016.BackoffPolicy\022\024\n\014backoffDelay\030\004 " +
-      "\001(\004\022\017\n\007timeout\030\005 \001(\004\"\256\002\n\006Egress\022\025\n\rconsu" +
+      "\001(\004\022\017\n\007timeout\030\005 \001(\004\"\315\002\n\006Egress\022\025\n\rconsu" +
       "merGroup\030\001 \001(\t\022\023\n\013destination\030\002 \001(\t\022\022\n\010r" +
       "eplyUrl\030\003 \001(\tH\000\022&\n\024replyToOriginalTopic\030" +
       "\004 \001(\0132\006.EmptyH\000\022\036\n\014discardReply\030\t \001(\0132\006." +
       "EmptyH\000\022\027\n\006filter\030\005 \001(\0132\007.Filter\022\013\n\003uid\030" +
       "\006 \001(\t\022#\n\014egressConfig\030\007 \001(\0132\r.EgressConf" +
       "ig\022%\n\rdeliveryOrder\030\010 \001(\0162\016.DeliveryOrde" +
-      "r\022\031\n\007keyType\030\n \001(\0162\010.KeyTypeB\017\n\rreplyStr" +
-      "ategy\"[\n\007Ingress\022!\n\013contentMode\030\001 \001(\0162\014." +
-      "ContentMode\022\016\n\004path\030\002 \001(\tH\000\022\016\n\004host\030\003 \001(" +
-      "\tH\000B\r\n\013ingressType\"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\017SecretReference\022\035\n\tre" +
-      "ference\030\001 \001(\0132\n.Reference\022.\n\022keyFieldRef" +
-      "erences\030\002 \003(\0132\022.KeyFieldReference\"C\n\021Key" +
-      "FieldReference\022\021\n\tsecretKey\030\002 \001(\t\022\033\n\005fie" +
-      "ld\030\003 \001(\0162\014.SecretField\"Y\n\024MultiSecretRef" +
-      "erence\022\033\n\010protocol\030\001 \001(\0162\t.Protocol\022$\n\nr" +
-      "eferences\030\002 \003(\0132\020.SecretReference\"\202\001\n\023Cl" +
-      "oudEventOverrides\0228\n\nextensions\030\001 \003(\0132$." +
-      "CloudEventOverrides.ExtensionsEntry\0321\n\017E" +
-      "xtensionsEntry\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\006topi" +
-      "cs\030\002 \003(\t\022\030\n\020bootstrapServers\030\003 \001(\t\022\031\n\007in" +
-      "gress\030\004 \001(\0132\010.Ingress\022#\n\014egressConfig\030\005 " +
-      "\001(\0132\r.EgressConfig\022\031\n\010egresses\030\006 \003(\0132\007.E" +
-      "gress\022\034\n\nabsentAuth\030\007 \001(\0132\006.EmptyH\000\022 \n\na" +
-      "uthSecret\030\010 \001(\0132\n.ReferenceH\000\0220\n\017multiAu" +
-      "thSecret\030\t \001(\0132\025.MultiSecretReferenceH\000\022" +
-      "1\n\023cloudEventOverrides\030\n \001(\0132\024.CloudEven" +
-      "tOverrides\022\035\n\treference\030\013 \001(\0132\n.Referenc" +
-      "eB\006\n\004Auth\"<\n\010Contract\022\022\n\ngeneration\030\001 \001(" +
-      "\004\022\034\n\tresources\030\002 \003(\0132\t.Resource*,\n\rBacko" +
-      "ffPolicy\022\017\n\013Exponential\020\000\022\n\n\006Linear\020\001*+\n" +
-      "\rDeliveryOrder\022\r\n\tUNORDERED\020\000\022\013\n\007ORDERED" +
-      "\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\013ContentMode" +
-      "\022\n\n\006BINARY\020\000\022\016\n\nSTRUCTURED\020\001*a\n\013SecretFi" +
-      "eld\022\022\n\016SASL_MECHANISM\020\000\022\n\n\006CA_CRT\020\001\022\014\n\010U" +
-      "SER_CRT\020\002\022\014\n\010USER_KEY\020\003\022\010\n\004USER\020\004\022\014\n\010PAS" +
-      "SWORD\020\005*D\n\010Protocol\022\r\n\tPLAINTEXT\020\000\022\022\n\016SA" +
-      "SL_PLAINTEXT\020\001\022\007\n\003SSL\020\002\022\014\n\010SASL_SSL\020\003B[\n" +
-      "*dev.knative.eventing.kafka.broker.contr" +
-      "actB\021DataPlaneContractZ\032control-plane/pk" +
-      "g/contractb\006proto3"
+      "r\022\031\n\007keyType\030\n \001(\0162\010.KeyType\022\035\n\treferenc" +
+      "e\030\013 \001(\0132\n.ReferenceB\017\n\rreplyStrategy\"[\n\007" +
+      "Ingress\022!\n\013contentMode\030\001 \001(\0162\014.ContentMo" +
+      "de\022\016\n\004path\030\002 \001(\tH\000\022\016\n\004host\030\003 \001(\tH\000B\r\n\013in" +
+      "gressType\"K\n\tReference\022\014\n\004uuid\030\001 \001(\t\022\021\n\t" +
+      "namespace\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\treference\030\001" +
+      " \001(\0132\n.Reference\022.\n\022keyFieldReferences\030\002" +
+      " \003(\0132\022.KeyFieldReference\"C\n\021KeyFieldRefe" +
+      "rence\022\021\n\tsecretKey\030\002 \001(\t\022\033\n\005field\030\003 \001(\0162" +
+      "\014.SecretField\"Y\n\024MultiSecretReference\022\033\n" +
+      "\010protocol\030\001 \001(\0162\t.Protocol\022$\n\nreferences" +
+      "\030\002 \003(\0132\020.SecretReference\"\202\001\n\023CloudEventO" +
+      "verrides\0228\n\nextensions\030\001 \003(\0132$.CloudEven" +
+      "tOverrides.ExtensionsEntry\0321\n\017Extensions" +
+      "Entry\022\013\n\003key\030\001 \001(\t\022\r\n\005value\030\002 \001(\t:\0028\001\"\350\002" +
+      "\n\010Resource\022\013\n\003uid\030\001 \001(\t\022\016\n\006topics\030\002 \003(\t\022" +
+      "\030\n\020bootstrapServers\030\003 \001(\t\022\031\n\007ingress\030\004 \001" +
+      "(\0132\010.Ingress\022#\n\014egressConfig\030\005 \001(\0132\r.Egr" +
+      "essConfig\022\031\n\010egresses\030\006 \003(\0132\007.Egress\022\034\n\n" +
+      "absentAuth\030\007 \001(\0132\006.EmptyH\000\022 \n\nauthSecret" +
+      "\030\010 \001(\0132\n.ReferenceH\000\0220\n\017multiAuthSecret\030" +
+      "\t \001(\0132\025.MultiSecretReferenceH\000\0221\n\023cloudE" +
+      "ventOverrides\030\n \001(\0132\024.CloudEventOverride" +
+      "s\022\035\n\treference\030\013 \001(\0132\n.ReferenceB\006\n\004Auth" +
+      "\"<\n\010Contract\022\022\n\ngeneration\030\001 \001(\004\022\034\n\treso" +
+      "urces\030\002 \003(\0132\t.Resource*,\n\rBackoffPolicy\022" +
+      "\017\n\013Exponential\020\000\022\n\n\006Linear\020\001*+\n\rDelivery" +
+      "Order\022\r\n\tUNORDERED\020\000\022\013\n\007ORDERED\020\001*=\n\007Key" +
+      "Type\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\013ContentMode\022\n\n\006BINAR" +
+      "Y\020\000\022\016\n\nSTRUCTURED\020\001*a\n\013SecretField\022\022\n\016SA" +
+      "SL_MECHANISM\020\000\022\n\n\006CA_CRT\020\001\022\014\n\010USER_CRT\020\002" +
+      "\022\014\n\010USER_KEY\020\003\022\010\n\004USER\020\004\022\014\n\010PASSWORD\020\005*D" +
+      "\n\010Protocol\022\r\n\tPLAINTEXT\020\000\022\022\n\016SASL_PLAINT" +
+      "EXT\020\001\022\007\n\003SSL\020\002\022\014\n\010SASL_SSL\020\003B[\n*dev.knat" +
+      "ive.eventing.kafka.broker.contractB\021Data" +
+      "PlaneContractZ\032control-plane/pkg/contrac" +
+      "tb\006proto3"
     };
     descriptor = com.google.protobuf.Descriptors.FileDescriptor
       .internalBuildGeneratedFileFrom(descriptorData,
@@ -16824,7 +17133,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", "ReplyStrategy", });
+        new java.lang.String[] { "ConsumerGroup", "Destination", "ReplyUrl", "ReplyToOriginalTopic", "DiscardReply", "Filter", "Uid", "EgressConfig", "DeliveryOrder", "KeyType", "Reference", "ReplyStrategy", });
     internal_static_Ingress_descriptor =
       getDescriptor().getMessageTypes().get(4);
     internal_static_Ingress_fieldAccessorTable = new
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java
index 2a3c2a896..f7dab59a0 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java
@@ -66,7 +66,7 @@ public class ResourcesReconcilerImpl implements ResourcesReconciler {
   }
 
   @Override
-  public Future<Void> reconcile(final Collection<DataPlaneContract.Resource> newResources) {
+  public synchronized Future<Void> reconcile(final Collection<DataPlaneContract.Resource> newResources) {
     if (isReconcilingIngress()) {
       return reconcileIngress(newResources);
     }
@@ -86,7 +86,7 @@ public class ResourcesReconcilerImpl implements ResourcesReconciler {
     final List<Future> futures = new ArrayList<>(egresses.size() + this.cachedEgresses.size());
 
     final var diff = CollectionsUtils.diff(this.cachedEgresses.keySet(), egresses.keySet());
-    logger.debug("Reconcile egress diff {}", diff);
+    logger.info("Reconcile egress diff {}", diff);
 
     diff.getRemoved().forEach(uid -> {
       final var entry = this.cachedEgresses.get(uid);
@@ -124,7 +124,7 @@ public class ResourcesReconcilerImpl implements ResourcesReconciler {
       final var oldResource = this.cachedResources.get(newResource.getUid());
 
       if (resourceEquals(newResource, oldResource) && egressEquals(newEgress, this.cachedEgresses.get(uid).getKey())) {
-        logger.debug("Nothing changed for egress {} {} {}",
+        logger.info("Nothing changed for egress {} {} {}",
           keyValue("id", newEgress.getUid()),
           keyValue("consumerGroup", newEgress.getConsumerGroup()),
           keyValue("destination", newEgress.getDestination())
@@ -165,7 +165,7 @@ public class ResourcesReconcilerImpl implements ResourcesReconciler {
     final List<Future> futures = new ArrayList<>(newResourcesMap.size() + this.cachedResources.size());
 
     final var diff = CollectionsUtils.diff(this.cachedResources.keySet(), newResourcesMap.keySet());
-    logger.debug("Reconcile ingress diff {}", diff);
+    logger.info("Reconcile ingress diff {}", diff);
 
     diff.getRemoved().stream()
       .map(this.cachedResources::get)
diff --git a/go.mod b/go.mod
index 615dea1ca..548f5352a 100644
--- a/go.mod
+++ b/go.mod
@@ -23,9 +23,9 @@ require (
 	k8s.io/apiserver v0.22.5
 	k8s.io/client-go v0.22.5
 	k8s.io/utils v0.0.0-20211208161948-7d6a63dca704
-	knative.dev/eventing v0.28.1-0.20220118072832-f6fec1088847
-	knative.dev/eventing-kafka v0.28.1-0.20220117141029-46249ce2a5f4
-	knative.dev/hack v0.0.0-20220111151514-59b0cf17578e
-	knative.dev/pkg v0.0.0-20220118025732-e989c22732aa
-	knative.dev/reconciler-test v0.0.0-20220117082429-6a9b91eef10c
+	knative.dev/eventing v0.28.1-0.20220124132429-9e0e90c9d603
+	knative.dev/eventing-kafka v0.28.1-0.20220121065504-fe10ecf57e92
+	knative.dev/hack v0.0.0-20220118141833-9b2ed8471e30
+	knative.dev/pkg v0.0.0-20220118160532-77555ea48cd4
+	knative.dev/reconciler-test v0.0.0-20220118183433-c8bfbe66bada
 )
diff --git a/go.sum b/go.sum
index d1086e479..a092cd9c4 100644
--- a/go.sum
+++ b/go.sum
@@ -63,20 +63,17 @@ github.com/Azure/azure-sdk-for-go v47.1.0+incompatible/go.mod h1:9XXNKU+eRnpl9mo
 github.com/Azure/azure-storage-blob-go v0.6.0/go.mod h1:oGfmITT1V6x//CswqY2gtAHND+xIP64/qL7a5QJix0Y=
 github.com/Azure/go-amqp v0.13.0/go.mod h1:qj+o8xPCz9tMSbQ83Vp8boHahuRDl5mkNHyt1xlxUTs=
 github.com/Azure/go-amqp v0.13.1/go.mod h1:qj+o8xPCz9tMSbQ83Vp8boHahuRDl5mkNHyt1xlxUTs=
-github.com/Azure/go-ansiterm v0.0.0-20170929234023-d6e3b3328b78/go.mod h1:LmzpDX56iTiv29bbRTIsUNlaFfuhWRQBWjQdVyAevI8=
 github.com/Azure/go-ansiterm v0.0.0-20210608223527-2377c96fe795/go.mod h1:LmzpDX56iTiv29bbRTIsUNlaFfuhWRQBWjQdVyAevI8=
 github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E=
 github.com/Azure/go-autorest v14.2.0+incompatible/go.mod h1:r+4oMnoxhatjLLJ6zxSWATqVooLgysK6ZNox3g/xq24=
 github.com/Azure/go-autorest/autorest v0.9.0/go.mod h1:xyHB1BMZT0cuDHU7I0+g046+BFDTQ8rEZB0s4Yfa6bI=
 github.com/Azure/go-autorest/autorest v0.9.3/go.mod h1:GsRuLYvwzLjjjRoWEIyMUaYq8GNUx2nRB378IPt/1p0=
 github.com/Azure/go-autorest/autorest v0.11.3/go.mod h1:JFgpikqFJ/MleTTxwepExTKnFUKKszPS8UavbQYUMuw=
-github.com/Azure/go-autorest/autorest v0.11.12/go.mod h1:eipySxLmqSyC5s5k1CLupqet0PSENBEDP93LQ9a8QYw=
 github.com/Azure/go-autorest/autorest v0.11.18/go.mod h1:dSiJPy22c3u0OtOKDNttNgqpNFY/GeWa7GH/Pz56QRA=
 github.com/Azure/go-autorest/autorest/adal v0.5.0/go.mod h1:8Z9fGy2MpX0PvDjB1pEgQTmVqjGhiHBW7RJJEciWzS0=
 github.com/Azure/go-autorest/autorest/adal v0.8.0/go.mod h1:Z6vX6WXXuyieHAXwMj0S6HY6e6wcHn37qQMBQlvY3lc=
 github.com/Azure/go-autorest/autorest/adal v0.8.1/go.mod h1:ZjhuQClTqx435SRJ2iMlOxPYt3d2C/T/7TiQCVZSn3Q=
 github.com/Azure/go-autorest/autorest/adal v0.9.0/go.mod h1:/c022QCutn2P7uY+/oQWWNcK9YU+MH96NgK+jErpbcg=
-github.com/Azure/go-autorest/autorest/adal v0.9.5/go.mod h1:B7KF7jKIeC9Mct5spmyCB/A8CG/sEz1vwIRGv/bbw7A=
 github.com/Azure/go-autorest/autorest/adal v0.9.13/go.mod h1:W/MM4U6nLxnIskrw4UwWzlHfGjwUS50aOsc/I3yuU8M=
 github.com/Azure/go-autorest/autorest/azure/auth v0.4.2/go.mod h1:90gmfKdlmKgfjUpnCEpOJzsUEjrWDSLwHIG73tSXddM=
 github.com/Azure/go-autorest/autorest/azure/cli v0.3.1/go.mod h1:ZG5p860J94/0kI9mNJVoIoLgXcirM2gF5i2kWloofxw=
@@ -188,7 +185,6 @@ github.com/cncf/xds/go v0.0.0-20210805033703-aa0b78936158/go.mod h1:eXthEFrGJvWH
 github.com/cncf/xds/go v0.0.0-20210922020428-25de7278fc84/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
 github.com/cncf/xds/go v0.0.0-20211011173535-cb28da3451f1 h1:zH8ljVhhq7yC0MIeUL/IviMtY8hx2mK8cN9wEYb8ggw=
 github.com/cncf/xds/go v0.0.0-20211011173535-cb28da3451f1/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
-github.com/cockroachdb/datadriven v0.0.0-20190809214429-80d97fb3cbaa/go.mod h1:zn76sxSg3SzpJ0PPJaLDCu+Bu0Lg3sKTORVIj19EIF8=
 github.com/cockroachdb/datadriven v0.0.0-20200714090401-bf6692d28da5/go.mod h1:h6jFvWxBdQXxjopDMZyH2UVceIRfR84bdzbkoKrsWNo=
 github.com/cockroachdb/errors v1.2.4/go.mod h1:rQD95gz6FARkaKkQXUksEje/d9a6wBJoCr5oaCLELYA=
 github.com/cockroachdb/logtags v0.0.0-20190617123548-eb05cc24525f/go.mod h1:i/u985jwjWRlyHXQbwatDASoW0RMlZ/3i9yJHE2xLkI=
@@ -199,15 +195,12 @@ github.com/coreos/go-etcd v2.0.0+incompatible/go.mod h1:Jez6KQU2B/sWsbdaef3ED8Nz
 github.com/coreos/go-oidc v2.1.0+incompatible/go.mod h1:CgnwVTmzoESiwO9qyAFEMiHoZ1nMCKZlZ9V6mm3/LKc=
 github.com/coreos/go-semver v0.2.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=
 github.com/coreos/go-semver v0.3.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=
-github.com/coreos/go-systemd v0.0.0-20180511133405-39ca1b05acc7/go.mod h1:F5haX7vjVVG0kc13fIWeqUViNPyEJxv/OmvnBo0Yme4=
 github.com/coreos/go-systemd v0.0.0-20190321100706-95778dfbb74e/go.mod h1:F5haX7vjVVG0kc13fIWeqUViNPyEJxv/OmvnBo0Yme4=
 github.com/coreos/go-systemd/v22 v22.3.2/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
-github.com/coreos/pkg v0.0.0-20160727233714-3ac0863d7acf/go.mod h1:E3G3o1h8I7cfcXa63jLwjI0eiQQMgzzUDFVpN/nH/eA=
 github.com/coreos/pkg v0.0.0-20180928190104-399ea9e2e55f/go.mod h1:E3G3o1h8I7cfcXa63jLwjI0eiQQMgzzUDFVpN/nH/eA=
 github.com/cpuguy83/go-md2man v1.0.10/go.mod h1:SmD6nW6nTyfqj6ABTjUi3V3JVMnlJmwcJI5acqYI6dE=
 github.com/cpuguy83/go-md2man/v2 v2.0.0-20190314233015-f79a8a8ca69d/go.mod h1:maD7wRr/U5Z6m/iR4s+kqSMx2CaBsrgA7czyZG/E6dU=
 github.com/cpuguy83/go-md2man/v2 v2.0.0/go.mod h1:maD7wRr/U5Z6m/iR4s+kqSMx2CaBsrgA7czyZG/E6dU=
-github.com/creack/pty v1.1.7/go.mod h1:lj5s0c3V2DBrqTV7llrYr5NG6My20zk30Fl46Y7DoTY=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/creack/pty v1.1.11/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -223,7 +216,6 @@ github.com/dgryski/go-sip13 v0.0.0-20181026042036-e10d5fee7954/go.mod h1:vAd38F8
 github.com/dimchansky/utfbom v1.1.0/go.mod h1:rO41eb7gLfo8SF1jd9F8HplJm1Fewwi4mQvIirEdv+8=
 github.com/docker/spdystream v0.0.0-20160310174837-449fdfce4d96/go.mod h1:Qh8CwZgvJUkLughtfhJv5dyTYa91l1fOUCrgjqmcifM=
 github.com/docopt/docopt-go v0.0.0-20180111231733-ee0de3bc6815/go.mod h1:WwZ+bS3ebgob9U8Nd0kOddGdZWjyMGR8Wziv+TBNwSE=
-github.com/dustin/go-humanize v0.0.0-20171111073723-bb3d318650d4/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
 github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
 github.com/eapache/go-resiliency v1.1.0/go.mod h1:kFI+JgMyC7bLPUVY133qvEBtVayf5mFgVsvEsIPBvNs=
 github.com/eapache/go-resiliency v1.2.0 h1:v7g92e/KSN71Rq7vSThKaWIq68fL4YHvWyiUKorFR1Q=
@@ -288,21 +280,15 @@ github.com/go-logr/logr v1.2.0/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbV
 github.com/go-logr/logr v1.2.2 h1:ahHml/yUpnlb96Rp8HCvtYVPY8ZYpxq3g7UYchIYwbs=
 github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
 github.com/go-openapi/jsonpointer v0.0.0-20160704185906-46af16f9f7b1/go.mod h1:+35s3my2LFTysnkMfxsJBAMHj/DoqoB9knIWoYG/Vk0=
-github.com/go-openapi/jsonpointer v0.19.2/go.mod h1:3akKfEdA7DF1sugOqz1dVQHBcuDBPKZGEoHC/NkiQRg=
 github.com/go-openapi/jsonpointer v0.19.3/go.mod h1:Pl9vOtqEWErmShwVjC8pYs9cog34VGT37dQOVbmoatg=
 github.com/go-openapi/jsonpointer v0.19.5 h1:gZr+CIYByUqjcgeLXnQu2gHYQC9o73G2XUeOFYEICuY=
 github.com/go-openapi/jsonpointer v0.19.5/go.mod h1:Pl9vOtqEWErmShwVjC8pYs9cog34VGT37dQOVbmoatg=
 github.com/go-openapi/jsonreference v0.0.0-20160704190145-13c6e3589ad9/go.mod h1:W3Z9FmVs9qj+KR4zFKmDPGiLdk1D9Rlm7cyMvf57TTg=
-github.com/go-openapi/jsonreference v0.19.2/go.mod h1:jMjeRr2HHw6nAVajTXJ4eiUwohSTlpa0o73RUL1owJc=
 github.com/go-openapi/jsonreference v0.19.3/go.mod h1:rjx6GuL8TTa9VaixXglHmQmIL98+wF9xc8zWvFonSJ8=
 github.com/go-openapi/jsonreference v0.19.5 h1:1WJP/wi4OjB4iV8KVbH73rQaoialJrqv8gitZLxGLtM=
 github.com/go-openapi/jsonreference v0.19.5/go.mod h1:RdybgQwPxbL4UEjuAruzK1x3nE69AqPYEJeo/TWfEeg=
 github.com/go-openapi/spec v0.0.0-20160808142527-6aced65f8501/go.mod h1:J8+jY1nAiCcj+friV/PDoE1/3eeccG9LYBs0tYvLOWc=
-github.com/go-openapi/spec v0.19.3/go.mod h1:FpwSN1ksY1eteniUU7X0N/BgJ7a4WvBFVA8Lj9mJglo=
-github.com/go-openapi/spec v0.19.5/go.mod h1:Hm2Jr4jv8G1ciIAo+frC/Ft+rR2kQDh8JHKHb3gWUSk=
-github.com/go-openapi/spec v0.19.6/go.mod h1:Hm2Jr4jv8G1ciIAo+frC/Ft+rR2kQDh8JHKHb3gWUSk=
 github.com/go-openapi/swag v0.0.0-20160704191624-1d0bd113de87/go.mod h1:DXUve3Dpr1UfpPtxFw+EFuQ41HhCWZfha5jSVRG7C7I=
-github.com/go-openapi/swag v0.19.2/go.mod h1:POnQmlKehdgb5mhVOsnJFsivZCEZ/vjK9gh66Z9tfKk=
 github.com/go-openapi/swag v0.19.5/go.mod h1:POnQmlKehdgb5mhVOsnJFsivZCEZ/vjK9gh66Z9tfKk=
 github.com/go-openapi/swag v0.19.14/go.mod h1:QYRuS/SOXUCsnplDa677K7+DxSOj6IPNl/eQntq43wQ=
 github.com/go-openapi/swag v0.19.15 h1:D2NRCBzS9/pEY3gP9Nl8aDqGUcPFrwG2p+CNFrLyrCM=
@@ -324,7 +310,6 @@ github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
 github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
 github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b h1:VKtxabqXZkF25pY9ekfRL6a582T4P37/31XEstQ5p58=
 github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
-github.com/golang/groupcache v0.0.0-20160516000752-02826c3e7903/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
 github.com/golang/groupcache v0.0.0-20190129154638-5b532d6fd5ef/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
 github.com/golang/groupcache v0.0.0-20190702054246-869f871628b6/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
 github.com/golang/groupcache v0.0.0-20191227052852-215e87163ea7/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
@@ -419,7 +404,6 @@ github.com/google/pprof v0.0.0-20210601050228-01bbb1931b22/go.mod h1:kpwsk12EmLe
 github.com/google/pprof v0.0.0-20210609004039-a478d1d731e9/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE=
 github.com/google/pprof v0.0.0-20210720184732-4bb14d4b1be1/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE=
 github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
-github.com/google/uuid v1.0.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
@@ -439,15 +423,12 @@ github.com/gorilla/mux v1.6.2/go.mod h1:1lud6UwP+6orDFRuTfBEV8e9/aOM/c4fVVCaMa2z
 github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So=
 github.com/gorilla/securecookie v1.1.1/go.mod h1:ra0sb63/xPlUeL+yeDciTfxMRAA+MP+HVt/4epWDjd4=
 github.com/gorilla/sessions v1.2.1/go.mod h1:dk2InVEVJ0sfLlnXv9EAgkf6ecYs/i80K/zI+bUmuGM=
-github.com/gorilla/websocket v0.0.0-20170926233335-4201258b820c/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ=
 github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/gregjones/httpcache v0.0.0-20180305231024-9cad4c3443a7/go.mod h1:FecbI9+v66THATjSRHfNgh1IVFe/9kFxbXtjV0ctIMA=
 github.com/grpc-ecosystem/go-grpc-middleware v1.0.0/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs=
-github.com/grpc-ecosystem/go-grpc-middleware v1.0.1-0.20190118093823-f849b5445de4/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs=
 github.com/grpc-ecosystem/go-grpc-middleware v1.3.0/go.mod h1:z0ButlSOZa5vEBq9m2m2hlwIgKw+rp3sdCBRoJY+30Y=
 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0/go.mod h1:8NvIoxWQoOIhqOTXgfV/d3M/q6VIi02HzZEHgUlZvzk=
 github.com/grpc-ecosystem/grpc-gateway v1.9.0/go.mod h1:vNeuVxBJEsws4ogUvrchl83t/GYV9WGTSLVdBhOQFDY=
-github.com/grpc-ecosystem/grpc-gateway v1.9.5/go.mod h1:vNeuVxBJEsws4ogUvrchl83t/GYV9WGTSLVdBhOQFDY=
 github.com/grpc-ecosystem/grpc-gateway v1.14.6/go.mod h1:zdiPV4Yse/1gnckTHtghG4GkDEdKCRJduHpTxT3/jcw=
 github.com/grpc-ecosystem/grpc-gateway v1.16.0 h1:gmcG1KaJ57LophUzW0Hy8NmPhnMZb4M0+kPpLofRdBo=
 github.com/grpc-ecosystem/grpc-gateway v1.16.0/go.mod h1:BDjrQk3hbvj6Nolgz8mAMFbcEtjT1g+wF4CSlocrBnw=
@@ -514,7 +495,6 @@ github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFF
 github.com/jpillora/backoff v0.0.0-20180909062703-3050d21c67d7/go.mod h1:2iMrUgbbvHEiQClaW2NsSzMyGHqN+rDFqY705q49KG0=
 github.com/jpillora/backoff v1.0.0/go.mod h1:J/6gKK9jxlEcS3zixgDgUAsiuZ7yrSoa/FX5e0EB2j4=
 github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
-github.com/json-iterator/go v1.1.7/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
 github.com/json-iterator/go v1.1.10/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
 github.com/json-iterator/go v1.1.11 h1:uVUAXhF2To8cbw/3xN3pxj6kk7TYKs98NIrTqPlMWAQ=
 github.com/json-iterator/go v1.1.11/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
@@ -540,7 +520,6 @@ github.com/kr/pretty v0.2.0/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfn
 github.com/kr/pretty v0.2.1 h1:Fmg33tUaq4/8ym9TJN1x7sLJnHVwhP33CNkpYV/7rwI=
 github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
-github.com/kr/pty v1.1.5/go.mod h1:9r2w37qlBe7rQ6e1fg1S/9xpWHSnaqNdHD3WcMdbPDA=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
@@ -558,8 +537,6 @@ github.com/mailru/easyjson v0.7.7 h1:UGYAvKxe3sBsEDzO8ZeWOSlIQfWFlxbzLZe7hwFURr0
 github.com/mailru/easyjson v0.7.7/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc=
 github.com/mattn/go-colorable v0.0.9/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaOChaDxuIBZU=
 github.com/mattn/go-isatty v0.0.3/go.mod h1:M+lRXTBqGeGNdLjl/ufCoiOlB5xdOkqRJdNxMWT7Zi4=
-github.com/mattn/go-isatty v0.0.4/go.mod h1:M+lRXTBqGeGNdLjl/ufCoiOlB5xdOkqRJdNxMWT7Zi4=
-github.com/mattn/go-runewidth v0.0.2/go.mod h1:LwmH8dsx7+W8Uxz3IHJYH5QSwggIsqBzpuz5H//U1FU=
 github.com/mattn/goveralls v0.0.2/go.mod h1:8d1ZMHsd7fW6IRPKQh46F2WRpyib5/X4FOpevwGNQEw=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369 h1:I0XW9+e1XWDxdcEniV4rQAIOPUGDq67JSCiRCgGCZLI=
@@ -577,7 +554,6 @@ github.com/mitchellh/mapstructure v0.0.0-20160808181253-ca63d7c062ee/go.mod h1:F
 github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
 github.com/mitchellh/mapstructure v1.3.3/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo=
 github.com/moby/spdystream v0.2.0/go.mod h1:f7i0iNDQJ059oMTcWxx8MA/zKFIuD/lY+0GqbN2Wy8c=
-github.com/moby/term v0.0.0-20201216013528-df9cb8a40635/go.mod h1:FBS0z0QWA44HXygs7VXDUOGoN/1TV3RuWkLO04am3wc=
 github.com/moby/term v0.0.0-20210610120745-9d4ed1856297/go.mod h1:vgPCkQMyxTZ7IDy8SXRufE172gr8+K/JE/7hHFxHW3A=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg=
@@ -595,7 +571,6 @@ github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI
 github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=
 github.com/nxadm/tail v1.4.8/go.mod h1:+ncqLTQzXmGhMZNUePPaPqPvBxHAIsmXswZKocGu+AU=
 github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U=
-github.com/olekukonko/tablewriter v0.0.0-20170122224234-a0225b3f23b5/go.mod h1:vsDQFd/mU46D+Z4whnwzcISnGGzXWMclvtLoiIKAKIo=
 github.com/onsi/ginkgo v0.0.0-20170829012221-11459a886d9c/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
 github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
 github.com/onsi/ginkgo v1.7.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
@@ -668,7 +643,6 @@ github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R
 github.com/prometheus/procfs v0.0.0-20190507164030-5867b95ac084/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA=
 github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA=
 github.com/prometheus/procfs v0.1.3/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4OA4YeYWdaU=
-github.com/prometheus/procfs v0.2.0/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4OA4YeYWdaU=
 github.com/prometheus/procfs v0.6.0 h1:mxy4L2jP6qMonqmq+aTtOx1ifVWUgG/TAmntgbh3xv4=
 github.com/prometheus/procfs v0.6.0/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1xBZuNvfVA=
 github.com/prometheus/statsd_exporter v0.21.0 h1:hA05Q5RFeIjgwKIYEdFd59xu5Wwaznf33yKI+pyX6T8=
@@ -711,13 +685,10 @@ github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasO
 github.com/spf13/afero v1.1.2/go.mod h1:j4pytiNVoe2o6bmDsKpLACNPDBIoEAkihy7loJ1B0CQ=
 github.com/spf13/afero v1.2.2/go.mod h1:9ZxEEn6pIJ8Rxe320qSDBk6AsU0r9pR7Q4OcevTdifk=
 github.com/spf13/cast v1.3.0/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE=
-github.com/spf13/cobra v0.0.3/go.mod h1:1l0Ry5zgKvJasoi3XT1TypsSe7PqH0Sj9dhYf7v3XqQ=
 github.com/spf13/cobra v0.0.5/go.mod h1:3K3wKZymM7VvHMDS9+Akkh4K60UwM26emMESw8tLCHU=
-github.com/spf13/cobra v1.1.1/go.mod h1:WnodtKOvamDL/PwE2M4iKs8aMDBZ5Q5klgD3qfVJQMI=
 github.com/spf13/cobra v1.1.3/go.mod h1:pGADOWyqRD/YMrPZigI/zbliZ2wVD/23d+is3pSWzOo=
 github.com/spf13/jwalterweatherman v1.0.0/go.mod h1:cQK4TGJAtQXfYWX+Ddv3mKDzgVb68N+wFjFa4jdeBTo=
 github.com/spf13/pflag v0.0.0-20170130214245-9ff6c6923cff/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4=
-github.com/spf13/pflag v1.0.1/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4=
 github.com/spf13/pflag v1.0.3/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4=
 github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA=
 github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
@@ -739,7 +710,6 @@ github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
 github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/subosito/gotenv v1.2.0/go.mod h1:N0PQaV/YGNqwC0u51sEeR/aUtSLEXKX9iv69rRypqCw=
-github.com/tmc/grpc-websocket-proxy v0.0.0-20170815181823-89b8d40f7ca8/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20190109142713-0ad062ec5ee5/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20201229170055-e5319fda7802/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=
 github.com/tsenart/go-tsz v0.0.0-20180814232043-cdeb9e1e981e/go.mod h1:SWZznP1z5Ki7hDT2ioqiFKEse8K9tU2OUvaRI0NeGQo=
@@ -747,7 +717,6 @@ github.com/tsenart/vegeta/v12 v12.8.3/go.mod h1:ZiJtwLn/9M4fTPdMY7bdbIeyNeFVE8/A
 github.com/tsenart/vegeta/v12 v12.8.4 h1:UQ7tG7WkDorKj0wjx78Z4/vsMBP8RJQMGJqRVrkvngg=
 github.com/tsenart/vegeta/v12 v12.8.4/go.mod h1:ZiJtwLn/9M4fTPdMY7bdbIeyNeFVE8/AHbWFqCsUuho=
 github.com/ugorji/go/codec v0.0.0-20181204163529-d75b2dcb6bc8/go.mod h1:VFNgLljTbGfSG7qAOspJ7OScBnGdDN/yBr0sguwnwf0=
-github.com/urfave/cli v1.20.0/go.mod h1:70zkFmudgCuE/ngEzBv17Jvp/497gISqfk5gWijbERA=
 github.com/urfave/cli/v2 v2.3.0/go.mod h1:LJmUH05zAU44vOAcrfzZQKsZbVcdbOG8rtL3/XcUArI=
 github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
 github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
@@ -770,10 +739,7 @@ github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9dec
 github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
 github.com/yuin/goldmark v1.4.1/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
 go.etcd.io/bbolt v1.3.2/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU=
-go.etcd.io/bbolt v1.3.3/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU=
-go.etcd.io/bbolt v1.3.5/go.mod h1:G5EMThwa9y8QZGBClrRx5EY+Yw9kAhnjy3bSjsnlVTQ=
 go.etcd.io/bbolt v1.3.6/go.mod h1:qXsaaIqmgQH0T+OPdb99Bf+PKfBBQVAdyD6TY9G8XM4=
-go.etcd.io/etcd v0.5.0-alpha.5.0.20200910180754-dd1b699fc489/go.mod h1:yVHk9ub3CSBatqGNg7GRmsnfLWtoW60w4eDYfh7vHDg=
 go.etcd.io/etcd/api/v3 v3.5.0/go.mod h1:cbVKeC6lCfl7j/8jBhAK6aIYO9XOjdptoxU/nLQcPvs=
 go.etcd.io/etcd/client/pkg/v3 v3.5.0/go.mod h1:IJHfcCEKxYu1Os13ZdwCwIUTUVGYTSAM3YSwc9/Ac1g=
 go.etcd.io/etcd/client/v2 v2.305.0/go.mod h1:h9puh54ZTgAKtEbut2oe9P4L/oqKCVB6xsXlzd7alYQ=
@@ -805,7 +771,6 @@ go.opentelemetry.io/otel/sdk/metric v0.20.0/go.mod h1:knxiS8Xd4E/N+ZqKmUPf3gTTZ4
 go.opentelemetry.io/otel/trace v0.20.0 h1:1DL6EXUdcg95gukhuRRvLDO/4X5THh/5dIV52lqtnbw=
 go.opentelemetry.io/otel/trace v0.20.0/go.mod h1:6GjCW8zgDjwGHGa6GkyeB8+/5vjT16gUEi0Nf1iBdgw=
 go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqeYNgFYFoEGnI=
-go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
 go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
 go.uber.org/atomic v1.6.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
 go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
@@ -834,7 +799,6 @@ golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACk
 golang.org/x/crypto v0.0.0-20190404164418-38d8ce5564a5/go.mod h1:WFFai1msRO1wXaEeE5yQxYXgSfI8pQAWXbQop6sCtWE=
 golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20190605123033-f99c8df09eb5/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
-golang.org/x/crypto v0.0.0-20190611184440-5c40567a22f8/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20190829043050-9756ffdc2472/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20191206172530-e9b2fee46413/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
@@ -882,7 +846,6 @@ golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzB
 golang.org/x/mod v0.1.1-0.20191107180719-034126e5016b/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
 golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
-golang.org/x/mod v0.3.1-0.20200828183125-ce943fd02449/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.4.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.4.1/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
@@ -906,7 +869,6 @@ golang.org/x/net v0.0.0-20190613194153-d28f0bde5980/go.mod h1:z5CRVTTTmAJ677TzLL
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190628185345-da137c7871d7/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190724013045-ca1201d0de80/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
-golang.org/x/net v0.0.0-20190813141303-74dc4d7220e7/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190827160401-ba9fcec4b297/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20191002035440-2ec189313ef0/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20191209160850-c0dbc17a3553/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
@@ -932,7 +894,6 @@ golang.org/x/net v0.0.0-20201110031124-69a78807bb2b/go.mod h1:sp8m0HH+o8qH0wwXwY
 golang.org/x/net v0.0.0-20201202161906-c7110b5ffcbb/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20201209123823-ac852fbbde11/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210119194325-5f4716e94777/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
-golang.org/x/net v0.0.0-20210224082022-3d97a244fca7/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210316092652-d523dce5a7f4/go.mod h1:RBQZq4jEuRlivfhVLdyRGr576XBO4/greRjx4P4O3yc=
 golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
@@ -942,7 +903,6 @@ golang.org/x/net v0.0.0-20210525063256-abc453219eb5/go.mod h1:9nx3DQGgdP8bBQD5qx
 golang.org/x/net v0.0.0-20210917221730-978cfadd31cf/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20211015210444-4f30a5c0130f/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
-golang.org/x/net v0.0.0-20211205041911-012df41ee64c/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20211209124913-491a49abca63 h1:iocB37TsdFuN6IBRZ+ry36wrkoV51/tl5vOWqkcPGvY=
 golang.org/x/net v0.0.0-20211209124913-491a49abca63/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
@@ -991,10 +951,8 @@ golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20190502145724-3ef323f4f1fd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190507160741-ecd444e8653b/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190606165138-5da285871e9c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20190616124812-15dcb6c0061f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190624142023-c5567b49c5d0/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190726091711-fc99dfbffb4e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20190826190057-c7b8b68b1456/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190904154756-749cb33beabd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191001151750-bb3f8db39f24/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191005200804-aed5e4c7ecf9/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -1038,7 +996,6 @@ golang.org/x/sys v0.0.0-20210320140829-1e4c9ba3b0c4/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210403161142-5e06dd20ab57/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20210426230700-d19ff857e887/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210514084401-e8d321eab015/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210603081109-ebe580a85c40/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
@@ -1067,7 +1024,6 @@ golang.org/x/text v0.3.5/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7 h1:olpwvP2KacW1ZWvsR7uQhoyTYvKAupfQrRGBFM352Gk=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
-golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
@@ -1090,7 +1046,6 @@ golang.org/x/tools v0.0.0-20190425150028-36563e24a262/go.mod h1:RgjU9mgBXZiqYHBn
 golang.org/x/tools v0.0.0-20190506145303-2d16b83fe98c/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
 golang.org/x/tools v0.0.0-20190524140312-2c0ae7006135/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
 golang.org/x/tools v0.0.0-20190606124116-d0a3d012864b/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
-golang.org/x/tools v0.0.0-20190614205625-5aca471b1d59/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
 golang.org/x/tools v0.0.0-20190621195816-6e04913cbbac/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
 golang.org/x/tools v0.0.0-20190624222133-a101b041ded4/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
 golang.org/x/tools v0.0.0-20190628153133-6cdbf07be9d0/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
@@ -1235,7 +1190,6 @@ google.golang.org/genproto v0.0.0-20200825200019-8632dd797987/go.mod h1:FWY/as6D
 google.golang.org/genproto v0.0.0-20200904004341-0bd0a958aa1d/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
 google.golang.org/genproto v0.0.0-20201019141844-1ed22bb0c154/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
 google.golang.org/genproto v0.0.0-20201109203340-2640f1f9cdfb/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
-google.golang.org/genproto v0.0.0-20201110150050-8816d57aaa9a/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
 google.golang.org/genproto v0.0.0-20201201144952-b05cb90ed32e/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
 google.golang.org/genproto v0.0.0-20201210142538-e3217bee35cc/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
 google.golang.org/genproto v0.0.0-20201214200347-8c77b98c765d/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
@@ -1317,7 +1271,6 @@ gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8
 gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
-gopkg.in/cheggaaa/pb.v1 v1.0.25/go.mod h1:V/YB90LKu/1FcN3WVnfiiE5oMCibMjukxqG/qStrOgw=
 gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
 gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
 gopkg.in/inf.v0 v0.9.1 h1:73M5CoZyi3ZLMOyDlQh031Cx6N9NDJ2Vvfl76EDAgDc=
@@ -1357,27 +1310,20 @@ honnef.co/go/tools v0.0.1-2019.2.3/go.mod h1:a3bituU0lyd329TUQxRnasdCoJDkEUEAqEt
 honnef.co/go/tools v0.0.1-2020.1.3/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k=
 honnef.co/go/tools v0.0.1-2020.1.4/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k=
 honnef.co/go/tools v0.0.1-2020.1.5/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k=
-k8s.io/api v0.21.4/go.mod h1:fTVGP+M4D8+00FN2cMnJqk/eb/GH53bvmNs2SVTmpFk=
 k8s.io/api v0.22.5 h1:xk7C+rMjF/EGELiD560jdmwzrB788mfcHiNbMQLIVI8=
 k8s.io/api v0.22.5/go.mod h1:mEhXyLaSD1qTOf40rRiKXkc+2iCem09rWLlFwhCEiAs=
-k8s.io/apiextensions-apiserver v0.21.4/go.mod h1:OoC8LhI9LnV+wKjZkXIBbLUwtnOGJiTRE33qctH5CIk=
 k8s.io/apiextensions-apiserver v0.22.5 h1:ML0QqT7FIlmZHN+9+2EtARJ3cJVHeoizt6GCteFRE0o=
 k8s.io/apiextensions-apiserver v0.22.5/go.mod h1:tIXeZ0BrDxUb1PoAz+tgOz43Zi1Bp4BEEqVtUccMJbE=
 k8s.io/apimachinery v0.19.2/go.mod h1:DnPGDnARWFvYa3pMHgSxtbZb7gpzzAZ1pTfaUNDVlmA=
 k8s.io/apimachinery v0.19.7/go.mod h1:6sRbGRAVY5DOCuZwB5XkqguBqpqLU6q/kOaOdk29z6Q=
-k8s.io/apimachinery v0.21.4/go.mod h1:H/IM+5vH9kZRNJ4l3x/fXP/5bOPJaVP/guptnZPeCFI=
 k8s.io/apimachinery v0.22.5 h1:cIPwldOYm1Slq9VLBRPtEYpyhjIm1C6aAMAoENuvN9s=
 k8s.io/apimachinery v0.22.5/go.mod h1:xziclGKwuuJ2RM5/rSFQSYAj0zdbci3DH8kj+WvyN0U=
-k8s.io/apiserver v0.21.4/go.mod h1:SErUuFBBPZUcD2nsUU8hItxoYheqyYr2o/pCINEPW8g=
 k8s.io/apiserver v0.22.5 h1:71krQxCUz218ecb+nPhfDsNB6QgP1/4EMvi1a2uYBlg=
 k8s.io/apiserver v0.22.5/go.mod h1:s2WbtgZAkTKt679sYtSudEQrTGWUSQAPe6MupLnlmaQ=
-k8s.io/client-go v0.21.4/go.mod h1:t0/eMKyUAq/DoQ7vW8NVVA00/nomlwC+eInsS8PxSew=
 k8s.io/client-go v0.22.5 h1:I8Zn/UqIdi2r02aZmhaJ1hqMxcpfJ3t5VqvHtctHYFo=
 k8s.io/client-go v0.22.5/go.mod h1:cs6yf/61q2T1SdQL5Rdcjg9J1ElXSwbjSrW2vFImM4Y=
-k8s.io/code-generator v0.21.4/go.mod h1:K3y0Bv9Cz2cOW2vXUrNZlFbflhuPvuadW6JdnN6gGKo=
 k8s.io/code-generator v0.22.5 h1:jn+mYXI5q7rzo7Bz/n8xZIgbe61SeXlIjU5jA8jLVps=
 k8s.io/code-generator v0.22.5/go.mod h1:sbdWCOVob+KaQ5O7xs8PNNaCTpbWVqNgA6EPwLOmRNk=
-k8s.io/component-base v0.21.4/go.mod h1:ZKG0eHVX+tUDcaoIGpU3Vtk4TIjMddN9uhEWDmW6Nyg=
 k8s.io/component-base v0.22.5/go.mod h1:VK3I+TjuF9eaa+Ln67dKxhGar5ynVbwnGrUiNF4MqCI=
 k8s.io/gengo v0.0.0-20200413195148-3a45101e95ac/go.mod h1:ezvh/TsK7cY6rbqRK0oQQ8IAqLxYwwyPxAX1Pzy0ii0=
 k8s.io/gengo v0.0.0-20201203183100-97869a43a9d9/go.mod h1:FiNAH4ZV3gBg2Kwh89tzAEV2be7d5xI0vBa/VySYy3E=
@@ -1389,38 +1335,28 @@ k8s.io/klog v1.0.0 h1:Pt+yjF5aB1xDSVbau4VsWe+dQNzA0qv1LlXdC2dF6Q8=
 k8s.io/klog v1.0.0/go.mod h1:4Bi6QPql/J/LkTDqv7R/cd3hPo4k2DG6Ptcz060Ez5I=
 k8s.io/klog/v2 v2.0.0/go.mod h1:PBfzABfn139FHAV07az/IF9Wp1bkk3vpT2XSJ76fSDE=
 k8s.io/klog/v2 v2.2.0/go.mod h1:Od+F08eJP+W3HUb4pSrPpgp9DGU4GzlpG/TmITuYh/Y=
-k8s.io/klog/v2 v2.8.0/go.mod h1:hy9LJ/NvuK+iVyP4Ehqva4HxZG/oXyIS3n3Jmire4Ec=
 k8s.io/klog/v2 v2.9.0/go.mod h1:hy9LJ/NvuK+iVyP4Ehqva4HxZG/oXyIS3n3Jmire4Ec=
 k8s.io/klog/v2 v2.40.1 h1:P4RRucWk/lFOlDdkAr3mc7iWFkgKrZY9qZMAgek06S4=
 k8s.io/klog/v2 v2.40.1/go.mod h1:y1WjHnz7Dj687irZUWR/WLkLc5N1YHtjLdmgWjndZn0=
 k8s.io/kube-openapi v0.0.0-20200805222855-6aeccd4b50c6/go.mod h1:UuqjUnNftUyPE5H64/qeyjQoUZhGpeFDVdxjTeEVN2o=
-k8s.io/kube-openapi v0.0.0-20210305001622-591a79e4bda7/go.mod h1:wXW5VT87nVfh/iLV8FpR2uDvrFyomxbtb1KivDbvPTE=
 k8s.io/kube-openapi v0.0.0-20211109043538-20434351676c h1:jvamsI1tn9V0S8jicyX82qaFC0H/NKxv2e5mbqsgR80=
 k8s.io/kube-openapi v0.0.0-20211109043538-20434351676c/go.mod h1:vHXdDvt9+2spS2Rx9ql3I8tycm3H9FDfdUoIuKCefvw=
-k8s.io/utils v0.0.0-20201110183641-67b214c5f920/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 k8s.io/utils v0.0.0-20210819203725-bdf08cb9a70a/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 k8s.io/utils v0.0.0-20211208161948-7d6a63dca704 h1:ZKMMxTvduyf5WUtREOqg5LiXaN1KO/+0oOQPRFrClpo=
 k8s.io/utils v0.0.0-20211208161948-7d6a63dca704/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
-knative.dev/control-protocol v0.0.0-20211208065539-a3e65c058156/go.mod h1:LR9JdrBn9RoCLboRN9a0OSiq0/PIoi8WlQQMKR+SXG8=
-knative.dev/eventing v0.28.1-0.20220117084229-65c4a6ff8ef4/go.mod h1:SKbTN3yj9f5ywji6CYOUy80cj8kvTPxiQ43r+gdEf9A=
-knative.dev/eventing v0.28.1-0.20220118072832-f6fec1088847 h1:PGdWief8mLiOe00o0is5XFyTI4y+Pb7M7p9Gy8Xq0Yw=
-knative.dev/eventing v0.28.1-0.20220118072832-f6fec1088847/go.mod h1:XqlxlmtiPimqTJ0FDMQuxRWbRqLjRl0bsEvpWKWP2lk=
-knative.dev/eventing-kafka v0.28.1-0.20220117141029-46249ce2a5f4 h1:2kwm4pcbZySntBDic6verkbm/k/HUVW89KFpyqHNnck=
-knative.dev/eventing-kafka v0.28.1-0.20220117141029-46249ce2a5f4/go.mod h1:V16boyvALgpKl9x+jh9hRo/fytc1mLct/QYTiYHWTqI=
-knative.dev/hack v0.0.0-20211122162614-813559cefdda/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/hack v0.0.0-20211203062838-e11ac125e707/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/hack v0.0.0-20220111151514-59b0cf17578e h1:tESsFI1iMD2Al+6FZueFN9CddxJ8Fa2cd9KmyNINe3w=
-knative.dev/hack v0.0.0-20220111151514-59b0cf17578e/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/hack/schema v0.0.0-20220111151514-59b0cf17578e/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
-knative.dev/pkg v0.0.0-20211206113427-18589ac7627e/go.mod h1:E6B4RTjZyxe55a0kxOlnEHEl71zuG7gghnqYvNBKwBw=
-knative.dev/pkg v0.0.0-20220113045912-c0e1594c2fb1/go.mod h1:X5B0/SR3tzOioS2WQtS9uJTO71M52qcrDkWQ5wtyLVc=
-knative.dev/pkg v0.0.0-20220114141842-0a429cba1c73/go.mod h1:X5B0/SR3tzOioS2WQtS9uJTO71M52qcrDkWQ5wtyLVc=
-knative.dev/pkg v0.0.0-20220118025732-e989c22732aa h1:bkUK53H+iNO1KfjHs8g01pcb9yud3xhBcG1kCWfyW3Y=
-knative.dev/pkg v0.0.0-20220118025732-e989c22732aa/go.mod h1:X5B0/SR3tzOioS2WQtS9uJTO71M52qcrDkWQ5wtyLVc=
-knative.dev/reconciler-test v0.0.0-20211207070557-0d138a88867b/go.mod h1:dCq1Fuu+eUISdnxABMvoDhefF91DYwE6O3rTYTraXbw=
-knative.dev/reconciler-test v0.0.0-20220113155112-af7be729d5e7/go.mod h1:zhcMJ0CfqI7de0Ir0yXTTB6BgShWvLEx+BqQdvg+1pU=
-knative.dev/reconciler-test v0.0.0-20220117082429-6a9b91eef10c h1:Vz1kKJ46VEExuIKu54nDBxT4XvWBcZ50Wg2HlZgAD9k=
-knative.dev/reconciler-test v0.0.0-20220117082429-6a9b91eef10c/go.mod h1:zhcMJ0CfqI7de0Ir0yXTTB6BgShWvLEx+BqQdvg+1pU=
+knative.dev/control-protocol v0.0.0-20220118185434-ad4ce9e9d455/go.mod h1:xiTC5Qt9GdPuhRknr0Oq744SsUqbkDx0oe04L9a+qYY=
+knative.dev/eventing v0.28.1-0.20220119171353-ad895de5fdd8/go.mod h1:u5T5NZTDUsLR7yJwp5MDnBnDX5MhywD3yK3Rq+7gTtI=
+knative.dev/eventing v0.28.1-0.20220124132429-9e0e90c9d603 h1:y1ms8nohoiqUD/Nmt0ARj4ZH79jrSjjIJ16tMstH+Eo=
+knative.dev/eventing v0.28.1-0.20220124132429-9e0e90c9d603/go.mod h1:u5T5NZTDUsLR7yJwp5MDnBnDX5MhywD3yK3Rq+7gTtI=
+knative.dev/eventing-kafka v0.28.1-0.20220121065504-fe10ecf57e92 h1:Saz5jV6m4MBhFJ5uc0fNnafp70zR/Uh1iFHaCU+87RU=
+knative.dev/eventing-kafka v0.28.1-0.20220121065504-fe10ecf57e92/go.mod h1:UVCY5kWcBfyp4N6pJ84dVR0mz8pd14jMTBP4GQDjOPY=
+knative.dev/hack v0.0.0-20220118141833-9b2ed8471e30 h1:UkNpCWCMM5C4AeQ8aTrPTuR/6OeARiqk+LEQ6tuMP7c=
+knative.dev/hack v0.0.0-20220118141833-9b2ed8471e30/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
+knative.dev/hack/schema v0.0.0-20220118141833-9b2ed8471e30/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
+knative.dev/pkg v0.0.0-20220118160532-77555ea48cd4 h1:b9aXVrcfM/ajjHE/lGvlJOHZNAR5FF2TOTLWG7eMhzQ=
+knative.dev/pkg v0.0.0-20220118160532-77555ea48cd4/go.mod h1:etVT7Tm8pSDf4RKhGk4r7j/hj3dNBpvT7bO6a6wpahs=
+knative.dev/reconciler-test v0.0.0-20220118183433-c8bfbe66bada h1:YSxoVY3Z2GF8QsNrWlJ8qQlJ6pUXKVsX7xkhY6odOXU=
+knative.dev/reconciler-test v0.0.0-20220118183433-c8bfbe66bada/go.mod h1:XV4cghzCtdASkfUsfMYSnGfGRyd/naDjy9h7Tnae22g=
 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 543f19938..45e9ed909 100644
--- a/proto/contract.proto
+++ b/proto/contract.proto
@@ -98,6 +98,13 @@ message Egress {
 
   // Kafka record key type.
   KeyType keyType = 10;
+
+  // Resource reference.
+  //
+  // This reference is used to reference the associated resource for data plane
+  // activities such as:
+  // - tagging metrics
+  Reference reference = 11;
 }
 
 // CloudEvent content mode
diff --git a/vendor/knative.dev/eventing/pkg/apis/eventing/v1/trigger_validation.go b/vendor/knative.dev/eventing/pkg/apis/eventing/v1/trigger_validation.go
index 043315d27..ac033c1d8 100644
--- a/vendor/knative.dev/eventing/pkg/apis/eventing/v1/trigger_validation.go
+++ b/vendor/knative.dev/eventing/pkg/apis/eventing/v1/trigger_validation.go
@@ -198,8 +198,7 @@ func ValidateCESQLExpression(ctx context.Context, expression string) (errs *apis
 	if expression == "" {
 		return nil
 	}
-	_, err := cesqlparser.Parse(expression)
-	if err != nil {
+	if _, err := cesqlparser.Parse(expression); err != nil {
 		return apis.ErrInvalidValue(expression, apis.CurrentField, err.Error())
 	}
 	return nil
diff --git a/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake/fake.go b/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake/fake.go
new file mode 100644
index 000000000..eb18a1fac
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake/fake.go
@@ -0,0 +1,40 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	context "context"
+
+	fake "knative.dev/eventing/pkg/client/injection/informers/factory/fake"
+	subscription "knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+)
+
+var Get = subscription.Get
+
+func init() {
+	injection.Fake.RegisterInformer(withInformer)
+}
+
+func withInformer(ctx context.Context) (context.Context, controller.Informer) {
+	f := fake.Get(ctx)
+	inf := f.Messaging().V1().Subscriptions()
+	return context.WithValue(ctx, subscription.Key{}, inf), inf.Informer()
+}
diff --git a/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/subscription.go b/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/subscription.go
new file mode 100644
index 000000000..37862d8a5
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/subscription.go
@@ -0,0 +1,116 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package subscription
+
+import (
+	context "context"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	cache "k8s.io/client-go/tools/cache"
+	apismessagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
+	versioned "knative.dev/eventing/pkg/client/clientset/versioned"
+	v1 "knative.dev/eventing/pkg/client/informers/externalversions/messaging/v1"
+	client "knative.dev/eventing/pkg/client/injection/client"
+	factory "knative.dev/eventing/pkg/client/injection/informers/factory"
+	messagingv1 "knative.dev/eventing/pkg/client/listers/messaging/v1"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+func init() {
+	injection.Default.RegisterInformer(withInformer)
+	injection.Dynamic.RegisterDynamicInformer(withDynamicInformer)
+}
+
+// Key is used for associating the Informer inside the context.Context.
+type Key struct{}
+
+func withInformer(ctx context.Context) (context.Context, controller.Informer) {
+	f := factory.Get(ctx)
+	inf := f.Messaging().V1().Subscriptions()
+	return context.WithValue(ctx, Key{}, inf), inf.Informer()
+}
+
+func withDynamicInformer(ctx context.Context) context.Context {
+	inf := &wrapper{client: client.Get(ctx), resourceVersion: injection.GetResourceVersion(ctx)}
+	return context.WithValue(ctx, Key{}, inf)
+}
+
+// Get extracts the typed informer from the context.
+func Get(ctx context.Context) v1.SubscriptionInformer {
+	untyped := ctx.Value(Key{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch knative.dev/eventing/pkg/client/informers/externalversions/messaging/v1.SubscriptionInformer from context.")
+	}
+	return untyped.(v1.SubscriptionInformer)
+}
+
+type wrapper struct {
+	client versioned.Interface
+
+	namespace string
+
+	resourceVersion string
+}
+
+var _ v1.SubscriptionInformer = (*wrapper)(nil)
+var _ messagingv1.SubscriptionLister = (*wrapper)(nil)
+
+func (w *wrapper) Informer() cache.SharedIndexInformer {
+	return cache.NewSharedIndexInformer(nil, &apismessagingv1.Subscription{}, 0, nil)
+}
+
+func (w *wrapper) Lister() messagingv1.SubscriptionLister {
+	return w
+}
+
+func (w *wrapper) Subscriptions(namespace string) messagingv1.SubscriptionNamespaceLister {
+	return &wrapper{client: w.client, namespace: namespace, resourceVersion: w.resourceVersion}
+}
+
+// SetResourceVersion allows consumers to adjust the minimum resourceVersion
+// used by the underlying client.  It is not accessible via the standard
+// lister interface, but can be accessed through a user-defined interface and
+// an implementation check e.g. rvs, ok := foo.(ResourceVersionSetter)
+func (w *wrapper) SetResourceVersion(resourceVersion string) {
+	w.resourceVersion = resourceVersion
+}
+
+func (w *wrapper) List(selector labels.Selector) (ret []*apismessagingv1.Subscription, err error) {
+	lo, err := w.client.MessagingV1().Subscriptions(w.namespace).List(context.TODO(), metav1.ListOptions{
+		LabelSelector:   selector.String(),
+		ResourceVersion: w.resourceVersion,
+	})
+	if err != nil {
+		return nil, err
+	}
+	for idx := range lo.Items {
+		ret = append(ret, &lo.Items[idx])
+	}
+	return ret, nil
+}
+
+func (w *wrapper) Get(name string) (*apismessagingv1.Subscription, error) {
+	return w.client.MessagingV1().Subscriptions(w.namespace).Get(context.TODO(), name, metav1.GetOptions{
+		ResourceVersion: w.resourceVersion,
+	})
+}
diff --git a/vendor/knative.dev/pkg/version/version.go b/vendor/knative.dev/pkg/version/version.go
index 56b7ae958..b304f2b0a 100644
--- a/vendor/knative.dev/pkg/version/version.go
+++ b/vendor/knative.dev/pkg/version/version.go
@@ -33,7 +33,7 @@ const (
 	// NOTE: If you are changing this line, please also update the minimum kubernetes
 	// version listed here:
 	// https://github.com/knative/docs/blob/mkdocs/docs/snippets/prerequisites.md
-	defaultMinimumVersion = "v1.20.0"
+	defaultMinimumVersion = "v1.21.0"
 )
 
 func getMinimumVersion() string {
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 8597fc4c5..1b0c9f168 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1124,7 +1124,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.28.1-0.20220118072832-f6fec1088847
+# knative.dev/eventing v0.28.1-0.20220124132429-9e0e90c9d603
 ## explicit
 knative.dev/eventing/pkg/apis/config
 knative.dev/eventing/pkg/apis/duck
@@ -1179,6 +1179,8 @@ knative.dev/eventing/pkg/client/injection/informers/eventing/v1/trigger
 knative.dev/eventing/pkg/client/injection/informers/eventing/v1/trigger/fake
 knative.dev/eventing/pkg/client/injection/informers/factory
 knative.dev/eventing/pkg/client/injection/informers/factory/fake
+knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription
+knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake
 knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker
 knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/trigger
 knative.dev/eventing/pkg/client/listers/eventing/v1
@@ -1259,7 +1261,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.28.1-0.20220117141029-46249ce2a5f4
+# knative.dev/eventing-kafka v0.28.1-0.20220121065504-fe10ecf57e92
 ## explicit
 knative.dev/eventing-kafka/pkg/apis/bindings
 knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1
@@ -1320,11 +1322,11 @@ knative.dev/eventing-kafka/test/rekt/features/kafkachannel
 knative.dev/eventing-kafka/test/rekt/resources/kafkachannel
 knative.dev/eventing-kafka/test/rekt/resources/resetoffset
 knative.dev/eventing-kafka/test/upgrade/continual
-# knative.dev/hack v0.0.0-20220111151514-59b0cf17578e
+# knative.dev/hack v0.0.0-20220118141833-9b2ed8471e30
 ## explicit
 knative.dev/hack
 knative.dev/hack/shell
-# knative.dev/pkg v0.0.0-20220118025732-e989c22732aa
+# knative.dev/pkg v0.0.0-20220118160532-77555ea48cd4
 ## explicit
 knative.dev/pkg/apis
 knative.dev/pkg/apis/duck
@@ -1419,7 +1421,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-20220117082429-6a9b91eef10c
+# knative.dev/reconciler-test v0.0.0-20220118183433-c8bfbe66bada
 ## explicit
 knative.dev/reconciler-test/cmd/eventshub
 knative.dev/reconciler-test/pkg/environment


Test patch:
diff --git a/control-plane/pkg/reconciler/channel/channel_test.go b/control-plane/pkg/reconciler/channel/channel_test.go
index 64f9e2831..ffd577dbc 100644
--- a/control-plane/pkg/reconciler/channel/channel_test.go
+++ b/control-plane/pkg/reconciler/channel/channel_test.go
@@ -22,6 +22,7 @@ import (
 	"testing"
 
 	"github.com/Shopify/sarama"
+	eventingrekttesting "knative.dev/eventing/pkg/reconciler/testing/v1"
 	"knative.dev/pkg/network"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
@@ -350,6 +351,9 @@ func TestReconcileKind(t *testing.T) {
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
+				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
+					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
+				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -372,6 +376,11 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
+								Reference: &contract.Reference{
+									Name:      Subscription1Name,
+									Namespace: ChannelNamespace,
+									Uuid:      Subscription1UUID,
+								},
 							}},
 						},
 					},
@@ -425,6 +434,9 @@ func TestReconcileKind(t *testing.T) {
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
+				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
+					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
+				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -447,6 +459,11 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
+								Reference: &contract.Reference{
+									Name:      Subscription1Name,
+									Namespace: ChannelNamespace,
+									Uuid:      Subscription1UUID,
+								},
 							}},
 						},
 					},
@@ -500,6 +517,9 @@ func TestReconcileKind(t *testing.T) {
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
+				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
+					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
+				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -522,6 +542,11 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
+								Reference: &contract.Reference{
+									Name:      Subscription1Name,
+									Namespace: ChannelNamespace,
+									Uuid:      Subscription1UUID,
+								},
 							}},
 						},
 					},
@@ -578,6 +603,12 @@ func TestReconcileKind(t *testing.T) {
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
+				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
+					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
+				),
+				eventingrekttesting.NewSubscription(Subscription2Name, ChannelNamespace,
+					eventingrekttesting.WithSubscriptionUID(Subscription2UUID),
+				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -600,12 +631,22 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
+								Reference: &contract.Reference{
+									Name:      Subscription1Name,
+									Namespace: ChannelNamespace,
+									Uuid:      Subscription1UUID,
+								},
 							}, {
 								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription2UUID,
 								Destination:   "http://" + Subscription2URI,
 								Uid:           Subscription2UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_DiscardReply{},
+								Reference: &contract.Reference{
+									Name:      Subscription2Name,
+									Namespace: ChannelNamespace,
+									Uuid:      Subscription2UUID,
+								},
 							}},
 						},
 					},
@@ -933,6 +974,9 @@ func TestReconcileKind(t *testing.T) {
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
 				NewSSLSecret(system.Namespace(), "secret-1"),
+				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
+					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
+				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -963,6 +1007,11 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
+								Reference: &contract.Reference{
+									Name:      Subscription1Name,
+									Namespace: ChannelNamespace,
+									Uuid:      Subscription1UUID,
+								},
 							}},
 						},
 					},
@@ -1217,8 +1266,9 @@ func useTable(t *testing.T, table TableTest, env config.Env) {
 				DispatcherLabel:             base.ChannelDispatcherLabel,
 				ReceiverLabel:               base.ChannelReceiverLabel,
 			},
-			Env:             env,
-			ConfigMapLister: listers.GetConfigMapLister(),
+			Env:                env,
+			ConfigMapLister:    listers.GetConfigMapLister(),
+			SubscriptionLister: listers.GetSubscriptionLister(),
 			InitOffsetsFunc: func(ctx context.Context, kafkaClient sarama.Client, kafkaAdminClient sarama.ClusterAdmin, topics []string, consumerGroup string) (int32, error) {
 				return 1, nil
 			},
diff --git a/control-plane/pkg/reconciler/channel/controller_test.go b/control-plane/pkg/reconciler/channel/controller_test.go
index cc3ca16ce..f3a49c227 100644
--- a/control-plane/pkg/reconciler/channel/controller_test.go
+++ b/control-plane/pkg/reconciler/channel/controller_test.go
@@ -24,6 +24,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	fakekubeclientset "k8s.io/client-go/kubernetes/fake"
 	_ "knative.dev/eventing-kafka/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/fake"
+	_ "knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake"
 	_ "knative.dev/pkg/client/injection/ducks/duck/v1/addressable/fake"
 	fakekubeclient "knative.dev/pkg/client/injection/kube/client/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/fake"
diff --git a/control-plane/pkg/reconciler/consumergroup/evictor_test.go b/control-plane/pkg/reconciler/consumergroup/evictor_test.go
new file mode 100644
index 000000000..e334ede0d
--- /dev/null
+++ b/control-plane/pkg/reconciler/consumergroup/evictor_test.go
@@ -0,0 +1,155 @@
+/*
+ * Copyright 2022 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package consumergroup
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/pointer"
+	eventingduckv1alpha1 "knative.dev/eventing/pkg/apis/duck/v1alpha1"
+	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
+	reconcilertesting "knative.dev/pkg/reconciler/testing"
+
+	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
+	_ "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client/fake"
+	kafkainternalsclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client/fake"
+)
+
+func TestNewEvictor(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	require.NotPanics(t, func() { newEvictor(ctx, zap.String("k", "n")) })
+}
+
+func TestEvictorEvictSuccess(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "name"},
+	}
+	cg := &kafkainternals.ConsumerGroup{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "ns-1", Name: "cg-name"},
+		Status: kafkainternals.ConsumerGroupStatus{
+			PlaceableStatus: eventingduckv1alpha1.PlaceableStatus{Placeable: eventingduckv1alpha1.Placeable{
+				MaxAllowedVReplicas: pointer.Int32(1),
+				Placements: []eventingduckv1alpha1.Placement{
+					{PodName: pod.GetName(), VReplicas: 1},
+				},
+			}},
+		},
+	}
+	cg.GetConditionSet().Manage(cg.GetStatus()).InitializeConditions()
+	cg.MarkScheduleSucceeded()
+
+	placement := &eventingduckv1alpha1.Placement{PodName: pod.GetName(), VReplicas: 1}
+
+	ctx, _ = kubeclient.With(ctx, pod)
+	ctx, _ = kafkainternalsclient.With(ctx, cg)
+
+	e := newEvictor(ctx)
+	err := e.evict(pod, cg, placement)
+
+	require.Nil(t, err)
+}
+
+func TestEvictorEvictSuccessConsumerGroupSchedulingInProgress(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "name"},
+	}
+	cg := &kafkainternals.ConsumerGroup{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "ns-1", Name: "cg-name"},
+		Status: kafkainternals.ConsumerGroupStatus{
+			PlaceableStatus: eventingduckv1alpha1.PlaceableStatus{Placeable: eventingduckv1alpha1.Placeable{
+				MaxAllowedVReplicas: pointer.Int32(1),
+				Placements: []eventingduckv1alpha1.Placement{
+					{PodName: pod.GetName(), VReplicas: 1},
+				},
+			}},
+		},
+	}
+	cg.GetConditionSet().Manage(cg.GetStatus()).InitializeConditions()
+
+	placement := &eventingduckv1alpha1.Placement{PodName: pod.GetName(), VReplicas: 1}
+
+	ctx, _ = kubeclient.With(ctx, pod)
+	ctx, _ = kafkainternalsclient.With(ctx, cg)
+
+	e := newEvictor(ctx)
+	err := e.evict(pod, cg, placement)
+
+	require.True(t, cg.IsNotScheduled())
+	require.Nil(t, err)
+}
+
+func TestEvictorEvictPodNotFound(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "name"},
+	}
+	cg := &kafkainternals.ConsumerGroup{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "ns-1", Name: "cg-name"},
+		Status: kafkainternals.ConsumerGroupStatus{
+			PlaceableStatus: eventingduckv1alpha1.PlaceableStatus{Placeable: eventingduckv1alpha1.Placeable{
+				MaxAllowedVReplicas: pointer.Int32(1),
+				Placements: []eventingduckv1alpha1.Placement{
+					{PodName: pod.GetName(), VReplicas: 1},
+				},
+			}},
+		},
+	}
+	placement := &eventingduckv1alpha1.Placement{PodName: pod.GetName(), VReplicas: 1}
+
+	ctx, _ = kafkainternalsclient.With(ctx, cg)
+
+	e := newEvictor(ctx)
+	err := e.evict(pod, cg, placement)
+
+	require.NotNil(t, err)
+}
+func TestEvictorEvictConsumerGroupNotFound(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "name"},
+	}
+	cg := &kafkainternals.ConsumerGroup{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "ns-1", Name: "cg-name"},
+		Status: kafkainternals.ConsumerGroupStatus{
+			PlaceableStatus: eventingduckv1alpha1.PlaceableStatus{Placeable: eventingduckv1alpha1.Placeable{
+				MaxAllowedVReplicas: pointer.Int32(1),
+				Placements: []eventingduckv1alpha1.Placement{
+					{PodName: pod.GetName(), VReplicas: 1},
+				},
+			}},
+		},
+	}
+	placement := &eventingduckv1alpha1.Placement{PodName: pod.GetName(), VReplicas: 1}
+
+	ctx, _ = kubeclient.With(ctx, pod)
+
+	e := newEvictor(ctx)
+	err := e.evict(pod, cg, placement)
+
+	require.Nil(t, err)
+}
diff --git a/control-plane/pkg/reconciler/source/source_test.go b/control-plane/pkg/reconciler/source/source_test.go
index d9b86eca8..b3736ca3a 100644
--- a/control-plane/pkg/reconciler/source/source_test.go
+++ b/control-plane/pkg/reconciler/source/source_test.go
@@ -105,10 +105,11 @@ func TestReconcileKind(t *testing.T) {
 									EgressConfig:  &DefaultEgressConfig,
 									DeliveryOrder: DefaultDeliveryOrder,
 									ReplyStrategy: &contract.Egress_DiscardReply{},
+									Reference:     SourceReference(),
 								},
 							},
 							Auth:      &contract.Resource_AbsentAuth{},
-							Reference: &contract.Reference{Namespace: SourceNamespace, Name: SourceName},
+							Reference: SourceReference(),
 						},
 					},
 				}),
@@ -170,10 +171,11 @@ func TestReconcileKind(t *testing.T) {
 									EgressConfig:  &DefaultEgressConfig,
 									DeliveryOrder: DefaultDeliveryOrder,
 									ReplyStrategy: &contract.Egress_DiscardReply{},
+									Reference:     SourceReference(),
 								},
 							},
 							Auth:      &contract.Resource_AbsentAuth{},
-							Reference: &contract.Reference{Namespace: SourceNamespace, Name: SourceName},
+							Reference: SourceReference(),
 						},
 					},
 				}),
@@ -234,10 +236,11 @@ func TestReconcileKind(t *testing.T) {
 									DeliveryOrder: DefaultDeliveryOrder,
 									KeyType:       contract.KeyType_String,
 									ReplyStrategy: &contract.Egress_DiscardReply{},
+									Reference:     SourceReference(),
 								},
 							},
 							Auth:      &contract.Resource_AbsentAuth{},
-							Reference: &contract.Reference{Namespace: SourceNamespace, Name: SourceName},
+							Reference: SourceReference(),
 						},
 					},
 				}),
@@ -296,10 +299,11 @@ func TestReconcileKind(t *testing.T) {
 									DeliveryOrder: DefaultDeliveryOrder,
 									KeyType:       contract.KeyType_Integer,
 									ReplyStrategy: &contract.Egress_DiscardReply{},
+									Reference:     SourceReference(),
 								},
 							},
 							Auth:      &contract.Resource_AbsentAuth{},
-							Reference: &contract.Reference{Namespace: SourceNamespace, Name: SourceName},
+							Reference: SourceReference(),
 						},
 					},
 				}),
@@ -358,10 +362,11 @@ func TestReconcileKind(t *testing.T) {
 									DeliveryOrder: DefaultDeliveryOrder,
 									KeyType:       contract.KeyType_ByteArray,
 									ReplyStrategy: &contract.Egress_DiscardReply{},
+									Reference:     SourceReference(),
 								},
 							},
 							Auth:      &contract.Resource_AbsentAuth{},
-							Reference: &contract.Reference{Namespace: SourceNamespace, Name: SourceName},
+							Reference: SourceReference(),
 						},
 					},
 				}),
@@ -420,10 +425,11 @@ func TestReconcileKind(t *testing.T) {
 									DeliveryOrder: DefaultDeliveryOrder,
 									KeyType:       contract.KeyType_Double,
 									ReplyStrategy: &contract.Egress_DiscardReply{},
+									Reference:     SourceReference(),
 								},
 							},
 							Auth:      &contract.Resource_AbsentAuth{},
-							Reference: &contract.Reference{Namespace: SourceNamespace, Name: SourceName},
+							Reference: SourceReference(),
 						},
 					},
 				}),
diff --git a/control-plane/pkg/reconciler/source/v2/controllerv2_test.go b/control-plane/pkg/reconciler/source/v2/controllerv2_test.go
index e378c42b1..ddf18dc2c 100644
--- a/control-plane/pkg/reconciler/source/v2/controllerv2_test.go
+++ b/control-plane/pkg/reconciler/source/v2/controllerv2_test.go
@@ -61,7 +61,7 @@ func TestNewControllerV2(t *testing.T) {
 
 	dynamicclient.With(ctx, dynamicScheme)
 
-	controller := NewControllerV2(ctx, configs)
+	controller := NewController(ctx, configs)
 	if controller == nil {
 		t.Error("failed to create controller: <nil>")
 	}
diff --git a/control-plane/pkg/reconciler/source/v2/sourcev2_test.go b/control-plane/pkg/reconciler/source/v2/sourcev2_test.go
new file mode 100644
index 000000000..f920ae1f9
--- /dev/null
+++ b/control-plane/pkg/reconciler/source/v2/sourcev2_test.go
@@ -0,0 +1,402 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package v2
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	clientgotesting "k8s.io/client-go/testing"
+	"knative.dev/pkg/kmeta"
+
+	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/logging"
+	. "knative.dev/pkg/reconciler/testing"
+
+	fakeconsumergroupinformer "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client/fake"
+	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	fakeeventingkafkaclient "knative.dev/eventing-kafka/pkg/client/injection/client/fake"
+	eventingkafkasourcereconciler "knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource"
+
+	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/testing"
+)
+
+func TestReconcileKind(t *testing.T) {
+
+	testKey := fmt.Sprintf("%s/%s", SourceNamespace, SourceName)
+
+	sources.RegisterAlternateKafkaConditionSet(conditionSet)
+
+	table := TableTest{
+		{
+			Name: "Reconciled normal",
+			Objects: []runtime.Object{
+				NewSource(),
+			},
+			Key: testKey,
+			WantCreates: []runtime.Object{
+				NewConsumerGroup(
+					WithConsumerGroupName(SourceUUID),
+					WithConsumerGroupNamespace(SourceNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(SourceTopics[0], SourceTopics[1]),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(SourceConsumerGroup),
+							ConsumerBootstrapServersConfig(SourceBootstrapServers),
+						),
+						ConsumerAuth(NewConsumerSpecAuth()),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+						ConsumerSubscriber(NewSourceSinkReference()),
+					)),
+					ConsumerGroupReplicas(1),
+				),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSource(
+						StatusSourceConsumerGroupUnknown(),
+						StatusSourceSinkResolved(""),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - ce overrides",
+			Objects: []runtime.Object{
+				NewSourceSinkObject(),
+				NewSource(WithCloudEventOverrides(&duckv1.CloudEventOverrides{
+					Extensions: map[string]string{"a": "foo", "b": "foo"},
+				})),
+			},
+			Key: testKey,
+			WantCreates: []runtime.Object{
+				NewConsumerGroup(
+					WithConsumerGroupName(SourceUUID),
+					WithConsumerGroupNamespace(SourceNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(SourceTopics[0], SourceTopics[1]),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(SourceConsumerGroup),
+							ConsumerBootstrapServersConfig(SourceBootstrapServers),
+						),
+						ConsumerAuth(NewConsumerSpecAuth()),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerCloudEventOverrides(&duckv1.CloudEventOverrides{
+							Extensions: map[string]string{"a": "foo", "b": "foo"},
+						}),
+					)),
+					ConsumerGroupReplicas(1),
+				),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSource(
+						WithCloudEventOverrides(&duckv1.CloudEventOverrides{
+							Extensions: map[string]string{"a": "foo", "b": "foo"},
+						}),
+						StatusSourceConsumerGroupUnknown(),
+						StatusSourceSinkResolved(""),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - existing cg with update",
+			Objects: []runtime.Object{
+				NewSource(),
+				NewConsumerGroup(
+					WithConsumerGroupName(SourceUUID),
+					WithConsumerGroupNamespace(SourceNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupReplicas(1),
+					ConsumerGroupReady,
+				),
+			},
+			Key:         testKey,
+			WantCreates: []runtime.Object{},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewConsumerGroup(
+						WithConsumerGroupName(SourceUUID),
+						WithConsumerGroupNamespace(SourceNamespace),
+						WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+						WithConsumerGroupLabels(nil),
+						ConsumerGroupConsumerSpec(NewConsumerSpec(
+							ConsumerTopics(SourceTopics[0], SourceTopics[1]),
+							ConsumerConfigs(
+								ConsumerGroupIdConfig(SourceConsumerGroup),
+								ConsumerBootstrapServersConfig(SourceBootstrapServers),
+							),
+							ConsumerAuth(NewConsumerSpecAuth()),
+							ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+							ConsumerSubscriber(NewSourceSinkReference()),
+						)),
+						ConsumerGroupReplicas(1),
+						ConsumerGroupReady,
+					),
+				},
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSource(
+						StatusSourceConsumerGroup(),
+						StatusSourceSinkResolved(""),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - existing cg with update but not ready",
+			Objects: []runtime.Object{
+				NewSource(),
+				NewConsumerGroup(
+					WithConsumerGroupName(SourceUUID),
+					WithConsumerGroupNamespace(SourceNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupReplicas(1),
+				),
+			},
+			Key:         testKey,
+			WantCreates: []runtime.Object{},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewConsumerGroup(
+						WithConsumerGroupName(SourceUUID),
+						WithConsumerGroupNamespace(SourceNamespace),
+						WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+						WithConsumerGroupLabels(nil),
+						ConsumerGroupConsumerSpec(NewConsumerSpec(
+							ConsumerTopics(SourceTopics[0], SourceTopics[1]),
+							ConsumerConfigs(
+								ConsumerGroupIdConfig(SourceConsumerGroup),
+								ConsumerBootstrapServersConfig(SourceBootstrapServers),
+							),
+							ConsumerAuth(NewConsumerSpecAuth()),
+							ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+							ConsumerSubscriber(NewSourceSinkReference()),
+						)),
+						ConsumerGroupReplicas(1),
+					),
+				},
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSource(
+						StatusSourceConsumerGroupUnknown(),
+						StatusSourceSinkResolved(""),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - existing cg without update",
+			Objects: []runtime.Object{
+				NewSource(),
+				NewConsumerGroup(
+					WithConsumerGroupName(SourceUUID),
+					WithConsumerGroupNamespace(SourceNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(SourceTopics[0], SourceTopics[1]),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(SourceConsumerGroup),
+							ConsumerBootstrapServersConfig(SourceBootstrapServers),
+						),
+						ConsumerAuth(NewConsumerSpecAuth()),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+						ConsumerSubscriber(NewSourceSinkReference()),
+					)),
+					ConsumerGroupReplicas(1),
+					ConsumerGroupReady,
+				),
+			},
+			Key:         testKey,
+			WantCreates: []runtime.Object{},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSource(
+						StatusSourceConsumerGroup(),
+						StatusSourceSinkResolved(""),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - existing cg without update but not ready",
+			Objects: []runtime.Object{
+				NewSource(),
+				NewConsumerGroup(
+					WithConsumerGroupName(SourceUUID),
+					WithConsumerGroupNamespace(SourceNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(SourceTopics[0], SourceTopics[1]),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(SourceConsumerGroup),
+							ConsumerBootstrapServersConfig(SourceBootstrapServers),
+						),
+						ConsumerAuth(NewConsumerSpecAuth()),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+						ConsumerSubscriber(NewSourceSinkReference()),
+					)),
+					ConsumerGroupReplicas(1),
+				),
+			},
+			Key:         testKey,
+			WantCreates: []runtime.Object{},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSource(
+						StatusSourceConsumerGroupUnknown(),
+						StatusSourceSinkResolved(""),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - existing cg but failed",
+			Objects: []runtime.Object{
+				NewSource(),
+				NewConsumerGroup(
+					WithConsumerGroupName(SourceUUID),
+					WithConsumerGroupNamespace(SourceNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(SourceTopics[0], SourceTopics[1]),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(SourceConsumerGroup),
+							ConsumerBootstrapServersConfig(SourceBootstrapServers),
+						),
+						ConsumerAuth(NewConsumerSpecAuth()),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+						ConsumerSubscriber(NewSourceSinkReference()),
+					)),
+					ConsumerGroupReplicas(1),
+					WithConsumerGroupFailed("failed", "failed"),
+				),
+			},
+			Key:         testKey,
+			WantCreates: []runtime.Object{},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSource(
+						StatusSourceConsumerGroupFailed("failed", "failed"),
+						StatusSourceSinkResolved(""),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - existing cg with replicas set in status",
+			Objects: []runtime.Object{
+				NewSource(),
+				NewConsumerGroup(
+					WithConsumerGroupName(SourceUUID),
+					WithConsumerGroupNamespace(SourceNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(SourceTopics[0], SourceTopics[1]),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(SourceConsumerGroup),
+							ConsumerBootstrapServersConfig(SourceBootstrapServers),
+						),
+						ConsumerAuth(NewConsumerSpecAuth()),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+						ConsumerSubscriber(NewSourceSinkReference()),
+					)),
+					ConsumerGroupReplicas(1),
+					ConsumerGroupReplicasStatus(1),
+					ConsumerGroupReady,
+				),
+			},
+			Key:         testKey,
+			WantCreates: []runtime.Object{},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSource(
+						StatusSourceConsumerGroup(),
+						StatusSourceSinkResolved(""),
+						StatusSourceConsumerGroupReplicas(1),
+					),
+				},
+			},
+		},
+	}
+
+	table.Test(t, NewFactory(nil, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
+
+		reconciler := &Reconciler{
+			ConsumerGroupLister: listers.GetConsumerGroupLister(),
+			InternalsClient:     fakeconsumergroupinformer.Get(ctx),
+		}
+
+		r := eventingkafkasourcereconciler.NewReconciler(
+			ctx,
+			logging.FromContext(ctx),
+			fakeeventingkafkaclient.Get(ctx),
+			listers.GetKafkaSourceLister(),
+			controller.GetEventRecorder(ctx),
+			reconciler,
+		)
+		return r
+	}))
+}
+
+func StatusSourceConsumerGroup() KRShapedOption {
+	return func(obj duckv1.KRShaped) {
+		ks := obj.(*sources.KafkaSource)
+		ks.GetConditionSet().Manage(ks.GetStatus()).MarkTrue(KafkaConditionConsumerGroup)
+	}
+}
+
+func StatusSourceConsumerGroupFailed(reason string, msg string) KRShapedOption {
+	return func(obj duckv1.KRShaped) {
+		ks := obj.(*sources.KafkaSource)
+		ks.GetConditionSet().Manage(ks.GetStatus()).MarkFalse(KafkaConditionConsumerGroup, reason, msg)
+	}
+}
+
+func StatusSourceConsumerGroupUnknown() KRShapedOption {
+	return func(obj duckv1.KRShaped) {
+		ks := obj.(*sources.KafkaSource)
+		ks.GetConditionSet().Manage(ks.GetStatus()).MarkUnknown(KafkaConditionConsumerGroup, "failed to reconcile consumer group", "consumer group not ready")
+	}
+}
+
+func StatusSourceConsumerGroupReplicas(replicas int32) KRShapedOption {
+	return func(obj duckv1.KRShaped) {
+		ks := obj.(*sources.KafkaSource)
+		ks.Status.Consumers = replicas
+	}
+}
diff --git a/control-plane/pkg/reconciler/testing/listers.go b/control-plane/pkg/reconciler/testing/listers.go
index 779496b29..3fd1175ef 100644
--- a/control-plane/pkg/reconciler/testing/listers.go
+++ b/control-plane/pkg/reconciler/testing/listers.go
@@ -24,8 +24,10 @@ import (
 	corelisters "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/tools/cache"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
+	messaging "knative.dev/eventing/pkg/apis/messaging/v1"
 	fakeeventingclientset "knative.dev/eventing/pkg/client/clientset/versioned/fake"
 	eventinglisters "knative.dev/eventing/pkg/client/listers/eventing/v1"
+	messaginglisters "knative.dev/eventing/pkg/client/listers/messaging/v1"
 	"knative.dev/pkg/reconciler/testing"
 
 	eventingkafkachannels "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
@@ -120,6 +122,10 @@ func (l *Listers) GetTriggerLister() eventinglisters.TriggerLister {
 	return eventinglisters.NewTriggerLister(l.indexerFor(&eventing.Trigger{}))
 }
 
+func (l *Listers) GetSubscriptionLister() messaginglisters.SubscriptionLister {
+	return messaginglisters.NewSubscriptionLister(l.indexerFor(&messaging.Subscription{}))
+}
+
 func (l *Listers) GetConfigMapLister() corelisters.ConfigMapLister {
 	return corelisters.NewConfigMapLister(l.indexerFor(&corev1.ConfigMap{}))
 }
diff --git a/control-plane/pkg/reconciler/testing/objects_broker.go b/control-plane/pkg/reconciler/testing/objects_broker.go
index 5e1e368bf..7ab0e6936 100644
--- a/control-plane/pkg/reconciler/testing/objects_broker.go
+++ b/control-plane/pkg/reconciler/testing/objects_broker.go
@@ -47,6 +47,9 @@ const (
 	BrokerNamespace = "test-namespace"
 	BrokerName      = "test-broker"
 
+	TriggerName      = "test-trigger"
+	TriggerNamespace = "test-namespace"
+
 	ConfigMapFinalizerName = "kafka.brokers.eventing.knative.dev/" + BrokerNamespace + "-" + BrokerName
 )
 
@@ -240,6 +243,14 @@ func BrokerReference() *contract.Reference {
 	}
 }
 
+func TriggerReference() *contract.Reference {
+	return &contract.Reference{
+		Uuid:      TriggerUUID,
+		Namespace: TriggerNamespace,
+		Name:      TriggerName,
+	}
+}
+
 func BrokerDLSResolved(uri string) func(broker *eventing.Broker) {
 	return func(broker *eventing.Broker) {
 		broker.Status.DeadLetterSinkURI, _ = apis.ParseURL(uri)
diff --git a/control-plane/pkg/reconciler/testing/objects_consumergroup.go b/control-plane/pkg/reconciler/testing/objects_consumergroup.go
index 72f40c1be..a569f4f9a 100644
--- a/control-plane/pkg/reconciler/testing/objects_consumergroup.go
+++ b/control-plane/pkg/reconciler/testing/objects_consumergroup.go
@@ -115,6 +115,12 @@ func ConsumerGroupReplicas(replicas int32) ConsumerGroupOption {
 	}
 }
 
+func ConsumerGroupReplicasStatus(replicas int32) ConsumerGroupOption {
+	return func(cg *kafkainternals.ConsumerGroup) {
+		cg.Status.Replicas = pointer.Int32Ptr(replicas)
+	}
+}
+
 func ConsumerGroupSelector(selector map[string]string) ConsumerGroupOption {
 	return func(cg *kafkainternals.ConsumerGroup) {
 		cg.Spec.Selector = selector
diff --git a/control-plane/pkg/reconciler/testing/objects_source.go b/control-plane/pkg/reconciler/testing/objects_source.go
index 04ce2639f..bb79545db 100644
--- a/control-plane/pkg/reconciler/testing/objects_source.go
+++ b/control-plane/pkg/reconciler/testing/objects_source.go
@@ -25,6 +25,7 @@ import (
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 )
 
@@ -156,3 +157,11 @@ func StatusSourceSinkNotResolved(err string) KRShapedOption {
 		)
 	}
 }
+
+func SourceReference() *contract.Reference {
+	return &contract.Reference{
+		Namespace: SourceNamespace,
+		Name:      SourceName,
+		Uuid:      SourceUUID,
+	}
+}
diff --git a/control-plane/pkg/reconciler/trigger/trigger_test.go b/control-plane/pkg/reconciler/trigger/trigger_test.go
index 437bdb85a..114f6a7db 100644
--- a/control-plane/pkg/reconciler/trigger/trigger_test.go
+++ b/control-plane/pkg/reconciler/trigger/trigger_test.go
@@ -24,12 +24,13 @@ import (
 
 	"google.golang.org/protobuf/testing/protocmp"
 	"k8s.io/utils/pointer"
-	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/tracker"
 
+	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 
 	"github.com/google/go-cmp/cmp"
@@ -53,13 +54,6 @@ import (
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/testing"
 )
 
-const (
-	// name of the trigger under test
-	triggerName = "test-trigger"
-	// namespace of the trigger under test
-	triggerNamespace = "test-namespace"
-)
-
 var DefaultEnv = &config.Env{
 	DataPlaneConfigMapNamespace: "knative-eventing",
 	DataPlaneConfigMapName:      "kafka-broker-brokers-triggers",
@@ -74,7 +68,7 @@ var (
 	finalizerUpdatedEvent = Eventf(
 		corev1.EventTypeNormal,
 		"FinalizerUpdate",
-		fmt.Sprintf(`Updated %q finalizers`, triggerName),
+		fmt.Sprintf(`Updated %q finalizers`, TriggerName),
 	)
 
 	url = &apis.URL{
@@ -98,7 +92,7 @@ func TestTriggerReconciler(t *testing.T) {
 
 func triggerReconciliation(t *testing.T, format string, env config.Env) {
 
-	testKey := fmt.Sprintf("%s/%s", triggerNamespace, triggerName)
+	testKey := fmt.Sprintf("%s/%s", TriggerNamespace, TriggerName)
 
 	env.DataPlaneConfigFormat = format
 
@@ -141,6 +135,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Reference:     TriggerReference(),
 								},
 							},
 						},
@@ -205,6 +200,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
 									EgressConfig:  &contract.EgressConfig{DeadLetter: ServiceURL},
+									Reference:     TriggerReference(),
 								},
 							},
 						},
@@ -267,6 +263,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Reference:     TriggerReference(),
 									EgressConfig: &contract.EgressConfig{
 										DeadLetter:    url.String(),
 										Retry:         3,
@@ -337,6 +334,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Reference:     TriggerReference(),
 									DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								},
 							},
@@ -401,6 +399,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Reference:     TriggerReference(),
 									DeliveryOrder: contract.DeliveryOrder_UNORDERED,
 								},
 							},
@@ -465,6 +464,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Reference:     TriggerReference(),
 									EgressConfig: &contract.EgressConfig{
 										DeadLetter:    url.String(),
 										Retry:         3,
@@ -517,6 +517,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Reference:     TriggerReference(),
 								},
 							},
 						},
@@ -540,6 +541,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Reference:     TriggerReference(),
 								},
 							},
 						},
@@ -567,6 +569,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Reference:     TriggerReference(),
 								},
 							},
 						},
@@ -587,6 +590,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Reference:     TriggerReference(),
 								},
 							},
 						},
@@ -807,6 +811,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Reference:     TriggerReference(),
 								},
 							},
 						},
@@ -874,6 +879,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Reference:     TriggerReference(),
 									Filter: &contract.Filter{Attributes: map[string]string{
 										"type": "type1",
 									}},
@@ -932,6 +938,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Reference:     TriggerReference(),
 								},
 								{
 									Destination:   "http://example.com/1",
@@ -1012,6 +1019,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Reference:     TriggerReference(),
 									Filter: &contract.Filter{Attributes: map[string]string{
 										"type": "type1",
 									}},
@@ -1248,6 +1256,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Reference:     TriggerReference(),
 									Filter: &contract.Filter{Attributes: map[string]string{
 										"ext": "extval",
 									}},
@@ -1459,6 +1468,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Reference:     TriggerReference(),
 								},
 								{
 									Destination:   "http://example.com/3",
@@ -1580,7 +1590,7 @@ func TestTriggerFinalizer(t *testing.T) {
 
 func triggerFinalizer(t *testing.T, format string, env config.Env) {
 
-	testKey := fmt.Sprintf("%s/%s", triggerNamespace, triggerName)
+	testKey := fmt.Sprintf("%s/%s", TriggerNamespace, TriggerName)
 
 	env.DataPlaneConfigFormat = format
 
@@ -2316,8 +2326,8 @@ func useTable(t *testing.T, table TableTest, env *config.Env) {
 
 func newTrigger(options ...reconcilertesting.TriggerOption) runtime.Object {
 	return reconcilertesting.NewTrigger(
-		triggerName,
-		triggerNamespace,
+		TriggerName,
+		TriggerNamespace,
 		BrokerName,
 		append(
 			options,
@@ -2368,8 +2378,8 @@ func withTriggerSubscriberResolvedSucceeded(deliveryOrder contract.DeliveryOrder
 
 func patchFinalizers() clientgotesting.PatchActionImpl {
 	action := clientgotesting.PatchActionImpl{}
-	action.Name = triggerName
-	action.Namespace = triggerNamespace
+	action.Name = TriggerName
+	action.Namespace = TriggerNamespace
 	patch := `{"metadata":{"finalizers":["` + FinalizerName + `"],"resourceVersion":""}}`
 	action.Patch = []byte(patch)
 	return action
diff --git a/vendor/knative.dev/eventing/test/e2e-rekt-tests.sh b/vendor/knative.dev/eventing/test/e2e-rekt-tests.sh
index 3ab65cff3..33c535b09 100644
--- a/vendor/knative.dev/eventing/test/e2e-rekt-tests.sh
+++ b/vendor/knative.dev/eventing/test/e2e-rekt-tests.sh
@@ -35,6 +35,6 @@ initialize $@ --skip-istio-addon
 export SKIP_UPLOAD_TEST_IMAGES="true"
 
 echo "Running E2E Reconciler Tests"
-go_test_e2e -timeout=30m -parallel=20 ./test/rekt || fail_test
+go_test_e2e -timeout=1h -parallel=20 ./test/rekt || fail_test
 
 success
diff --git a/vendor/knative.dev/eventing/test/rekt/features/broker/control_plane.go b/vendor/knative.dev/eventing/test/rekt/features/broker/control_plane.go
index a2592181d..936f1c4e4 100644
--- a/vendor/knative.dev/eventing/test/rekt/features/broker/control_plane.go
+++ b/vendor/knative.dev/eventing/test/rekt/features/broker/control_plane.go
@@ -27,8 +27,14 @@ import (
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/wait"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/ptr"
+	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/state"
+	"knative.dev/reconciler-test/resources/svc"
+
 	v1 "knative.dev/eventing/pkg/apis/duck/v1"
 	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
 	eventingclientsetv1 "knative.dev/eventing/pkg/client/clientset/versioned/typed/eventing/v1"
@@ -39,14 +45,6 @@ import (
 	brokerresources "knative.dev/eventing/test/rekt/resources/broker"
 	"knative.dev/eventing/test/rekt/resources/delivery"
 	triggerresources "knative.dev/eventing/test/rekt/resources/trigger"
-	"knative.dev/pkg/apis"
-	duckv1 "knative.dev/pkg/apis/duck/v1"
-	"knative.dev/pkg/injection/clients/dynamicclient"
-	"knative.dev/pkg/ptr"
-	"knative.dev/reconciler-test/pkg/environment"
-	"knative.dev/reconciler-test/pkg/feature"
-	"knative.dev/reconciler-test/pkg/state"
-	"knative.dev/reconciler-test/resources/svc"
 )
 
 func ControlPlaneConformance(brokerName string) *feature.FeatureSet {
@@ -314,12 +312,18 @@ func addControlPlaneDelivery(fs *feature.FeatureSet) {
 		t2FailCount: 4, // Should end up in DLQ.
 	}, {
 		name: "When `TriggerSpec.Delivery` is configured, then `TriggerSpec.Delivery` SHOULD be used. (Retry)",
+		brokerDS: &v1.DeliverySpec{ // Disable delivery spec defaulting
+			Retry: ptr.Int32(0),
+		},
 		t1DS: &v1.DeliverySpec{
 			DeadLetterSink: new(duckv1.Destination),
 			Retry:          ptr.Int32(3),
 		},
+		t2DS: &v1.DeliverySpec{
+			Retry: ptr.Int32(1),
+		},
 		t1FailCount: 3, // Should get event.
-		t2FailCount: 1, // Should be dropped.
+		t2FailCount: 2, // Should be dropped.
 	}, {
 		name: "When both `BrokerSpec.Delivery` and `TriggerSpec.Delivery` is configured, then `TriggerSpec.Delivery` SHOULD be used. (Retry)",
 		brokerDS: &v1.DeliverySpec{
@@ -385,30 +389,11 @@ func addControlPlaneDelivery(fs *feature.FeatureSet) {
 			}
 		})
 		f.Stable("Conformance").Should(tt.name, knconf.AssertEventPatterns(prober, expectedEvents))
-		f.Teardown("Delete feature resources", deleteFeatureResources())
+		f.Teardown("Delete feature resources", f.DeleteResources)
 		fs.Features = append(fs.Features, *f)
 	}
 }
 
-func deleteFeatureResources() feature.StepFn {
-	return func(ctx context.Context, t feature.T) {
-		dc := dynamicclient.Get(ctx)
-		f := feature.FromContext(ctx)
-		for _, ref := range f.References() {
-			gv, err := schema.ParseGroupVersion(ref.APIVersion)
-			if err != nil {
-				t.Errorf("Could not parse GroupVersion for %+v", ref.APIVersion)
-			} else {
-				resource := apis.KindToResource(gv.WithKind(ref.Kind))
-				t.Logf("Deleting %s/%s of GVR: %+v", ref.Namespace, ref.Name, resource)
-				if err := dc.Resource(resource).Namespace(ref.Namespace).Delete(ctx, ref.Name, *metav1.NewDeleteOptions(0)); err != nil {
-					t.Logf("Failed to delete during cleanup %s/%s of GVR: %+v", ref.Namespace, ref.Name, resource)
-				}
-			}
-		}
-	}
-}
-
 func addControlPlaneEventRouting(fs *feature.FeatureSet) {
 
 	fullEvent := cetest.FullEvent()
@@ -551,7 +536,7 @@ func addControlPlaneEventRouting(fs *feature.FeatureSet) {
 		})
 
 		f.Stable("Conformance").Should(tt.name, assertExpectedRoutedEvents(prober, expectedEvents))
-		f.Teardown("Delete feature resources", deleteFeatureResources())
+		f.Teardown("Delete feature resources", f.DeleteResources)
 		fs.Features = append(fs.Features, *f)
 	}
 }
diff --git a/vendor/knative.dev/eventing/test/rekt/features/broker/delivery.go b/vendor/knative.dev/eventing/test/rekt/features/broker/delivery.go
new file mode 100644
index 000000000..39141ce8e
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/rekt/features/broker/delivery.go
@@ -0,0 +1,43 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package broker
+
+import (
+	"knative.dev/reconciler-test/pkg/feature"
+
+	"knative.dev/eventing/test/rekt/resources/broker"
+)
+
+func DefaultDeliverySpec() *feature.Feature {
+	f := feature.NewFeatureNamed("default delivery spec")
+
+	brokerName := feature.MakeRandomK8sName("broker")
+
+	f.Setup("install broker", broker.Install(brokerName, broker.WithEnvConfig()...))
+
+	f.Requirement("broker is ready", broker.IsReady(brokerName))
+	f.Requirement("broker is addressable", broker.IsAddressable(brokerName))
+
+	f.Assert("broker has delivery spec set", broker.WaitForCondition(brokerName,
+		broker.HasDelivery().
+			And(broker.HasDeliveryRetry()).
+			And(broker.HasDeliveryBackoffDelay()).
+			And(broker.HasDeliveryBackoffPolicy()),
+	))
+
+	return f
+}
diff --git a/vendor/knative.dev/eventing/test/rekt/features/broker/readyness.go b/vendor/knative.dev/eventing/test/rekt/features/broker/readyness.go
index bc5811242..86b01a6ff 100644
--- a/vendor/knative.dev/eventing/test/rekt/features/broker/readyness.go
+++ b/vendor/knative.dev/eventing/test/rekt/features/broker/readyness.go
@@ -22,11 +22,11 @@ import (
 	"knative.dev/reconciler-test/pkg/eventshub"
 	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/manifest"
+	"knative.dev/reconciler-test/resources/svc"
 
 	"knative.dev/eventing/test/rekt/resources/broker"
 	"knative.dev/eventing/test/rekt/resources/delivery"
 	"knative.dev/eventing/test/rekt/resources/trigger"
-	"knative.dev/reconciler-test/resources/svc"
 )
 
 // TriggerGoesReady returns a feature that tests after the creation of a
@@ -67,7 +67,7 @@ func GoesReady(name string, cfg ...manifest.CfgFn) *feature.Feature {
 	return f
 }
 
-// GoesReady returns a feature that will create a Broker of the given
+// GoesReadyWithProbeReceiver returns a feature that will create a Broker of the given
 // name and class, with an event receiver as its DLS
 // and confirm it becomes ready with an address.
 func GoesReadyWithProbeReceiver(name, sinkName string, prober *eventshub.EventProber, cfg ...manifest.CfgFn) *feature.Feature {
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/broker/broker.go b/vendor/knative.dev/eventing/test/rekt/resources/broker/broker.go
index 6d84ae245..e94b10cf7 100644
--- a/vendor/knative.dev/eventing/test/rekt/resources/broker/broker.go
+++ b/vendor/knative.dev/eventing/test/rekt/resources/broker/broker.go
@@ -24,13 +24,20 @@ import (
 	"time"
 
 	"github.com/kelseyhightower/envconfig"
+	"go.uber.org/multierr"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
-	"knative.dev/eventing/test/rekt/resources/addressable"
-	"knative.dev/eventing/test/rekt/resources/delivery"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"knative.dev/pkg/apis"
+	"knative.dev/reconciler-test/pkg/environment"
 	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/k8s"
 	"knative.dev/reconciler-test/pkg/manifest"
+
+	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
+	eventingclient "knative.dev/eventing/pkg/client/injection/client"
+	"knative.dev/eventing/test/rekt/resources/addressable"
+	"knative.dev/eventing/test/rekt/resources/delivery"
 )
 
 //go:embed *.yaml
@@ -134,3 +141,90 @@ func IsAddressable(name string, timings ...time.Duration) feature.StepFn {
 func Address(ctx context.Context, name string, timings ...time.Duration) (*apis.URL, error) {
 	return addressable.Address(ctx, GVR(), name, timings...)
 }
+
+type Condition struct {
+	Name      string
+	Condition func(br eventingv1.Broker) (bool, error)
+}
+
+func (c Condition) And(other Condition) Condition {
+	return c.compose(other, func(c1, c2 bool) bool { return c1 && c2 })
+}
+
+func (c Condition) compose(other Condition, combineFunc func(c1, c2 bool) bool) Condition {
+	return Condition{
+		Name: c.Name + " + " + other.Name,
+		Condition: func(br eventingv1.Broker) (bool, error) {
+			c1, err1 := c.Condition(br)
+			c2, err2 := other.Condition(br)
+			return combineFunc(c1, c2), multierr.Append(err1, err2)
+		},
+	}
+}
+
+func WaitForCondition(name string, condition Condition, timing ...time.Duration) feature.StepFn {
+	return func(ctx context.Context, t feature.T) {
+		env := environment.FromContext(ctx)
+		interval, timeout := k8s.PollTimings(ctx, timing)
+		var lastErr error
+		var lastBroker *eventingv1.Broker
+		err := wait.PollImmediate(interval, timeout, func() (done bool, err error) {
+			br, err := eventingclient.Get(ctx).
+				EventingV1().
+				Brokers(env.Namespace()).
+				Get(ctx, name, metav1.GetOptions{})
+			if err != nil {
+				lastErr = err
+				return false, nil
+			}
+
+			lastBroker = br
+			return condition.Condition(*br)
+		})
+		if err != nil {
+			t.Fatalf("failed to verify condition %s %v: %v\n%+v\n", condition.Name, err, lastErr, lastBroker)
+		}
+	}
+}
+
+func HasDelivery() Condition {
+	return Condition{
+		Name: "has delivery",
+		Condition: func(br eventingv1.Broker) (bool, error) {
+			return br.Spec.Delivery != nil, nil
+		},
+	}
+}
+
+func HasDeliveryRetry() Condition {
+	return Condition{
+		Name: "has delivery retry",
+		Condition: func(br eventingv1.Broker) (bool, error) {
+			return br.Spec.Delivery != nil &&
+				br.Spec.Delivery.Retry != nil &&
+				*br.Spec.Delivery.Retry > 0, nil
+		},
+	}
+}
+
+func HasDeliveryBackoffDelay() Condition {
+	return Condition{
+		Name: "has delivery backoff delay",
+		Condition: func(br eventingv1.Broker) (bool, error) {
+			return br.Spec.Delivery != nil &&
+				br.Spec.Delivery.BackoffDelay != nil &&
+				len(*br.Spec.Delivery.BackoffDelay) > 0, nil
+		},
+	}
+}
+
+func HasDeliveryBackoffPolicy() Condition {
+	return Condition{
+		Name: "has delivery backoff policy",
+		Condition: func(br eventingv1.Broker) (bool, error) {
+			return br.Spec.Delivery != nil &&
+				br.Spec.Delivery.BackoffPolicy != nil &&
+				len(*br.Spec.Delivery.BackoffPolicy) > 0, nil
+		},
+	}
+}
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/trigger/trigger.yaml b/vendor/knative.dev/eventing/test/rekt/resources/trigger/trigger.yaml
index 797414940..1bcc87c87 100644
--- a/vendor/knative.dev/eventing/test/rekt/resources/trigger/trigger.yaml
+++ b/vendor/knative.dev/eventing/test/rekt/resources/trigger/trigger.yaml
@@ -34,6 +34,10 @@ spec:
       {{ $key }}: "{{ $value }}"
     {{ end }}
   {{ end }}
+  {{ if .filters }}
+  filters:
+  {{ .filters }}
+  {{ end }}
   {{ if .subscriber }}
   subscriber:
     {{ if .subscriber.ref }}
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/configuration.go b/vendor/knative.dev/eventing/test/upgrade/prober/configuration.go
index 0307daf8b..963963c45 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/configuration.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/configuration.go
@@ -40,7 +40,6 @@ const (
 	defaultHomedir           = "/home/nonroot"
 	defaultConfigFilename    = "config.toml"
 	defaultHealthEndpoint    = "/healthz"
-	defaultFinishedSleep     = 5 * time.Second
 
 	// Silence will suppress notification about event duplicates.
 	Silence DuplicateAction = "silence"
@@ -63,12 +62,11 @@ type DuplicateAction string
 // Config represents a configuration for prober.
 type Config struct {
 	Wathola
-	Interval      time.Duration
-	FinishedSleep time.Duration
-	Serving       ServingConfig
-	FailOnErrors  bool
-	OnDuplicate   DuplicateAction
-	Ctx           context.Context
+	Interval     time.Duration
+	Serving      ServingConfig
+	FailOnErrors bool
+	OnDuplicate  DuplicateAction
+	Ctx          context.Context
 }
 
 // Wathola represents options related strictly to wathola testing tool.
@@ -109,11 +107,10 @@ func NewConfigOrFail(c pkgupgrade.Context) *Config {
 // NewConfig will create a prober.Config or return error.
 func NewConfig() (*Config, error) {
 	config := &Config{
-		Interval:      Interval,
-		FinishedSleep: defaultFinishedSleep,
-		FailOnErrors:  true,
-		OnDuplicate:   Warn,
-		Ctx:           context.Background(),
+		Interval:     Interval,
+		FailOnErrors: true,
+		OnDuplicate:  Warn,
+		Ctx:          context.Background(),
 		Serving: ServingConfig{
 			Use:         false,
 			ScaleToZero: true,
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/prober.go b/vendor/knative.dev/eventing/test/upgrade/prober/prober.go
index bb8908684..f63424b5b 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/prober.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/prober.go
@@ -67,7 +67,6 @@ func (p *probeRunner) Verify(ctx pkgupgrade.Context) {
 	p.client.T = ctx.T
 	t := ctx.T
 	p.Finish()
-	waitAfterFinished(p.prober)
 
 	errors, events := p.prober.Verify()
 	if len(errors) == 0 {
@@ -137,8 +136,3 @@ func (p *prober) remove() {
 	}
 	p.ensureNoError(p.client.Tracker.Clean(true))
 }
-
-func waitAfterFinished(p *prober) {
-	p.log.Infof("Waiting %v after sender finished...", p.config.FinishedSleep)
-	time.Sleep(p.config.FinishedSleep)
-}
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/sender.go b/vendor/knative.dev/eventing/test/upgrade/prober/sender.go
index 5f3bdd4ec..ed5376dd3 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/sender.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/sender.go
@@ -16,13 +16,18 @@
 package prober
 
 import (
+	"encoding/json"
 	"fmt"
+	"time"
 
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	testlib "knative.dev/eventing/test/lib"
+	"k8s.io/apimachinery/pkg/util/wait"
 	pkgTest "knative.dev/pkg/test"
+
+	testlib "knative.dev/eventing/test/lib"
 )
 
 var senderName = "wathola-sender"
@@ -88,8 +93,30 @@ func (p *prober) deploySender() {
 func (p *prober) removeSender() {
 	p.log.Info("Remove of sender deployment: ", senderName)
 
+	foreground := metav1.DeletePropagationForeground
+	dOpts := metav1.DeleteOptions{PropagationPolicy: &foreground}
 	err := p.client.Kube.AppsV1().
 		Deployments(p.client.Namespace).
-		Delete(p.config.Ctx, senderName, metav1.DeleteOptions{})
+		Delete(p.config.Ctx, senderName, dOpts)
 	p.ensureNoError(err)
+
+	var d *appsv1.Deployment
+	pollErr := wait.PollImmediate(time.Second, time.Minute, func() (done bool, err error) {
+		// Save err and deployment for error reporting.
+		d, err = p.client.Kube.AppsV1().
+			Deployments(p.client.Namespace).
+			Get(p.config.Ctx, senderName, metav1.GetOptions{})
+		if apierrors.IsNotFound(err) {
+			return true, nil
+		}
+		if err != nil {
+			return false, err
+		}
+		return false, nil
+	})
+
+	if pollErr != nil {
+		b, _ := json.MarshalIndent(d, "", " ")
+		p.client.T.Fatalf("Failed while waiting for sender deletion %v: %v\nDeployment: \n%s\n", pollErr, err, string(b))
+	}
 }
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/verify.go b/vendor/knative.dev/eventing/test/upgrade/prober/verify.go
index 90ce927a6..0b1e5eda3 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/verify.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/verify.go
@@ -37,23 +37,29 @@ import (
 const (
 	fetcherName     = "wathola-fetcher"
 	jobWaitInterval = time.Second
-	jobWaitTimeout  = 5 * time.Minute
+	jobWaitTimeout  = 10 * time.Minute
 )
 
 // Verify will verify prober state after finished has been sent.
 func (p *prober) Verify() (eventErrs []error, eventsSent int) {
-	report := p.fetchReport()
+	var report *receiver.Report
+	p.log.Info("Waiting for complete report from receiver...")
+	start := time.Now()
+	if err := wait.PollImmediate(jobWaitInterval, jobWaitTimeout, func() (bool, error) {
+		report = p.fetchReport()
+		return report.State != "active", nil
+	}); err != nil {
+		p.client.T.Fatalf("Error fetching complete/inactive report: %v\nReport: %+v", err, report)
+	}
+	elapsed := time.Since(start)
 	availRate := 0.0
 	if report.TotalRequests != 0 {
 		availRate = float64(report.EventsSent*100) / float64(report.TotalRequests)
 	}
-	p.log.Infof("Fetched receiver report. Events propagated: %v. State: %v.",
-		report.EventsSent, report.State)
+	p.log.Infof("Fetched receiver report after %s. Events propagated: %v. State: %v.",
+		elapsed, report.EventsSent, report.State)
 	p.log.Infof("Availability: %.3f%%, Requests sent: %d.",
 		availRate, report.TotalRequests)
-	if report.State == "active" {
-		p.client.T.Fatal("report fetched too early, receiver is in active state")
-	}
 	for _, t := range report.Thrown.Missing {
 		eventErrs = append(eventErrs, errors.New(t))
 	}
@@ -102,7 +108,7 @@ func replayLogs(log *zap.SugaredLogger, exec *fetcher.Execution) {
 func (p *prober) fetchExecution() *fetcher.Execution {
 	ns := p.client.Namespace
 	job := p.deployFetcher()
-	defer p.deleteFetcher()
+	defer p.deleteFetcher(job.Name)
 	pod, err := p.findSucceededPod(job)
 	p.ensureNoError(err)
 	bytes, err := pkgTest.PodLogs(p.config.Ctx, p.client.Kube, pod.Name, fetcherName, ns)
@@ -132,8 +138,8 @@ func (p *prober) deployFetcher() *batchv1.Job {
 	var replicas int32 = 1
 	fetcherJob := &batchv1.Job{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      fetcherName,
-			Namespace: p.client.Namespace,
+			GenerateName: fetcherName + "-",
+			Namespace:    p.client.Namespace,
 		},
 		Spec: batchv1.JobSpec{
 			Completions: &replicas,
@@ -172,16 +178,16 @@ func (p *prober) deployFetcher() *batchv1.Job {
 	created, err := jobs.Create(p.config.Ctx, fetcherJob, metav1.CreateOptions{})
 	p.ensureNoError(err)
 	p.log.Info("Waiting for fetcher job to succeed: ", fetcherName)
-	err = waitForJobToComplete(p.config.Ctx, p.client.Kube, fetcherName, p.client.Namespace)
+	err = waitForJobToComplete(p.config.Ctx, p.client.Kube, created.Name, p.client.Namespace)
 	p.ensureNoError(err)
 
 	return created
 }
 
-func (p *prober) deleteFetcher() {
+func (p *prober) deleteFetcher(name string) {
 	ns := p.client.Namespace
 	jobs := p.client.Kube.BatchV1().Jobs(ns)
-	err := jobs.Delete(p.config.Ctx, fetcherName, metav1.DeleteOptions{})
+	err := jobs.Delete(p.config.Ctx, name, metav1.DeleteOptions{})
 	p.ensureNoError(err)
 }
 
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/wathola/config/logger.go b/vendor/knative.dev/eventing/test/upgrade/prober/wathola/config/logger.go
index a933bd636..c102392ab 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/wathola/config/logger.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/wathola/config/logger.go
@@ -17,6 +17,7 @@ package config
 
 import (
 	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
 )
 
 // Log is a default logger for wathola
@@ -24,6 +25,8 @@ var Log = newLogger()
 var logConfig = zap.NewProductionConfig()
 
 func newLogger() *zap.SugaredLogger {
+	// Get timestamps readable and comparable with wathola-fetcher
+	logConfig.EncoderConfig.EncodeTime = zapcore.RFC3339NanoTimeEncoder
 	z, err := logConfig.Build()
 	if err != nil {
 		panic(err)
diff --git a/vendor/knative.dev/reconciler-test/pkg/milestone/emitter_log.go b/vendor/knative.dev/reconciler-test/pkg/milestone/emitter_log.go
index 594ad4136..f15b64c9c 100644
--- a/vendor/knative.dev/reconciler-test/pkg/milestone/emitter_log.go
+++ b/vendor/knative.dev/reconciler-test/pkg/milestone/emitter_log.go
@@ -45,18 +45,18 @@ func NewLogEmitter(ctx context.Context, namespace string, t feature.T) Emitter {
 func (l logEmitter) Environment(env map[string]string) {
 	bytes, err := json.MarshalIndent(env, " ", " ")
 	if err != nil {
-		l.log(err)
+		l.log(nil, err)
 		return
 	}
-	l.log("Environment", "Namespace", l.namespace, "\n", string(bytes))
+	l.log(nil, "Environment", "Namespace", l.namespace, "\n", string(bytes))
 }
 
 func (l logEmitter) NamespaceCreated(namespace string) {
-	l.log("Namespace created", namespace)
+	l.log(nil, "Namespace created", namespace)
 }
 
 func (l logEmitter) NamespaceDeleted(namespace string) {
-	l.log("Namespace deleted", namespace)
+	l.log(nil, "Namespace deleted", namespace)
 }
 
 func (l logEmitter) TestStarted(feature string, t feature.T) {
@@ -103,12 +103,12 @@ func (l logEmitter) TestSetFinished(featureSet string, t feature.T) {
 }
 
 func (l logEmitter) Finished() {
-	l.log("Finished")
+	l.log(nil, "Finished")
 	l.dumpEvents()
 }
 
 func (l logEmitter) Exception(reason, messageFormat string, messageA ...interface{}) {
-	l.log("Exception", reason, fmt.Sprintf(messageFormat, messageA...))
+	l.log(nil, "Exception", reason, fmt.Sprintf(messageFormat, messageA...))
 }
 
 func (l logEmitter) dumpEvents() {
@@ -117,15 +117,15 @@ func (l logEmitter) dumpEvents() {
 	}
 	events, err := kubeclient.Get(l.ctx).CoreV1().Events(l.namespace).List(l.ctx, metav1.ListOptions{})
 	if err != nil {
-		l.log("failed to list events", err)
+		l.log(nil, "failed to list events", err)
 		return
 	}
 	if len(events.Items) == 0 {
-		l.log("No events found")
+		l.log(nil, "No events found")
 		return
 	}
 	for _, e := range sortEventsByTime(events.Items) {
-		l.log(formatEvent(e))
+		l.log(nil, formatEvent(e))
 	}
 }
 
@@ -166,12 +166,15 @@ func formatEvent(e corev1.Event) string {
 	}, "\n")
 }
 
-func (l logEmitter) log(args ...interface{}) {
+func (l logEmitter) log(t feature.T, args ...interface{}) {
 	ctxArgs := []interface{}{
 		"namespace", l.namespace,
 		"timestamp", time.Now().Format(time.RFC3339),
 		"\n",
 	}
+	if t == nil {
+		t = l.t
+	}
 	args = append(ctxArgs, args...)
 	if l.t != nil {
 		l.t.Log(args...)