Repository: knative-sandbox/eventing-kafka-broker
Pull request #2598 (2022-09-05 11:05:45)

[Automated] Update eventing-kafka-broker-eventing nightly
/assign @knative/delivery-wg-leads
/cc @knative/delivery-wg-leads

Produced via:
```shell
for x in eventing-crds.yaml eventing-core.yaml; do
  curl https://storage.googleapis.com/knative-nightly/eventing/latest/$x > ${GITHUB_WORKSPACE}/./third_party/eventing-latest/$x
done
```

Details:
```

```

knative-prow[bot]: @knative-automation: GitHub didn't allow me to request PR reviews from the following users: knative/delivery-wg-leads.

Note that only [knative-sandbox members](https://github.com/orgs/knative-sandbox/people) and repo collaborators can review this PR, and authors cannot review their own PRs.

<details>

In response to [this](https://github.com/knative-sandbox/eventing-kafka-broker/pull/2598):

>/assign @knative/delivery-wg-leads
>/cc @knative/delivery-wg-leads
>
>Produced via:
>```shell
>for x in eventing-crds.yaml eventing-core.yaml; do
>  curl https://storage.googleapis.com/knative-nightly/eventing/latest/$x > ${GITHUB_WORKSPACE}/./third_party/eventing-latest/$x
>done
>```
>
>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.
</details>
knative-prow[bot]: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2598#" title="Author self-approved">knative-automation</a>*
**Once this PR has been reviewed and has the lgtm label**, please assign matzew for approval by writing `/assign @matzew` in a comment. For more information see:[The Kubernetes Code Review Process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process).

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).

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

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

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":["matzew"]} -->
knative-prow[bot]: @knative-automation: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
reconciler-tests_eventing-kafka-broker_main | ab54be52a2bcc299508bd50158f03f5772d0d4d7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2598/reconciler-tests_eventing-kafka-broker_main/1566744395170975744) | true | `/test reconciler-tests_eventing-kafka-broker_main`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
reconciler-tests_eventing-kafka-broker_main | ab54be52a2bcc299508bd50158f03f5772d0d4d7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2598/reconciler-tests_eventing-kafka-broker_main/1566744395170975744) | true | `/test reconciler-tests_eventing-kafka-broker_main`
channel-integration-tests-ssl_eventing-kafka-broker_main | ab54be52a2bcc299508bd50158f03f5772d0d4d7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2598/channel-integration-tests-ssl_eventing-kafka-broker_main/1566744395233890304) | true | `/test channel-integration-tests-ssl_eventing-kafka-broker_main`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
reconciler-tests_eventing-kafka-broker_main | ab54be52a2bcc299508bd50158f03f5772d0d4d7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2598/reconciler-tests_eventing-kafka-broker_main/1566744395170975744) | true | `/test reconciler-tests_eventing-kafka-broker_main`
channel-integration-tests-ssl_eventing-kafka-broker_main | ab54be52a2bcc299508bd50158f03f5772d0d4d7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2598/channel-integration-tests-ssl_eventing-kafka-broker_main/1566744395233890304) | true | `/test channel-integration-tests-ssl_eventing-kafka-broker_main`
integration-tests_eventing-kafka-broker_main | ab54be52a2bcc299508bd50158f03f5772d0d4d7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2598/integration-tests_eventing-kafka-broker_main/1566744395053535232) | true | `/test integration-tests_eventing-kafka-broker_main`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
reconciler-tests_eventing-kafka-broker_main | ab54be52a2bcc299508bd50158f03f5772d0d4d7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2598/reconciler-tests_eventing-kafka-broker_main/1566744395170975744) | true | `/test reconciler-tests_eventing-kafka-broker_main`
channel-integration-tests-ssl_eventing-kafka-broker_main | ab54be52a2bcc299508bd50158f03f5772d0d4d7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2598/channel-integration-tests-ssl_eventing-kafka-broker_main/1566744395233890304) | true | `/test channel-integration-tests-ssl_eventing-kafka-broker_main`
integration-tests_eventing-kafka-broker_main | ab54be52a2bcc299508bd50158f03f5772d0d4d7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2598/integration-tests_eventing-kafka-broker_main/1566744395053535232) | true | `/test integration-tests_eventing-kafka-broker_main`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | ab54be52a2bcc299508bd50158f03f5772d0d4d7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2598/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1566744395355525120) | true | `/test channel-integration-tests-sasl-plain_eventing-kafka-broker_main`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
reconciler-tests_eventing-kafka-broker_main | ab54be52a2bcc299508bd50158f03f5772d0d4d7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2598/reconciler-tests_eventing-kafka-broker_main/1566744395170975744) | true | `/test reconciler-tests_eventing-kafka-broker_main`
channel-integration-tests-ssl_eventing-kafka-broker_main | ab54be52a2bcc299508bd50158f03f5772d0d4d7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2598/channel-integration-tests-ssl_eventing-kafka-broker_main/1566744395233890304) | true | `/test channel-integration-tests-ssl_eventing-kafka-broker_main`
integration-tests_eventing-kafka-broker_main | ab54be52a2bcc299508bd50158f03f5772d0d4d7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2598/integration-tests_eventing-kafka-broker_main/1566744395053535232) | true | `/test integration-tests_eventing-kafka-broker_main`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | ab54be52a2bcc299508bd50158f03f5772d0d4d7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2598/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1566744395355525120) | true | `/test channel-integration-tests-sasl-plain_eventing-kafka-broker_main`
channel-integration-tests-sasl-ssl_eventing-kafka-broker_main | ab54be52a2bcc299508bd50158f03f5772d0d4d7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2598/channel-integration-tests-sasl-ssl_eventing-kafka-broker_main/1566744395296804864) | true | `/test channel-integration-tests-sasl-ssl_eventing-kafka-broker_main`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
reconciler-tests_eventing-kafka-broker_main | ab54be52a2bcc299508bd50158f03f5772d0d4d7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2598/reconciler-tests_eventing-kafka-broker_main/1566744395170975744) | true | `/test reconciler-tests_eventing-kafka-broker_main`
channel-integration-tests-ssl_eventing-kafka-broker_main | ab54be52a2bcc299508bd50158f03f5772d0d4d7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2598/channel-integration-tests-ssl_eventing-kafka-broker_main/1566744395233890304) | true | `/test channel-integration-tests-ssl_eventing-kafka-broker_main`
integration-tests_eventing-kafka-broker_main | ab54be52a2bcc299508bd50158f03f5772d0d4d7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2598/integration-tests_eventing-kafka-broker_main/1566744395053535232) | true | `/test integration-tests_eventing-kafka-broker_main`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | ab54be52a2bcc299508bd50158f03f5772d0d4d7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2598/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1566744395355525120) | true | `/test channel-integration-tests-sasl-plain_eventing-kafka-broker_main`
channel-integration-tests-sasl-ssl_eventing-kafka-broker_main | ab54be52a2bcc299508bd50158f03f5772d0d4d7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2598/channel-integration-tests-sasl-ssl_eventing-kafka-broker_main/1566744395296804864) | true | `/test channel-integration-tests-sasl-ssl_eventing-kafka-broker_main`
channel-reconciler-tests-ssl_eventing-kafka-broker_main | ab54be52a2bcc299508bd50158f03f5772d0d4d7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2598/channel-reconciler-tests-ssl_eventing-kafka-broker_main/1566744395414245376) | true | `/test channel-reconciler-tests-ssl_eventing-kafka-broker_main`

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

<details>

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

Patch:
diff --git a/control-plane/config/eventing-kafka-broker/200-controller/500-controller.yaml b/control-plane/config/eventing-kafka-broker/200-controller/500-controller.yaml
index 6e9570ac7..26234a4a8 100644
--- a/control-plane/config/eventing-kafka-broker/200-controller/500-controller.yaml
+++ b/control-plane/config/eventing-kafka-broker/200-controller/500-controller.yaml
@@ -69,24 +69,24 @@ spec:
             - name: SOURCE_DATA_PLANE_CONFIG_MAP_NAMESPACE
               value: knative-eventing
 
-            - name: BROKER_DATA_PLANE_CONFIG_MAP_NAME
+            - name: BROKER_CONTRACT_CONFIG_MAP_NAME
               value: kafka-broker-brokers-triggers
-            - name: CHANNEL_DATA_PLANE_CONFIG_MAP_NAME
+            - name: CHANNEL_CONTRACT_CONFIG_MAP_NAME
               value: kafka-channel-channels-subscriptions
-            - name: SINK_DATA_PLANE_CONFIG_MAP_NAME
+            - name: SINK_CONTRACT_CONFIG_MAP_NAME
               value: kafka-sink-sinks
-            - name: SOURCE_DATA_PLANE_CONFIG_MAP_NAME
+            - name: SOURCE_CONTRACT_CONFIG_MAP_NAME
               value: kafka-source-sources
 
-            - name: BROKER_DATA_PLANE_CONFIG_FORMAT
+            - name: BROKER_CONTRACT_CONFIG_MAP_FORMAT
               value: json
-            - name: CHANNEL_DATA_PLANE_CONFIG_FORMAT
+            - name: CHANNEL_CONTRACT_CONFIG_MAP_FORMAT
               value: json
-            - name: SINK_DATA_PLANE_CONFIG_FORMAT
+            - name: SINK_CONTRACT_CONFIG_MAP_FORMAT
               value: json
-            - name: SOURCE_DATA_PLANE_CONFIG_FORMAT
+            - name: SOURCE_CONTRACT_CONFIG_MAP_FORMAT
               value: json
-            - name: CONSUMER_DATA_PLANE_CONFIG_FORMAT
+            - name: CONSUMER_CONTRACT_CONFIG_MAP_FORMAT
               value: json
 
             - name: BROKER_INGRESS_NAME
diff --git a/control-plane/config/eventing-kafka-source/200-controller/500-controller.yaml b/control-plane/config/eventing-kafka-source/200-controller/500-controller.yaml
index 2c227b991..3aded98b1 100644
--- a/control-plane/config/eventing-kafka-source/200-controller/500-controller.yaml
+++ b/control-plane/config/eventing-kafka-source/200-controller/500-controller.yaml
@@ -62,9 +62,9 @@ spec:
           env:
             - name: SOURCE_DATA_PLANE_CONFIG_MAP_NAMESPACE
               value: knative-eventing
-            - name: SOURCE_DATA_PLANE_CONFIG_MAP_NAME
+            - name: SOURCE_CONTRACT_CONFIG_MAP_NAME
               value: kafka-source-sources
-            - name: SOURCE_DATA_PLANE_CONFIG_FORMAT
+            - name: SOURCE_CONTRACT_CONFIG_MAP_FORMAT
               value: json
             - name: SOURCE_INGRESS_NAME
               value: kafka-source-ingress
@@ -74,7 +74,7 @@ spec:
               valueFrom:
                 fieldRef:
                   fieldPath: metadata.namespace
-            - name: CONSUMER_DATA_PLANE_CONFIG_FORMAT
+            - name: CONSUMER_CONTRACT_CONFIG_FORMAT
               value: json
 
             - name: SYSTEM_NAMESPACE
diff --git a/control-plane/pkg/config/env_config.go b/control-plane/pkg/config/env_config.go
index bfdc80fb4..0bd404f76 100644
--- a/control-plane/pkg/config/env_config.go
+++ b/control-plane/pkg/config/env_config.go
@@ -28,19 +28,19 @@ type Env struct {
 	// namespace
 	DataPlaneConfigMapNamespace string `required:"true" split_words:"true"`
 
-	// DataPlaneConfigMapName is the name of the configmap that holds the contract between the control plane
+	// ContractConfigMapName is the name of the configmap that holds the contract between the control plane
 	// and the data plane.
-	DataPlaneConfigMapName string `required:"true" split_words:"true"` // example: kafka-broker-triggers
+	ContractConfigMapName string `required:"true" split_words:"true"` // example: kafka-broker-brokers-triggers
 
 	// GeneralConfigMapName is the name of the configmap that holds configuration that affects the control plane
 	// and the data plane. For example, Kafka bootstrap server information could be in here for broker configuration.
 	GeneralConfigMapName string `required:"true" split_words:"true"` // example: kafka-broker-config
 
-	IngressName           string `required:"true" split_words:"true"` // example: kafka-broker-ingress
-	IngressPodPort        string `required:"false" split_words:"true"`
-	SystemNamespace       string `required:"true" split_words:"true"`
-	DataPlaneConfigFormat string `required:"true" split_words:"true"`
-	DefaultBackoffDelayMs uint64 `required:"false" split_words:"true"`
+	IngressName             string `required:"true" split_words:"true"` // example: kafka-broker-ingress
+	IngressPodPort          string `required:"false" split_words:"true"`
+	SystemNamespace         string `required:"true" split_words:"true"`
+	ContractConfigMapFormat string `required:"true" split_words:"true"`
+	DefaultBackoffDelayMs   uint64 `required:"false" split_words:"true"`
 }
 
 // ValidationOption represents a function to validate the Env configurations.
@@ -63,5 +63,5 @@ func GetEnvConfig(prefix string, validations ...ValidationOption) (*Env, error)
 }
 
 func (c *Env) DataPlaneConfigMapAsString() string {
-	return fmt.Sprintf("%s/%s", c.DataPlaneConfigMapNamespace, c.DataPlaneConfigMapName)
+	return fmt.Sprintf("%s/%s", c.DataPlaneConfigMapNamespace, c.ContractConfigMapName)
 }
diff --git a/control-plane/pkg/reconciler/broker/controller.go b/control-plane/pkg/reconciler/broker/controller.go
index e8bdff712..2dcdb5449 100644
--- a/control-plane/pkg/reconciler/broker/controller.go
+++ b/control-plane/pkg/reconciler/broker/controller.go
@@ -62,8 +62,8 @@ func NewController(ctx context.Context, watcher configmap.Watcher, env *config.E
 			PodLister:                   podinformer.Get(ctx).Lister(),
 			SecretLister:                secretinformer.Get(ctx).Lister(),
 			DataPlaneConfigMapNamespace: env.DataPlaneConfigMapNamespace,
-			DataPlaneConfigMapName:      env.DataPlaneConfigMapName,
-			DataPlaneConfigFormat:       env.DataPlaneConfigFormat,
+			DataPlaneConfigMapName:      env.ContractConfigMapName,
+			DataPlaneConfigFormat:       env.ContractConfigMapFormat,
 			DataPlaneNamespace:          env.SystemNamespace,
 			DispatcherLabel:             base.BrokerDispatcherLabel,
 			ReceiverLabel:               base.BrokerReceiverLabel,
@@ -103,7 +103,7 @@ func NewController(ctx context.Context, watcher configmap.Watcher, env *config.E
 	}
 
 	configmapInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
-		FilterFunc: controller.FilterWithNameAndNamespace(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName),
+		FilterFunc: controller.FilterWithNameAndNamespace(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName),
 		Handler: cache.ResourceEventHandlerFuncs{
 			AddFunc: func(obj interface{}) {
 				globalResync(obj)
diff --git a/control-plane/pkg/reconciler/broker/namespaced_controller.go b/control-plane/pkg/reconciler/broker/namespaced_controller.go
index 7879fd5b6..148560c05 100644
--- a/control-plane/pkg/reconciler/broker/namespaced_controller.go
+++ b/control-plane/pkg/reconciler/broker/namespaced_controller.go
@@ -75,8 +75,8 @@ func NewNamespacedController(ctx context.Context, watcher configmap.Watcher, env
 			PodLister:                   podinformer.Get(ctx).Lister(),
 			SecretLister:                secretinformer.Get(ctx).Lister(),
 			DataPlaneConfigMapNamespace: env.DataPlaneConfigMapNamespace,
-			DataPlaneConfigMapName:      env.DataPlaneConfigMapName,
-			DataPlaneConfigFormat:       env.DataPlaneConfigFormat,
+			DataPlaneConfigMapName:      env.ContractConfigMapName,
+			DataPlaneConfigFormat:       env.ContractConfigMapFormat,
 			DataPlaneNamespace:          env.SystemNamespace,
 			DispatcherLabel:             base.BrokerDispatcherLabel,
 			ReceiverLabel:               base.BrokerReceiverLabel,
diff --git a/control-plane/pkg/reconciler/channel/controller.go b/control-plane/pkg/reconciler/channel/controller.go
index 9dcf79f7f..d91a7b5fd 100644
--- a/control-plane/pkg/reconciler/channel/controller.go
+++ b/control-plane/pkg/reconciler/channel/controller.go
@@ -59,8 +59,8 @@ func NewController(ctx context.Context, configs *config.Env) *controller.Impl {
 			PodLister:                   podinformer.Get(ctx).Lister(),
 			SecretLister:                secretinformer.Get(ctx).Lister(),
 			DataPlaneConfigMapNamespace: configs.DataPlaneConfigMapNamespace,
-			DataPlaneConfigMapName:      configs.DataPlaneConfigMapName,
-			DataPlaneConfigFormat:       configs.DataPlaneConfigFormat,
+			DataPlaneConfigMapName:      configs.ContractConfigMapName,
+			DataPlaneConfigFormat:       configs.ContractConfigMapFormat,
 			DataPlaneNamespace:          configs.SystemNamespace,
 			DispatcherLabel:             base.ChannelDispatcherLabel,
 			ReceiverLabel:               base.ChannelReceiverLabel,
@@ -104,7 +104,7 @@ func NewController(ctx context.Context, configs *config.Env) *controller.Impl {
 	})
 
 	configmapInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
-		FilterFunc: controller.FilterWithNameAndNamespace(configs.DataPlaneConfigMapNamespace, configs.DataPlaneConfigMapName),
+		FilterFunc: controller.FilterWithNameAndNamespace(configs.DataPlaneConfigMapNamespace, configs.ContractConfigMapName),
 		Handler: cache.ResourceEventHandlerFuncs{
 			AddFunc:    globalResync,
 			DeleteFunc: globalResync,
diff --git a/control-plane/pkg/reconciler/channel/v2/controllerv2.go b/control-plane/pkg/reconciler/channel/v2/controllerv2.go
index db8befc7c..898692497 100644
--- a/control-plane/pkg/reconciler/channel/v2/controllerv2.go
+++ b/control-plane/pkg/reconciler/channel/v2/controllerv2.go
@@ -61,8 +61,8 @@ func NewController(ctx context.Context, configs *config.Env) *controller.Impl {
 			PodLister:                   podinformer.Get(ctx).Lister(),
 			SecretLister:                secretinformer.Get(ctx).Lister(),
 			DataPlaneConfigMapNamespace: configs.DataPlaneConfigMapNamespace,
-			DataPlaneConfigMapName:      configs.DataPlaneConfigMapName,
-			DataPlaneConfigFormat:       configs.DataPlaneConfigFormat,
+			DataPlaneConfigMapName:      configs.ContractConfigMapName,
+			DataPlaneConfigFormat:       configs.ContractConfigMapFormat,
 			DataPlaneNamespace:          configs.SystemNamespace,
 		},
 		NewKafkaClient:             sarama.NewClient,
diff --git a/control-plane/pkg/reconciler/consumer/controller.go b/control-plane/pkg/reconciler/consumer/controller.go
index bd4fe5759..35bca4952 100644
--- a/control-plane/pkg/reconciler/consumer/controller.go
+++ b/control-plane/pkg/reconciler/consumer/controller.go
@@ -37,7 +37,7 @@ import (
 )
 
 type ControllerConfig struct {
-	DataPlaneConfigFormat string `required:"true" split_words:"true"`
+	ContractConfigMapFormat string `required:"true" split_words:"true"`
 }
 
 func NewController(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
@@ -50,7 +50,7 @@ func NewController(ctx context.Context, watcher configmap.Watcher) *controller.I
 	consumerInformer := consumer.Get(ctx)
 
 	r := &Reconciler{
-		SerDe:               formatSerDeFromString(controllerConfig.DataPlaneConfigFormat),
+		SerDe:               formatSerDeFromString(controllerConfig.ContractConfigMapFormat),
 		ConsumerGroupLister: consumergroup.Get(ctx).Lister(),
 		SecretLister:        secretinformer.Get(ctx).Lister(),
 		PodLister:           podinformer.Get(ctx).Lister(),
diff --git a/control-plane/pkg/reconciler/sink/controller.go b/control-plane/pkg/reconciler/sink/controller.go
index bf7f086ab..44a8616de 100644
--- a/control-plane/pkg/reconciler/sink/controller.go
+++ b/control-plane/pkg/reconciler/sink/controller.go
@@ -57,8 +57,8 @@ func NewController(ctx context.Context, _ configmap.Watcher, configs *config.Env
 			PodLister:                   podinformer.Get(ctx).Lister(),
 			SecretLister:                secretinformer.Get(ctx).Lister(),
 			DataPlaneConfigMapNamespace: configs.DataPlaneConfigMapNamespace,
-			DataPlaneConfigMapName:      configs.DataPlaneConfigMapName,
-			DataPlaneConfigFormat:       configs.DataPlaneConfigFormat,
+			DataPlaneConfigMapName:      configs.ContractConfigMapName,
+			DataPlaneConfigFormat:       configs.ContractConfigMapFormat,
 			DataPlaneNamespace:          configs.SystemNamespace,
 			ReceiverLabel:               base.SinkReceiverLabel,
 		},
@@ -89,7 +89,7 @@ func NewController(ctx context.Context, _ configmap.Watcher, configs *config.Env
 	}
 
 	configmapInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
-		FilterFunc: controller.FilterWithNameAndNamespace(configs.DataPlaneConfigMapNamespace, configs.DataPlaneConfigMapName),
+		FilterFunc: controller.FilterWithNameAndNamespace(configs.DataPlaneConfigMapNamespace, configs.ContractConfigMapName),
 		Handler: cache.ResourceEventHandlerFuncs{
 			AddFunc: func(obj interface{}) {
 				globalResync(obj)
diff --git a/control-plane/pkg/reconciler/trigger/controller.go b/control-plane/pkg/reconciler/trigger/controller.go
index 5b4443ecd..d0e1494a0 100644
--- a/control-plane/pkg/reconciler/trigger/controller.go
+++ b/control-plane/pkg/reconciler/trigger/controller.go
@@ -67,8 +67,8 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *conf
 			PodLister:                   podinformer.Get(ctx).Lister(),
 			SecretLister:                secretinformer.Get(ctx).Lister(),
 			DataPlaneConfigMapNamespace: configs.DataPlaneConfigMapNamespace,
-			DataPlaneConfigMapName:      configs.DataPlaneConfigMapName,
-			DataPlaneConfigFormat:       configs.DataPlaneConfigFormat,
+			DataPlaneConfigMapName:      configs.ContractConfigMapName,
+			DataPlaneConfigFormat:       configs.ContractConfigMapFormat,
 			DataPlaneNamespace:          configs.SystemNamespace,
 			DispatcherLabel:             base.BrokerDispatcherLabel,
 			ReceiverLabel:               base.BrokerReceiverLabel,
@@ -112,7 +112,7 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *conf
 	}
 
 	configmapInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
-		FilterFunc: controller.FilterWithNameAndNamespace(configs.DataPlaneConfigMapNamespace, configs.DataPlaneConfigMapName),
+		FilterFunc: controller.FilterWithNameAndNamespace(configs.DataPlaneConfigMapNamespace, configs.ContractConfigMapName),
 		Handler: cache.ResourceEventHandlerFuncs{
 			AddFunc:    globalResync,
 			DeleteFunc: globalResync,
diff --git a/control-plane/pkg/reconciler/trigger/namespaced_controller.go b/control-plane/pkg/reconciler/trigger/namespaced_controller.go
index e4735e4e2..3f0fd15a0 100644
--- a/control-plane/pkg/reconciler/trigger/namespaced_controller.go
+++ b/control-plane/pkg/reconciler/trigger/namespaced_controller.go
@@ -60,8 +60,8 @@ func NewNamespacedController(ctx context.Context, watcher configmap.Watcher, con
 			PodLister:                   podinformer.Get(ctx).Lister(),
 			SecretLister:                secretinformer.Get(ctx).Lister(),
 			DataPlaneConfigMapNamespace: configs.DataPlaneConfigMapNamespace,
-			DataPlaneConfigMapName:      configs.DataPlaneConfigMapName,
-			DataPlaneConfigFormat:       configs.DataPlaneConfigFormat,
+			DataPlaneConfigMapName:      configs.ContractConfigMapName,
+			DataPlaneConfigFormat:       configs.ContractConfigMapFormat,
 			DataPlaneNamespace:          configs.SystemNamespace,
 			DispatcherLabel:             base.BrokerDispatcherLabel,
 			ReceiverLabel:               base.BrokerReceiverLabel,
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java
index 3a955e233..aa4acc35f 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java
@@ -34,6 +34,7 @@ import io.vertx.core.Context;
 import io.vertx.core.Future;
 import io.vertx.core.Promise;
 import io.vertx.core.Vertx;
+import io.vertx.core.buffer.Buffer;
 import io.vertx.ext.web.client.HttpResponse;
 import io.vertx.kafka.client.common.tracing.ConsumerTracer;
 import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
@@ -44,6 +45,7 @@ import org.slf4j.LoggerFactory;
 
 import javax.annotation.Nullable;
 
+import java.util.Base64;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
@@ -290,12 +292,14 @@ public class RecordDispatcherImpl implements RecordDispatcher {
     extensions.put(KN_ERROR_DEST_EXT_NAME, destination);
     extensions.put(KN_ERROR_CODE_EXT_NAME, String.valueOf(response.statusCode()));
 
-    var data = response.bodyAsString();
+    // we extract the response as byte array as we do not need a string
+    // representation of it
+    var data = response.bodyAsBuffer();
     if (data != null) {
       if (data.length() > KN_ERROR_DATA_MAX_BYTES) {
-        data = data.substring(0, KN_ERROR_DATA_MAX_BYTES);
+        data = Buffer.buffer(data.getBytes(0, KN_ERROR_DATA_MAX_BYTES));
       }
-      extensions.put(KN_ERROR_DATA_EXT_NAME, data);
+      extensions.put(KN_ERROR_DATA_EXT_NAME, Base64.getEncoder().encodeToString(data.getBytes()));
     }
 
     return addExtensions(recordContext, extensions);
diff --git a/go.mod b/go.mod
index 003a1e214..f1528abeb 100644
--- a/go.mod
+++ b/go.mod
@@ -29,11 +29,11 @@ require (
 	k8s.io/apiserver v0.23.9
 	k8s.io/client-go v0.23.9
 	k8s.io/utils v0.0.0-20220210201930-3a6ce19ff2f9
-	knative.dev/eventing v0.34.1
-	knative.dev/eventing-kafka v0.34.0
-	knative.dev/hack v0.0.0-20220823140917-8d1e4ccf9dc3
-	knative.dev/pkg v0.0.0-20220818004048-4a03844c0b15
-	knative.dev/reconciler-test v0.0.0-20220901014218-ff8756131bd5
+	knative.dev/eventing v0.34.1-0.20220906111743-7f1f49f5b4e0
+	knative.dev/eventing-kafka v0.34.1-0.20220907020318-2f1fbd6dda1b
+	knative.dev/hack v0.0.0-20220906140443-f9dc7220d362
+	knative.dev/pkg v0.0.0-20220826162920-93b66e6a8700
+	knative.dev/reconciler-test v0.0.0-20220905144823-18e7400195c1
 	sigs.k8s.io/yaml v1.3.0
 )
 
diff --git a/go.sum b/go.sum
index 97e1d445c..400ba0002 100644
--- a/go.sum
+++ b/go.sum
@@ -1561,20 +1561,23 @@ k8s.io/utils v0.0.0-20210819203725-bdf08cb9a70a/go.mod h1:jPW/WVKK9YHAvNhRxK0md/
 k8s.io/utils v0.0.0-20211116205334-6203023598ed/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 k8s.io/utils v0.0.0-20220210201930-3a6ce19ff2f9 h1:HNSDgDCrr/6Ly3WEGKZftiE7IY19Vz2GdbOCyI4qqhc=
 k8s.io/utils v0.0.0-20220210201930-3a6ce19ff2f9/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
-knative.dev/control-protocol v0.0.0-20220818153549-f18dbde7d9bd/go.mod h1:vO3Xc0k0h6fFVsVG9kNMUMcVKG7MAx7jMbZDvgSuzwI=
-knative.dev/eventing v0.34.1 h1:r6QuQmDCmmSANdTRLdbKb6YcYaJoeNpEuFfS8Bq0ZgQ=
-knative.dev/eventing v0.34.1/go.mod h1:6UnNnPrEUNAM9PfCpf7L8N7G/1vq+HQlpOjzndY6ryw=
-knative.dev/eventing-kafka v0.34.0 h1:TOYpkJbgsBWTfEgKZ6wgcywF5OPD+UPfv4csec287BE=
-knative.dev/eventing-kafka v0.34.0/go.mod h1:SzSr5SGIgZ/R/XbIPKWecBoym+hM2L2GBE1zpFj5iyE=
+knative.dev/control-protocol v0.0.0-20220905133527-ca2ee058ed65/go.mod h1:kV+UoX4DjjwMlS37wHV1ZC7OL7r5IPAOhUvTHWi347M=
+knative.dev/eventing v0.34.1-0.20220906111743-7f1f49f5b4e0 h1:IVLZkOQlBL8MMwSYw28WVMru36UdXTZW+FkKuZACMhs=
+knative.dev/eventing v0.34.1-0.20220906111743-7f1f49f5b4e0/go.mod h1:6UnNnPrEUNAM9PfCpf7L8N7G/1vq+HQlpOjzndY6ryw=
+knative.dev/eventing-kafka v0.34.1-0.20220907020318-2f1fbd6dda1b h1:r+411BL8hsD3C52NIS+Yv9WNJIccjxWKTugWlbOFjM4=
+knative.dev/eventing-kafka v0.34.1-0.20220907020318-2f1fbd6dda1b/go.mod h1:ONepeihHUOP1rFaGIJTYYlphtOGQltLa0+WX1UoGds4=
 knative.dev/hack v0.0.0-20220815132133-e9a8475f4329/go.mod h1:t/azP8I/Cygaw+87O7rkAPrNRjCelmtfSzWzu/9TM7I=
-knative.dev/hack v0.0.0-20220823140917-8d1e4ccf9dc3 h1:umaeMRecA0g5g48L9tnEAkTBIitr9eKWMyJYo9YttAA=
 knative.dev/hack v0.0.0-20220823140917-8d1e4ccf9dc3/go.mod h1:t/azP8I/Cygaw+87O7rkAPrNRjCelmtfSzWzu/9TM7I=
+knative.dev/hack v0.0.0-20220902220419-664eac5c391e/go.mod h1:yk2OjGDsbEnQjfxdm0/HJKS2WqTLEFg/N6nUs6Rqx3Q=
+knative.dev/hack v0.0.0-20220906140443-f9dc7220d362 h1:z3MkroifHD8RBAJv3zD6PDYhE/XRRJdrMpuIYHd3cFI=
+knative.dev/hack v0.0.0-20220906140443-f9dc7220d362/go.mod h1:yk2OjGDsbEnQjfxdm0/HJKS2WqTLEFg/N6nUs6Rqx3Q=
 knative.dev/hack/schema v0.0.0-20220823140917-8d1e4ccf9dc3/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
-knative.dev/pkg v0.0.0-20220818004048-4a03844c0b15 h1:GNmzHVaUo3zoi/wtIN71LPQaWy6DdoYzmb+GIq2s4fw=
 knative.dev/pkg v0.0.0-20220818004048-4a03844c0b15/go.mod h1:YLjXbkQLlGHok+u0FLfMbBHFzY9WGu3GHhnrptoAy8I=
+knative.dev/pkg v0.0.0-20220826162920-93b66e6a8700 h1:6xjS69vf5Pbwu+dWckqPALmm5pc3p0/XScPpRBhugRc=
+knative.dev/pkg v0.0.0-20220826162920-93b66e6a8700/go.mod h1:WIQZyPTLzT+z0V3/gB91aWugIi3E1mkU9rjCrLfjOwY=
 knative.dev/reconciler-test v0.0.0-20220818122349-177f8264c28c/go.mod h1:A437yxlDVDVKQv779WlB9Nj9lWAMoOKHQXFXls24Sps=
-knative.dev/reconciler-test v0.0.0-20220901014218-ff8756131bd5 h1:Z17OIrS0nusy3sgqWNSWGWqyFvYhhMaTXqDU32vTLuk=
-knative.dev/reconciler-test v0.0.0-20220901014218-ff8756131bd5/go.mod h1:LTqPpT5I7bSUToTpKpNkW6avSRFnMt4hby1y53MEBLE=
+knative.dev/reconciler-test v0.0.0-20220905144823-18e7400195c1 h1:l+cCnJdBI26lLo42pN0FO7dsmnbSDxtNnj4kVXsNavM=
+knative.dev/reconciler-test v0.0.0-20220905144823-18e7400195c1/go.mod h1:XBrYVW7mHvm9ogENdh4qS3xFHacT7GQGZOCtDb0dqXs=
 pgregory.net/rapid v0.3.3 h1:jCjBsY4ln4Atz78QoBWxUEvAHaFyNDQg9+WU62aCn1U=
 pgregory.net/rapid v0.3.3/go.mod h1:UYpPVyjFHzYBGHIxLFoupi8vwk6rXNzRY9OMvVxFIOU=
 rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8=
diff --git a/vendor/knative.dev/hack/go.work b/vendor/knative.dev/hack/go.work
new file mode 100644
index 000000000..b20aabbbc
--- /dev/null
+++ b/vendor/knative.dev/hack/go.work
@@ -0,0 +1,7 @@
+go 1.18
+
+use (
+	.
+	schema
+	test/e2e
+)
diff --git a/vendor/knative.dev/hack/library.sh b/vendor/knative.dev/hack/library.sh
index 26cd00cb8..ade3303ba 100644
--- a/vendor/knative.dev/hack/library.sh
+++ b/vendor/knative.dev/hack/library.sh
@@ -97,7 +97,7 @@ function patch_version() {
 # Print error message and exit 1
 # Parameters: $1..$n - error message to be displayed
 function abort() {
-  echo "error: $*"
+  echo "error: $*" >&2
   exit 1
 }
 
@@ -126,7 +126,7 @@ function subheader() {
 
 # Simple warning banner for logging purposes.
 function warning() {
-  make_banner "!" "$1"
+  make_banner '!' "$*" >&2
 }
 
 # Checks whether the given function exists.
@@ -567,7 +567,6 @@ function go_run() {
   export GORUN_PATH
   GOPATH="${GORUN_PATH}" \
   GOFLAGS='' \
-  GO111MODULE='' \
     go run "$package" "$@"
 }
 
@@ -577,6 +576,7 @@ function go_run() {
 #             $3..$n - parameters passed to the tool.
 # Deprecated: use go_run instead
 function run_go_tool() {
+  warning 'The "run_go_tool" function is deprecated. Use "go_run" instead.'
   local package=$1
   # If no `@version` is provided, default to adding `@latest`
   if [[ "$package" != *@* ]]; then
@@ -601,6 +601,26 @@ function add_trap {
   done
 }
 
+# Run a command, described by $1, for every go module in the project.
+# Parameters: $1      - Description of the command being run,
+#             $2 - $n - Arguments to pass to the command.
+function foreach_go_module() {
+  local failed=0
+  local -r cmd="$1"
+  shift
+  local gomod_filepath gomod_dir
+  while read -r gomod_filepath; do
+    gomod_dir="$(dirname "$gomod_filepath")"
+    pushd "$gomod_dir" > /dev/null
+    "$cmd" "$@" || failed=$?
+    popd > /dev/null
+    if (( failed )); then
+      echo "Command '${cmd}' failed in module $gomod_dir: $failed" >&2
+      return $failed
+    fi
+  done < <(find . -name go.mod -type f ! -path "*/vendor/*" ! -path "*/third_party/*")
+}
+
 # Update go deps.
 # Parameters (parsed as flags):
 #   "--upgrade", bool, do upgrade.
@@ -613,14 +633,18 @@ function add_trap {
 # global env var: FLOATING_DEPS
 # --upgrade will set GOPROXY to direct unless it is already set.
 function go_update_deps() {
-  cd "${REPO_ROOT_DIR}" || return 1
+  foreach_go_module __go_update_deps_for_module "$@"
+}
+
+function __go_update_deps_for_module() {
+  ( # do not modify the environment
+  set -Eeuo pipefail
 
-  export GO111MODULE=on
   export GOFLAGS=""
   export GONOSUMDB="${GONOSUMDB:-},knative.dev/*"
   export GONOPROXY="${GONOPROXY:-},knative.dev/*"
 
-  echo "=== Update Deps for Golang"
+  echo "=== Update Deps for Golang module: $(go_mod_module_name)"
 
   local UPGRADE=0
   local RELEASE="v9000.1" # release v9000 is so far in the future, it will always pick the default branch.
@@ -646,7 +670,7 @@ function go_update_deps() {
     else
       group "Upgrading to release ${RELEASE}"
     fi
-    FLOATING_DEPS+=( $(go_run knative.dev/test-infra/buoy@latest float ${REPO_ROOT_DIR}/go.mod "${buoyArgs[@]}") )
+    FLOATING_DEPS+=( $(go_run knative.dev/test-infra/buoy@latest float ./go.mod "${buoyArgs[@]}") )
     if [[ ${#FLOATING_DEPS[@]} > 0 ]]; then
       echo "Floating deps to ${FLOATING_DEPS[@]}"
       go get -d ${FLOATING_DEPS[@]}
@@ -664,6 +688,9 @@ function go_update_deps() {
   go mod vendor 2>&1 |  grep -v "ignoring symlink" || true
   eval "$orig_pipefail_opt"
 
+  if ! [ -d vendor ]; then
+    return 0
+  fi
   group "Removing unwanted vendor files"
 
   # Remove unwanted vendor files
@@ -680,13 +707,15 @@ function go_update_deps() {
 
   group "Removing broken symlinks"
   remove_broken_symlinks ./vendor
+  )
 }
 
+
 # Return the go module name of the current module.
 # Intended to be used like:
 #   export MODULE_NAME=$(go_mod_module_name)
 function go_mod_module_name() {
-  go mod graph | cut -d' ' -f 1 | grep -v '@' | head -1
+  grep -E '^module ' go.mod | cut -d' ' -f2
 }
 
 # Return a GOPATH to a temp directory. Works around the out-of-GOPATH issues
@@ -717,11 +746,10 @@ function run_kntest() {
 # Parameters: $1 - output file, relative to repo root dir.
 #             $2 - directory to inspect.
 function update_licenses() {
-  cd "${REPO_ROOT_DIR}" || return 1
   local dst=$1
   local dir=$2
   shift
-  go_run github.com/google/go-licenses@v1.2.0 \
+  go_run github.com/google/go-licenses@v1.2.1 \
     save "${dir}" --save_path="${dst}" --force || \
     { echo "--- FAIL: go-licenses failed to update licenses"; return 1; }
 }
@@ -729,7 +757,7 @@ function update_licenses() {
 # Run go-licenses to check for forbidden licenses.
 function check_licenses() {
   # Check that we don't have any forbidden licenses.
-  go_run github.com/google/go-licenses@v1.2.0 \
+  go_run github.com/google/go-licenses@v1.2.1 \
     check "${REPO_ROOT_DIR}/..." || \
     { echo "--- FAIL: go-licenses failed the license check"; return 1; }
 }
@@ -762,7 +790,7 @@ function is_protected_project() {
 # Remove symlinks in a path that are broken or lead outside the repo.
 # Parameters: $1 - path name, e.g. vendor
 function remove_broken_symlinks() {
-  for link in $(find $1 -type l); do
+  for link in $(find "$1" -type l); do
     # Remove broken symlinks
     if [[ ! -e ${link} ]]; then
       unlink ${link}
diff --git a/vendor/knative.dev/pkg/webhook/admission.go b/vendor/knative.dev/pkg/webhook/admission.go
index ea5518260..db131b902 100644
--- a/vendor/knative.dev/pkg/webhook/admission.go
+++ b/vendor/knative.dev/pkg/webhook/admission.go
@@ -17,10 +17,13 @@ limitations under the License.
 package webhook
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
 	"net/http"
+	"strings"
 	"time"
 
 	"go.uber.org/zap"
@@ -90,10 +93,12 @@ func admissionHandler(rootLogger *zap.SugaredLogger, stats StatsReporter, c Admi
 		logger.Infof("Webhook ServeHTTP request=%#v", r)
 
 		var review admissionv1.AdmissionReview
-		if err := json.NewDecoder(r.Body).Decode(&review); err != nil {
+		bodyBuffer := bytes.Buffer{}
+		if err := json.NewDecoder(io.TeeReader(r.Body, &bodyBuffer)).Decode(&review); err != nil {
 			http.Error(w, fmt.Sprint("could not decode body:", err), http.StatusBadRequest)
 			return
 		}
+		r.Body = io.NopCloser(&bodyBuffer)
 
 		logger = logger.With(
 			logkey.Kind, review.Request.Kind.String(),
@@ -124,6 +129,18 @@ func admissionHandler(rootLogger *zap.SugaredLogger, stats StatsReporter, c Admi
 		if !reviewResponse.Allowed || reviewResponse.PatchType != nil || response.Response == nil {
 			response.Response = reviewResponse
 		}
+
+		// If warnings contain newlines, which they will do by default if
+		// using Knative apis.FieldError, split them based on newlines
+		// and create a new warning. This is because any control characters
+		// in the warnings will cause the warning to be dropped silently.
+		if reviewResponse.Warnings != nil {
+			cleanedWarnings := make([]string, 0, len(reviewResponse.Warnings))
+			for _, w := range reviewResponse.Warnings {
+				cleanedWarnings = append(cleanedWarnings, strings.Split(w, "\n")...)
+			}
+			reviewResponse.Warnings = cleanedWarnings
+		}
 		response.Response.UID = review.Request.UID
 
 		logger = logger.With(
@@ -141,7 +158,7 @@ func admissionHandler(rootLogger *zap.SugaredLogger, stats StatsReporter, c Admi
 
 		if stats != nil {
 			// Only report valid requests
-			stats.ReportRequest(review.Request, response.Response, time.Since(ttStart))
+			stats.ReportAdmissionRequest(review.Request, response.Response, time.Since(ttStart))
 		}
 	}
 }
diff --git a/vendor/knative.dev/pkg/webhook/conversion.go b/vendor/knative.dev/pkg/webhook/conversion.go
index 201ff41f4..2e876c14f 100644
--- a/vendor/knative.dev/pkg/webhook/conversion.go
+++ b/vendor/knative.dev/pkg/webhook/conversion.go
@@ -21,6 +21,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http"
+	"time"
 
 	"go.uber.org/zap"
 	apixv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
@@ -37,8 +38,9 @@ type ConversionController interface {
 	Convert(context.Context, *apixv1.ConversionRequest) *apixv1.ConversionResponse
 }
 
-func conversionHandler(rootLogger *zap.SugaredLogger, _ StatsReporter, c ConversionController) http.HandlerFunc {
+func conversionHandler(rootLogger *zap.SugaredLogger, stats StatsReporter, c ConversionController) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
+		var ttStart = time.Now()
 		logger := rootLogger
 		logger.Infof("Webhook ServeHTTP request=%#v", r)
 
@@ -69,10 +71,9 @@ func conversionHandler(rootLogger *zap.SugaredLogger, _ StatsReporter, c Convers
 			return
 		}
 
-		// TODO(dprotaso) - figure out what metrics we want reported
-		// if stats != nil {
-		// 	// Only report valid requests
-		// 	stats.ReportRequest(review.Request, response.Response, time.Since(ttStart))
-		// }
+		if stats != nil {
+			// Only report valid requests
+			stats.ReportConversionRequest(review.Request, response.Response, time.Since(ttStart))
+		}
 	}
 }
diff --git a/vendor/knative.dev/pkg/webhook/resourcesemantics/interface.go b/vendor/knative.dev/pkg/webhook/resourcesemantics/interface.go
index 7f216b101..aa989894d 100644
--- a/vendor/knative.dev/pkg/webhook/resourcesemantics/interface.go
+++ b/vendor/knative.dev/pkg/webhook/resourcesemantics/interface.go
@@ -17,6 +17,7 @@ limitations under the License.
 package resourcesemantics
 
 import (
+	admissionregistrationv1 "k8s.io/api/admissionregistration/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"knative.dev/pkg/apis"
 )
@@ -28,3 +29,22 @@ type GenericCRD interface {
 	apis.Validatable
 	runtime.Object
 }
+
+// VerbLimited defines which Verbs you want to have the webhook invoked on.
+type VerbLimited interface {
+	// SupportedVerbs define which operations (verbs) webhook is called on.
+	SupportedVerbs() []admissionregistrationv1.OperationType
+}
+
+// SubResourceLimited defines which subresources you want to have the webhook
+// invoked on. For example "status", "scale", etc.
+type SubResourceLimited interface {
+	// SupportedSubResources are the subresources that will be registered
+	// for the resource validation.
+	// If you wanted to add for example scale validation for Deployments, you'd
+	// do:
+	// []string{"", "/status", "/scale"}
+	// And to get just the main resource, you would do:
+	// []string{""}
+	SupportedSubResources() []string
+}
diff --git a/vendor/knative.dev/pkg/webhook/resourcesemantics/validation/controller.go b/vendor/knative.dev/pkg/webhook/resourcesemantics/validation/controller.go
index 02e47032b..8e8e008e6 100644
--- a/vendor/knative.dev/pkg/webhook/resourcesemantics/validation/controller.go
+++ b/vendor/knative.dev/pkg/webhook/resourcesemantics/validation/controller.go
@@ -35,34 +35,21 @@ import (
 	"knative.dev/pkg/webhook/resourcesemantics"
 )
 
-// NewAdmissionController constructs a reconciler
-func NewAdmissionController(
+// NewAdmissionControllerWithConfig constructs a reconciler and registers the
+// provided handlers with specified verbs and SubResources
+func NewAdmissionControllerWithConfig(
 	ctx context.Context,
 	name, path string,
 	handlers map[schema.GroupVersionKind]resourcesemantics.GenericCRD,
 	wc func(context.Context) context.Context,
 	disallowUnknownFields bool,
-	callbacks ...map[schema.GroupVersionKind]Callback,
+	callbacks map[schema.GroupVersionKind]Callback,
 ) *controller.Impl {
-
 	client := kubeclient.Get(ctx)
 	vwhInformer := vwhinformer.Get(ctx)
 	secretInformer := secretinformer.Get(ctx)
 	options := webhook.GetOptions(ctx)
 
-	// This not ideal, we are using a variadic argument to effectively make callbacks optional
-	// This allows this addition to be non-breaking to consumers of /pkg
-	// TODO: once all sub-repos have adopted this, we might move this back to a traditional param.
-	var unwrappedCallbacks map[schema.GroupVersionKind]Callback
-	switch len(callbacks) {
-	case 0:
-		unwrappedCallbacks = map[schema.GroupVersionKind]Callback{}
-	case 1:
-		unwrappedCallbacks = callbacks[0]
-	default:
-		panic("NewAdmissionController may not be called with multiple callback maps")
-	}
-
 	wh := &reconciler{
 		LeaderAwareFuncs: pkgreconciler.LeaderAwareFuncs{
 			// Have this reconciler enqueue our singleton whenever it becomes leader.
@@ -77,7 +64,7 @@ func NewAdmissionController(
 		},
 		path:      path,
 		handlers:  handlers,
-		callbacks: unwrappedCallbacks,
+		callbacks: callbacks,
 
 		withContext:           wc,
 		disallowUnknownFields: disallowUnknownFields,
@@ -109,4 +96,29 @@ func NewAdmissionController(
 	})
 
 	return c
+
+}
+
+// NewAdmissionController constructs a reconciler
+func NewAdmissionController(
+	ctx context.Context,
+	name, path string,
+	handlers map[schema.GroupVersionKind]resourcesemantics.GenericCRD,
+	wc func(context.Context) context.Context,
+	disallowUnknownFields bool,
+	callbacks ...map[schema.GroupVersionKind]Callback,
+) *controller.Impl {
+	// This not ideal, we are using a variadic argument to effectively make callbacks optional
+	// This allows this addition to be non-breaking to consumers of /pkg
+	// TODO: once all sub-repos have adopted this, we might move this back to a traditional param.
+	var unwrappedCallbacks map[schema.GroupVersionKind]Callback
+	switch len(callbacks) {
+	case 0:
+		unwrappedCallbacks = map[schema.GroupVersionKind]Callback{}
+	case 1:
+		unwrappedCallbacks = callbacks[0]
+	default:
+		panic("NewAdmissionController may not be called with multiple callback maps")
+	}
+	return NewAdmissionControllerWithConfig(ctx, name, path, handlers, wc, disallowUnknownFields, unwrappedCallbacks)
 }
diff --git a/vendor/knative.dev/pkg/webhook/resourcesemantics/validation/reconcile_config.go b/vendor/knative.dev/pkg/webhook/resourcesemantics/validation/reconcile_config.go
index 0078b61ce..a838351c8 100644
--- a/vendor/knative.dev/pkg/webhook/resourcesemantics/validation/reconcile_config.go
+++ b/vendor/knative.dev/pkg/webhook/resourcesemantics/validation/reconcile_config.go
@@ -100,23 +100,54 @@ func (ac *reconciler) reconcileValidatingWebhook(ctx context.Context, caCert []b
 	logger := logging.FromContext(ctx)
 
 	rules := make([]admissionregistrationv1.RuleWithOperations, 0, len(ac.handlers))
-	for gvk := range ac.handlers {
+	for gvk, config := range ac.handlers {
 		plural := strings.ToLower(flect.Pluralize(gvk.Kind))
 
+		// If SupportedVerbs has not been given, provide the legacy defaults
+		// of Create, Update, and Delete
+		supportedVerbs := []admissionregistrationv1.OperationType{
+			admissionregistrationv1.Create,
+			admissionregistrationv1.Update,
+			admissionregistrationv1.Delete,
+		}
+
+		if vl, ok := config.(resourcesemantics.VerbLimited); ok {
+			logging.FromContext(ctx).Debugf("Using custom Verbs")
+			supportedVerbs = vl.SupportedVerbs()
+		}
+		logging.FromContext(ctx).Debugf("Registering verbs: %s", supportedVerbs)
+
+		resources := []string{}
+		// If SupportedSubResources has not been given, provide the legacy
+		// defaults of main resource, and status
+		if srl, ok := config.(resourcesemantics.SubResourceLimited); ok {
+			logging.FromContext(ctx).Debugf("Using custom SubResources")
+			for _, subResource := range srl.SupportedSubResources() {
+				if subResource == "" {
+					// Special case the actual plural if given
+					resources = append(resources, plural)
+				} else {
+					resources = append(resources, plural+subResource)
+				}
+			}
+		} else {
+			resources = append(resources, plural, plural+"/status")
+		}
+		logging.FromContext(ctx).Debugf("Registering SubResources: %s", resources)
 		rules = append(rules, admissionregistrationv1.RuleWithOperations{
-			Operations: []admissionregistrationv1.OperationType{
-				admissionregistrationv1.Create,
-				admissionregistrationv1.Update,
-				admissionregistrationv1.Delete,
-			},
+			Operations: supportedVerbs,
 			Rule: admissionregistrationv1.Rule{
 				APIGroups:   []string{gvk.Group},
 				APIVersions: []string{gvk.Version},
-				Resources:   []string{plural, plural + "/status"},
+				Resources:   resources,
 			},
 		})
 	}
 
+	for _, r := range rules {
+		logging.FromContext(ctx).Debugf("Rule: %+v", r)
+	}
+
 	// Sort the rules by Group, Version, Kind so that things are deterministically ordered.
 	sort.Slice(rules, func(i, j int) bool {
 		lhs, rhs := rules[i], rules[j]
diff --git a/vendor/knative.dev/pkg/webhook/resourcesemantics/validation/validation_admit.go b/vendor/knative.dev/pkg/webhook/resourcesemantics/validation/validation_admit.go
index 8ac55e805..2b05a7cd0 100644
--- a/vendor/knative.dev/pkg/webhook/resourcesemantics/validation/validation_admit.go
+++ b/vendor/knative.dev/pkg/webhook/resourcesemantics/validation/validation_admit.go
@@ -41,7 +41,6 @@ type Callback struct {
 	function func(ctx context.Context, unstructured *unstructured.Unstructured) error
 
 	// supportedVerbs are the verbs supported for the callback.
-	// The function will only be called on these actions.
 	supportedVerbs map[webhook.Operation]struct{}
 }
 
diff --git a/vendor/knative.dev/pkg/webhook/stats_reporter.go b/vendor/knative.dev/pkg/webhook/stats_reporter.go
index 9d64634fe..1fe30e8af 100644
--- a/vendor/knative.dev/pkg/webhook/stats_reporter.go
+++ b/vendor/knative.dev/pkg/webhook/stats_reporter.go
@@ -25,6 +25,7 @@ import (
 	"go.opencensus.io/stats/view"
 	"go.opencensus.io/tag"
 	admissionv1 "k8s.io/api/admission/v1"
+	apixv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"knative.dev/pkg/metrics"
 )
 
@@ -57,11 +58,17 @@ var (
 	resourceResourceKey  = tag.MustNewKey("resource_resource")
 	resourceNamespaceKey = tag.MustNewKey("resource_namespace")
 	admissionAllowedKey  = tag.MustNewKey("admission_allowed")
+
+	desiredAPIVersionKey = tag.MustNewKey("desired_api_version")
+	resultStatusKey      = tag.MustNewKey("result_status")
+	resultReasonKey      = tag.MustNewKey("result_reason")
+	resultCodeKey        = tag.MustNewKey("result_code")
 )
 
 // StatsReporter reports webhook metrics
 type StatsReporter interface {
-	ReportRequest(request *admissionv1.AdmissionRequest, response *admissionv1.AdmissionResponse, d time.Duration) error
+	ReportAdmissionRequest(request *admissionv1.AdmissionRequest, response *admissionv1.AdmissionResponse, d time.Duration) error
+	ReportConversionRequest(request *apixv1.ConversionRequest, response *apixv1.ConversionResponse, d time.Duration) error
 }
 
 // reporter implements StatsReporter interface
@@ -82,7 +89,7 @@ func NewStatsReporter() (StatsReporter, error) {
 }
 
 // Captures req count metric, recording the count and the duration
-func (r *reporter) ReportRequest(req *admissionv1.AdmissionRequest, resp *admissionv1.AdmissionResponse, d time.Duration) error {
+func (r *reporter) ReportAdmissionRequest(req *admissionv1.AdmissionRequest, resp *admissionv1.AdmissionResponse, d time.Duration) error {
 	ctx, err := tag.New(
 		r.ctx,
 		tag.Insert(requestOperationKey, string(req.Operation)),
@@ -105,6 +112,25 @@ func (r *reporter) ReportRequest(req *admissionv1.AdmissionRequest, resp *admiss
 	return nil
 }
 
+// Captures req count metric, recording the count and the duration
+func (r *reporter) ReportConversionRequest(req *apixv1.ConversionRequest, resp *apixv1.ConversionResponse, d time.Duration) error {
+	ctx, err := tag.New(
+		r.ctx,
+		tag.Insert(desiredAPIVersionKey, req.DesiredAPIVersion),
+		tag.Insert(resultStatusKey, resp.Result.Status),
+		tag.Insert(resultReasonKey, string(resp.Result.Reason)),
+		tag.Insert(resultCodeKey, strconv.Itoa(int(resp.Result.Code))),
+	)
+	if err != nil {
+		return err
+	}
+
+	metrics.RecordBatch(ctx, requestCountM.M(1),
+		// Convert time.Duration in nanoseconds to milliseconds
+		responseTimeInMsecM.M(float64(d.Milliseconds())))
+	return nil
+}
+
 func RegisterMetrics() {
 	tagKeys := []tag.Key{
 		requestOperationKey,
@@ -115,7 +141,11 @@ func RegisterMetrics() {
 		resourceVersionKey,
 		resourceResourceKey,
 		resourceNamespaceKey,
-		admissionAllowedKey}
+		admissionAllowedKey,
+		desiredAPIVersionKey,
+		resultStatusKey,
+		resultReasonKey,
+		resultCodeKey}
 
 	if err := view.Register(
 		&view.View{
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 9acd0aebb..da0d92ff8 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1284,7 +1284,7 @@ k8s.io/utils/internal/third_party/forked/golang/net
 k8s.io/utils/net
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.34.1
+# knative.dev/eventing v0.34.1-0.20220906111743-7f1f49f5b4e0
 ## explicit; go 1.18
 knative.dev/eventing/cmd/event_display
 knative.dev/eventing/pkg/apis/config
@@ -1435,7 +1435,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.34.0
+# knative.dev/eventing-kafka v0.34.1-0.20220907020318-2f1fbd6dda1b
 ## explicit; go 1.16
 knative.dev/eventing-kafka/pkg/apis/bindings
 knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1
@@ -1502,11 +1502,11 @@ knative.dev/eventing-kafka/test/rekt/resources/resetoffset
 knative.dev/eventing-kafka/test/upgrade
 knative.dev/eventing-kafka/test/upgrade/continual
 knative.dev/eventing-kafka/test/upgrade/installation
-# knative.dev/hack v0.0.0-20220823140917-8d1e4ccf9dc3
-## explicit; go 1.17
+# knative.dev/hack v0.0.0-20220906140443-f9dc7220d362
+## explicit; go 1.18
 knative.dev/hack
 knative.dev/hack/shell
-# knative.dev/pkg v0.0.0-20220818004048-4a03844c0b15
+# knative.dev/pkg v0.0.0-20220826162920-93b66e6a8700
 ## explicit; go 1.18
 knative.dev/pkg/apiextensions/storageversion
 knative.dev/pkg/apiextensions/storageversion/cmd/migrate
@@ -1613,7 +1613,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-20220901014218-ff8756131bd5
+# knative.dev/reconciler-test v0.0.0-20220905144823-18e7400195c1
 ## explicit; go 1.18
 knative.dev/reconciler-test/cmd/eventshub
 knative.dev/reconciler-test/pkg/environment


Test patch:
diff --git a/control-plane/pkg/config/env_config_test.go b/control-plane/pkg/config/env_config_test.go
index 54328c467..541675dd0 100644
--- a/control-plane/pkg/config/env_config_test.go
+++ b/control-plane/pkg/config/env_config_test.go
@@ -38,19 +38,19 @@ func TestGetEnvConfig(t *testing.T) {
 			prefix: "BROKER",
 			want: &Env{
 				DataPlaneConfigMapNamespace: "knative-eventing",
-				DataPlaneConfigMapName:      "kafka-brokers-triggers",
+				ContractConfigMapName:       "kafka-brokers-triggers",
 				GeneralConfigMapName:        "kafka-config",
 				IngressName:                 "kafka-broker-ingress",
 				SystemNamespace:             "knative-eventing",
-				DataPlaneConfigFormat:       "json",
+				ContractConfigMapFormat:     "json",
 			},
 			setEnv: func() {
 				_ = os.Setenv("BROKER_DATA_PLANE_CONFIG_MAP_NAMESPACE", "knative-eventing")
-				_ = os.Setenv("BROKER_DATA_PLANE_CONFIG_MAP_NAME", "kafka-brokers-triggers")
+				_ = os.Setenv("BROKER_CONTRACT_CONFIG_MAP_NAME", "kafka-brokers-triggers")
 				_ = os.Setenv("BROKER_GENERAL_CONFIG_MAP_NAME", "kafka-config")
 				_ = os.Setenv("BROKER_INGRESS_NAME", "kafka-broker-ingress")
 				_ = os.Setenv("BROKER_SYSTEM_NAMESPACE", "knative-eventing")
-				_ = os.Setenv("BROKER_DATA_PLANE_CONFIG_FORMAT", "json")
+				_ = os.Setenv("BROKER_CONTRACT_CONFIG_MAP_FORMAT", "json")
 			},
 			wantErr: false,
 		},
@@ -59,19 +59,19 @@ func TestGetEnvConfig(t *testing.T) {
 			prefix: "SINK",
 			want: &Env{
 				DataPlaneConfigMapNamespace: "knative-eventing",
-				DataPlaneConfigMapName:      "kafka-sinks",
+				ContractConfigMapName:       "kafka-sinks",
 				GeneralConfigMapName:        "kafka-config",
 				IngressName:                 "kafka-sink-ingress",
 				SystemNamespace:             "knative-eventing",
-				DataPlaneConfigFormat:       "json",
+				ContractConfigMapFormat:     "json",
 			},
 			setEnv: func() {
 				_ = os.Setenv("SINK_DATA_PLANE_CONFIG_MAP_NAMESPACE", "knative-eventing")
-				_ = os.Setenv("SINK_DATA_PLANE_CONFIG_MAP_NAME", "kafka-sinks")
+				_ = os.Setenv("SINK_CONTRACT_CONFIG_MAP_NAME", "kafka-sinks")
 				_ = os.Setenv("SINK_GENERAL_CONFIG_MAP_NAME", "kafka-config")
 				_ = os.Setenv("SINK_INGRESS_NAME", "kafka-sink-ingress")
 				_ = os.Setenv("SINK_SYSTEM_NAMESPACE", "knative-eventing")
-				_ = os.Setenv("SINK_DATA_PLANE_CONFIG_FORMAT", "json")
+				_ = os.Setenv("SINK_CONTRACT_CONFIG_MAP_FORMAT", "json")
 			},
 			wantErr: false,
 		},
@@ -79,11 +79,11 @@ func TestGetEnvConfig(t *testing.T) {
 			name:   "missing required variable - SINK_DATA_PLANE_CONFIG_MAP_NAMESPACE",
 			prefix: "SINK",
 			setEnv: func() {
-				_ = os.Setenv("SINK_DATA_PLANE_CONFIG_MAP_NAME", "kafka-sinks")
+				_ = os.Setenv("SINK_CONTRACT_CONFIG_MAP_NAME", "kafka-sinks")
 				_ = os.Setenv("SINK_GENERAL_CONFIG_MAP_NAME", "kafka-config")
 				_ = os.Setenv("SINK_INGRESS_NAME", "kafka-sink-ingress")
 				_ = os.Setenv("SINK_SYSTEM_NAMESPACE", "knative-eventing")
-				_ = os.Setenv("SINK_DATA_PLANE_CONFIG_FORMAT", "json")
+				_ = os.Setenv("SINK_DATA_PLANE_CONFIG_MAP_FORMAT", "json")
 			},
 			wantErr: true,
 		},
@@ -100,11 +100,11 @@ func TestGetEnvConfig(t *testing.T) {
 			},
 			setEnv: func() {
 				_ = os.Setenv("BROKER_DATA_PLANE_CONFIG_MAP_NAMESPACE", "knative-eventing")
-				_ = os.Setenv("BROKER_DATA_PLANE_CONFIG_MAP_NAME", "kafka-brokers-triggers")
+				_ = os.Setenv("BROKER_CONTRACT_CONFIG_MAP_NAME", "kafka-brokers-triggers")
 				_ = os.Setenv("BROKER_GENERAL_CONFIG_MAP_NAME", "kafka-config")
 				_ = os.Setenv("BROKER_INGRESS_NAME", "kafka-broker-ingress")
 				_ = os.Setenv("BROKER_SYSTEM_NAMESPACE", "knative-eventing")
-				_ = os.Setenv("BROKER_DATA_PLANE_CONFIG_FORMAT", "json")
+				_ = os.Setenv("BROKER_CONTRACT_CONFIG_MAP_FORMAT", "json")
 				_ = os.Setenv("BROKER_DEFAULT_BACKOFF_DELAY_MS", "0")
 			},
 			wantErr: true,
@@ -122,11 +122,11 @@ func TestGetEnvConfig(t *testing.T) {
 			},
 			setEnv: func() {
 				_ = os.Setenv("BROKER_DATA_PLANE_CONFIG_MAP_NAMESPACE", "knative-eventing")
-				_ = os.Setenv("BROKER_DATA_PLANE_CONFIG_MAP_NAME", "kafka-brokers-triggers")
+				_ = os.Setenv("BROKER_CONTRACT_CONFIG_MAP_NAME", "kafka-brokers-triggers")
 				_ = os.Setenv("BROKER_GENERAL_CONFIG_MAP_NAME", "kafka-config")
 				_ = os.Setenv("BROKER_INGRESS_NAME", "kafka-broker-ingress")
 				_ = os.Setenv("BROKER_SYSTEM_NAMESPACE", "knative-eventing")
-				_ = os.Setenv("BROKER_DATA_PLANE_CONFIG_FORMAT", "json")
+				_ = os.Setenv("BROKER_CONTRACT_CONFIG_MAP_FORMAT", "json")
 			},
 			wantErr: true,
 		},
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index 52f214a9b..2e32e4b91 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -90,11 +90,11 @@ var (
 
 var DefaultEnv = &config.Env{
 	DataPlaneConfigMapNamespace: "knative-eventing",
-	DataPlaneConfigMapName:      "kafka-broker-brokers-triggers",
+	ContractConfigMapName:       "kafka-broker-brokers-triggers",
 	GeneralConfigMapName:        "kafka-broker-config",
 	IngressName:                 "kafka-broker-ingress",
 	SystemNamespace:             "knative-eventing",
-	DataPlaneConfigFormat:       base.Json,
+	ContractConfigMapFormat:     base.Json,
 	DefaultBackoffDelayMs:       1000,
 }
 
@@ -112,7 +112,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 
 	testKey := fmt.Sprintf("%s/%s", BrokerNamespace, BrokerName)
 
-	env.DataPlaneConfigFormat = format
+	env.ContractConfigMapFormat = format
 
 	table := TableTest{
 		{
@@ -120,7 +120,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 			Objects: []runtime.Object{
 				NewBroker(),
 				BrokerConfig(bootstrapServers, 20, 5),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
@@ -136,7 +136,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              BrokerUUID,
@@ -181,7 +181,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					WithExternalTopic(ExternalTopicName),
 				),
 				BrokerConfig(bootstrapServers, 20, 5),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
@@ -197,7 +197,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              BrokerUUID,
@@ -264,7 +264,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
@@ -290,7 +290,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 			Objects: []runtime.Object{
 				NewBroker(),
 				BrokerConfig(bootstrapServers, 20, 5),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
@@ -306,7 +306,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              BrokerUUID,
@@ -352,7 +352,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 			Objects: []runtime.Object{
 				NewBroker(),
 				BrokerConfig(bootstrapServers, 20, 5),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
@@ -368,7 +368,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              BrokerUUID,
@@ -418,7 +418,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Generation: 1,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "3"}),
 				BrokerDispatcherPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
@@ -428,7 +428,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              BrokerUUID,
@@ -477,7 +477,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Generation: 1,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				NewService(WithServiceNamespace(BrokerNamespace)),
 				BrokerReceiverPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "3"}),
 				BrokerDispatcherPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
@@ -487,7 +487,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              BrokerUUID,
@@ -548,7 +548,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
@@ -581,7 +581,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				&corev1.ConfigMap{
 					ObjectMeta: metav1.ObjectMeta{
 						Namespace: env.DataPlaneConfigMapNamespace,
-						Name:      env.DataPlaneConfigMapName + "a", // Use a different name
+						Name:      env.ContractConfigMapName + "a", // Use a different name
 					},
 				},
 			},
@@ -591,10 +591,10 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              BrokerUUID,
@@ -639,7 +639,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 			Objects: []runtime.Object{
 				NewBroker(),
 				BrokerConfig(bootstrapServers, 20, 5),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, []byte(`{"hello": "world"}`)),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, []byte(`{"hello": "world"}`)),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, nil),
 				BrokerDispatcherPod(env.SystemNamespace, map[string]string{
@@ -651,7 +651,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              BrokerUUID,
@@ -707,7 +707,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 							EgressConfig: &contract.EgressConfig{DeadLetter: "http://www.my-sink.com"},
 						},
 					},
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "2",
@@ -721,7 +721,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:          "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
@@ -799,7 +799,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 							EgressConfig: &contract.EgressConfig{DeadLetter: "http://www.my-sink.com"},
 						},
 					},
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "5",
@@ -813,7 +813,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:          "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
@@ -891,7 +891,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 							EgressConfig: &contract.EgressConfig{DeadLetter: "http://www.my-sink.com"},
 						},
 					},
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "5",
@@ -905,7 +905,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:          "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
@@ -971,7 +971,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						},
 					},
 					Generation: 1,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "5",
@@ -985,7 +985,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:          "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
@@ -1056,7 +1056,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						},
 					},
 					Generation: 1,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				BrokerReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "5",
 				}),
@@ -1103,7 +1103,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				NewConfigMapFromContract(&contract.Contract{
 					Resources:  []*contract.Resource{},
 					Generation: 1,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				BrokerReceiverPod(env.SystemNamespace, nil),
 				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
@@ -1139,7 +1139,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					),
 				),
 				BrokerConfig(bootstrapServers, 20, 5),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
@@ -1155,7 +1155,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              BrokerUUID,
@@ -1213,7 +1213,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				),
 				NewSSLSecret(ConfigMapNamespace, "secret-1"),
 				BrokerConfig(bootstrapServers, 20, 5, BrokerAuthConfig("secret-1")),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, map[string]string{
 					"annotation_to_preserve": "value_to_preserve",
@@ -1228,7 +1228,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
 				SecretFinalizerUpdate("secret-1", SecretFinalizerName),
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              BrokerUUID,
@@ -1308,7 +1308,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
@@ -1362,7 +1362,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
@@ -1423,7 +1423,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						},
 					},
 					Generation: 1,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, nil),
 				BrokerDispatcherPod(env.SystemNamespace, nil),
@@ -1433,7 +1433,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:    BrokerUUID,
@@ -1514,7 +1514,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
@@ -1538,7 +1538,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Generation: 1,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
 				BrokerDispatcherPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
@@ -1548,7 +1548,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              BrokerUUID,
@@ -1604,7 +1604,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Generation: 1,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
 				BrokerDispatcherPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "0"}),
@@ -1614,7 +1614,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              BrokerUUID,
@@ -1670,7 +1670,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Generation: 1,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
 				BrokerDispatcherPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
@@ -1680,7 +1680,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              BrokerUUID,
@@ -1733,7 +1733,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Generation: 1,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
 				BrokerDispatcherPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
@@ -1743,7 +1743,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              BrokerUUID,
@@ -1809,7 +1809,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						},
 					},
 					Generation: 1,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
 				BrokerDispatcherPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
@@ -1857,7 +1857,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						},
 					},
 					Generation: 1,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "0"}),
 				BrokerDispatcherPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "0"}),
@@ -1935,7 +1935,7 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 
 	testKey := fmt.Sprintf("%s/%s", BrokerNamespace, BrokerName)
 
-	env.DataPlaneConfigFormat = format
+	env.ContractConfigMapFormat = format
 
 	table := TableTest{
 		{
@@ -1952,11 +1952,11 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 						},
 					},
 					Generation: 1,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources:  []*contract.Resource{},
 					Generation: 2,
 				}),
@@ -1980,11 +1980,11 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 						},
 					},
 					Generation: 1,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources:  []*contract.Resource{},
 					Generation: 2,
 				}),
@@ -2006,11 +2006,11 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 						},
 					},
 					Generation: 1,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources:  []*contract.Resource{},
 					Generation: 2,
 				}),
@@ -2033,11 +2033,11 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 						},
 					},
 					Generation: 1,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources:  []*contract.Resource{},
 					Generation: 2,
 				}),
@@ -2064,11 +2064,11 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 						},
 					},
 					Generation: 1,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 2,
 				}),
 			},
@@ -2097,7 +2097,7 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 						},
 					},
 					Generation: 1,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, map[string]string{
 					"annotation_to_preserve": "value_to_preserve",
@@ -2108,7 +2108,7 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources:  []*contract.Resource{},
 					Generation: 2,
 				}),
@@ -2141,7 +2141,7 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 						},
 					},
 					Generation: 1,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 			},
 			Key:     testKey,
 			WantErr: true,
@@ -2154,7 +2154,7 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 				),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 2,
 				}),
 			},
@@ -2175,7 +2175,7 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 			Key:         testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{},
 			WantCreates: []runtime.Object{
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			OtherTestData: map[string]interface{}{
@@ -2201,11 +2201,11 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 						},
 					},
 					Generation: 5,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:          "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
@@ -2239,11 +2239,11 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 						},
 					},
 					Generation: 5,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:          "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
@@ -2273,7 +2273,7 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 						},
 					},
 					Generation: 5,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 			},
 			Key:         testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{},
@@ -2337,8 +2337,8 @@ func useTable(t *testing.T, table TableTest, env *config.Env) {
 				PodLister:                   listers.GetPodLister(),
 				SecretLister:                listers.GetSecretLister(),
 				DataPlaneConfigMapNamespace: env.DataPlaneConfigMapNamespace,
-				DataPlaneConfigMapName:      env.DataPlaneConfigMapName,
-				DataPlaneConfigFormat:       env.DataPlaneConfigFormat,
+				DataPlaneConfigMapName:      env.ContractConfigMapName,
+				DataPlaneConfigFormat:       env.ContractConfigMapFormat,
 				DataPlaneNamespace:          env.SystemNamespace,
 				DispatcherLabel:             base.BrokerDispatcherLabel,
 				ReceiverLabel:               base.BrokerReceiverLabel,
diff --git a/control-plane/pkg/reconciler/channel/channel_test.go b/control-plane/pkg/reconciler/channel/channel_test.go
index fea4692b7..41adb3899 100644
--- a/control-plane/pkg/reconciler/channel/channel_test.go
+++ b/control-plane/pkg/reconciler/channel/channel_test.go
@@ -72,11 +72,11 @@ var finalizerUpdatedEvent = Eventf(
 
 var DefaultEnv = &config.Env{
 	DataPlaneConfigMapNamespace: "knative-eventing",
-	DataPlaneConfigMapName:      "kafka-channel-channels-subscriptions",
+	ContractConfigMapName:       "kafka-channel-channels-subscriptions",
 	GeneralConfigMapName:        "kafka-channel-config",
 	IngressName:                 "kafka-channel-ingress",
 	SystemNamespace:             "knative-eventing",
-	DataPlaneConfigFormat:       base.Json,
+	ContractConfigMapFormat:     base.Json,
 }
 
 func TestReconcileKind(t *testing.T) {
@@ -113,7 +113,7 @@ func TestReconcileKind(t *testing.T) {
 				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 			},
 			OtherTestData: map[string]interface{}{
 				testProber: probertesting.MockProber(prober.StatusNotReady),
@@ -129,7 +129,7 @@ func TestReconcileKind(t *testing.T) {
 				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 			},
 			WantErr: true,
 			OtherTestData: map[string]interface{}{
@@ -155,7 +155,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -180,7 +180,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				NewPerChannelService(DefaultEnv),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
@@ -228,7 +228,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -257,7 +257,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -303,7 +303,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -328,7 +328,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				NewPerChannelService(&env),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
@@ -372,7 +372,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -397,7 +397,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				NewPerChannelService(&env),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
@@ -438,11 +438,11 @@ func TestReconcileKind(t *testing.T) {
 				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -509,11 +509,11 @@ func TestReconcileKind(t *testing.T) {
 				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -582,11 +582,11 @@ func TestReconcileKind(t *testing.T) {
 				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -655,11 +655,11 @@ func TestReconcileKind(t *testing.T) {
 				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -731,11 +731,11 @@ func TestReconcileKind(t *testing.T) {
 				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -1013,7 +1013,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -1038,7 +1038,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -1082,12 +1082,12 @@ func TestReconcileKind(t *testing.T) {
 					security.AuthSecretNameKey:         "secret-1",
 					security.AuthSecretNamespaceKey:    "ns-1",
 				}),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				NewLegacySSLSecret("ns-1", "secret-1"),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -1176,12 +1176,12 @@ func TestReconcileKind(t *testing.T) {
 					security.AuthSecretNameKey:         "secret-1",
 					security.AuthSecretNamespaceKey:    "ns-1",
 				}),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				NewLegacySASLSSLSecret("ns-1", "secret-1"),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -1273,12 +1273,12 @@ func TestReconcileKind(t *testing.T) {
 					security.AuthSecretNameKey:         "secret-1",
 					security.AuthSecretNamespaceKey:    "ns-1",
 				}),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				NewLegacySASLSecret("ns-1", "secret-1"),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -1368,7 +1368,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -1393,7 +1393,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -1432,11 +1432,11 @@ func TestReconcileKind(t *testing.T) {
 				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -1496,7 +1496,7 @@ func TestReconcileKind(t *testing.T) {
 				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, []byte("corrupt")),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, []byte("corrupt")),
 			},
 			Key:                     testKey,
 			WantErr:                 true,
@@ -1559,7 +1559,7 @@ func TestFinalizeKind(t *testing.T) {
 							},
 						},
 					},
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
@@ -1574,7 +1574,7 @@ func TestFinalizeKind(t *testing.T) {
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 2,
 					Resources:  []*contract.Resource{},
 				}),
@@ -1621,8 +1621,8 @@ func useTable(t *testing.T, table TableTest, env config.Env) {
 				PodLister:                   listers.GetPodLister(),
 				SecretLister:                listers.GetSecretLister(),
 				DataPlaneConfigMapNamespace: env.DataPlaneConfigMapNamespace,
-				DataPlaneConfigMapName:      env.DataPlaneConfigMapName,
-				DataPlaneConfigFormat:       env.DataPlaneConfigFormat,
+				DataPlaneConfigMapName:      env.ContractConfigMapName,
+				DataPlaneConfigFormat:       env.ContractConfigMapFormat,
 				DataPlaneNamespace:          env.SystemNamespace,
 				DispatcherLabel:             base.ChannelDispatcherLabel,
 				ReceiverLabel:               base.ChannelReceiverLabel,
diff --git a/control-plane/pkg/reconciler/channel/v2/channelv2_test.go b/control-plane/pkg/reconciler/channel/v2/channelv2_test.go
index 814b55f84..44f7df994 100644
--- a/control-plane/pkg/reconciler/channel/v2/channelv2_test.go
+++ b/control-plane/pkg/reconciler/channel/v2/channelv2_test.go
@@ -72,11 +72,11 @@ var finalizerUpdatedEvent = Eventf(
 
 var DefaultEnv = &config.Env{
 	DataPlaneConfigMapNamespace: "knative-eventing",
-	DataPlaneConfigMapName:      "kafka-channel-channels-subscriptions",
+	ContractConfigMapName:       "kafka-channel-channels-subscriptions",
 	GeneralConfigMapName:        "kafka-channel-config",
 	IngressName:                 "kafka-channel-ingress",
 	SystemNamespace:             "knative-eventing",
-	DataPlaneConfigFormat:       base.Json,
+	ContractConfigMapFormat:     base.Json,
 }
 
 func TestReconcileKind(t *testing.T) {
@@ -111,7 +111,7 @@ func TestReconcileKind(t *testing.T) {
 				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 			},
 			OtherTestData: map[string]interface{}{
 				testProber: probertesting.MockProber(prober.StatusNotReady),
@@ -127,7 +127,7 @@ func TestReconcileKind(t *testing.T) {
 				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 			},
 			WantErr: true,
 			OtherTestData: map[string]interface{}{
@@ -157,10 +157,10 @@ func TestReconcileKind(t *testing.T) {
 							},
 						},
 					},
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 				}),
 			},
@@ -186,7 +186,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -211,7 +211,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				NewPerChannelService(&env),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
@@ -257,7 +257,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -286,7 +286,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				NewPerChannelService(&env),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
@@ -324,7 +324,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -341,7 +341,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				NewPerChannelService(&env),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
@@ -376,7 +376,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -393,7 +393,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				NewPerChannelService(&env),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
@@ -425,7 +425,7 @@ func TestReconcileKind(t *testing.T) {
 				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 			},
 			Key: testKey,
 			WantCreates: []runtime.Object{
@@ -448,7 +448,7 @@ func TestReconcileKind(t *testing.T) {
 				),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -492,7 +492,7 @@ func TestReconcileKind(t *testing.T) {
 				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 			},
 			Key: testKey,
 			WantCreates: []runtime.Object{
@@ -515,7 +515,7 @@ func TestReconcileKind(t *testing.T) {
 				),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -559,7 +559,7 @@ func TestReconcileKind(t *testing.T) {
 				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 			},
 			Key: testKey,
 			WantCreates: []runtime.Object{
@@ -582,7 +582,7 @@ func TestReconcileKind(t *testing.T) {
 				),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -626,7 +626,7 @@ func TestReconcileKind(t *testing.T) {
 				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				NewConsumerGroup(
 					WithConsumerGroupName(Subscription1UUID),
 					WithConsumerGroupNamespace(ChannelNamespace),
@@ -640,7 +640,7 @@ func TestReconcileKind(t *testing.T) {
 				NewPerChannelService(&env),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -702,7 +702,7 @@ func TestReconcileKind(t *testing.T) {
 				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				NewConsumerGroup(
 					WithConsumerGroupName(Subscription1UUID),
 					WithConsumerGroupNamespace(ChannelNamespace),
@@ -740,7 +740,7 @@ func TestReconcileKind(t *testing.T) {
 				},
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -772,7 +772,7 @@ func TestReconcileKind(t *testing.T) {
 				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 			},
 			Key: testKey,
 			WantCreates: []runtime.Object{
@@ -811,7 +811,7 @@ func TestReconcileKind(t *testing.T) {
 				),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -857,7 +857,7 @@ func TestReconcileKind(t *testing.T) {
 				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				NewConsumerGroup(
 					WithConsumerGroupName(Subscription2UUID), //removed subscriber
 					WithConsumerGroupNamespace(ChannelNamespace),
@@ -896,7 +896,7 @@ func TestReconcileKind(t *testing.T) {
 				),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -1053,7 +1053,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -1070,7 +1070,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				NewPerChannelService(&env),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
@@ -1105,7 +1105,7 @@ func TestReconcileKind(t *testing.T) {
 					security.AuthSecretNameKey:         "secret-1",
 					security.AuthSecretNamespaceKey:    "ns-1",
 				}),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				NewSSLSecret("ns-1", "secret-1"),
 				NewConsumerGroup(
 					WithConsumerGroupName(Subscription1UUID),
@@ -1130,7 +1130,7 @@ func TestReconcileKind(t *testing.T) {
 				NewPerChannelService(&env),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -1185,7 +1185,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -1202,7 +1202,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				NewPerChannelService(&env),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
@@ -1232,14 +1232,14 @@ func TestReconcileKind(t *testing.T) {
 				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 			},
 			Key: testKey,
 			WantCreates: []runtime.Object{
 				NewPerChannelService(&env),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -1281,7 +1281,7 @@ func TestReconcileKind(t *testing.T) {
 				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, []byte("corrupt")),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, []byte("corrupt")),
 			},
 			Key:                     testKey,
 			WantErr:                 true,
@@ -1336,8 +1336,8 @@ func TestReconcileKind(t *testing.T) {
 				PodLister:                   listers.GetPodLister(),
 				SecretLister:                listers.GetSecretLister(),
 				DataPlaneConfigMapNamespace: env.DataPlaneConfigMapNamespace,
-				DataPlaneConfigMapName:      env.DataPlaneConfigMapName,
-				DataPlaneConfigFormat:       env.DataPlaneConfigFormat,
+				DataPlaneConfigMapName:      env.ContractConfigMapName,
+				DataPlaneConfigFormat:       env.ContractConfigMapFormat,
 				DataPlaneNamespace:          env.SystemNamespace,
 				DispatcherLabel:             base.ChannelDispatcherLabel,
 				ReceiverLabel:               base.ChannelReceiverLabel,
diff --git a/control-plane/pkg/reconciler/consumer/controller_test.go b/control-plane/pkg/reconciler/consumer/controller_test.go
index 039f7a920..2d2171361 100644
--- a/control-plane/pkg/reconciler/consumer/controller_test.go
+++ b/control-plane/pkg/reconciler/consumer/controller_test.go
@@ -50,7 +50,7 @@ func TestNewController(t *testing.T) {
 
 	dynamicclient.With(ctx, dynamicScheme)
 
-	t.Setenv("CONSUMER_DATA_PLANE_CONFIG_FORMAT", "json")
+	t.Setenv("CONSUMER_CONTRACT_CONFIG_MAP_FORMAT", "json")
 
 	controller := NewController(ctx, configmap.NewStaticWatcher(&corev1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink_test.go b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
index 6ef1e2180..dc88866ea 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink_test.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
@@ -85,11 +85,11 @@ var (
 
 var DefaultEnv = &config.Env{
 	DataPlaneConfigMapNamespace: "knative-eventing",
-	DataPlaneConfigMapName:      "kafka-sink-sinks",
+	ContractConfigMapName:       "kafka-sink-sinks",
 	GeneralConfigMapName:        "kafka-broker-config",
 	IngressName:                 "kafka-sink-ingress",
 	SystemNamespace:             "knative-eventing",
-	DataPlaneConfigFormat:       base.Json,
+	ContractConfigMapFormat:     base.Json,
 }
 
 func TestSinkReconciler(t *testing.T) {
@@ -107,7 +107,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 
 	testKey := fmt.Sprintf("%s/%s", SinkNamespace, SinkName)
 
-	env.DataPlaneConfigFormat = format
+	env.ContractConfigMapFormat = format
 
 	table := TableTest{
 		{
@@ -116,7 +116,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 				NewSink(
 					StatusControllerOwnsTopic(reconciler.ControllerTopicOwner),
 				),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					"annotation_to_preserve": "value_to_preserve",
 				}),
@@ -126,7 +126,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              SinkUUID,
@@ -169,7 +169,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 					StatusControllerOwnsTopic(reconciler.ControllerTopicOwner),
 					SinkContentMode(v1alpha1.ModeStructured),
 				),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					"annotation_to_preserve": "value_to_preserve",
 				}),
@@ -179,7 +179,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              SinkUUID,
@@ -224,7 +224,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 					SinkAuthSecretRef("secret-1"),
 				),
 				NewSSLSecret(SinkNamespace, "secret-1"),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
@@ -235,7 +235,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              SinkUUID,
@@ -291,7 +291,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						s.Spec.NumPartitions = nil
 					},
 				),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
@@ -302,7 +302,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              SinkUUID,
@@ -357,7 +357,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						s.Spec.NumPartitions = nil
 					},
 				),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 					"annotation_to_preserve":           "value_to_preserve",
@@ -407,7 +407,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						s.Spec.BootstrapServers = []string{"kafka-broker:10000"}
 					},
 				),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
@@ -418,7 +418,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              SinkUUID,
@@ -512,7 +512,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 				&corev1.ConfigMap{
 					ObjectMeta: metav1.ObjectMeta{
 						Namespace: env.DataPlaneConfigMapNamespace,
-						Name:      env.DataPlaneConfigMapName + "a", // Use a different name
+						Name:      env.ContractConfigMapName + "a", // Use a different name
 					},
 				},
 			},
@@ -522,10 +522,10 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              SinkUUID,
@@ -567,7 +567,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 					StatusControllerOwnsTopic(reconciler.ControllerTopicOwner),
 					BootstrapServers(bootstrapServersArr),
 				),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, []byte(`{"hello": "world"}`)),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, []byte(`{"hello": "world"}`)),
 				SinkReceiverPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
@@ -575,7 +575,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              SinkUUID,
@@ -629,7 +629,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 							Topics: []string{"my-existing-topic-b"},
 						},
 					},
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "2",
 				}),
@@ -639,7 +639,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:     "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
@@ -705,7 +705,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY},
 						},
 					},
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "5",
 				}),
@@ -715,7 +715,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:          "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
@@ -809,7 +809,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						},
 					},
 					Generation: 1,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
@@ -864,7 +864,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						},
 					},
 					Generation: 1,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
 				}),
@@ -903,7 +903,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 				NewSink(
 					StatusControllerOwnsTopic(reconciler.ControllerTopicOwner),
 				),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					"annotation_to_preserve": "value_to_preserve",
 				}),
@@ -913,7 +913,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              SinkUUID,
@@ -957,7 +957,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 				NewSink(
 					StatusControllerOwnsTopic(reconciler.ControllerTopicOwner),
 				),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					"annotation_to_preserve": "value_to_preserve",
 				}),
@@ -967,7 +967,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              SinkUUID,
@@ -1029,7 +1029,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 
 	testKey := fmt.Sprintf("%s/%s", SinkNamespace, SinkName)
 
-	env.DataPlaneConfigFormat = format
+	env.ContractConfigMapFormat = format
 
 	table := TableTest{
 		{
@@ -1060,11 +1060,11 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 						},
 					},
 					Generation: 1,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:    SinkUUID + "a",
@@ -1111,11 +1111,11 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 						},
 					},
 					Generation: 1,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:    SinkUUID + "a",
@@ -1162,11 +1162,11 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 						},
 					},
 					Generation: 1,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:    SinkUUID + "a",
@@ -1213,11 +1213,11 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 						},
 					},
 					Generation: 1,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:    SinkUUID + "a",
@@ -1269,7 +1269,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 						},
 					},
 					Generation: 1,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 			},
 			Key:     testKey,
 			WantErr: true,
@@ -1282,7 +1282,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 				),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:    SinkUUID + "a",
@@ -1377,8 +1377,8 @@ func useTable(t *testing.T, table TableTest, env *config.Env) {
 				PodLister:                   listers.GetPodLister(),
 				SecretLister:                listers.GetSecretLister(),
 				DataPlaneConfigMapNamespace: env.DataPlaneConfigMapNamespace,
-				DataPlaneConfigMapName:      env.DataPlaneConfigMapName,
-				DataPlaneConfigFormat:       env.DataPlaneConfigFormat,
+				DataPlaneConfigMapName:      env.ContractConfigMapName,
+				DataPlaneConfigFormat:       env.ContractConfigMapFormat,
 				DataPlaneNamespace:          env.SystemNamespace,
 				ReceiverLabel:               base.SinkReceiverLabel,
 			},
diff --git a/control-plane/pkg/reconciler/trigger/namespaced_trigger_test.go b/control-plane/pkg/reconciler/trigger/namespaced_trigger_test.go
index 5868aeeed..ba3df6765 100644
--- a/control-plane/pkg/reconciler/trigger/namespaced_trigger_test.go
+++ b/control-plane/pkg/reconciler/trigger/namespaced_trigger_test.go
@@ -57,7 +57,7 @@ func namespacedTriggerReconciliation(t *testing.T, format string, env config.Env
 
 	testKey := fmt.Sprintf("%s/%s", TriggerNamespace, TriggerName)
 
-	env.DataPlaneConfigFormat = format
+	env.ContractConfigMapFormat = format
 
 	table := TableTest{
 		{
@@ -76,7 +76,7 @@ func namespacedTriggerReconciliation(t *testing.T, format string, env config.Env
 							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 						},
 					},
-				}, BrokerNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, BrokerNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				BrokerDispatcherPod(BrokerNamespace, nil),
 			},
 			Key: testKey,
@@ -87,7 +87,7 @@ func namespacedTriggerReconciliation(t *testing.T, format string, env config.Env
 				patchFinalizers(),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(BrokerNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(BrokerNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:     BrokerUUID,
@@ -143,8 +143,8 @@ func useNamespacedTable(t *testing.T, table TableTest, env *config.Env) {
 				PodLister:                   listers.GetPodLister(),
 				SecretLister:                listers.GetSecretLister(),
 				DataPlaneConfigMapNamespace: env.DataPlaneConfigMapNamespace,
-				DataPlaneConfigMapName:      env.DataPlaneConfigMapName,
-				DataPlaneConfigFormat:       env.DataPlaneConfigFormat,
+				DataPlaneConfigMapName:      env.ContractConfigMapName,
+				DataPlaneConfigFormat:       env.ContractConfigMapFormat,
 				DataPlaneNamespace:          env.SystemNamespace,
 				DispatcherLabel:             base.BrokerDispatcherLabel,
 				ReceiverLabel:               base.BrokerReceiverLabel,
diff --git a/control-plane/pkg/reconciler/trigger/trigger_test.go b/control-plane/pkg/reconciler/trigger/trigger_test.go
index 9372627a0..68a7f5c3d 100644
--- a/control-plane/pkg/reconciler/trigger/trigger_test.go
+++ b/control-plane/pkg/reconciler/trigger/trigger_test.go
@@ -57,11 +57,11 @@ import (
 
 var DefaultEnv = &config.Env{
 	DataPlaneConfigMapNamespace: "knative-eventing",
-	DataPlaneConfigMapName:      "kafka-broker-brokers-triggers",
+	ContractConfigMapName:       "kafka-broker-brokers-triggers",
 	GeneralConfigMapName:        "kafka-broker-config",
 	IngressName:                 "kafka-broker-ingress",
 	SystemNamespace:             "knative-eventing",
-	DataPlaneConfigFormat:       base.Json,
+	ContractConfigMapFormat:     base.Json,
 	DefaultBackoffDelayMs:       1000,
 }
 
@@ -118,7 +118,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 
 	testKey := fmt.Sprintf("%s/%s", TriggerNamespace, TriggerName)
 
-	env.DataPlaneConfigFormat = format
+	env.ContractConfigMapFormat = format
 
 	table := TableTest{
 		{
@@ -137,7 +137,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 						},
 					},
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
@@ -148,7 +148,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 				patchFinalizers(),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:     BrokerUUID,
@@ -201,7 +201,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 						},
 					},
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
@@ -212,7 +212,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 				patchFinalizers(),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:     BrokerUUID,
@@ -265,7 +265,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 						},
 					},
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
@@ -276,7 +276,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 				patchFinalizers(),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:     BrokerUUID,
@@ -336,7 +336,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 						},
 					},
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
@@ -347,7 +347,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 				patchFinalizers(),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:     BrokerUUID,
@@ -401,7 +401,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 						},
 					},
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
@@ -412,7 +412,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 				patchFinalizers(),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:     BrokerUUID,
@@ -466,7 +466,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 						},
 					},
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
@@ -477,7 +477,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 				patchFinalizers(),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:     BrokerUUID,
@@ -593,7 +593,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						},
 					},
 					Generation: 2,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
@@ -604,7 +604,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 				patchFinalizers(),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:     BrokerUUID + "z",
@@ -688,7 +688,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 			WantErr:                 true,
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 			},
 			WantEvents: []string{
 				finalizerUpdatedEvent,
@@ -719,7 +719,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 				NewBroker(BrokerReady),
 				newTrigger(),
 				NewService(),
-				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
@@ -801,7 +801,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 				NewDeletedBroker(),
 				NewConfigMapFromContract(&contract.Contract{
 					Generation: 8,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
@@ -831,7 +831,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						},
 					},
 					Generation: 8,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
@@ -874,7 +874,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						},
 					},
 					Generation: 8,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			WantPatches: []clientgotesting.PatchActionImpl{
@@ -885,7 +885,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:    BrokerUUID,
@@ -963,7 +963,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 							},
 						},
 					},
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
@@ -1133,7 +1133,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 							},
 						},
 					},
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
@@ -1144,7 +1144,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 				patchFinalizers(),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:     BrokerUUID,
@@ -1431,7 +1431,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 							},
 						},
 					},
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
@@ -1442,7 +1442,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 				patchFinalizers(),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:     BrokerUUID + "a",
@@ -1791,7 +1791,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 							},
 						},
 					},
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
@@ -1802,7 +1802,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 				patchFinalizers(),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:     BrokerUUID + "a",
@@ -2053,7 +2053,7 @@ func triggerFinalizer(t *testing.T, format string, env config.Env) {
 
 	testKey := fmt.Sprintf("%s/%s", TriggerNamespace, TriggerName)
 
-	env.DataPlaneConfigFormat = format
+	env.ContractConfigMapFormat = format
 
 	table := TableTest{
 		{
@@ -2081,7 +2081,7 @@ func triggerFinalizer(t *testing.T, format string, env config.Env) {
 						},
 					},
 					Generation: 8,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			WantPatches: []clientgotesting.PatchActionImpl{
@@ -2092,7 +2092,7 @@ func triggerFinalizer(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:    BrokerUUID,
@@ -2133,7 +2133,7 @@ func triggerFinalizer(t *testing.T, format string, env config.Env) {
 						},
 					},
 					Generation: 8,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
@@ -2157,7 +2157,7 @@ func triggerFinalizer(t *testing.T, format string, env config.Env) {
 				NewDeletedBroker(),
 				NewConfigMapFromContract(&contract.Contract{
 					Generation: 8,
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
@@ -2251,7 +2251,7 @@ func triggerFinalizer(t *testing.T, format string, env config.Env) {
 							},
 						},
 					},
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
@@ -2262,7 +2262,7 @@ func triggerFinalizer(t *testing.T, format string, env config.Env) {
 				patchFinalizers(),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:     BrokerUUID,
@@ -2417,7 +2417,7 @@ func triggerFinalizer(t *testing.T, format string, env config.Env) {
 							},
 						},
 					},
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
@@ -2428,7 +2428,7 @@ func triggerFinalizer(t *testing.T, format string, env config.Env) {
 				patchFinalizers(),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:     BrokerUUID + "a",
@@ -2616,7 +2616,7 @@ func triggerFinalizer(t *testing.T, format string, env config.Env) {
 							},
 						},
 					},
-				}, env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat),
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
 				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
@@ -2627,7 +2627,7 @@ func triggerFinalizer(t *testing.T, format string, env config.Env) {
 				patchFinalizers(),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:     BrokerUUID + "a",
@@ -2759,8 +2759,8 @@ func useTableWithFlags(t *testing.T, table TableTest, env *config.Env, flags fea
 				PodLister:                   listers.GetPodLister(),
 				SecretLister:                listers.GetSecretLister(),
 				DataPlaneConfigMapNamespace: env.DataPlaneConfigMapNamespace,
-				DataPlaneConfigMapName:      env.DataPlaneConfigMapName,
-				DataPlaneConfigFormat:       env.DataPlaneConfigFormat,
+				DataPlaneConfigMapName:      env.ContractConfigMapName,
+				DataPlaneConfigFormat:       env.ContractConfigMapFormat,
 				DataPlaneNamespace:          env.SystemNamespace,
 				DispatcherLabel:             base.BrokerDispatcherLabel,
 				ReceiverLabel:               base.BrokerReceiverLabel,
diff --git a/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go b/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go
index 417d586a8..fd0fd9b43 100644
--- a/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go
+++ b/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go
@@ -54,7 +54,7 @@ const (
 
 var DefaultEnv = &config.Env{
 	DataPlaneConfigMapNamespace: "knative-eventing",
-	DataPlaneConfigMapName:      "kafka-broker-brokers-triggers",
+	ContractConfigMapName:       "kafka-broker-brokers-triggers",
 	GeneralConfigMapName:        "kafka-broker-config",
 	SystemNamespace:             "knative-eventing",
 }
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherTest.java
index c11d828de..a94c0e919 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherTest.java
@@ -47,6 +47,7 @@ import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.ArgumentCaptor;
 
+import java.util.Base64;
 import java.util.Collections;
 import java.util.concurrent.atomic.AtomicBoolean;
 
@@ -291,7 +292,7 @@ public class RecordDispatcherTest {
     assertEquals(record.value().getData(), failedRecord.value().getData());
     assertEquals("testdest", failedRecord.value().getExtension("knativeerrordest"));
     assertEquals(String.valueOf(errorCode), failedRecord.value().getExtension("knativeerrorcode"));
-    assertEquals(errorBody, failedRecord.value().getExtension("knativeerrordata"));
+    assertEquals(Base64.getEncoder().encodeToString(errorBody.getBytes()), failedRecord.value().getExtension("knativeerrordata"));
 
     assertEventDispatchLatency();
     assertEventProcessingLatency();
@@ -352,7 +353,7 @@ public class RecordDispatcherTest {
     assertEquals(record.value().getData(), failedRecord.value().getData());
     assertEquals("testdest", failedRecord.value().getExtension("knativeerrordest"));
     assertEquals(String.valueOf(errorCode), failedRecord.value().getExtension("knativeerrorcode"));
-    assertEquals(errorBody, failedRecord.value().getExtension("knativeerrordata"));
+    assertEquals(Base64.getEncoder().encodeToString(errorBody.getBytes()), failedRecord.value().getExtension("knativeerrordata"));
 
     assertEventDispatchLatency();
     assertEventProcessingLatency();
diff --git a/test/cmd/watch-cm/main.go b/test/cmd/watch-cm/main.go
index c162b943b..1b1c73789 100644
--- a/test/cmd/watch-cm/main.go
+++ b/test/cmd/watch-cm/main.go
@@ -37,9 +37,9 @@ func main() {
 	testobservability.WatchDataPlaneConfigMap(
 		types.NamespacedName{
 			Namespace: envConfig.DataPlaneConfigMapNamespace,
-			Name:      envConfig.DataPlaneConfigMapName,
+			Name:      envConfig.ContractConfigMapName,
 		},
-		envConfig.DataPlaneConfigFormat,
+		envConfig.ContractConfigMapFormat,
 	)
 
 }
diff --git a/test/config/cm-watcher-broker.yaml b/test/config/cm-watcher-broker.yaml
index 4b428304e..a4d7d3032 100644
--- a/test/config/cm-watcher-broker.yaml
+++ b/test/config/cm-watcher-broker.yaml
@@ -46,7 +46,7 @@ spec:
           env:
             - name: DATA_PLANE_CONFIG_MAP_NAMESPACE
               value: knative-eventing
-            - name: DATA_PLANE_CONFIG_MAP_NAME
+            - name: CONTRACT_CONFIG_MAP_NAME
               value: kafka-broker-brokers-triggers
             - name: CONFIG_LOGGING_NAME
               value: config-logging
@@ -54,7 +54,7 @@ spec:
               value: config-observability
             - name: METRICS_DOMAIN
               value: knative.dev/eventing
-            - name: DATA_PLANE_CONFIG_FORMAT
+            - name: CONTRACT_CONFIG_MAP_FORMAT
               value: json
             - name: INGRESS_NAME
               value: kafka-broker-ingress
diff --git a/test/config/cm-watcher-channel.yaml b/test/config/cm-watcher-channel.yaml
index 5167bcc46..13c10941f 100644
--- a/test/config/cm-watcher-channel.yaml
+++ b/test/config/cm-watcher-channel.yaml
@@ -46,7 +46,7 @@ spec:
           env:
             - name: DATA_PLANE_CONFIG_MAP_NAMESPACE
               value: knative-eventing
-            - name: DATA_PLANE_CONFIG_MAP_NAME
+            - name: CONTRACT_CONFIG_MAP_NAME
               value: kafka-channel-channels-subscriptions
             - name: CONFIG_LOGGING_NAME
               value: config-logging
@@ -54,7 +54,7 @@ spec:
               value: config-observability
             - name: METRICS_DOMAIN
               value: knative.dev/eventing
-            - name: DATA_PLANE_CONFIG_FORMAT
+            - name: CONTRACT_CONFIG_MAP_FORMAT
               value: json
             - name: INGRESS_NAME
               value: kafka-channel-ingress
diff --git a/test/config/cm-watcher-sink.yaml b/test/config/cm-watcher-sink.yaml
index e98238b3d..d26a23422 100644
--- a/test/config/cm-watcher-sink.yaml
+++ b/test/config/cm-watcher-sink.yaml
@@ -46,7 +46,7 @@ spec:
           env:
             - name: DATA_PLANE_CONFIG_MAP_NAMESPACE
               value: knative-eventing
-            - name: DATA_PLANE_CONFIG_MAP_NAME
+            - name: CONTRACT_CONFIG_MAP_NAME
               value: kafka-sink-sinks
             - name: CONFIG_LOGGING_NAME
               value: config-logging
@@ -54,7 +54,7 @@ spec:
               value: config-observability
             - name: METRICS_DOMAIN
               value: knative.dev/eventing
-            - name: DATA_PLANE_CONFIG_FORMAT
+            - name: CONTRACT_CONFIG_MAP_FORMAT
               value: json
             - name: INGRESS_NAME
               value: kafka-sink-ingress
diff --git a/test/e2e_new/dls_extensions_test.go b/test/e2e_new/dls_extensions_test.go
index 1dd14bc99..893b14f6d 100644
--- a/test/e2e_new/dls_extensions_test.go
+++ b/test/e2e_new/dls_extensions_test.go
@@ -21,6 +21,8 @@ package e2e_new
 
 import (
 	"context"
+	"encoding/base64"
+	"html"
 	"strings"
 	"testing"
 	"time"
@@ -56,6 +58,7 @@ func TestDeadLetterSinkExtensions(t *testing.T) {
 	env.Test(ctx, t, SubscriberReturnedErrorNoData())
 	env.Test(ctx, t, SubscriberReturnedErrorSmallData())
 	env.Test(ctx, t, SubscriberReturnedErrorLargeData())
+	env.Test(ctx, t, SubscriberReturnedHtmlWebpage())
 }
 
 func SubscriberUnreachable() *feature.Feature {
@@ -224,7 +227,7 @@ func SubscriberReturnedErrorSmallData() *feature.Feature {
 			return cetest.HasExtension("knativeerrorcode", "422")
 		},
 		func(ctx context.Context) cetest.EventMatcher {
-			return cetest.HasExtension("knativeerrordata", errorData)
+			return cetest.HasExtension("knativeerrordata", base64.StdEncoding.EncodeToString([]byte(errorData)))
 		},
 	))
 
@@ -288,7 +291,70 @@ func SubscriberReturnedErrorLargeData() *feature.Feature {
 			return cetest.HasExtension("knativeerrorcode", "422")
 		},
 		func(ctx context.Context) cetest.EventMatcher {
-			return cetest.HasExtension("knativeerrordata", errorDataTruncated)
+			return cetest.HasExtension("knativeerrordata", base64.StdEncoding.EncodeToString([]byte(errorDataTruncated)))
+		},
+	))
+
+	return f
+}
+
+func SubscriberReturnedHtmlWebpage() *feature.Feature {
+	f := feature.NewFeature()
+
+	sourceName := feature.MakeRandomK8sName("source")
+	sinkName := feature.MakeRandomK8sName("sink")
+	deadLetterSinkName := feature.MakeRandomK8sName("dls")
+	triggerName := feature.MakeRandomK8sName("trigger")
+	brokerName := feature.MakeRandomK8sName("broker")
+
+	ev := cetest.FullEvent()
+
+	f.Setup("install one partition configuration", single_partition_config.Install)
+	f.Setup("install broker", broker.Install(
+		brokerName,
+		broker.WithBrokerClass(kafka.BrokerClass),
+		broker.WithConfig(single_partition_config.ConfigMapName),
+	))
+	f.Setup("broker is ready", broker.IsReady(brokerName))
+	f.Setup("broker is addressable", broker.IsAddressable(brokerName))
+
+	errorData := "<!doctype html>\n<html>\n<head>\n    <title>Error Page(tm)</title>\n</head>\n<body>\n<p>Quoth the server, 404!\n</body></html>"
+	f.Setup("install sink", eventshub.Install(
+		sinkName,
+		eventshub.StartReceiver,
+		eventshub.DropFirstN(1),
+		eventshub.DropEventsResponseCode(404),
+		eventshub.DropEventsResponseBody(errorData),
+	))
+	f.Setup("install dead letter sink", eventshub.Install(
+		deadLetterSinkName,
+		eventshub.StartReceiver,
+	))
+	f.Setup("install trigger", trigger.Install(
+		triggerName,
+		brokerName,
+		trigger.WithSubscriber(svc.AsKReference(sinkName), ""),
+		trigger.WithDeadLetterSink(svc.AsKReference(deadLetterSinkName), ""),
+	))
+	f.Setup("trigger is ready", trigger.IsReady(triggerName))
+
+	f.Requirement("install source", eventshub.Install(
+		sourceName,
+		eventshub.StartSenderToResource(broker.GVR(), brokerName),
+		eventshub.InputEvent(ev),
+	))
+
+	f.Assert("knativeerrordest, knativeerrorcode, knativeerrordata added", assertEnhancedWithKnativeErrorExtensions(
+		deadLetterSinkName,
+		func(ctx context.Context) cetest.EventMatcher {
+			sinkAddress, _ := svc.Address(ctx, sinkName)
+			return cetest.HasExtension("knativeerrordest", sinkAddress.String())
+		},
+		func(ctx context.Context) cetest.EventMatcher {
+			return cetest.HasExtension("knativeerrorcode", "404")
+		},
+		func(ctx context.Context) cetest.EventMatcher {
+			return cetest.HasExtension("knativeerrordata", html.UnescapeString(base64.StdEncoding.EncodeToString([]byte(errorData))))
 		},
 	))
 
diff --git a/third_party/eventing-latest/eventing-core.yaml b/third_party/eventing-latest/eventing-core.yaml
index 3709147ac..7f371dadb 100644
--- a/third_party/eventing-latest/eventing-core.yaml
+++ b/third_party/eventing-latest/eventing-core.yaml
@@ -16,8 +16,8 @@ kind: Namespace
 metadata:
   name: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 
 ---
@@ -41,8 +41,8 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 ---
 apiVersion: rbac.authorization.k8s.io/v1
@@ -50,8 +50,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -67,8 +67,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-resolver
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -84,8 +84,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-source-observer
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -101,8 +101,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-sources-controller
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -118,8 +118,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-manipulator
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -151,8 +151,8 @@ metadata:
   name: pingsource-mt-adapter
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 ---
 apiVersion: rbac.authorization.k8s.io/v1
@@ -160,8 +160,8 @@ kind: ClusterRoleBinding
 metadata:
   name: knative-eventing-pingsource-mt-adapter
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -193,8 +193,8 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 ---
 apiVersion: rbac.authorization.k8s.io/v1
@@ -202,8 +202,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -220,8 +220,8 @@ metadata:
   namespace: knative-eventing
   name: eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -237,8 +237,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-webhook-resolver
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -254,8 +254,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-webhook-podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -287,8 +287,8 @@ metadata:
   name: config-br-default-channel
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 data:
   channel-template-spec: |
@@ -316,8 +316,8 @@ metadata:
   name: config-br-defaults
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 data:
   # Configures the default for any Broker that does not specify a spec.config or Broker class.
@@ -354,8 +354,8 @@ metadata:
   name: default-ch-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 data:
   # Configuration for defaulting channels that do not specify CRD implementations.
@@ -389,10 +389,10 @@ metadata:
   name: config-ping-defaults
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
   annotations:
     knative.dev/example-checksum: "9185c153"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 data:
   _example: |
@@ -436,10 +436,10 @@ metadata:
   name: config-features
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 data:
   # ALPHA feature: The kreference-group allows you to use the Group field in KReferences.
@@ -483,7 +483,7 @@ metadata:
   name: config-kreference-mapping
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
   annotations:
@@ -532,8 +532,8 @@ metadata:
   name: config-leader-election
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
   annotations:
     knative.dev/example-checksum: "f7948630"
@@ -594,10 +594,10 @@ metadata:
   name: config-logging
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 data:
   # Common configuration for all Knative codebase
@@ -648,10 +648,10 @@ metadata:
   name: config-observability
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
   annotations:
     knative.dev/example-checksum: "f46cf09d"
@@ -725,8 +725,8 @@ metadata:
   name: config-sugar
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
   annotations:
     knative.dev/example-checksum: "62dfac6f"
@@ -783,10 +783,10 @@ metadata:
   name: config-tracing
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
   annotations:
     knative.dev/example-checksum: "0492ceb0"
@@ -841,10 +841,10 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     knative.dev/high-availability: "true"
     app.kubernetes.io/component: eventing-controller
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 spec:
   selector:
@@ -854,9 +854,9 @@ spec:
     metadata:
       labels:
         app: eventing-controller
-        eventing.knative.dev/release: "v20220831-ceaae2455"
+        eventing.knative.dev/release: "v20220905-e1866d766"
         app.kubernetes.io/component: eventing-controller
-        app.kubernetes.io/version: "20220831-ceaae2455"
+        app.kubernetes.io/version: "20220905-e1866d766"
         app.kubernetes.io/name: knative-eventing
     spec:
       # To avoid node becoming SPOF, spread our replicas to different nodes.
@@ -874,7 +874,7 @@ spec:
       containers:
         - name: eventing-controller
           terminationMessagePolicy: FallbackToLogsOnError
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:1e32e86f679fb6213dcbb0efdbffa93337031082502bc4a15646d09ffbed20f3
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:0c8bc06b4bcf31c9dc32144618e8a6e3f403780af8f545d264c027d169ce3367
           resources:
             requests:
               cpu: 100m
@@ -892,7 +892,7 @@ spec:
               value: knative.dev/eventing
             # APIServerSource
             - name: APISERVER_RA_IMAGE
-              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:2eb36c88da61e919d9a0da5b65d0bfc073a8ccbf9a0195f9644a430e21af5ce1
+              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:415bb3d37f9f28851103420184a9fbe13a7b6d64032cbb9331cb593d9021b816
             - name: POD_NAME
               valueFrom:
                 fieldRef:
@@ -941,9 +941,9 @@ metadata:
   name: pingsource-mt-adapter
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     app.kubernetes.io/component: pingsource-mt-adapter
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 spec:
   # when set to 0 (and only 0) will be set to 1 when the first PingSource is created.
@@ -956,9 +956,9 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20220831-ceaae2455"
+        eventing.knative.dev/release: "v20220905-e1866d766"
         app.kubernetes.io/component: pingsource-mt-adapter
-        app.kubernetes.io/version: "20220831-ceaae2455"
+        app.kubernetes.io/version: "20220905-e1866d766"
         app.kubernetes.io/name: knative-eventing
     spec:
       affinity:
@@ -972,7 +972,7 @@ spec:
       enableServiceLinks: false
       containers:
         - name: dispatcher
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:b2687bbab50e79a6aeeb6845945c8ab012553fdbbb62bf0b2b43b8eb63460d6f
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:0991c5e3b7b1c459f7ce660d5a964ab0f154dba4e03da02018462f7071af6088
           env:
             - name: SYSTEM_NAMESPACE
               value: ''
@@ -1043,9 +1043,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 spec:
   scaleTargetRef:
@@ -1069,9 +1069,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 spec:
   minAvailable: 80%
@@ -1100,9 +1100,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 spec:
   selector:
@@ -1113,9 +1113,9 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20220831-ceaae2455"
+        eventing.knative.dev/release: "v20220905-e1866d766"
         app.kubernetes.io/component: eventing-webhook
-        app.kubernetes.io/version: "20220831-ceaae2455"
+        app.kubernetes.io/version: "20220905-e1866d766"
         app.kubernetes.io/name: knative-eventing
     spec:
       # To avoid node becoming SPOF, spread our replicas to different nodes.
@@ -1135,7 +1135,7 @@ spec:
           terminationMessagePolicy: FallbackToLogsOnError
           # This is the Go import path for the binary that is containerized
           # and substituted here.
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/webhook@sha256:e5de1bf706aabca1406ad462062388cf459f85d54a03ed79fdade6e687273ede
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/webhook@sha256:a7c42383a9df7e3d5fdb6123bc6ec09f282ef4b23e7dc9b68e664306f420db5d
           resources:
             requests:
               # taken from serving.
@@ -1204,10 +1204,10 @@ apiVersion: v1
 kind: Service
 metadata:
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     role: eventing-webhook
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
   name: eventing-webhook
   namespace: knative-eventing
@@ -1239,11 +1239,11 @@ kind: CustomResourceDefinition
 metadata:
   creationTimestamp: null
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -1455,10 +1455,10 @@ kind: CustomResourceDefinition
 metadata:
   name: brokers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -1622,11 +1622,11 @@ kind: CustomResourceDefinition
 metadata:
   name: channels.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -1897,11 +1897,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
   name: containersources.sources.knative.dev
 spec:
@@ -2051,9 +2051,9 @@ kind: CustomResourceDefinition
 metadata:
   name: eventtypes.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -2180,10 +2180,10 @@ kind: CustomResourceDefinition
 metadata:
   name: parallels.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: flows.knative.dev
@@ -2435,11 +2435,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -2626,10 +2626,10 @@ kind: CustomResourceDefinition
 metadata:
   name: sequences.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: flows.knative.dev
@@ -2946,12 +2946,12 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
   name: sinkbindings.sources.knative.dev
 spec:
@@ -3138,9 +3138,9 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -3350,9 +3350,9 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -3536,8 +3536,8 @@ kind: ClusterRole
 metadata:
   name: addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -3550,9 +3550,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: service-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3570,9 +3570,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: serving-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3593,9 +3593,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: channel-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3620,9 +3620,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: broker-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3641,9 +3641,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: flows-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3679,8 +3679,8 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-filter
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3706,8 +3706,8 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-ingress
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3724,8 +3724,8 @@ kind: ClusterRole
 metadata:
   name: eventing-config-reader
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3758,8 +3758,8 @@ kind: ClusterRole
 metadata:
   name: channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -3772,9 +3772,9 @@ kind: ClusterRole
 metadata:
   name: meta-channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     duck.knative.dev/channelable: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "channelable-manipulator" role.
 rules:
@@ -3812,9 +3812,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-eventing-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev"]
@@ -3826,9 +3826,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-messaging-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["messaging.knative.dev"]
@@ -3840,9 +3840,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-flows-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["flows.knative.dev"]
@@ -3854,9 +3854,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-sources-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["sources.knative.dev"]
@@ -3868,9 +3868,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-bindings-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["bindings.knative.dev"]
@@ -3883,8 +3883,8 @@ metadata:
   name: knative-eventing-namespaced-edit
   labels:
     rbac.authorization.k8s.io/aggregate-to-edit: "true"
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
@@ -3897,8 +3897,8 @@ metadata:
   name: knative-eventing-namespaced-view
   labels:
     rbac.authorization.k8s.io/aggregate-to-view: "true"
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
@@ -3925,8 +3925,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-controller
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -4061,8 +4061,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-pingsource-mt-adapter
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -4129,8 +4129,8 @@ kind: ClusterRole
 metadata:
   name: podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -4143,9 +4143,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: builtin-podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     duck.knative.dev/podspecable: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "podspecable-binding role.
 rules:
@@ -4191,8 +4191,8 @@ kind: ClusterRole
 metadata:
   name: source-observer
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -4205,9 +4205,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: eventing-sources-source-observer
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     duck.knative.dev/source: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "source-observer" role.
 rules:
@@ -4243,8 +4243,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-sources-controller
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -4330,8 +4330,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 rules:
   # For watching logging configuration and getting certs.
@@ -4430,8 +4430,8 @@ metadata:
   namespace: knative-eventing
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 rules:
   # For manipulating certs into secrets.
@@ -4467,8 +4467,8 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: config.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4505,8 +4505,8 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4539,8 +4539,8 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: validation.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4574,8 +4574,8 @@ metadata:
   name: eventing-webhook-certs
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 # The data is populated at install time.
 
@@ -4599,8 +4599,8 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: sinkbindings.webhook.sources.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
diff --git a/third_party/eventing-latest/eventing-crds.yaml b/third_party/eventing-latest/eventing-crds.yaml
index 9d5365c67..f68f673eb 100644
--- a/third_party/eventing-latest/eventing-crds.yaml
+++ b/third_party/eventing-latest/eventing-crds.yaml
@@ -17,11 +17,11 @@ kind: CustomResourceDefinition
 metadata:
   creationTimestamp: null
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -233,10 +233,10 @@ kind: CustomResourceDefinition
 metadata:
   name: brokers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -400,11 +400,11 @@ kind: CustomResourceDefinition
 metadata:
   name: channels.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -675,11 +675,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
   name: containersources.sources.knative.dev
 spec:
@@ -829,9 +829,9 @@ kind: CustomResourceDefinition
 metadata:
   name: eventtypes.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -958,10 +958,10 @@ kind: CustomResourceDefinition
 metadata:
   name: parallels.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: flows.knative.dev
@@ -1213,11 +1213,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -1404,10 +1404,10 @@ kind: CustomResourceDefinition
 metadata:
   name: sequences.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: flows.knative.dev
@@ -1724,12 +1724,12 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
   name: sinkbindings.sources.knative.dev
 spec:
@@ -1916,9 +1916,9 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -2128,9 +2128,9 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220831-ceaae2455"
+    eventing.knative.dev/release: "v20220905-e1866d766"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220831-ceaae2455"
+    app.kubernetes.io/version: "20220905-e1866d766"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
diff --git a/vendor/knative.dev/eventing/test/e2e/helpers/broker_prefer_reply_header_helper.go b/vendor/knative.dev/eventing/test/e2e/helpers/broker_prefer_reply_header_helper.go
deleted file mode 100644
index cf52359ca..000000000
--- a/vendor/knative.dev/eventing/test/e2e/helpers/broker_prefer_reply_header_helper.go
+++ /dev/null
@@ -1,97 +0,0 @@
-//go:build e2e
-// +build e2e
-
-/*
-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 helpers
-
-import (
-	"context"
-	"testing"
-
-	cloudevents "github.com/cloudevents/sdk-go/v2"
-	"github.com/google/uuid"
-
-	testlib "knative.dev/eventing/test/lib"
-	"knative.dev/eventing/test/lib/recordevents"
-	"knative.dev/eventing/test/lib/resources"
-)
-
-func BrokerPreferHeaderCheck(
-	ctx context.Context,
-	t *testing.T,
-	creator BrokerCreator,
-	options ...testlib.SetupClientOption) {
-	const (
-		recorderName = "event-recorder"
-		triggerName  = "test-trigger"
-		senderName   = "request-sender"
-		eventSource  = "source1"
-		eventType    = "type1"
-		eventBody    = `{"msg":"test msg"}`
-	)
-
-	tests := []struct {
-		name string
-	}{
-		{
-			name: "test message without explicit prefer header should have the header",
-		},
-	}
-	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			client := testlib.Setup(t, true, options...)
-			defer testlib.TearDown(client)
-
-			brokerName := creator(client, "v1")
-
-			// Create event tracker that should receive all events.
-			eventTracker, _ := recordevents.StartEventRecordOrFail(
-				ctx,
-				client,
-				recorderName,
-			)
-
-			client.CreateTriggerOrFail(triggerName,
-				resources.WithSubscriberServiceRefForTrigger(recorderName),
-				resources.WithBroker(brokerName),
-			)
-
-			client.WaitForAllTestResourcesReadyOrFail(ctx)
-
-			eventToSend := cloudevents.NewEvent()
-			eventToSend.SetID(uuid.New().String())
-			eventToSend.SetType(eventType)
-			eventToSend.SetSource(eventSource)
-			if err := eventToSend.SetData(cloudevents.ApplicationJSON, []byte(eventBody)); err != nil {
-				t.Fatal("Cannot set the payload of the event:", err.Error())
-			}
-			client.SendEventToAddressable(
-				ctx,
-				senderName,
-				brokerName,
-				testlib.BrokerTypeMeta,
-				eventToSend,
-			)
-
-			eventTracker.AssertAtLeast(
-				1,
-				recordevents.HasAdditionalHeader("Prefer", "reply"),
-			)
-		})
-	}
-}
diff --git a/vendor/knative.dev/eventing/test/e2e/helpers/channel_prefer_reply_header_helper.go b/vendor/knative.dev/eventing/test/e2e/helpers/channel_prefer_reply_header_helper.go
deleted file mode 100644
index 2556267d2..000000000
--- a/vendor/knative.dev/eventing/test/e2e/helpers/channel_prefer_reply_header_helper.go
+++ /dev/null
@@ -1,99 +0,0 @@
-//go:build e2e
-// +build e2e
-
-/*
-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 helpers
-
-import (
-	"context"
-	"testing"
-
-	cloudevents "github.com/cloudevents/sdk-go/v2"
-	"github.com/google/uuid"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-
-	testlib "knative.dev/eventing/test/lib"
-	"knative.dev/eventing/test/lib/recordevents"
-	"knative.dev/eventing/test/lib/resources"
-)
-
-func ChannelPreferHeaderCheck(
-	ctx context.Context,
-	t *testing.T,
-	component metav1.TypeMeta,
-	options ...testlib.SetupClientOption) {
-	const (
-		recorderName     = "event-recorder"
-		channelName      = "test-channel"
-		subscriptionName = "test-subscription"
-		senderName       = "request-sender"
-		eventSource      = "source1"
-		eventType        = "type1"
-		eventBody        = `{"msg":"test msg"}`
-	)
-
-	tests := []struct {
-		name string
-	}{
-		{
-			name: "test message without explicit prefer header should have it after fanout",
-		},
-	}
-	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			client := testlib.Setup(t, true, options...)
-			defer testlib.TearDown(client)
-
-			// create channels
-			client.CreateChannelOrFail(channelName, &component)
-			client.WaitForResourcesReadyOrFail(&component)
-
-			// create loggerPod and expose it as a service
-			eventTracker, _ := recordevents.StartEventRecordOrFail(ctx, client, recorderName)
-
-			client.CreateSubscriptionOrFail(
-				subscriptionName,
-				channelName,
-				&component,
-				resources.WithSubscriberForSubscription(recorderName),
-			)
-
-			client.WaitForAllTestResourcesReadyOrFail(ctx)
-
-			eventToSend := cloudevents.NewEvent()
-			eventToSend.SetID(uuid.New().String())
-			eventToSend.SetType(eventType)
-			eventToSend.SetSource(eventSource)
-			if err := eventToSend.SetData(cloudevents.ApplicationJSON, []byte(eventBody)); err != nil {
-				t.Fatal("Cannot set the payload of the event:", err.Error())
-			}
-			client.SendEventToAddressable(
-				ctx,
-				senderName,
-				channelName,
-				&component,
-				eventToSend,
-			)
-
-			eventTracker.AssertAtLeast(
-				1,
-				recordevents.HasAdditionalHeader("Prefer", "reply"),
-			)
-		})
-	}
-}
diff --git a/vendor/knative.dev/eventing/test/rekt/features/broker/feature.go b/vendor/knative.dev/eventing/test/rekt/features/broker/feature.go
new file mode 100644
index 000000000..71f59804b
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/rekt/features/broker/feature.go
@@ -0,0 +1,76 @@
+/*
+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 broker
+
+import (
+	cloudevents "github.com/cloudevents/sdk-go/v2"
+	"github.com/google/uuid"
+	"knative.dev/reconciler-test/pkg/eventshub"
+	eventasssert "knative.dev/reconciler-test/pkg/eventshub/assert"
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/resources/svc"
+
+	"knative.dev/eventing/test/rekt/resources/broker"
+	"knative.dev/eventing/test/rekt/resources/trigger"
+	"knative.dev/reconciler-test/pkg/manifest"
+)
+
+func BrokerPreferHeaderCheck() *feature.Feature {
+	f := feature.NewFeatureNamed("Channel PreferHeader Check")
+
+	source := feature.MakeRandomK8sName("source")
+	sink := feature.MakeRandomK8sName("sink")
+	via := feature.MakeRandomK8sName("via")
+
+	eventSource := "source1"
+	eventType := "type1"
+	eventBody := `{"msg":"test msg"}`
+	event := cloudevents.NewEvent()
+	event.SetID(uuid.New().String())
+	event.SetType(eventType)
+	event.SetSource(eventSource)
+	event.SetData(cloudevents.ApplicationJSON, []byte(eventBody))
+
+	//Install the broker
+	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.Setup("install sink", eventshub.Install(sink, eventshub.StartReceiver))
+
+	// Point the Trigger subscriber to the sink svc.
+	cfg := []manifest.CfgFn{trigger.WithSubscriber(svc.AsKReference(sink), "")}
+
+	// Install the trigger
+	f.Setup("install trigger", trigger.Install(via, brokerName, cfg...))
+	f.Setup("trigger goes ready", trigger.IsReady(via))
+
+	f.Requirement("install source", eventshub.Install(
+		source,
+		eventshub.StartSenderToResource(broker.GVR(), brokerName),
+		eventshub.InputEvent(event),
+	))
+
+	f.Stable("test message without explicit prefer header should have the header").
+		Must("delivers events",
+			eventasssert.OnStore(sink).Match(
+				eventasssert.HasAdditionalHeader("Prefer", "reply"),
+			).AtLeast(1))
+
+	return f
+}
diff --git a/vendor/knative.dev/eventing/test/rekt/features/channel/features.go b/vendor/knative.dev/eventing/test/rekt/features/channel/features.go
index 352bd2926..dd20720ec 100644
--- a/vendor/knative.dev/eventing/test/rekt/features/channel/features.go
+++ b/vendor/knative.dev/eventing/test/rekt/features/channel/features.go
@@ -23,6 +23,7 @@ import (
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	"github.com/cloudevents/sdk-go/v2/binding"
 	"github.com/cloudevents/sdk-go/v2/test"
+	"github.com/google/uuid"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/reconciler-test/pkg/eventshub"
 	"knative.dev/reconciler-test/pkg/eventshub/assert"
@@ -38,6 +39,7 @@ import (
 	"knative.dev/eventing/test/rekt/resources/pingsource"
 	"knative.dev/eventing/test/rekt/resources/source"
 	"knative.dev/eventing/test/rekt/resources/subscription"
+	eventasssert "knative.dev/reconciler-test/pkg/eventshub/assert"
 )
 
 func ChannelChain(length int, createSubscriberFn func(ref *duckv1.KReference, uri string) manifest.CfgFn) *feature.Feature {
@@ -275,3 +277,47 @@ func SingleEventWithEncoding(encoding binding.Encoding) *feature.Feature {
 
 	return f
 }
+
+func ChannelPreferHeaderCheck(createSubscriberFn func(ref *duckv1.KReference, uri string) manifest.CfgFn) *feature.Feature {
+	f := feature.NewFeatureNamed("Channel PreferHeader Check")
+
+	channelName := feature.MakeRandomK8sName("channel")
+	sub := feature.MakeRandomK8sName("subscription")
+	source := feature.MakeRandomK8sName("source")
+	sink := feature.MakeRandomK8sName("sink")
+
+	eventSource := "source1"
+	eventType := "type1"
+	eventBody := `{"msg":"test msg"}`
+	event := cloudevents.NewEvent()
+	event.SetID(uuid.New().String())
+	event.SetType(eventType)
+	event.SetSource(eventSource)
+	event.SetData(cloudevents.ApplicationJSON, []byte(eventBody))
+
+	f.Setup("install sink", eventshub.Install(sink, eventshub.StartReceiver))
+	f.Setup("install channel", channel.Install(channelName,
+		channel.WithTemplate(),
+	))
+	f.Setup("install subscription", subscription.Install(sub,
+		subscription.WithChannel(channel.AsRef(channelName)),
+		createSubscriberFn(svc.AsKReference(sink), ""),
+	))
+
+	f.Setup("subscription is ready", subscription.IsReady(sub))
+	f.Setup("channel is ready", channel.IsReady(channelName))
+
+	f.Requirement("install source", eventshub.Install(
+		source,
+		eventshub.StartSenderToResource(channel.GVR(), channelName),
+		eventshub.InputEvent(event),
+	))
+
+	f.Stable("test message without explicit prefer header should have the header").
+		Must("delivers events",
+			eventasssert.OnStore(sink).Match(
+				eventasssert.HasAdditionalHeader("Prefer", "reply"),
+			).AtLeast(1))
+
+	return f
+}
diff --git a/vendor/knative.dev/eventing/test/rekt/features/trigger/feature.go b/vendor/knative.dev/eventing/test/rekt/features/trigger/feature.go
new file mode 100644
index 000000000..6b2ac4794
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/rekt/features/trigger/feature.go
@@ -0,0 +1,85 @@
+/*
+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 trigger
+
+import (
+	"context"
+
+	"github.com/cloudevents/sdk-go/v2/test"
+	"knative.dev/eventing/test/rekt/resources/broker"
+	"knative.dev/eventing/test/rekt/resources/pingsource"
+	"knative.dev/eventing/test/rekt/resources/trigger"
+	"knative.dev/reconciler-test/pkg/eventshub"
+	"knative.dev/reconciler-test/pkg/eventshub/assert"
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/manifest"
+	"knative.dev/reconciler-test/resources/svc"
+)
+
+// This test is for avoiding regressions on the trigger dependency annotation functionality.
+func TriggerDependencyAnnotation() *feature.Feature {
+	sink := feature.MakeRandomK8sName("sink")
+	triggerName := feature.MakeRandomK8sName("triggerName")
+
+	f := new(feature.Feature)
+
+	//Install the broker
+	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))
+
+	psourcename := "test-ping-source-annotation"
+	dependencyAnnotation := `{"kind":"PingSource","name":"test-ping-source-annotation","apiVersion":"sources.knative.dev/v1"}`
+	annotations := map[string]interface{}{
+		"knative.dev/dependency": dependencyAnnotation,
+	}
+
+	// Add the annotation to trigger and point the Trigger subscriber to the sink svc.
+	f.Setup("install sink", eventshub.Install(sink, eventshub.StartReceiver))
+	cfg := []manifest.CfgFn{
+		trigger.WithSubscriber(svc.AsKReference(sink), ""),
+		trigger.WithAnnotations(annotations),
+	}
+
+	// Install the trigger
+	f.Setup("install trigger", trigger.Install(triggerName, brokerName, cfg...))
+
+	f.Setup("trigger goes ready", trigger.IsReady(triggerName))
+
+	f.Setup("install pingsource", func(ctx context.Context, t feature.T) {
+		brokeruri, err := broker.Address(ctx, brokerName)
+		if err != nil {
+			t.Error("failed to get address of broker", err)
+		}
+		cfg := []manifest.CfgFn{
+			pingsource.WithSchedule("*/1 * * * *"),
+			pingsource.WithSink(nil, brokeruri.String()),
+			pingsource.WithData("text/plain", "Test trigger-annotation"),
+		}
+		pingsource.Install(psourcename, cfg...)(ctx, t)
+	})
+	f.Setup("PingSource goes ready", pingsource.IsReady(psourcename))
+
+	f.Stable("pingsource as event source to test trigger with annotations").
+		Must("delivers events on broker with URI", assert.OnStore(sink).MatchEvent(
+			test.HasType("dev.knative.sources.ping"),
+			test.DataContains("Test trigger-annotation"),
+		).AtLeast(1))
+
+	return f
+}
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/containersource/containersource.go b/vendor/knative.dev/eventing/test/rekt/resources/containersource/containersource.go
index a9d44bbbe..ceb0a474e 100644
--- a/vendor/knative.dev/eventing/test/rekt/resources/containersource/containersource.go
+++ b/vendor/knative.dev/eventing/test/rekt/resources/containersource/containersource.go
@@ -22,6 +22,8 @@ import (
 	"time"
 
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"knative.dev/eventing/test/rekt/resources/source"
+
 	"knative.dev/reconciler-test/pkg/environment"
 	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/k8s"
@@ -59,6 +61,9 @@ func Install(name string, opts ...manifest.CfgFn) feature.StepFn {
 	}
 }
 
+// WithSink adds the sink related config to a ContainerSource spec.
+var WithSink = source.WithSink
+
 // WithExtensions adds the ceOverrides related config to a ContainerSource spec.
 func WithExtensions(extensions map[string]interface{}) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
@@ -79,6 +84,15 @@ func WithExtensions(extensions map[string]interface{}) manifest.CfgFn {
 	}
 }
 
+// WithArgs add template.spec.containers.args to a ContainerSource spec.
+func WithArgs(args string) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if args != "" {
+			cfg["args"] = args
+		}
+	}
+}
+
 func registerImage(ctx context.Context) error {
 	im := manifest.ImagesFromFS(ctx, yaml)
 	reg := environment.RegisterPackage(im...)
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/containersource/containersource.yaml b/vendor/knative.dev/eventing/test/rekt/resources/containersource/containersource.yaml
index f480b459b..7482dd1ef 100644
--- a/vendor/knative.dev/eventing/test/rekt/resources/containersource/containersource.yaml
+++ b/vendor/knative.dev/eventing/test/rekt/resources/containersource/containersource.yaml
@@ -43,10 +43,12 @@ spec:
       containers:
       - name: heartbeats
         image: ko://knative.dev/eventing/test/test_images/heartbeats
+        imagePullPolicy: IfNotPresent
         args:
-        - --period=1
+        - {{ .args }}
         env:
         - name: POD_NAME
           value: heartbeats
         - name: POD_NAMESPACE
           value: {{ .namespace }}
+
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/pingsource/pingsource.go b/vendor/knative.dev/eventing/test/rekt/resources/pingsource/pingsource.go
index a102ff762..2d49ba35b 100644
--- a/vendor/knative.dev/eventing/test/rekt/resources/pingsource/pingsource.go
+++ b/vendor/knative.dev/eventing/test/rekt/resources/pingsource/pingsource.go
@@ -82,3 +82,12 @@ func WithDataBase64(contentType, dataBase64 string) manifest.CfgFn {
 		}
 	}
 }
+
+//WithSchedule add the schedule config to a Pingsource spec.
+func WithSchedule(schedule string) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if schedule != "" {
+			cfg["schedule"] = schedule
+		}
+	}
+}
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/trigger/trigger.go b/vendor/knative.dev/eventing/test/rekt/resources/trigger/trigger.go
index c714a5ecb..5cbbf8406 100644
--- a/vendor/knative.dev/eventing/test/rekt/resources/trigger/trigger.go
+++ b/vendor/knative.dev/eventing/test/rekt/resources/trigger/trigger.go
@@ -79,6 +79,7 @@ func WithSubscriber(ref *duckv1.KReference, uri string) manifest.CfgFn {
 	}
 }
 
+// Deprecated
 // WithAnnotation adds an annotation to the trigger
 func WithAnnotation(key string, value string) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
@@ -89,6 +90,46 @@ func WithAnnotation(key string, value string) manifest.CfgFn {
 	}
 }
 
+// WithAnnotations adds annotations to the trigger
+func WithAnnotations(annotations map[string]interface{}) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if _, set := cfg["ceOverrides"]; !set {
+			cfg["ceOverrides"] = map[string]interface{}{}
+		}
+		ceOverrides := cfg["ceOverrides"].(map[string]interface{})
+
+		if annotations != nil {
+			if _, set := ceOverrides["annotations"]; !set {
+				ceOverrides["annotations"] = map[string]interface{}{}
+			}
+			ceExt := ceOverrides["annotations"].(map[string]interface{})
+			for k, v := range annotations {
+				ceExt[k] = v
+			}
+		}
+	}
+}
+
+// WithExtensions adds the ceOverrides related config to a ContainerSource spec.
+func WithExtensions(extensions map[string]interface{}) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if _, set := cfg["ceOverrides"]; !set {
+			cfg["ceOverrides"] = map[string]interface{}{}
+		}
+		ceOverrides := cfg["ceOverrides"].(map[string]interface{})
+
+		if extensions != nil {
+			if _, set := ceOverrides["extensions"]; !set {
+				ceOverrides["extensions"] = map[string]interface{}{}
+			}
+			ceExt := ceOverrides["extensions"].(map[string]interface{})
+			for k, v := range extensions {
+				ceExt[k] = v
+			}
+		}
+	}
+}
+
 // WithDeadLetterSink adds the dead letter sink related config to a Trigger spec.
 var WithDeadLetterSink = delivery.WithDeadLetterSink
 
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 1bcc87c87..2be65de5f 100644
--- a/vendor/knative.dev/eventing/test/rekt/resources/trigger/trigger.yaml
+++ b/vendor/knative.dev/eventing/test/rekt/resources/trigger/trigger.yaml
@@ -20,7 +20,7 @@ metadata:
   {{ if .annotations }}
   annotations:
   {{ range $key, $value := .annotations }}
-    {{ $key }}: "{{ $value }}"
+    {{ $key }}: {{ $value }}
   {{ end }}
   {{ end }}
 spec:
diff --git a/vendor/knative.dev/hack/presubmit-tests.sh b/vendor/knative.dev/hack/presubmit-tests.sh
index d9549f51b..539facf73 100644
--- a/vendor/knative.dev/hack/presubmit-tests.sh
+++ b/vendor/knative.dev/hack/presubmit-tests.sh
@@ -122,7 +122,12 @@ function report_build_test() {
 # * `go build` on the entire repo
 # * check licenses in all go packages
 function default_build_test_runner() {
+  foreach_go_module __build_test_runner_for_module
+}
+
+function __build_test_runner_for_module() {
   local failed=0
+  subheader "Build tests for $(go_mod_module_name)"
   # Run verify-codegen check
   if [[ -f ./hack/verify-codegen.sh ]]; then
     subheader "Checking autogenerated code is up-to-date"
@@ -132,18 +137,26 @@ function default_build_test_runner() {
   # verify-codegen (as md files can be auto-generated in some repos).
   (( IS_DOCUMENTATION_PR )) && return ${failed}
   # Don't merge these two lines, or return code will always be 0.
+  # Get all build tags in go code (ignore /vendor, /hack and /third_party)
+  local tags
+  tags="$(grep -I  -r '// +build' . | \
+    grep -v '/vendor/' | \
+    grep -v '/hack/' | \
+    grep -v '/third_party' | \
+    cut -f3 -d' ' | \
+    tr ',' '\n' | \
+    sort | uniq | \
+    grep -v '^!' | \
+    tr '\n' ' ')"
   local go_pkg_dirs
-  go_pkg_dirs="$(go list ./...)" || return 1
-  # Skip build test if there is no go code
-  [[ -z "${go_pkg_dirs}" ]] && return ${failed}
+  go_pkg_dirs="$(go list -tags "${tags}" ./...)" || return $?
+  if [[ -z "${go_pkg_dirs}" ]]; then
+    subheader "No golang code found, skipping build tests"
+    return 0
+  fi
   # Ensure all the code builds
   subheader "Checking that go code builds"
-  # Get all build tags in go code (ignore /vendor, /hack and /third_party)
-  local tags
-  tags="$(find . \
-    -path './vendor' -prune -o -path './hack' -prune -o -path './third_party' -prune \
-    -o -type f -name '*.go' -exec grep '// +build' {} + \
-    | cut -f3 -d' ' | tr ',' '\n' | uniq | sort | tr '\n' ' ')"
+
   report_build_test Build_Go \
     go test -vet=off -tags "${tags}" -exec echo ./... || failed=2
 
@@ -185,6 +198,11 @@ function run_unit_tests() {
 
 # Default unit test runner that runs all go tests in the repo.
 function default_unit_test_runner() {
+  foreach_go_module __unit_test_runner_for_module
+}
+
+function __unit_test_runner_for_module() {
+  subheader "Unit tests for $(go_mod_module_name)"
   report_go_test -short -race -count 1 ./...
 }
 
diff --git a/vendor/knative.dev/pkg/test/kube_checks.go b/vendor/knative.dev/pkg/test/kube_checks.go
index 30698e694..3e295a55f 100644
--- a/vendor/knative.dev/pkg/test/kube_checks.go
+++ b/vendor/knative.dev/pkg/test/kube_checks.go
@@ -34,6 +34,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/kubernetes"
 	k8styped "k8s.io/client-go/kubernetes/typed/core/v1"
+
 	"knative.dev/pkg/test/logging"
 )
 
@@ -141,6 +142,9 @@ func WaitForServiceEndpoints(ctx context.Context, client kubernetes.Interface, s
 	waitErr := wait.PollImmediate(interval, podTimeout, func() (bool, error) {
 		var err error
 		endpoints, err = endpointsService.Get(ctx, svcName, metav1.GetOptions{})
+		if apierrs.IsNotFound(err) {
+			return false, nil
+		}
 		if err != nil {
 			return false, err
 		}